企业官网建设流程全解析

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

本文基于 ADK（Agent Development Kit）官方示例 node_as_tool 展开，演示如何把一个普通的@node节点和一个Workflow图直接挂到父 Agent 的tools列表中，让 LLM 像调用普通函数一样调用它们。读完本文，你将掌握：工具声明如何从节点的input_schema自动生成、隔离子分支（sub-branch）如何防止中间事件污染父 Agent 上下文，以及节点工具如何通过RequestInput暂停执行并借助ResumabilityConfig在下一轮对话中恢复运行。

示例总览：客服场景下的两级工具调用

该示例构建了一个客服场景：父 Agent 收到「客户 c123 能享受什么折扣」的询问后，分两步行动——

1. 先调用customer_lookup_workflow（一个被包装为工具的Workflow）查出客户的会员等级；
2. 再调用calculate_discount（一个用@node装饰器定义、被包装为工具的普通节点）根据等级计算折扣；
3. 如果客户是 VIP，calculate_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. 定义输入 Schema：CustomerLookupArgs

from pydantic import BaseModel from pydantic import Field # 1. Define schemas class CustomerLookupArgs(BaseModel): user_id: str = Field(description="The customer's unique identifier.")

Pydantic 模型是Workflow作为工具时参数声明的来源：工具适配器会把它转成模型函数调用所需的 JSON Schema，每个字段的description会原样进入模型提示词，帮助 LLM 正确填参。

2. 用@node定义可暂停的 Node 工具：calculate_discount

