基于 LiveKit Agents 实现客服热线 Warm Transfer 监督转接:从示例到源码级原理 📅 发布时间:2026/9/14 22:26:22 👁 浏览次数: 基于 LiveKit Agents 实现客服热线 Warm Transfer 监督转接从示例到源码级原理【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agents导读本文以仓库 examples/warm-transfer 示例为骨架讲解如何在实时语音 AI 客服中实现Warm Transfer监督转接 / 主管升级工作流当客户要求转接人工时AI 客服将其保持在线、主动呼叫主管、用对话摘要进行简报再把主管无缝并入客户所在房间。你将掌握WarmTransferTask与TwilioConnectorWarmTransferTask两个开箱即用的 Beta 工作流任务的完整用法、全部关键参数以及其底层独立 Room SIP 外呼 独立 AgentSession MoveParticipant的实现机制可直接复制到自己的呼叫中心场景中。Warm Transfer 是什么五步转接流程Warm Transfer监督转接与冷转接Cold Transfer的本质区别在于在把主管接入客户之前AI 助手先与主管进行一段简报对话确保主管掌握足够的上下文避免客户向主管重复描述问题。关联文档 README 给出了完整流程客户请求升级Customer requests escalation——例如客户说我要找你们主管Agent 将客户置于保持状态Agent places the customer on hold——播放保持音/音乐同时暂停客户侧的语音输入输出Agent 联系下一级升级点主管Agent contacts the next escalation point——通过 SIP 或 Twilio 外呼主管Agent 向主管简报Agent briefs the supervisor with a summary——AI 基于此前与客户的对话历史生成摘要向主管说明谁、为何来电、为何需要人工Agent 将主管接入客户Agent connects the supervisor to the customer——主管同意后双方在同一房间直接通话AI 助手退出。这五步在示例 support_agent.py 中由transfer_to_human工具函数驱动其工具描述明确要求模型先与用户确认再发起转接避免误转接function_tool async def transfer_to_human(self) - None: Called when the user asks to speak to a human agent. This will put the user on hold while the supervisor is connected. Ensure that the user has confirmed that they wanted to be transferred. ... await self.session.say( Please hold while I connect you to a human agent., allow_interruptionsFalse ) try: result await self._start_transfer() ... await self.session.say( you are on the line with my supervisor. Ill be hanging up now., allow_interruptionsFalse, ) self.session.shutdown()注意两处细节转接过程中所有播报都带allow_interruptionsFalse防打断转接完成后 AI 主动shutdown()结束自己的会话。LiveKit 如何支撑监督转接关联文档 README 用三条要点概括了 LiveKit 侧的机制结合源码 warm_transfer.py 可以还原出完整的实现细节为主管新建一个独立 Room转接任务启动时_dial_human_agentwarm_transfer.py#L291-L365以客户房间名 -human-agent为名创建新房间并用带agentkind 的 AccessToken 连接with_kind(agent)用CreateSIPParticipant发起 SIP 外呼_originate_human_agentwarm_transfer.py#L367-L385构造api.CreateSIPParticipantRequest把主管的电话号码拨入该新房间wait_until_answeredTrue表示等主管接听后才算建立连接用独立的AgentSession与主管共享上下文在新房间中再启动一个AgentSession其模型/STT/TTS/VAD 均从当前会话继承self.session.vad or NOT_GIVEN等并注入包含客户对话历史的提示词——于是AI 与主管的简报对话得以发生主管同意后用MoveParticipant迁移_merge_callswarm_transfer.py#L387-L403调用api.room.move_participant把主管从human-agent房间搬到客户所在房间两路通话即合并为一路。整个过程可用下图概括文字示意客户 ──► 客服 Agent 房间保持中播放保持音、IO 禁用 │ CreateSIPParticipant / Twilio 外呼 ▼ 主管新房间「客户房间-human-agent」 │ 独立 AgentSession简报对话共享聊天历史 │ 主管同意 → connect_to_caller 工具 ▼ MoveParticipant 把主管迁入客户房间 ──► 客户与主管直接通话快速上手一行调用 WarmTransferTask关联文档 README 强调你不需要自己实现转接逻辑livekit.agents.beta.workflows中导出的WarmTransferTask已经封装好上述全部流程只需传入目标号码与 SIP trunk IDresult await WarmTransferTask( target_phone_numberSUPERVISOR_PHONE_NUMBER, sip_trunk_idSIP_TRUNK_ID, chat_ctxself.chat_ctx, # Provides conversation history to the supervisor )需要说明的是示例中的target_phone_number参数在当前仓库源码中已被标记为deprecatedwarm_transfer.py#L155-L158虽然仍会兼容性地回退到sip_call_to但新代码应直接使用sip_call_to。仓库示例 warm_transfer.py 给出了推荐写法class SIPSupportAgent(SupportAgent): async def _start_transfer(self) - WarmTransferResult: assert SIP_TRUNK_ID is not None assert SUPERVISOR_PHONE_NUMBER is not None return await WarmTransferTask( sip_call_toSUPERVISOR_PHONE_NUMBER, sip_trunk_idSIP_TRUNK_ID, sip_numberSIP_NUMBER, # 主管侧看到的来电号码Caller ID chat_ctxself.chat_ctx, # 客户对话历史供主管简报使用 # 转接到 IVR 后的分机时可发送 DTMF 音每个 w 暂停约 0.5s # dtmfwwww1234#, # 主管 25 秒内不接听则放弃 # ringing_timeout25, # 追加简报提示词 extra_instructionsSUMMARY_INSTRUCTIONS, )WarmTransferTask 参数详解WarmTransferTask继承自AgentTask[WarmTransferResult]其完整构造签名位于 warm_transfer.py#L40-L115。除转接专用参数外它还透传了AgentTask的标准能力stt/vad/llm/tts/turn_detection/tools/allow_interruptions等不传则沿用宿主AgentSession的配置。参数类型说明sip_call_tostr主管的电话号码如15105550123或 SIP URI如sip:userexample.com必填sip_trunk_idstr \| None已配置的 LiveKit SIP 出站 trunk ID缺省时回退到环境变量LIVEKIT_SIP_OUTBOUND_TRUNKsip_connectionapi.SIPOutboundConfig底层 SIP 连接配置用于自定义 SIP 域名而非已保存的 trunk 发起呼叫可指定自定义 hostname、传输方式与鉴权凭据sip_numberstr外呼时展示给主管的 Caller ID 号码缺省读取环境变量LIVEKIT_SIP_NUMBERsip_headersdict[str, str]附加的 SIP 头dtmfstr \| None接通后发送的 DTMF 音用于拨分机或穿越 IVR 菜单如1234#w表示约 0.5 秒停顿wwww1234#约等待 2 秒适合目的地先播欢迎语再接收按键的场景ringing_timeoutfloat \| None等待主管接听的最长秒数超时后任务以ToolError结束客户会话恢复hold_audioAudioSource \| AudioConfig \| list[AudioConfig] \| None客户保持期间播放的音频默认使用内置AudioConfig(BuiltinAudioClip.HOLD_MUSIC, volume0.8)保持音乐instructionsWorkflowInstructions \| Instructions \| str自定义简报 Agent 的提示词提供了extra_instructions时需注意两者同时设置会以instructions为准并打印告警chat_ctxllm.ChatContext客户对话上下文用于生成给主管的简报extra_instructionsstr已弃用倾向追加到默认简报提示词末尾的额外指令target_phone_numberstr已弃用等价于sip_call_to仅供旧代码兼容源码中值得注意的默认行为sip_trunk_id未显式传入时会读取LIVEKIT_SIP_OUTBOUND_TRUNK环境变量warm_transfer.py#L165-L176若 trunk 与sip_connection均未提供构造时直接抛出ValueError防止运行时才暴露配置缺失。转接中的三个内置工具函数WarmTransferTask向简报 Agent 暴露了三个function_tool(flagsToolFlag.IGNORE_ON_ENTER)工具warm_transfer.py#L229-L253由 AI 在与主管的简报对话中自主决策connect_to_caller主管确认愿意接听后调用。它会触发_merge_calls用MoveParticipant把主管迁入客户房间并以WarmTransferResult(human_agent_identity...)完成任务随后监听客户房间的participant_disconnected任一方挂断时删除房间。decline_transfer(reason)主管明确拒绝接听时调用任务以ToolError(human agent declined to connect: ...)结束转接失败回到客户对话。voicemail_detected()听到语音信箱问候语后调用用于让 AI 判断对方是真人还是语音信箱同样以ToolError(voicemail detected)结束任务。任务结束的统一路径是_set_resultwarm_transfer.py#L276-L289关闭主管侧的AgentSession、停止保持音、恢复客户侧 IO再complete(result)。简报是怎么生成的对话历史与提示词模板转接成功的关键在于简报质量。WarmTransferTask内部做了两件事格式化对话历史_format_conversation_historywarm_transfer.py#L184-L196遍历chat_ctx中user/assistant角色的文本消息分别标注为Caller:与Assistant:拼成文本渲染提示词模板默认PERSONA声明你是正在向人类代理求助的 AgentINSTRUCTIONS_TEMPLATEwarm_transfer.py#L565-L594把 persona、对话历史与extra拼接成完整指令并明确要求Once the human agent has confirmed, you should call the toolconnect_to_caller、以给主管总结对话开场并回答他们的问题。示例 support_agent.py 中的SUMMARY_INSTRUCTIONS进一步规定了摘要结构WHO通话对象、WHY来电目的、WHY为何需要人工、以及 100-200 字符的第一人称简述。这种模板 业务补充的组合正是WorkflowInstructions定义于 utils.py的设计用途保留工作流内置默认按需覆盖 persona 或追加 extra 段。基于 Twilio Connector 的变体对于使用 Twilio 而非 SIP trunk 的呼叫中心仓库提供了TwilioConnectorWarmTransferTaskwarm_transfer.py#L437-L562。它继承WarmTransferTask但外呼路径完全不同通过connector.connect_twilio_call获取connect_url拼装 TwiMLStream url.../再用 Twilio REST SDKpip install twilio发起client.calls.create把主管通话音频流回 connector需要TWILIO_ACCOUNT_SID、TWILIO_AUTH_TOKEN也可直接传参与TWILIO_FROM_NUMBER主管侧显示的号码由于 Twilio 的未接听/失败只通过异步状态 webhook 上报默认强制ringing_timeout30.0_TWILIO_RINGING_TIMEOUT见 warm_transfer.py#L432-L434避免主管一直不接听导致客户无限等待超时后还会主动canceled取消仍在振铃的呼叫warm_transfer.py#L526-L532接通判定通过监听房间里主管身份的 audio track 发布事件完成_wait_for_human_agent。示例 twilio_connector_warm_transfer.py 展示了完整用法return await TwilioConnectorWarmTransferTask( SUPERVISOR_PHONE_NUMBER, twilio_from_numberTWILIO_FROM_NUMBER, twilio_account_sidTWILIO_ACCOUNT_SID, twilio_auth_tokenTWILIO_AUTH_TOKEN, chat_ctxself.chat_ctx, # ringing_timeout25, extra_instructionsSUMMARY_INSTRUCTIONS, )完整客服 Agent 骨架示例将业务逻辑与传输层解耦support_agent.py 定义抽象基类SupportAgent含transfer_to_human工具与_start_transfer抽象方法warm_transfer.py 与 twilio_connector_warm_transfer.py 分别实现 SIP 与 Twilio 两种传输。入口函数统一为def run(create_agent: Callable[[], Agent]) - None: server AgentServer() server.rtc_session(agent_namesip-inbound) async def entrypoint(ctx: JobContext) - None: session AgentSession( llmopenai/gpt-4.1-mini, sttdeepgram/nova-3:en, ttscartesia/sonic-3:9626c31c-bec5-4cca-baa8-f8ba9e84c8bc, ) await session.start( agentcreate_agent(), roomctx.room, room_optionsroom_io.RoomOptions( audio_inputroom_io.AudioInputOptions( noise_cancellationnoise_cancellation.BVCTelephony(), # Krisp 电话降噪 ), delete_room_on_closeFalse, # 保持房间存活等待客户与主管通话 ), ) cli.run_app(server)两个要点一是使用命名 Agentnamed agent显式派发注册名为sip-inbound因为主管会被放进独立房间不能让默认 agent 误派发到主管房间二是delete_room_on_closeFalse确保 AI 会话结束后房间仍保留给客户与主管。INSTRUCTIONS常量则定义了客服的人格与转接边界用户要求转人工时必须先确认再转接。运行前提与启动方式关联文档 README 的 Usage 部分给出如下要求结合示例源码补充完整Prerequisites一个 LiveKit Cloud 账号或自建 LiveKit 服务SDK 通过LIVEKIT_URL/LIVEKIT_API_KEY/LIVEKIT_API_SECRET读取凭据已配置的 SIP trunk入站与出站两个电话号码一个拨打 AI 客服一个作为主管转接目标一条 SIP dispatch rule拨打时触发名为sip-inbound的 agent。环境变量变量用途示例LIVEKIT_SIP_OUTBOUND_TRUNK出站 SIP trunk IDST_abcxyzLIVEKIT_SUPERVISOR_PHONE_NUMBER主管电话号码含与国家码12003004000LIVEKIT_SIP_NUMBER外呼时展示给主管的 Caller ID15005006000TWILIO_ACCOUNT_SID/TWILIO_AUTH_TOKEN/TWILIO_FROM_NUMBERTwilio 变体专用凭据与主叫号码ACxxxx.../xxxx.../15005006000启动 Agentpython warm_transfer.py devdev是 LiveKit Agents CLI 的开发模式子命令示例通过 support_agent.py 中的cli.run_app(server)接入 CLI 框架。生产环境可改用python warm_transfer.py start等正式模式也可在两个文件间切换SIP 场景运行 warm_transfer.pyTwilio 场景运行 twilio_connector_warm_transfer.py。边界情况与失败恢复从源码可以梳理出转接任务对异常路径的处理策略直接决定生产可用性外呼失败/超时on_enter中通过asyncio.wait并发等待拨号任务与房间关闭事件warm_transfer.py#L209-L227任一侧失败即以ToolError(could not dial human agent)结束任务主管挂断_on_human_agent_room_close会触发任务失败客户侧会话自动恢复warm_transfer.py#L255-L263主管拒绝/语音信箱对应decline_transfer与voicemail_detected工具均以ToolError收尾客户侧 IO 管理任务期间_set_io_enabled(False)会记住原始状态audio/video 输入输出、转写输出并整体禁用结束后精确还原warm_transfer.py#L405-L429客户挂断participant_disconnected监听触发delete_room清理资源warm_transfer.py#L265-L274。示例侧同样做了防御transfer_to_human把底层异常包装为ToolError上抛让 LLM 能感知转接失败这一工具执行结果从而用自然语言安抚客户support_agent.py。小结examples/warm-transfer示例与livekit.agents.beta.workflows模块共同给出了一条完整、可复制的客服升级人工技术路径五步监督转接流程、SIP 与 Twilio 双通道支持、基于聊天历史的自动简报、三个内置决策工具以及完备的失败恢复。对于要在实时语音客服中落地人工兜底的团队直接复用WarmTransferTask或TwilioConnectorWarmTransferTaskSupportAgent骨架再按业务补充SUMMARY_INSTRUCTIONS与转接确认话术即可在很短时间内跑通完整的监督转接能力。【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agents创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考