ADK-Python 长时运行工具与人工审批使用 FunctionResponse 回传实现 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-python 仓库中的contributing/samples/hitl/human_in_loop示例为蓝本系统讲解如何在 Google ADKAgent Development Kit中实现「长时运行工具Long-Running Tool」当工具无法立即给出最终结果例如需要等待人工审批时Agent 如何先收到一个pending状态的初始响应再由应用侧在外部流程完成后用携带相同id与name的types.FunctionResponse以roleuser消息回传驱动 Agent 继续执行后续动作如调用报销工具。读完本文你将掌握 ADK 长时运行工具从定义、调用、监听事件到回传结果的完整闭环并理解其在报销审批这类 Human-in-the-Loop 场景中的落地方式。示例概览报销审批 Agent示例位于 contributing/samples/hitl/human_in_loop目录结构如下agent.py定义报销审批 Agent 及其工具main.py演示完整的「调用 → 捕获 pending → 模拟外部审批 → 回传结果」流程tests/reimburse_dinner.json 与 tests/auto_reimburse_coffee.jsonAgent 测试脚本用的会话事件序列。业务场景是员工申请报销若金额低于 100 美元Agent 自动调用reimburse()若金额超过 100 美元Agent 调用ask_for_approval()等待经理审批审批通过后再执行报销拒绝则告知员工。定义两个工具agent.py 中定义了两个普通 Python 函数def reimburse(purpose: str, amount: float) - dict[str, str]: Reimburse the amount of money to the employee. return { status: ok, } def ask_for_approval( purpose: str, amount: float, tool_context: ToolContext ) - dict[str, Any]: Ask for approval for the reimbursement. return { status: pending, amount: amount, ticketId: reimbursement-ticket-001, }ask_for_approval是长时运行工具它立即返回pending状态并附上ticketId用于后续追踪审批请求。tool_context: ToolContext参数由框架注入ToolContext中携带了本次函数调用的function_call_id等上下文信息。把工具挂到 Agent 上root_agent Agent( namereimbursement_agent, instruction..., tools[reimburse, LongRunningFunctionTool(funcask_for_approval)], generate_content_configtypes.GenerateContentConfig(temperature0.1), )关键点是ask_for_approval必须用 LongRunningFunctionTool 包装而普通工具reimburse直接传入即可。长时运行工具的底层原理LongRunningFunctionTool 继承自FunctionTool构造时仅多做一件事def __init__(self, func: Callable[..., Any]): super().__init__(func) self.is_long_running Trueis_long_running True是框架识别长时运行工具的标志。此外它覆写了_get_declaration()会在模型的函数声明描述末尾追加一段提示NOTE: This is a long-running operation. Do not call this tool again if it has already returned some intermediate or pending status.这段提示用于引导模型对于已经返回pending状态的调用不要重复调用同一工具。从源码结构看ADK 中is_long_running True还被用于其它需要外部回传的场景例如 get_user_choice_tool.py 中的get_user_choice_tool和 _request_input_tool.py 中的request_input它们同样通过LongRunningFunctionTool包装实现「等待用户输入」的能力。事件如何标记长时调用当模型发出一个指向长时运行工具的函数调用时该调用会以 FunctionCall 事件的形式出现在事件流中并且事件上带有long_running_tool_ids字段。Event 模型 对该字段的注释说明了它的用途long_running_tool_ids: set[str] | None None Set of ids of the long running function calls. Agent client will know from this field about which function call is long running. only valid for function call event 客户端也就是你的应用代码通过判断某个FunctionCall的id是否出现在event.long_running_tool_ids中即可识别出它是长时运行调用。这一点在 main.py 中有直接体现if not long_running_function_call and part.function_call.id in ( event.long_running_tool_ids or [] ): long_running_function_call part.function_call在 tests/reimburse_dinner.json 的测试事件中也能看到同样的结构——模型事件e-2携带longRunningToolIds: [fc-1]序列化时采用 camelCase 别名随后e-3就是携带 pending 数据的 FunctionResponse 事件。长时运行工具的关键流程原示例文档将完整流程归纳为 6 个步骤每一步都不可或缺步骤 1初始调用Agent 在推理中决定调用长时运行工具例如ask_for_approval模型侧产生一个types.FunctionCall其中包含工具名name与参数args。步骤 2初始工具响应工具立即返回初始响应典型内容是pending状态加一个追踪标识如ticketId。框架会将其包装为types.FunctionResponse送回给 Agent供其进入下一轮推理。此时应用侧无需等待外部任务完成。步骤 3Agent 确认状态Agent 处理这个初始响应通常会向用户说明任务处于 pending 状态。测试事件e-4中 Agent 的回复就是典型例子Your reimbursement request for $150 for a client dinner has been sent for approval. Your ticket ID is reimbursement-ticket-001. I will let you know once I have an update.步骤 4外部流程推进长时任务在 Agent 之外的世界推进——例如经理在审批系统中点击了「同意」。步骤 5核心回传更新的工具响应外部流程完成后你的应用必须构造一个新的types.FunctionResponse并遵守以下三个要求使用与原始FunctionCall相同的id和name让 Agent 能将该响应与之前的调用对应起来response字段放入更新后的数据例如{status: approved, ...}以roleuser的新消息将包含该响应的Part发送回 Agent。示例 main.py 中的实现updated_tool_output_data { status: approved, ticketId: ticket_id, approver_feedback: ( Approved by manager at str(asyncio.get_event_loop().time()) ), } updated_function_response_part types.Part( function_responsetypes.FunctionResponse( idlong_running_function_call.id, # 原始调用的 ID namelong_running_function_call.name, # 原始调用的名称 responseupdated_tool_output_data, ) ) async for _ in runner.run_async( session_idsession.id, user_idUSER_ID, new_messagetypes.Content( parts[updated_function_response_part], roleuser ), ): pass # 消费生成器或处理事件步骤 6Agent 基于更新继续行动Agent 收到这条携带types.FunctionResponse的消息后根据指令推进下一步审批通过则调用reimburse()拒绝则告知员工。为什么要回传这个 FunctionResponse这是整个模式中最容易忽略、也最关键的一点Agent 依赖这条以roleuser消息形式回传的types.FunctionResponse来感知长时任务的结论或状态变化。如果不回传Agent 将永远不知道 pending 任务的结果无法继续后续动作。示例中main.py的第二轮runner.run_async正是把更新后的 FunctionResponse 当作新的user消息输入驱动 Agent 继续执行审批通过后的reimburse()流程。从 main.py 的监听逻辑可以看到应用侧通过事件流识别长时调用与初始响应遍历事件中的parts当part.function_call的id命中event.long_running_tool_ids时记录为long_running_function_call当part.function_response.id与该调用 id 一致时记录为initial_tool_response并从中取出ticketId当初始响应状态为pending时进入外部审批模拟与结果回传分支。运行示例示例的 main.py 依次执行两个查询来演示两种分支await call_agent(Please reimburse $50 for meals) # 低于 $100自动报销 print( * 70) await call_agent(Please reimburse $200 for conference travel) # 超过 $100走审批运行前需设置环境变量GOOGLE_CLOUD_PROJECT用于 Cloud Trace 导出因为 main.py 在入口处初始化了 OpenTelemetry 的TracerProvider与CloudTraceSpanExporter将运行轨迹导出到指定的 GCP 项目同时通过load_dotenv(overrideTrue)从.env文件加载模型凭据等配置。运行需要配置好 Google 生成式 AI 的访问凭据并安装google-adk及示例所需的dotenv、OpenTelemetry 相关依赖。测试事件文件reimburse_dinner.json 记录了一次 150 美元报销的完整事件序列正好对应「长时审批」分支事件author内容说明e-1user文本申请 $150 客户晚餐报销用户消息e-2reimbursement_agentFunctionCallask_for_approvallongRunningToolIds: [fc-1]模型发起长时调用e-3reimbursement_agentFunctionResponseidfc-1statuspending、ticketIdreimbursement-ticket-001初始 pending 响应e-4reimbursement_agent文本告知已提交审批与 ticket IDAgent 向用户确认状态而 auto_reimburse_coffee.json 则对应金额低于 100 美元时直接自动报销的分支两个文件放在一起可以对比「需要审批」与「自动通过」两种路径在事件流上的差异。这类 JSON 事件脚本可直接用于 ADK 的 Agent 测试工具如adk test验证 Agent 行为是否符合预期。实践要点总结用LongRunningFunctionTool包装任何需要外部异步确认的工具都必须用LongRunningFunctionTool包装框架才能将其标记为长时运行is_long_running True并在函数声明中提示模型不要重复调用。初始响应必须包含状态与追踪标识立即返回pending状态与ticketId让 Agent 先向用户交代清楚也让应用侧有依据关联后续结果。回传时 id/name 必须一致更新后的types.FunctionResponse必须复用原始FunctionCall的id与name这是 Agent 将结果与调用关联起来的唯一凭据。回传消息 role 必须是user携带function_responsePart 的types.Content必须以roleuser发送通过runner.run_async(..., new_message...)开启新一轮运行。借助event.long_running_tool_ids识别长时调用在事件流中通过该字段判断哪些 FunctionCall 是长时运行调用从而只对这些调用等待后续回传。该模式是 ADK 中实现 Human-in-the-LoopHITL的核心手段之一把「等待人类审批」建模为一次长时运行工具调用既保持了 Agent 会话的连续性又允许外部系统在任何时间点完成审批后无缝地把结果交还给 Agent 继续执行。更多相关能力可参考仓库中的 request_input 示例 与 tool_confirmation 示例。【免费下载链接】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），仅供参考