from 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 tool's response schema stays small. @node(rerun_on_resume=True) def calculate_discount(tier: str, ctx: Context) -> Generator[str, None, None]: """Calculates the discount percentage based on customer tier. Args: tier: The customer's membership tier (e.g., VIP, Standard). """ yield Event(message=f"Checking 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_id="confirm_vip_discount", message=f"Apply 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_resume=True。这是 HITL 恢复的前提。节点暂停后，恢复时该节点会整体重跑（而非从暂停指令处断点续执），因此函数必须先检查ctx.resume_inputs.get("confirm_vip_discount")：如果没有恢复输入，就yield RequestInput并return退出；如果有，就读取用户回答（可能是 dict 取text字段，也可能是裸字符串）决定给 "20% off" 还是 "5% off (VIP declined)"。
• RequestInput中断。interrupt_id="confirm_vip_discount"与恢复输入中的键一一对应，用户回复会被路由回该节点。
• 参数签名即接口。tier: str参数和 docstring 中的Args:描述会被框架自动推断为工具的参数名与描述。从 node 装饰器源码 可以看到，当节点作为 Agent 工具使用时，parameter_binding会取值为'node_input'，参数即从工具入参绑定，并从函数签名推断input_schema/output_schema。

3. 用Workflow定义第二个工具：customer_lookup_workflow

def 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( name="customer_lookup_workflow", description="Looks up customer status and tier by user_id.", input_schema=CustomerLookupArgs, edges=[ ("START", lookup_customer_data), ], )

这里演示了 Workflow 作为工具的完整写法：

• name会成为模型看到的函数调用名；description进入提示词；input_schema提供参数 JSON Schema；
• edges用元组链描述图：("START", lookup_customer_data)表示从入口直接执行lookup_customer_data；
• 函数参数命名为node_input: CustomerLookupArgs，即接收整个 Pydantic 输入对象。

注意Workflow本身也是一个 Node，它可以作为节点嵌套进更大的图（参见 Workflow 指南），而本示例直接把它平铺为父 Agent 的工具。

4. 组装父 Agent 并启用可恢复性

root_agent = Agent( name="customer_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( name="node_as_tool", root_agent=root_agent, resumability_config=ResumabilityConfig(is_resumable=True), )

两个要点：

• 直接传对象即可。把Workflow实例和@node装饰后的FunctionNode直接放进tools列表，LlmAgent会自动将它们包装为NodeTool，无需手写任何适配代码——这正是 README「How To」部分给出的两条规则：定义节点/工作流并赋予input_schema与description，然后把它们直接传入tools列表。
• ResumabilityConfig(is_resumable=True)是 HITL 的硬性前提。跨用户轮次暂停与恢复要求 Runner 保存并还原执行状态，因此 Agent 必须包在启用了可恢复性的App中运行。

底层机制：NodeTool 如何把节点变成工具

LlmAgent包装节点工具的实际逻辑在 NodeTool 源码，示例中的每个行为都能在其中找到对应实现。

声明生成：name、description、input_schema 三要素

_build_node_declaration从节点属性构造模型的FunctionDeclaration：

其中有一个细节值得注意：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_schema

NodeTool.__init__中有三道闸门（见 _node_tool.py 第 84-109 行）：

1. 拒绝包装BaseAgent。如果传入的是 Agent 实例，直接抛ValueError提示应改用 Sub-Agent——对话型 Agent 有自己的轮次与会话语义，不属于任务型工具；
2. 自动对齐parameter_binding。若FunctionNode的parameter_binding不是'node_input'，会调用node._as_tool_node()重新绑定，保证参数来自工具入参而非会话 state；
3. 强制显式input_schema。非FunctionNode（即Workflow等）如果没有input_schema，抛ValueError要求提供 Pydantic Schema——这解释了为什么示例中customer_lookup_workflow必须声明CustomerLookupArgs。

执行期：隔离子分支与中断传播

run_async 是工具真正被调用时的入口，其执行流程：

1. 参数校验。若input_schema是 Pydantic 类，先model_validate(args)；校验失败不抛异常，而是把错误信息作为字符串返回给模型，让它自行修正参数重试。
2. 构造隔离子分支。fc_id = tool_context.function_call_id，子分支段为{工具名}@{函数调用ID}，拼接在父分支之后。这正是 Node as tool 指南 所描述的「isolated sub-branch」：节点执行中产生的中间事件、状态增量都归属该子分支，父 Agent 构建后续提示词时会过滤掉子分支事件，只保留工具最终返回值——所以calculate_discount里那句 "Checking discount rules for tier '...'..." 的进度消息不会干扰父 Agent 的上下文。
3. 以raise_on_wait=True运行节点，并通过override_branch=tool_branch注入子分支。关键点在第 173-174 行：普通的执行异常被捕获为错误字符串返回给模型，而NodeInterruptedError会被原样向上抛出——这就是RequestInput暂停信号能够穿透工具层、冒泡到 Runner 的机制。
4. NodeTool在构造时设置is_long_running = True（见 _node_tool.py 第 118 行），与 go.json 测试文件中的longRunningToolIds字段相呼应：Runner 知道这些工具调用可能跨轮次暂停。

恢复流程：从暂停点到ctx.resume_inputs

用户第二轮回复后，Runner 基于会话历史重建执行树，把回复直接路由到工具分支内被暂停的节点。rerun_on_resume=True让calculate_discount整体重跑：这次ctx.resume_inputs.get("confirm_vip_discount")有值，节点跳过RequestInput，按用户回答计算折扣并yield最终结果。

测试数据印证：一次真实的 HITL 事件序列

示例附带的事件轨迹文件 tests/go.json 完整记录了上面这套机制落地后的事件流，可以对照源码理解每个字段的含义：

1. 父 Agent 发出functionCall（id: fc-1）调用customer_lookup_workflow，args为{"user_id": "c123"}，且该调用被标记进longRunningToolIds；
2. customer_lookup_workflow返回{"user_id": "c123", "tier": "Verified VIP Member"}，其nodeInfo.path为customer_lookup_workflow@1/lookup_customer_data@1，branch为customer_lookup_workflow@fc-1——即上文说的{工具名}@{函数调用ID}子分支；
3. 父 Agent 再发functionCall（id: fc-2）调用calculate_discount，args为{"tier": "Verified VIP Member"}，分支calculate_discount@fc-2；
4. calculate_discount先产出进度事件 "Checking discount rules..."，随后产出author为calculate_discount的adk_request_input函数调用事件（id: fc-3），消息为 "Apply VIP discount for tier 'Verified VIP Member'?"，同样进入longRunningToolIds，本轮执行到此暂停；
5. 第二轮，用户事件以functionResponse形式回复fc-3：{"text": "yes"}，author为user；
6. 节点恢复后输出"20% off"（nodeInfo.outputFor指回calculate_discount@1），父 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_resumable=True)的App。如果节点只是确定性计算、不 yieldRequestInput，则不需要可恢复性配置。
• rerun_on_resume=True意味着恢复即重跑：节点函数必须写成幂等友好的形式——先读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