ADK 中将 Workflow 与 Node 包装为 Agent 工具的实战从 customer_lookup_workflow 到 Human-in-the-Loop 折扣确认【免费下载链接】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官方示例 node_as_tool 展开演示如何把一个普通的node节点和一个Workflow图直接挂到父 Agent 的tools列表中让 LLM 像调用普通函数一样调用它们。读完本文你将掌握工具声明如何从节点的input_schema自动生成、隔离子分支sub-branch如何防止中间事件污染父 Agent 上下文以及节点工具如何通过RequestInput暂停执行并借助ResumabilityConfig在下一轮对话中恢复运行。示例总览客服场景下的两级工具调用该示例构建了一个客服场景父 Agent 收到「客户 c123 能享受什么折扣」的询问后分两步行动——先调用customer_lookup_workflow一个被包装为工具的Workflow查出客户的会员等级再调用calculate_discount一个用node装饰器定义、被包装为工具的普通节点根据等级计算折扣如果客户是 VIPcalculate_discount会yield一个RequestInput事件向用户确认「是否应用 VIP 折扣」整个调用在此暂停用户下一轮回复后执行树被重建从暂停点继续最终输出 20% off由父 Agent 汇总答复。Agent 拓扑如下继承自 README 原文档推荐的试跑输入是What discount does customer c123 get?这是一次典型的多轮 Human-in-the-Loop 交互第 1 轮触发RequestInput暂停第 2 轮回复yes恢复执行。完整实现代码解析示例的完整实现在 agent.py下面按「Schema 定义 → Node 工具 → Workflow 工具 → 父 Agent → App 可恢复性」的顺序逐段拆解。1. 定义输入 SchemaCustomerLookupArgsfrom pydantic import BaseModel from pydantic import Field # 1. Define schemas class CustomerLookupArgs(BaseModel): user_id: str Field(descriptionThe customers unique identifier.)Pydantic 模型是Workflow作为工具时参数声明的来源工具适配器会把它转成模型函数调用所需的 JSON Schema每个字段的description会原样进入模型提示词帮助 LLM 正确填参。2. 用node定义可暂停的 Node 工具calculate_discountfrom google.adk import Context from google.adk.events import RequestInput # 2. Define a regular Node using the node decorator. # This Node is wrapped as a NodeTool automatically by the Agent. # As a NodeTool, it has the ability to yield intermediate Events during execution. # Annotate the yield type with the data the tool returns, not the Event and # RequestInput control-flow items, so the tools response schema stays small. node(rerun_on_resumeTrue) def calculate_discount(tier: str, ctx: Context) - Generator[str, None, None]: Calculates the discount percentage based on customer tier. Args: tier: The customers membership tier (e.g., VIP, Standard). yield Event(messagefChecking discount rules for tier {tier}...) resume_input ctx.resume_inputs.get(confirm_vip_discount) if VIP in tier: if not resume_input: yield RequestInput( interrupt_idconfirm_vip_discount, messagefApply VIP discount for tier {tier}?, ) return user_response ( resume_input.get(text) if isinstance(resume_input, dict) else resume_input ) if str(user_response).lower() in (yes, y, true): discount 20% off else: discount 5% off (VIP declined) else: discount 5% off yield discount这段代码承载了示例的四个关键机制生成器语义。函数声明为Generator[str, None, None]可以yield三种东西中间进度Event、控制流事件RequestInput、最终返回值字符串 discount。注意注释中的建议——yield 类型标注应写工具真正返回的数据类型这里是str而不是Event/RequestInput这类控制流类型这样工具响应 Schema 会保持精简。rerun_on_resumeTrue。这是 HITL 恢复的前提。节点暂停后恢复时该节点会整体重跑而非从暂停指令处断点续执因此函数必须先检查ctx.resume_inputs.get(confirm_vip_discount)如果没有恢复输入就yield RequestInput并return退出如果有就读取用户回答可能是 dict 取text字段也可能是裸字符串决定给 20% off 还是 5% off (VIP declined)。RequestInput中断。interrupt_idconfirm_vip_discount与恢复输入中的键一一对应用户回复会被路由回该节点。参数签名即接口。tier: str参数和 docstring 中的Args:描述会被框架自动推断为工具的参数名与描述。从 node 装饰器源码 可以看到当节点作为 Agent 工具使用时parameter_binding会取值为node_input参数即从工具入参绑定并从函数签名推断input_schema/output_schema。3. 用Workflow定义第二个工具customer_lookup_workflowdef lookup_customer_data(node_input: CustomerLookupArgs, ctx) - dict[str, str]: return {user_id: node_input.user_id, tier: Verified VIP Member} customer_lookup_workflow Workflow( namecustomer_lookup_workflow, descriptionLooks up customer status and tier by user_id., input_schemaCustomerLookupArgs, edges[ (START, lookup_customer_data), ], )这里演示了 Workflow 作为工具的完整写法name会成为模型看到的函数调用名description进入提示词input_schema提供参数 JSON Schemaedges用元组链描述图(START, lookup_customer_data)表示从入口直接执行lookup_customer_data函数参数命名为node_input: CustomerLookupArgs即接收整个 Pydantic 输入对象。注意Workflow本身也是一个 Node它可以作为节点嵌套进更大的图参见 Workflow 指南而本示例直接把它平铺为父 Agent 的工具。4. 组装父 Agent 并启用可恢复性root_agent Agent( namecustomer_service_agent, instruction You are a customer service assistant. 1. First, call customer_lookup_workflow using the user_id to get their membership tier. 2. Then, call calculate_discount node with that tier to find out what discount they get. Summarize these details for the customer. , tools[customer_lookup_workflow, calculate_discount], ) # Wrap the agent in an App and enable resumability. This is required because # the calculate_discount tool yields a RequestInput event which pauses # execution, and we need to resume the agent in a subsequent turn. app App( namenode_as_tool, root_agentroot_agent, resumability_configResumabilityConfig(is_resumableTrue), )两个要点直接传对象即可。把Workflow实例和node装饰后的FunctionNode直接放进tools列表LlmAgent会自动将它们包装为NodeTool无需手写任何适配代码——这正是 README「How To」部分给出的两条规则定义节点/工作流并赋予input_schema与description然后把它们直接传入tools列表。ResumabilityConfig(is_resumableTrue)是 HITL 的硬性前提。跨用户轮次暂停与恢复要求 Runner 保存并还原执行状态因此 Agent 必须包在启用了可恢复性的App中运行。底层机制NodeTool 如何把节点变成工具LlmAgent包装节点工具的实际逻辑在 NodeTool 源码示例中的每个行为都能在其中找到对应实现。声明生成name、description、input_schema 三要素_build_node_declaration从节点属性构造模型的FunctionDeclaration工具配置项来源说明工具名node.name展示给模型的 function call 标识符本例为customer_lookup_workflow、calculate_discount描述node.description或函数 docstring向模型描述工具用途的提示词参数node.input_schema或函数签名模型函数调用参数的 JSON Schema其中有一个细节值得注意GenAI API 要求parameters_json_schema必须是object类型如果节点的输入 Schema 是原始类型如str、int框架会自动包一层{type: object, properties: {request: ...}, required: [request]}见 _node_tool.py 第 58-65 行。另外若节点声明了output_schema也会同步写入response_json_schema。构造期校验拒绝 Agent、强制 input_schemaNodeTool.__init__中有三道闸门见 _node_tool.py 第 84-109 行拒绝包装BaseAgent。如果传入的是 Agent 实例直接抛ValueError提示应改用 Sub-Agent——对话型 Agent 有自己的轮次与会话语义不属于任务型工具自动对齐parameter_binding。若FunctionNode的parameter_binding不是node_input会调用node._as_tool_node()重新绑定保证参数来自工具入参而非会话 state强制显式input_schema。非FunctionNode即Workflow等如果没有input_schema抛ValueError要求提供 Pydantic Schema——这解释了为什么示例中customer_lookup_workflow必须声明CustomerLookupArgs。执行期隔离子分支与中断传播run_async 是工具真正被调用时的入口其执行流程参数校验。若input_schema是 Pydantic 类先model_validate(args)校验失败不抛异常而是把错误信息作为字符串返回给模型让它自行修正参数重试。构造隔离子分支。fc_id tool_context.function_call_id子分支段为{工具名}{函数调用ID}拼接在父分支之后。这正是 Node as tool 指南 所描述的「isolated sub-branch」节点执行中产生的中间事件、状态增量都归属该子分支父 Agent 构建后续提示词时会过滤掉子分支事件只保留工具最终返回值——所以calculate_discount里那句 Checking discount rules for tier ...... 的进度消息不会干扰父 Agent 的上下文。以raise_on_waitTrue运行节点并通过override_branchtool_branch注入子分支。关键点在第 173-174 行普通的执行异常被捕获为错误字符串返回给模型而NodeInterruptedError会被原样向上抛出——这就是RequestInput暂停信号能够穿透工具层、冒泡到 Runner 的机制。NodeTool在构造时设置is_long_running True见 _node_tool.py 第 118 行与 go.json 测试文件中的longRunningToolIds字段相呼应Runner 知道这些工具调用可能跨轮次暂停。恢复流程从暂停点到ctx.resume_inputs用户第二轮回复后Runner 基于会话历史重建执行树把回复直接路由到工具分支内被暂停的节点。rerun_on_resumeTrue让calculate_discount整体重跑这次ctx.resume_inputs.get(confirm_vip_discount)有值节点跳过RequestInput按用户回答计算折扣并yield最终结果。测试数据印证一次真实的 HITL 事件序列示例附带的事件轨迹文件 tests/go.json 完整记录了上面这套机制落地后的事件流可以对照源码理解每个字段的含义父 Agent 发出functionCallid: fc-1调用customer_lookup_workflowargs为{user_id: c123}且该调用被标记进longRunningToolIdscustomer_lookup_workflow返回{user_id: c123, tier: Verified VIP Member}其nodeInfo.path为customer_lookup_workflow1/lookup_customer_data1branch为customer_lookup_workflowfc-1——即上文说的{工具名}{函数调用ID}子分支父 Agent 再发functionCallid: fc-2调用calculate_discountargs为{tier: Verified VIP Member}分支calculate_discountfc-2calculate_discount先产出进度事件 Checking discount rules...随后产出author为calculate_discount的adk_request_input函数调用事件id: fc-3消息为 Apply VIP discount for tier Verified VIP Member?同样进入longRunningToolIds本轮执行到此暂停第二轮用户事件以functionResponse形式回复fc-3{text: yes}author为user节点恢复后输出20% offnodeInfo.outputFor指回calculate_discount1父 Agent 收到functionResponse后给出最终答复 Customer c123 is a Verified VIP Member and gets a 20% discount.这个文件是理解节点工具暂停/恢复事件语义的最佳素材branch字段标记事件归属的子分支nodeInfo.path标记执行树位置longRunningToolIds标记跨轮次未完成的调用。适用边界与注意事项只包装任务型节点。NodeTool明确拒绝把BaseAgent包成工具见 源码第 87-91 行要把工作委托给另一个 Agent应配置sub_agents而非tools。Workflow必须有 Pydanticinput_schema否则构造NodeTool时抛异常而node函数则直接从签名与 docstring 推断参数两者风格不同示例中同时展示了两种写法。HITL 节点必须配合ResumabilityConfig(is_resumableTrue)的App。如果节点只是确定性计算、不 yieldRequestInput则不需要可恢复性配置。rerun_on_resumeTrue意味着恢复即重跑节点函数必须写成幂等友好的形式——先读ctx.resume_inputs判断是否已恢复未恢复才发中断这也是示例代码中查恢复输入 → 无则中断 → 有则消费三段式结构的由来。延伸阅读Workflow 指南讲解edges图定义、动态调度、并行分支与工作流输出规则单终端节点取输出、多终端节点需要JoinNode聚合。Node as tool 指南更系统地说明 schema 生成、参数校验、隔离子分支作用域与 HITL 恢复的机制并附退款审批等进阶示例。示例入口 agent.py 与事件轨迹 tests/go.json前者可直接复制到你的 ADK 项目中改造成业务场景后者可作为调试多轮暂停/恢复行为时的对参考数据。【免费下载链接】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），仅供参考