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-PythonGoogle 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 暴露为工具下面示例构建了一个客户验证工作流并将其暴露给父级客户服务 Agentfrom google.adk import Agent from google.adk import Workflow from pydantic import BaseModel, Field class CustomerLookupArgs(BaseModel): user_id: str Field(descriptionThe 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( namelookup_customer_tier, descriptionLook up membership status and account tier for a customer., input_schemaCustomerLookupArgs, edges[(START, fetch_tier)], ) root_agent Agent( namesupport_agent, instructionAnswer customer questions using the available lookup tools., tools[verification_workflow], )关键点Workflow必须显式指定input_schema一个 PydanticBaseModelrunner 才能据此为模型生成合法的参数声明name与description会被自动转成工具的调用标识与提示上下文详见下文配置选项edges[(START, fetch_tier)]表示从入口节点直接执行fetch_tier这是最简工作流形态。底层原理NodeTool 的包装与执行链路自动包装发生在哪里从源码结构看节点的工具化由两个入口共同完成均在 llm_agent.py 中Pydantic 模型校验器_pre_validate_toolsllm_agent.py在构造LlmAgent时遍历tools列表凡isinstance(t, BaseNode)的元素都会被替换为NodeTool(nodet, descriptiont.description)同时若t是BaseAgent会直接抛出ValueError。运行时工具展开llm_agent.pyBaseNode实例被包装为NodeToolBaseTool实例原样保留普通可调用对象包装为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的执行链路参数校验若input_schema是 PydanticBaseModel子类则调用input_schema.model_validate(args)校验模型传入的参数失败时返回错误字符串而不会中断整个流程。构造隔离分支以{tool_name}{function_call_id}为段名追加到父分支路径之后得到工具分支tool_branchfunction_call_id缺省时退化为纯tool_name。运行节点调用tool_context.run_node(...)传入override_branchtool_branch、use_sub_branchFalse、raise_on_waitTrue。所有中间事件、状态增量和进度日志都归属于该子分支父级 Agent 在构造后续模型提示时过滤掉子分支事件只保留工具返回的最终输出。中断透传若节点内部抛出NodeInterruptedError例如等待用户输入异常会被原样向上传播供上层在合适的时机恢复。配置选项工具属性与节点属性的映射当 Agent 将节点或工作流暴露为工具时工具配置完全由节点属性派生属性来源说明工具名Tool namenode.name呈现给模型的函数调用标识。描述Descriptionnode.description或 docstring描述工具用途的提示上下文。参数Parametersnode.input_schema或函数签名供模型函数调用使用的 JSON Schema。可被包装的节点包括任何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( nameorder_assistant, instructionHelp users check their order status., tools[check_order], )进阶应用二Human-in-the-loop 中断与恢复作为工具使用的节点可以产出RequestInput等交互式控制流事件。由于跨用户轮次的暂停与恢复要求 agent runner 保存并还原会话状态因此 Agent 必须被包装在配置了ResumabilityConfig(is_resumableTrue)的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_resumeTrue) 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_idmanager_approval, messagefAuthorize 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( namefinance_agent, instructionProcess customer refund requests using the refund tool., tools[process_refund], ) app App( namefinance_app, root_agentservice_agent, resumability_configResumabilityConfig(is_resumableTrue), )这里必须注意两个细节node(rerun_on_resumeTrue)节点被中断后再次恢复运行时会重新执行FunctionNode构造时也强制要求带auth_config的节点必须rerun_on_resumeTrue见 _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。限制与边界节点工具化面向任务导向、有界的工作流与确定性节点存在以下限制禁止包装会话式 AgentBaseAgent实例不能作为工具因为会话式 Agent 需要独立的轮流对话、多消息历史与子 Agent 交接。若需委托给另一个 Agent请配置sub_agents而非tools。Workflow 必须有 Pydantic 输入 schema任何用作工具的 Workflow 必须定义 PydanticBaseModel作为input_schema否则无法为模型函数调用生成合法参数声明。同时非FunctionNode且没有input_schema的节点在构造NodeTool时也会被拒绝_node_tool.py。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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考