企业官网建设流程全解析

ADK-Python Node as Tool 深度指南：将 Workflow 与执行节点封装为 Agent 工具

【免费下载链接】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-Python（Google Agent Development Kit）中的 Node as Tool 能力展开：在多智能体架构中，父级 Agent 需要把确定性的多步流程（如数据处理管线、专业计算步骤）委托给 Workflow 或单个执行节点执行。本文讲解如何将Workflow与@node装饰的函数直接传入 Agent 的tools列表，由框架自动生成函数声明（Function Declaration）、校验参数、在隔离的子分支（sub-branch）中运行节点，并支持 Human-in-the-Loop 的暂停与恢复。读完本文，你将掌握节点工具化配置的完整写法、底层实现原理、HITL 中断恢复方案及限制边界。

为什么需要把节点暴露为工具

在多智能体架构中，Agent 经常需要委托确定性的工作流、数据处理管线或专门的计算步骤。将这些多步例程暴露为工具后，父级 Agent 模型可以像调用普通函数一样动态调用它们，而不是把流程逻辑硬编码进模型提示词。

把节点或工作流直接传入 Agent 的tools列表，相当于把"工作流执行单元"与"工具子系统"桥接起来。当 Agent 调用基于节点的工具时，runner 会在一个隔离的子分支中执行底层的节点或工作流。这个子分支的设计目的是：

• 隔离节点执行期间产生的中间进度消息与内部状态变化，避免污染父级 Agent 的上下文；
• 同时仍然允许 Human-in-the-loop 的暂停以中断（interrupt）形式浮出到调用方，供上层处理。

任何Workflow或BaseNode实例只要被放入LlmAgent的tools列表，就会被自动包装成工具，无需手动实例化包装类。

快速开始：把 Workflow 暴露为工具

下面示例构建了一个客户验证工作流，并将其暴露给父级客户服务 Agent：

from google.adk import Agent from google.adk import Workflow from pydantic import BaseModel, Field class CustomerLookupArgs(BaseModel): user_id: str = Field(description="The unique identifier of the customer.") def fetch_tier(node_input: CustomerLookupArgs, ctx) -> dict[str, str]: return {"user_id": node_input.user_id, "tier": "Gold Member"} verification_workflow = Workflow( name="lookup_customer_tier", description="Look up membership status and account tier for a customer.", input_schema=CustomerLookupArgs, edges=[("START", fetch_tier)], ) root_agent = Agent( name="support_agent", instruction="Answer customer questions using the available lookup tools.", tools=[verification_workflow], )

关键点：

• Workflow必须显式指定input_schema（一个 PydanticBaseModel），runner 才能据此为模型生成合法的参数声明；
• name与description会被自动转成工具的调用标识与提示上下文（详见下文"配置选项"）；
• edges=[("START", fetch_tier)]表示从入口节点直接执行fetch_tier，这是最简工作流形态。

底层原理：NodeTool 的包装与执行链路

自动包装发生在哪里

从源码结构看，节点的工具化由两个入口共同完成，均在 llm_agent.py 中：

1. Pydantic 模型校验器（_pre_validate_tools，llm_agent.py）：在构造LlmAgent时遍历tools列表，凡isinstance(t, BaseNode)的元素都会被替换为NodeTool(node=t, description=t.description)；同时若t是BaseAgent会直接抛出ValueError。
2. 运行时工具展开（llm_agent.py）：BaseNode实例被包装为NodeTool，BaseTool实例原样保留，普通可调用对象包装为FunctionTool。

核心包装类NodeTool定义于 _node_tool.py，它继承BaseTool，持有被包装的节点，并设置了self.is_long_running = True——节点工具被标记为长任务，这与 Agent Tool 的长时间运行语义一致。

函数声明的生成规则

_build_node_declaration（_node_tool.py）负责为模型构造FunctionDeclaration：

• 名称取自node.name，描述取自node.description（缺省时回退为'Executes the node: {node.name}'）；
• input_schema通过schema_to_json_schema转成 JSON Schema 后写入parameters_json_schema；若节点输入是str、int等原始类型，GenAI API 要求 schema 必须是object类型，因此会被自动包成带request属性的对象 schema；
• 若节点定义了output_schema，还会写入response_json_schema。

对于@node装饰的函数节点（FunctionNode），工具参数名与类型直接从函数签名与 docstring 推断；若其parameter_binding不是node_input，则调用FunctionNode._as_tool_node()对齐绑定方式（见 _function_node.py）。

参数校验与子分支执行

run_async（_node_tool.py）的执行链路：

