在 ADK Workflow 中嵌入 LLM Agent 节点：Task 模式与 Single-Turn 模式实战（agent_in_workflow 示例解析）

在 ADK Workflow 中嵌入 LLM Agent 节点Task 模式与 Single-Turn 模式实战agent_in_workflow 示例解析【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python本文基于 ADKAgent Development Kit官方示例agent_in_workflow展开讲解如何将task模式与single_turn模式的 LLM Agent 直接作为Workflow的节点参与图编排并结合一个医疗实验室预约登记场景演示多轮对话采集结构化数据、函数节点条件路由retry 回路以及工具执行前的人工确认tool confirmation。读完本文你将掌握在 ADK 工作流中混合编排会聊天收数据的 Agent与单轮执行任务的 Agent并理解底层 Runner 如何根据mode调度它们。示例概览一个医学实验室登记工作流示例位于 contributing/samples/workflows/agent_in_workflow其完整实现只有一个文件 agent.py。整个工作流模拟医疗实验室的登记intake流程由三个节点构成intake_agent一个task模式 Agent与用户进行多轮对话收集用户的name和phone_number。它一直与用户交互直到输出结构PatientIdentity被填满即任务完成。check_identity一个普通 Python 函数节点接收PatientIdentity模拟数据库校验如果姓名不是Jane Doe则产生一条retry路由把用户送回intake_agent如果姓名是Jane Doe则沿默认路由DEFAULT_ROUTE走向generate_instruction。generate_instruction一个single_turn模式 Agent使用find_orders工具查询订单且该工具在真正执行前需要用户确认tool confirmation。工作流图如下[ START ] | v [ intake_agent ] ----. | | v | [ check_identity ] --- retry | | (DEFAULT_ROUTE) v [ generate_instruction ]从源码看check_identity是一个生成器函数节点它把输入PatientIdentity校验结果以Event的形式产出并通过routeretry携带路由信息由工作流调度器据此决定下一步走向agent.py。推荐的测试输入示例 README 提供了三组典型输入覆盖了工作流的全部三条路径输入预期行为Hi, I am Jane Doe, my phone number is 555-1234.系统一次性拿到姓名和电话直接返回模拟的实验室订单CBC、Lipid Panel以及 AI 生成的准备指引Im here for my blood work.信息不完整系统会追问姓名和电话号码task 模式多轮对话My name is John Doe, and my number is 123-456-7890.数据库校验失败系统把用户路由回intake_agent重新采集这三条路径在示例自带的测试用例中都有对应的完整事件轨迹可对照阅读tests/jane_doe_and_phone_number.json一次输入完整信息走通全流程tests/wrong_name.json姓名错误触发retry回路第二次输入正确姓名后成功tests/go_approve.json 与 tests/go_decline.json分别演示工具确认被批准与被拒绝的场景tests/jane_doe.json、tests/phone_number.json分步补充信息的交互过程。如何在 Workflow 中嵌入 Agent 节点在 ADK 工作流中LLM Agent 可以直接作为节点嵌入图。ADK Runner 会根据 Agent 的mode决定如何调度它modetaskAgent 自己处理多轮对话直到任务完成才把控制权交给下一个节点modesingle_turn省略时的默认值只执行一次 LLM 调用常用于内联文本生成、摘要或分类不与用户闲聊。在源码中mode的类型定义于 src/google/adk/agents/llm_agent.py取值为Literal[chat, task, single_turn]。从源码结构看src/google/adk/agents/llm_agent.py 附近的模式分支逻辑task模式与single_turn模式在输出处理、子 Agent 委托等方面有专门的代码路径这正是本示例想要展示的核心机制。1. Task 模式 Agent多轮对话收集结构化数据class PatientIdentity(BaseModel): name: str Field(descriptionThe patients full name.) phone_number: str Field(descriptionThe patients phone number.) intake_agent Agent( nameintake_agent, modetask, # 停下来与用户对话直到 schema 被填满 output_schemaPatientIdentity, instruction\ You are a medical lab intake assistant. Your job is to chat with the user to get their full name and phone number. Do not make up information. Once you have both, finish your task. If identity check failed, ask for another name. , )关键点output_schema即任务的完成条件task模式 Agent 会持续与用户交互直到解析出的输出符合PatientIdentityname、phone_number两个必填字段为止。当数据齐备后它会调用内部finish_task工具结束任务。输出自动向下传递解析好的output_schema对象会自动作为node_input传给图中的下一个节点。在示例中check_identity(node_input: PatientIdentity)直接以类型注解接收了这个对象agent.py工作流框架据此把上游输出绑定到下游函数参数上。指令中的回环提示instruction里写明如果身份校验失败请再询问一个名字配合check_identity的retry路由形成了采集 → 校验 → 失败 → 重新采集的闭环。从测试轨迹 tests/jane_doe.json 可以看到task模式的实际行为Agent 先追问Hi Jane, what is your phone number?用户补充号码后模型发出finish_task函数调用事件中携带output: {name: Jane Doe, phone_number: 555-1234}随后check_identity节点才被触发——这正是任务完成后才移交控制权的运行证据。2. Single-Turn 模式 Agent一次调用完成查询与生成def find_orders() - list[str]: Finds orders for the patient. return [CBC (Complete Blood Count), Lipid Panel] generate_instruction Agent( namegenerate_instruction, tools[FunctionTool(find_orders, require_confirmationTrue)], instruction Use the find_orders tool to get the patients orders. List the orders found, and then generate a concise instruction about how to prepare based on those orders. , )generate_instruction省略了mode参数因此默认就是single_turn它接收上游check_identity传下的PatientIdentity此时check_identity的事件消息Hello Jane Doe! Let me look up your orders.已在上下文中调用find_orders拿到订单列表再生成一条简明的检查前准备指引一次调用即产出最终回复。从测试轨迹看该 Agent 的最终输出事件带有messageAsOutput: true且outputFor同时指向自身与工作流根节点例如 tests/jane_doe_and_phone_number.json说明single_turn节点的消息会被提升为工作流的整体输出。3. 工具确认Tool Confirmation执行前的人工审批示例还演示了tool confirmationFunctionTool(find_orders, require_confirmationTrue)表示调用find_orders前必须征得用户同意。从事件轨迹可以看到完整的交互协议tests/go_approve.json模型发出find_orders函数调用fc-2框架将其转为名为adk_request_confirmation的确认请求携带toolConfirmation提示与longRunningToolIds同时给原调用返回需要确认的错误响应用户以FunctionResponse回复{confirmed: true}或false确认通过后框架才真正执行find_orders返回[CBC (Complete Blood Count), Lipid Panel]。若用户拒绝见 tests/go_decline.json工具会收到This tool call is rejected.的错误Agent 据此向用户说明无法获取订单。这种机制为Agent 自动调用工具加上了一道人工闸门特别适合订单查询、数据修改等敏感操作。完整工作流装配边与路由三个节点通过Workflow的edges组装成图root_agent Workflow( nametask_in_workflow, edges[ (START, intake_agent, check_identity), ( check_identity, {retry: intake_agent, DEFAULT_ROUTE: generate_instruction}, ), ], )解读第一条边(START, intake_agent, check_identity)表示从START进入intake_agent完成后顺序进入check_identity函数节点默认会把上游输出作为node_input。第二条边以路由字典形式声明分支check_identity产出的routeretry事件把控制权送回intake_agent其余情况走DEFAULT_ROUTE到generate_instruction。DEFAULT_ROUTE在源码中定义为DEFAULT_ROUTE __DEFAULT__src/google/adk/workflow/_graph.py从 src/google/adk/workflow/_graph.py 附近的边匹配逻辑可以推断路由事件与边的route精确匹配未命中任何显式路由时回落到DEFAULT_ROUTE。默认路由因此充当else 分支保证图始终有确定的下一步。运行与验证方式ADK 提供了adk run命令行工具可交互式运行一个 agent 目录。以本仓库中其他示例 README 的用法为参考例如 contributing/samples/context_management/cache_analysis/README.md 中的adk run 目录本示例可类似地通过如下命令启动交互会话adk run contributing/samples/workflows/agent_in_workflow运行时可以输入上文推荐的测试输入中的任意一句观察三种典型路径完整信息直达订单结果、信息不全触发追问、错误姓名触发重试回路。示例自带的 JSON 测试文件则提供了确定性的事件级验证可与本地运行结果相互对照用于理解每一步用户消息、模型回复、finish_task、路由事件、工具确认、最终输出在图中的确切流向。小结agent_in_workflow示例浓缩了 ADK 工作流编排的三种核心能力Agent 即节点task与single_turn模式 Agent 可以直接作为Workflow节点由 Runner 依据mode自动调度——多轮对话采集数据用task单轮查询/生成用single_turn结构化输出接力task模式 Agent 解析出的output_schema对象自动成为下游节点的node_input实现对话收数据 → 代码做校验的无缝衔接路由与人工闸门函数节点通过route事件驱动重试回路DEFAULT_ROUTE兜底默认走向require_confirmationTrue为工具调用增加了人工确认环节。如需进一步了解工作流节点、路由与动态编排的底层机制可继续阅读仓库中的 src/google/adk/workflow/_workflow.py工作流调度主循环、src/google/adk/workflow/_graph.py图结构与DEFAULT_ROUTE定义以及 src/google/adk/workflow/_llm_agent_wrapper.pyLLM Agent 作为节点的包装与finish_task处理。【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考