基于 LiveKit Agents 构建医疗客服语音智能体：AgentTasks 结构化工作流与动态工具编排实战

基于 LiveKit Agents 构建医疗客服语音智能体AgentTasks 结构化工作流与动态工具编排实战【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agents医疗行业对话场景对智能体的要求远高于普通闲聊需要在查询敏感信息前完成身份认证、需要按保险范围动态筛选医生、需要避免 LLM 幻觉出数据库中不存在的预约时间、还需要在超出职责范围时无缝转接人工。examples/healthcare示例给出了一个完整的参考实现——一个基于 LiveKit Agents 框架的医疗助理它同时支持文本与语音交互用AgentTask将身份认证、预约管理、账单处理拆解为可复用的结构化工作流并通过动态注入 Function Tool 的方式彻底消除参数幻觉。读完本文你将掌握 AgentTask/TaskGroup 的编排模型、任务完成回调实现认证快进、按运行时数据动态构建工具 schema 的完整套路以及如何在自己的语音 Agent 中落地这些模式。示例概览一个模态无关Modality-Agnostic的医疗助理本示例的核心定位见 examples/healthcare/README.md是A full healthcare assistant providing secure appointment management and billing handling.它利用多种AgentTask组合出结构化工作流来收集信息并且模态无关modality-agnostic——用户可以通过文本或语音任意一种方式交互并能无缝切换。一旦对话超出智能体的职责范围用户会被转接给真人warm transfer。整套智能体只由 4 个文件构成麻雀虽小五脏俱全文件职责examples/healthcare/agent.py全部 Agent 逻辑主 Agent、各类 AgentTask、动态工具构建、服务入口examples/healthcare/fake_database.py内存版患者库 医生库模拟真实诊所数据库行为examples/healthcare/pyproject.toml依赖声明livekit-agents1.6等examples/healthcare/Dockerfile可独立部署的容器镜像从源码入口examples/healthcare/agent.py#L750-L797可以看到HealthcareAgent通过AgentSession装配 STT / LLM / TTS全部走 LiveKit Inference托管推理无需自持各家 API Keyserver AgentServer() server.rtc_session() async def entrypoint(ctx: JobContext): db FakeDatabase() userdata UserData(databasedb, profileNone) session AgentSession( userdatauserdata, sttinference.STT(deepgram/nova-3, languagemulti), llminference.LLM(google/gemma-4-31b-it), ttsinference.TTS( inworld/inworld-tts-2, voiceLuna, extra_kwargs{delivery_mode: CREATIVE, speaking_rate: 1.1}, ), preemptive_generationTrue, # Flip user_state to away after 10s of mutual silence so we can # check whether theyre still there (default is 15s). user_away_timeout10.0, )这里还能看到一个实用的空闲唤醒细节user_away_timeout10.0将用户离开判定从默认 15 秒缩短到 10 秒并通过监听user_state_changed事件启动一个每 10 秒一次的 nudge 循环examples/healthcare/agent.py#L772-L792用户一旦重新说话状态离开away该任务即被取消。运行与配置从克隆到上线环境变量根据 examples/README.md 的说明在examples目录创建.env并填入连接信息LIVEKIT_URLwss://your-project.livekit.cloud LIVEKIT_API_KEYyour_api_key LIVEKIT_API_SECRETyour_api_secret本示例额外使用 3 个环境变量来支撑转人工warm transfer能力examples/healthcare/agent.py#L42-L45SIP_TRUNK_ID os.getenv(LIVEKIT_SIP_OUTBOUND_TRUNK) # ST_abcxyz SUPERVISOR_PHONE_NUMBER os.getenv(LIVEKIT_SUPERVISOR_PHONE_NUMBER) # 12003004000 SIP_NUMBER os.getenv(LIVEKIT_SIP_NUMBER) # 15005006000 - caller ID shown to supervisorLIVEKIT_SIP_OUTBOUND_TRUNKSIP 出站中继 ID用于拨出电话给主管LIVEKIT_SUPERVISOR_PHONE_NUMBER主管电话号码转接目标LIVEKIT_SIP_NUMBER显示给主管的来电号码。若这三个变量未配置transfer_to_human工具会抛出ToolError(SIP_TRUNK_ID is not configured)之类的错误因此仅当你要测试转人工功能时才需要设置它们其余功能不受影响。安装与启动在仓库根目录同步依赖uv sync --all-extras --dev示例目录是独立的 uv workspace 成员也可以把目录复制出去独立运行cd examples/healthcare uv sync uv run agent.py consoleconsole子命令让 Agent 在本机终端里跑起来适合快速调试部署到 LiveKit Cloud 时使用lk agent deploy .容器镜像由目录内的 Dockerfile 提供镜像内会预执行python -m livekit.agents download-files预下载 silero VAD 等模型权重避免上线时冷启动卡顿。依赖声明examples/healthcare/pyproject.toml要求 Python 3.10核心依赖为livekit-agents1.6另有livekit-plugins-sileroVAD、python-dotenv读取.env与pydantic工具参数 schema。核心概念一AgentTask——把对话拆成可复用的任务整个示例的架构基石是AgentTask。每个任务封装了一段完整的信息收集流程包含自己的系统指令、工具集与完成条件可以被顶层 Agent 或其他任务嵌套调用。以本示例自定义的GetInsuranceTask为例examples/healthcare/agent.py#L198-L237class GetInsuranceTask(AgentTask[GetInsuranceResult]): def __init__( self, extra_instructions: str , chat_ctx: llm.ChatContext | None None, require_confirmation: bool False, ): extra f\n{extra_instructions} if extra_instructions else super().__init__( instructionsInstructions( _GET_INSURANCE_BASE_INSTRUCTIONS.format( modality_specific_GET_INSURANCE_AUDIO_SPECIFIC, extra_instructionsextra, ), text_GET_INSURANCE_BASE_INSTRUCTIONS.format( modality_specific_GET_INSURANCE_TEXT_SPECIFIC, extra_instructionsextra, ), ), tools[transfer_to_human], chat_ctxchat_ctx, ) async def on_enter(self): await self.session.generate_reply( instructionsCollect the users health insurance, inform them of the accepted insurances if they ask. ) function_tool() async def record_health_insurance( self, context: RunContext, insurance: Annotated[str, Field(json_schema_extra{enum: VALID_INSURANCES})], ): Record the users health insurance. self.complete(GetInsuranceResult(insuranceinsurance))这里有几个值得注意的模式模态双指令Instructions 二元结构Instructions同时传入语音版与文本版指令。语音场景强调避免使用破折号和特殊字符、口头确认后再记录文本场景则直接接受输入。这让同一套任务逻辑在电话与聊天中都能自然运转正是模态无关的实现细节。on_enter主动开场任务进入时主动生成一句话引导收集信息而不是被动等用户开口。self.complete(...)显式收尾一旦工具被调用即产出结构化结果GetInsuranceResult任务随即结束并返回给调用方。参数枚举约束insurance参数的enum被硬编码为VALID_INSURANCES [Anthem, Aetna, EmblemHealth, HealthFirst]examples/healthcare/agent.py#L47LLM 只能在白名单内取值从 schema 层面杜绝了虚构保险公司。核心概念二Profile 认证——TaskGroup 编排与任务完成回调快进这是 README 重点讲解的第一块能力。任何敏感信息查询预约、账单之前用户必须先完成认证而认证每次通话只执行一次。如果用户提供的姓名 生日已存在于数据库中流程会被快进跳过。用 TaskGroup 并行编排四个收集子任务认证逻辑集中在profile_authenticatorexamples/healthcare/agent.py#L591-L642。它先判断当前会话是否已有 profileself.session.userdata.profile没有则构造一个TaskGroup把姓名、生日、电话、保险四个收集任务全部注册进去async def profile_authenticator(self) - None: Creates a TaskGroup that collects user information logger.info(Authenticating user information) if not self.session.userdata.profile: task_group TaskGroup( chat_ctxself.chat_ctx, return_exceptionsFalse, on_task_completedlambda event: self.task_completed_callback(event, task_group), ) task_group.add( lambda: GetNameTask(last_nameTrue), idget_name_task, descriptionGathers the users name, ) task_group.add( lambda: GetDOBTask(), idget_dob_task, descriptionGathers the users date of birth, ) task_group.add( lambda: GetPhoneNumberTask(), idget_phone_number_task, descriptionGathers the users phone number, ) task_group.add( lambda: GetInsuranceTask(), idget_insurance_task, descriptionGathers the users insurance, ) try: results await task_group except ProfileFound: await self.session.generate_reply( instructionsInform the user that an existing profile has been found with their details. ) else: patient_name f{results.task_results[get_name_task].first_name} {results.task_results[get_name_task].last_name} profile { name: patient_name, date_of_birth: results.task_results[get_dob_task].date_of_birth, phone_number: results.task_results[get_phone_number_task].phone_number, insurance: results.task_results[get_insurance_task].insurance, } self.session.userdata.profile profile self._database.add_patient_record(infoprofile)TaskGroup的底层实现位于 livekit-agents/livekit/agents/beta/workflows/task_group.py关键能力包括顺序编排 可回退按注册顺序执行各子任务并允许用户中途回到上一个问题重新回答on_task_completed回调每个子任务完成时触发一次接收TaskCompletedEvent含task_id与result这是实现认证快进的关键钩子return_exceptionsFalse默认直接传播异常配合下方ProfileFound的使用方式设为True时则把异常收进结果字典继续执行上下文压缩默认summarize_chat_ctxTrue把组内交互摘要合并进上下文控制 token 消耗。框架内预置的GetNameTask、GetDOBTask、GetPhoneNumberTask等实现同样位于 livekit-agents/livekit/agents/beta/workflows/如name.py、dob.py、phone_number.py本示例直接复用不必自己从零编写问姓名、问生日这类琐碎收集逻辑。用自定义异常快进认证task_completed_callbackexamples/healthcare/agent.py#L575-L589演示了快进的实现async def task_completed_callback(self, event, task_group): if event.task_id get_name_task: self._pending_name event.result.first_name event.result.last_name if self.session.userdata.profile: # in the case that the user creates a new profile or restarts, the recorded session profile is cleared self.session.userdata.profile {} if event.task_id get_dob_task: existing_record self._database.get_patient_by_name_and_dob( self._pending_name, event.result.date_of_birth ) if existing_record: logger.info(fFound existing patient profile for {self._pending_name}) self.session.userdata.profile existing_record raise ProfileFound()当用户说出姓名与生日后get_dob_task完成回调立刻拿着(姓名, 生日)去数据库查证。命中则直接抛出自定义的ProfileFound(ToolError)examples/healthcare/agent.py#L76-L78异常向上传播终止TaskGroup后面问电话、问保险两个任务被跳过——这就是 README 所述如果用户提供的信息已存在于数据库流程被快进。同时已找到的existing_record被写入session.userdata.profile作为本次通话的会话身份。若用户是首次就诊未命中四个任务完整走完组装出 profile 后调用self._database.add_patient_record(infoprofile)落库FakeDatabase.add_patient_record还会自动为每位新患者生成 20~3000 美元随机的未结余额见 examples/healthcare/fake_database.py#L125-L127。认证后动态追加更新档案工具认证结束时示例会把update_record工具动态挂载到主 Agent 上examples/healthcare/agent.py#L640-L642current_tools [t for t in self.tools if t.id ! update_record] current_tools.append(build_update_record()) await self.update_tools(current_tools)build_update_recordexamples/healthcare/agent.py#L240-L276是一个工具工厂默认允许修改[dob, phone, insurance]三个字段且支持按需裁剪如预约任务里只允许改[dob, phone]。用户要求改保险时它会内部再嵌套调用GetInsuranceTask重新收集新值写库后同步更新会话内 profile。注意更新保险后ScheduleAppointmentTask里兼容医生列表会随之刷新见下文。核心概念三动态工具编排——预约管理零幻觉README 强调Function tools 是动态添加的因此 LLM 既不能凭空捏造参数也不能过早调用工具。预约流程被拆成四个递进的阶段每个阶段只在用户完成当前步骤后才把下一阶段所需工具其参数枚举来自实时数据库数据注入到模型可见的工具集中。阶段 1按保险筛医生ScheduleAppointmentTask.on_enter首先调用_setup_doctor_selectionexamples/healthcare/agent.py#L303-L330async def _setup_doctor_selection(self): database self.session.userdata.database insurance self.session.userdata.profile[insurance] self._compatible_doctor_records database.get_compatible_doctors(insuranceinsurance) available_doctors [doctor[name] for doctor in self._compatible_doctor_records] doctor_confirmation_tool self._build_doctor_selection_tool( available_doctorsavailable_doctors ) current_tools [t for t in self.tools if t.id ! confirm_doctor_selection] current_tools.append(doctor_confirmation_tool) await self.update_tools(current_tools) chat_ctx self.chat_ctx.copy() chat_ctx.add_message( rolesystem, contentfThese doctors are now compatible with the users insurance: {available_doctors}, ) await self.update_chat_ctx(chat_ctx)它按用户保险从FakeDatabase.get_compatible_doctors拉取兼容医生把医生名单编译进工具参数的enumexamples/healthcare/agent.py#L358-L395function_tool() async def confirm_doctor_selection( selected_doctor: Annotated[ str, Field( descriptionThe names of the available doctors, json_schema_extra{enum: available_doctors}, ), ], ) - None: Call to confirm the users doctor selection. self._selected_doctor selected_doctor doctor_record self.session.userdata.database.get_doctor_by_name(selected_doctor) available_times doctor_record[availability] schedule_appointment_tool self._build_schedule_appointment_tool( available_timesavailable_times ) ...同时把医生名单以 system 消息注入chat_ctxupdate_chat_ctx让模型知道这份名单。用户选定医生后才轮到预约时间。阶段 2按医生可用时间排预约_build_schedule_appointment_toolexamples/healthcare/agent.py#L397-L429把医生档案里的可用时间槽转成 ISO 字符串枚举iso_times [ datetime.combine(slot[date], slot[time]).isoformat() for slot in available_times ] function_tool() async def schedule_appointment( appointment_time: Annotated[ str, Field( descriptionThe available appointment times in ISO format, json_schema_extra{enum: iso_times}, ), ], ): Call to confirm the users selected appointment time. self._appointment_time datetime.fromisoformat(appointment_time) ...appointment_time的候选值只包含该医生在数据库中真实存在的空闲时段模型无法编造一个数据库里不存在的时间。这正是 README 所述appointment scheduling tool is built dynamically with the availabilities——工具 schema 由运行时数据实时生成。阶段 3收集就诊原因并落库选定时间后confirm_visit_reason工具被挂上examples/healthcare/agent.py#L431-L447收集自由文本的就诊原因并self.complete(...)产出ScheduleAppointmentResult(doctor_name, appointment_time, visit_reason)。顶层schedule_appointment工具examples/healthcare/agent.py#L644-L660随后调用self._database.add_appointment(...)落库。注意FakeDatabase.add_appointmentexamples/healthcare/fake_database.py#L87-L102在写入预约的同时会调用remove_doctor_availability把该时间槽从医生可用列表中移除——即 README 所述医生可用性被移除保证下一个用户看到的候选时间永远是最新的。而cancel_appointmentexamples/healthcare/fake_database.py#L104-L123会反向把时间槽归还给医生。阶段 4改签 / 取消复用预约任务ModifyAppointmentTaskexamples/healthcare/agent.py#L450-L540处理已有预约的修改进入任务后先查患者名下预约没有则直接告知您当前没有在档预约并结束有预约时把[reschedule, cancel]与患者真实预约时间列表一起编译进confirm_appointment_selection的参数枚举无论改签还是取消都先执行self._database.cancel_appointment(...)取消原预约释放医生时间槽若选择改签reschedule会先把聊天历史摘要压缩chat_ctx._summarize(self.session.llm)然后复用同一个ScheduleAppointmentTask走一遍选医生 → 选时间 → 填原因的完整流程examples/healthcare/agent.py#L529-L538if function cancel: self.complete( ModifyAppointmentResult(new_appointmentNone, old_appointmentappointment) ) else: chat_ctx await self.chat_ctx.copy()._summarize(self.session.llm) result await ScheduleAppointmentTask(chat_ctxchat_ctx) self.complete( ModifyAppointmentResult(new_appointmentresult, old_appointmentappointment) )这正是 README 强调的用户希望改签时原预约被取消并复用ScheduleAppointmentTask——任务即组件可被反复嵌套调用且由于复用了同一任务类改签流程与首次预约保持一致体验。值得一提的还有ScheduleAppointmentTask内部的update_insurance工具examples/healthcare/agent.py#L332-L356当用户在预约中途要求更换保险它会重新执行GetInsuranceTask收集新保险、写库然后重新调用_setup_doctor_selection刷新兼容医生列表并明确指示模型在新医生列表出来前不要确认医生——严格约束了对话状态机的转换时机。核心概念四账单处理——GetCreditCardTask 与余额校验README 展示的最后一块能力是GetCreditCardTask()。顶层handle_billing工具examples/healthcare/agent.py#L701-L715先做认证查询患者未结余额然后动态挂载confirm_payment_proceeds工具并让模型告知余额、询问是否现在支付。confirm_payment_proceedsexamples/healthcare/agent.py#L717-L747在真正收单前做三层校验function_tool() async def confirm_payment_proceeds(amount: float) - str | None: Call to proceed with payment steps regarding the users bill. name self.session.userdata.profile[name] balance self._database.get_outstanding_balance(name) if amount 0: return The payment amount must be greater than zero. if amount balance: return fThe payment amount exceeds the outstanding balance of ${balance}. result await GetCreditCardTask() last_four_digits result.card_number[-4:] remaining self._database.apply_payment(name, amount) logger.info( fPayment of ${amount} confirmed for {name}, card ending in {last_four_digits}, remaining balance: ${remaining} ) await self.session.generate_reply( instructionsfInform the user that the payment method ending in {last_four_digits} has been successfully charged ${amount}. Remaining balance: ${remaining}. ) current_tools [t for t in self.tools if t.id ! confirm_payment_proceeds] await self.update_tools(current_tools)金额 0拒绝无效/负数金额金额 ≤ 余额拒绝超额支付通过校验后才调用框架内置的GetCreditCardTask()livekit-agents/livekit/agents/beta/workflows/credit_card.py#L571收集卡号、有效期、安全码与持卡人姓名。GetCreditCardTask内部同样用TaskGroup编排多个子任务并且其收集顺序经过专门设计先问卡号用户最自然先报出的信息、再问有效期与安全码、持卡人姓名最后问从而避免用户先说数字被误认成姓名的失败模式。取到卡号后示例只记录后四位用于对用户播报并调用apply_payment扣减余额同时把临时挂载的支付工具从工具集中移除update_tools清掉confirm_payment_proceeds防止流程结束后模型误调用。从框架语义看GetCreditCardTask这类收集敏感信息的任务天然承担了合规边界卡号等数据只在任务内部流转结构化结果仅保留必要字段这为接入真实 PCI 合规支付网关留下了清晰的替换点。越界转人工WarmTransferTask 与安全确认无论主 Agent 还是各子任务都挂载了transfer_to_human工具examples/healthcare/agent.py#L81-L129。该工具的系统提示词明确要求模型必须先获得用户确认才能转接Ensure that the user has confirmed that they wanted to be transferred. Do not start transfer until the user has confirmed.这避免了一听到我要投诉就擅自挂机转接的误操作。转接流程先播报 Please hold while I connect you to a human agent.allow_interruptionsFalse防止被插话打断然后调用框架的WarmTransferTask实现在 livekit-agents/livekit/agents/beta/workflows/warm_transfer.py基于出站 SIP 中继result await WarmTransferTask( target_phone_numberSUPERVISOR_PHONE_NUMBER, sip_trunk_idSIP_TRUNK_ID, sip_numberSIP_NUMBER, chat_ctxcontext.session.history, )注意chat_ctxcontext.session.history把完整会话历史带给主管实现warm转接——主管接起电话时已经了解用户前面说了什么。转接成功后Agent 播报 you are on the line with my supervisor. Ill be hanging up now. 并主动context.session.shutdown()结束会话。全局系统指令GLOBAL_INSTRUCTIONSexamples/healthcare/agent.py#L49也始终约束着模型绝不提供医疗建议或诊断凡是超出协助范围的请求一律升级给真人。从示例到实战关键模式总结读完整个示例可以提炼出几条可直接迁移到任意语音智能体的工程模式任务即组件把问姓名问生日收集保险收信用卡等高频流程封装成AgentTask通过TaskGroup编排、通过complete()产出结构化结果、通过嵌套await复用改签复用ScheduleAppointmentTask复杂对话被分解为可测试、可复用的状态机。动态工具消除幻觉工具参数枚举在运行时从数据库读取并编译进 JSON Schema医生名单、可用时间槽、患者预约列表模型只能从真实数据中选值配合用户完成上一步才挂载下一步工具的节奏彻底杜绝参数幻觉与过早调用。异常即流程控制用自定义ToolError子类ProfileFound配合TaskGroup的return_exceptionsFalse实现认证快进把是否命中老客户的业务判断变成一次异常抛出代码路径清晰且易于扩展。模态双指令Instructions(语音版, 文本版)让同一任务在电话与聊天场景各说各话是模态无关体验的落地关键。敏感的转接转人工前强制确认、转接时传递完整history、转接成功后主动结束会话形成完整的客服升级闭环。进一步阅读本仓库还有其他可对照学习的示例例如 examples/frontdesk前台接待 日历预约、examples/drive_thru免下车点餐 订单管理与 examples/warm-transfer专门的转人工场景演示框架侧的AgentTask/TaskGroup等底层实现可继续阅读 livekit-agents/livekit/agents/beta/workflows/ 目录下的源码与对应测试。【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考