1. 参数校验：若input_schema是 PydanticBaseModel子类，则调用input_schema.model_validate(args)校验模型传入的参数；失败时返回错误字符串而不会中断整个流程。
2. 构造隔离分支：以{tool_name}@{function_call_id}为段名，追加到父分支路径之后，得到工具分支tool_branch（function_call_id缺省时退化为纯tool_name）。
3. 运行节点：调用tool_context.run_node(...)，传入override_branch=tool_branch、use_sub_branch=False、raise_on_wait=True。所有中间事件、状态增量和进度日志都归属于该子分支；父级 Agent 在构造后续模型提示时过滤掉子分支事件，只保留工具返回的最终输出。
4. 中断透传：若节点内部抛出NodeInterruptedError（例如等待用户输入），异常会被原样向上传播，供上层在合适的时机恢复。

配置选项：工具属性与节点属性的映射

当 Agent 将节点或工作流暴露为工具时，工具配置完全由节点属性派生：

可被包装的节点包括任何BaseNode派生实例：Workflow图、@node装饰的函数等。直接包装 Agent 作为工具会被拒绝——因为 Agent 使用对话式会话语义，应放在sub_agents中（此校验同时存在于 _node_tool.py 与 llm_agent.py）。

两种输入推断路径的差异：

• Workflow：必须通过Workflow(..., input_schema=...)显式指定输入 schema；
• 独立@node函数：参数名、类型与 docstring 描述直接从函数签名推断，无需额外定义 Pydantic 模型。

进阶应用一：@node函数直接作为工具

将@node装饰的函数直接传入 Agent 的tools参数即可自动包装为工具：

from google.adk import Agent from google.adk.workflow import node @node def check_order(order_id: str) -> dict[str, str]: """Checks shipping status for an existing order identifier. Args: order_id: The identifier of the order to check. """ return {"status": "shipped"} agent = Agent( name="order_assistant", instruction="Help users check their order status.", tools=[check_order], )

进阶应用二：Human-in-the-loop 中断与恢复

作为工具使用的节点可以产出RequestInput等交互式控制流事件。由于跨用户轮次的暂停与恢复要求 agent runner 保存并还原会话状态，因此 Agent 必须被包装在配置了ResumabilityConfig(is_resumable=True)的App中：

from typing import Generator from google.adk import Agent from google.adk import Context from google.adk.apps import App from google.adk.apps import ResumabilityConfig from google.adk.events import RequestInput from google.adk.workflow import node @node(rerun_on_resume=True) def process_refund( amount: float, ctx: Context ) -> Generator[str, None, None]: """Processes customer refund requests with manager approval. Args: amount: The refund amount in dollars. """ resume_input = ctx.resume_inputs.get("manager_approval") if not resume_input: yield RequestInput( interrupt_id="manager_approval", message=f"Authorize refund of ${amount}?", ) return decision = str(resume_input).strip().lower() if decision in ("approved", "yes"): yield "Refund processed successfully." else: yield "Refund request rejected." service_agent = Agent( name="finance_agent", instruction="Process customer refund requests using the refund tool.", tools=[process_refund], ) app = App( name="finance_app", root_agent=service_agent, resumability_config=ResumabilityConfig(is_resumable=True), )

这里必须注意两个细节：

• @node(rerun_on_resume=True)：节点被中断后再次恢复运行时会重新执行；FunctionNode构造时也强制要求"带auth_config的节点必须rerun_on_resume=True"（见 _function_node.py）；
• 中断恢复路径：当用户以响应事件恢复调用时，runner 会重建执行树，并把恢复响应直接路由到工具分支内被暂停的节点。

完整可运行的端到端示例位于 node_as_tool/agent.py：父 Agent 先调用customer_lookup_workflow获取客户等级，再调用calculate_discount节点；当客户是 VIP 时节点产出RequestInput请求确认，运行暂停，下一轮输入yes后恢复并计算出 "20% off" 折扣。对应拓扑图与多轮输入说明见 node_as_tool/README.md。

限制与边界

节点工具化面向任务导向、有界的工作流与确定性节点，存在以下限制：

1. 禁止包装会话式 Agent：BaseAgent实例不能作为工具，因为会话式 Agent 需要独立的轮流对话、多消息历史与子 Agent 交接。若需委托给另一个 Agent，请配置sub_agents而非tools。
2. Workflow 必须有 Pydantic 输入 schema：任何用作工具的 Workflow 必须定义 PydanticBaseModel作为input_schema，否则无法为模型函数调用生成合法参数声明。同时，非FunctionNode且没有input_schema的节点在构造NodeTool时也会被拒绝（_node_tool.py）。
3. @node函数则相反：独立节点的参数名与类型提示直接声明在函数签名上，无需 Pydantic 模型。

相关资源

• Node as Tool 示例（agent.py）：演示 Agent 同时将 Workflow 与交互式 HITL 节点作为工具调用；
• Node as Tool 示例说明（README.md）：包含拓扑图与多轮输入演练；
• 核心实现：NodeTool 工具包装类、LlmAgent 工具自动包装逻辑、FunctionNode 节点实现；
• 相关工作流指南：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