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-PythonAgent Development Kit中父 Agent 经常需要把多步的确定性流程、数据处理管线或带审批步骤的环节委托出去。LlmAgent会自动把传入其tools列表的Workflow或BaseNode实例包装成工具父模型的函数调用即可触发这些执行单元如果节点在执行中yield一个RequestInput事件请求人工输入整个调用可以暂停并在用户下一轮回复后恢复。本文按“定义工具 → 声明人工介入点 → 配置可恢复的 App → 运行验证”这条路径走一遍示例来自项目内置样例 node_as_tool环境要求为 Python 3.10安装方式见根目录 READMEpip 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(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], )工具配置是从节点属性派生的见 Node as tool 指南属性来源作用工具名node.name呈现给模型的函数调用标识。描述node.description或 docstring描述工具用途的 prompt 上下文。参数node.input_schema或函数签名模型函数调用参数的 JSON Schema。把 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( nameorder_assistant, instructionHelp users check their order status., tools[check_order], )注意不要直接把会话型 Agent 塞进tools框架禁止把BaseAgent包装成工具因为对话型 Agent 有自己的轮次与消息历史语义应通过sub_agents委托。在工具节点中暂停并恢复人工介入节点作为工具时可以yieldRequestInput这类交互控制流事件。由于跨轮次暂停和恢复需要 Runner 保存并恢复会话状态Agent 要包在配置了ResumabilityConfig(is_resumableTrue)的App里并且节点要带node(rerun_on_resumeTrue)这样恢复时节点会重新执行到暂停点而不是直接以恢复输入作为输出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), )这套机制的底层过程在 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 时会过滤掉子分支事件只保留工具返回的最终输出人工暂停事件则可以穿透子分支向上浮出给调用方。运行与验证如果模块内定义了模块级变量appadk 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(namenode_as_tool, resumability_configResumabilityConfig(is_resumableTrue))包装。按样例 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_discountfc-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_nameapp.name, user_iduser ) runner Runner(appapp, session_servicesession_service) async for event in runner.run_async( user_iduser, session_idsession.id, new_messagetypes.Content( roleuser, parts[types.Part(textWhat 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的RequestInputADK 负责恢复时的解析但校验用户输入是否符合 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 ADK-Python 如何把 Workflow 和 @node 节点直接作为 Agent 工具暴露并保留人工介入恢复
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-PythonAgent Development Kit中父 Agent 经常需要把多步的确定性流程、数据处理管线或带审批步骤的环节委托出去。LlmAgent会自动把传入其tools列表的Workflow或BaseNode实例包装成工具父模型的函数调用即可触发这些执行单元如果节点在执行中yield一个RequestInput事件请求人工输入整个调用可以暂停并在用户下一轮回复后恢复。本文按“定义工具 → 声明人工介入点 → 配置可恢复的 App → 运行验证”这条路径走一遍示例来自项目内置样例 node_as_tool环境要求为 Python 3.10安装方式见根目录 READMEpip 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(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], )工具配置是从节点属性派生的见 Node as tool 指南属性来源作用工具名node.name呈现给模型的函数调用标识。描述node.description或 docstring描述工具用途的 prompt 上下文。参数node.input_schema或函数签名模型函数调用参数的 JSON Schema。把 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( nameorder_assistant, instructionHelp users check their order status., tools[check_order], )注意不要直接把会话型 Agent 塞进tools框架禁止把BaseAgent包装成工具因为对话型 Agent 有自己的轮次与消息历史语义应通过sub_agents委托。在工具节点中暂停并恢复人工介入节点作为工具时可以yieldRequestInput这类交互控制流事件。由于跨轮次暂停和恢复需要 Runner 保存并恢复会话状态Agent 要包在配置了ResumabilityConfig(is_resumableTrue)的App里并且节点要带node(rerun_on_resumeTrue)这样恢复时节点会重新执行到暂停点而不是直接以恢复输入作为输出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), )这套机制的底层过程在 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 时会过滤掉子分支事件只保留工具返回的最终输出人工暂停事件则可以穿透子分支向上浮出给调用方。运行与验证如果模块内定义了模块级变量appadk 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(namenode_as_tool, resumability_configResumabilityConfig(is_resumableTrue))包装。按样例 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_discountfc-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_nameapp.name, user_iduser ) runner Runner(appapp, session_servicesession_service) async for event in runner.run_async( user_iduser, session_idsession.id, new_messagetypes.Content( roleuser, parts[types.Part(textWhat 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的RequestInputADK 负责恢复时的解析但校验用户输入是否符合 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考