企业官网建设流程全解析

ADK-Python 如何把 Workflow 和 @node 节点直接作为 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（Agent Development Kit）中，父 Agent 经常需要把多步的确定性流程、数据处理管线或带审批步骤的环节委托出去。LlmAgent会自动把传入其tools列表的Workflow或BaseNode实例包装成工具，父模型的函数调用即可触发这些执行单元；如果节点在执行中yield一个RequestInput事件请求人工输入，整个调用可以暂停并在用户下一轮回复后恢复。本文按“定义工具 → 声明人工介入点 → 配置可恢复的 App → 运行验证”这条路径走一遍，示例来自项目内置样例 node_as_tool，环境要求为 Python 3.10+，安装方式见根目录 README：

pip install google-adk

把 Workflow 暴露为 Agent 工具

Workflow本身是一个节点（graph-based 编排节点，详见 Workflow 指南），可以直接放进 Agent 的tools。作为工具时，它必须用一个 PydanticBaseModel声明input_schema，框架据此为模型生成函数声明，并在调用前按该 schema 校验参数：

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], )

工具配置是从节点属性派生的（见 Node as tool 指南）：

把 @node 函数暴露为工具

被@node装饰的函数可以直接传给tools，自动包装成工具。与 Workflow 不同，独立@node函数不要求单独的input_schema，参数名、类型和 docstring 描述直接从函数签名推断：

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], )

注意不要直接把会话型 Agent 塞进tools：框架禁止把BaseAgent包装成工具，因为对话型 Agent 有自己的轮次与消息历史语义，应通过sub_agents委托。

在工具节点中暂停并恢复人工介入

节点作为工具时可以yieldRequestInput这类交互控制流事件。由于跨轮次暂停和恢复需要 Runner 保存并恢复会话状态，Agent 要包在配置了ResumabilityConfig(is_resumable=True)的App里，并且节点要带@node(rerun_on_resume=True)，这样恢复时节点会重新执行到暂停点而不是直接以恢复输入作为输出：

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), )

这套机制的底层过程在 RequestInput 指南 中有完整说明：节点yield的RequestInput会被包装成携带名为adk_request_input的函数调用的Event，客户端展示其中的message；恢复时客户端回发一个名为adk_request_input、id匹配interrupt_id的FunctionResponse，响应载荷放在response字典里。Runner 收到恢复事件后会重建执行树，把恢复响应直接路由到工具分支内被暂停的节点。

隔离执行是这条路径能保持父上下文干净的关键：Agent 生成针对节点/工作流的工具调用后，Runner 在父分支下构造一个隔离子分支，路径形如{tool_name}@{function_call_id}。工具执行中产生的中间事件、状态变更和进度日志都归属这个子分支，父 Agent 构建后续模型 prompt 时会过滤掉子分支事件，只保留工具返回的最终输出；人工暂停事件则可以穿透子分支向上浮出给调用方。

运行与验证

如果模块内定义了模块级变量app，adk run、adk web和adk api_server会优先取用它来构建 Runner（只有找不到App时才回退到root_agent），上面的 HITL 样例因此只需导出app变量即可被这些命令识别。App 指南另有一个注意点：加载 Agent 时若 App 名与所在目录名不一致，Runner 会记录警告，改名保持一致即可。

完整可运行版本是内置样例 node_as_tool：父 Agentcustomer_service_agent同时挂了一个Workflow工具（按user_id查客户等级）和一个带 HITL 的@node工具（按等级算折扣），外层用App(name="node_as_tool", resumability_config=ResumabilityConfig(is_resumable=True))包装。按样例 README 的说明，发送输入What discount does customer c123 get?后的预期交互是（以下为文档示例输出）：

• 第 1 轮：父 Agent 先调用customer_lookup_workflow拿到等级，再调用calculate_discount；后者yield出RequestInput，消息为Apply VIP discount for tier 'Verified VIP Member'?，本次运行暂停。
• 第 2 轮：回复yes恢复运行，节点算出20% off，父 Agent 汇总为Customer c123 is a Verified VIP Member and gets a 20% discount.

完整的两轮事件序列（含子分支标记calculate_discount@fc-2与恢复用的adk_request_input函数响应）可对照样例的测试记录 tests/go.json，其中branch字段能直接看出中间事件落在哪个子分支里。

如果想绕过 CLI 自己驱动，可以创建会话服务、把App交给Runner逐事件打印（取自 App 指南）：

import asyncio from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.genai import types async def main() -> None: session_service = InMemorySessionService() session = await session_service.create_session( app_name=app.name, user_id="user" ) runner = Runner(app=app, session_service=session_service) async for event in runner.run_async( user_id="user", session_id=session.id, new_message=types.Content( role="user", parts=[types.Part(text="What discount does customer c123 get?")], ), ): if event.content and event.content.parts: print(event.author, event.content.parts[0].text) if __name__ == "__main__": asyncio.run(main())

注意会话按app.name索引，create_session与查找会话用的名称必须一致。

限制与排查

• Workflow 作为工具必须有 Pydanticinput_schema，否则 Runner 无法为模型函数调用生成有效的参数声明；独立@node函数则依赖签名和类型注解。
• 恢复是尽力而为的：按 App 指南 的说明，resumption 只保证 at-least-once 执行，可能恢复的工具需要幂等，暂停期间的内存状态会丢失。
• ResumabilityConfig属于实验性配置，构造时会输出 experimental 警告，且可能不提前通知就变更。
• 带response_schema的RequestInput：ADK 负责恢复时的解析，但校验用户输入是否符合 schema 是客户端的责任（见 RequestInput 指南）。
• 函数节点的更多参数与rerun_on_resume等包装选项见 Function nodes 指南；图的结构、START导入和校验规则见 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