企业官网建设流程全解析

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 ADK（Agent Development Kit）中实现「长时运行工具（Long-Running Tool）」：当工具无法立即给出最终结果（例如需要等待人工审批）时，Agent 如何先收到一个pending状态的初始响应，再由应用侧在外部流程完成后，用携带相同id与name的types.FunctionResponse以role="user"消息回传，驱动 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.json：Agent 测试脚本用的会话事件序列。

业务场景是：员工申请报销，若金额低于 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( name='reimbursement_agent', instruction="""...""", tools=[reimburse, LongRunningFunctionTool(func=ask_for_approval)], generate_content_config=types.GenerateContentConfig(temperature=0.1), )

关键点是ask_for_approval必须用 LongRunningFunctionTool 包装，而普通工具reimburse直接传入即可。

长时运行工具的底层原理

LongRunningFunctionTool 继承自FunctionTool，构造时仅多做一件事：

def __init__(self, func: Callable[..., Any]): super().__init__(func) self.is_long_running = True

is_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，供其进入下一轮推理。此时应用侧无需等待外部任务完成。

步骤 3：Agent 确认状态

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，并遵守以下三个要求：

1. 使用与原始FunctionCall相同的id和name，让 Agent 能将该响应与之前的调用对应起来；
2. response字段放入更新后的数据（例如{'status': 'approved', ...}）；
3. 以role="user"的新消息将包含该响应的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_response=types.FunctionResponse( id=long_running_function_call.id, # 原始调用的 ID name=long_running_function_call.name, # 原始调用的名称 response=updated_tool_output_data, ) ) async for _ in runner.run_async( session_id=session.id, user_id=USER_ID, new_message=types.Content( parts=[updated_function_response_part], role="user" ), ): pass # 消费生成器（或处理事件）

步骤 6：Agent 基于更新继续行动

Agent 收到这条携带types.FunctionResponse的消息后，根据指令推进下一步：审批通过则调用reimburse()，拒绝则告知员工。

为什么要回传这个 FunctionResponse

这是整个模式中最容易忽略、也最关键的一点：Agent 依赖这条以role="user"消息形式回传的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(override=True)从.env文件加载模型凭据等配置。运行需要配置好 Google 生成式 AI 的访问凭据，并安装google-adk及示例所需的dotenv、OpenTelemetry 相关依赖。

测试事件文件

reimburse_dinner.json 记录了一次 150 美元报销的完整事件序列，正好对应「长时审批」分支：

而 auto_reimburse_coffee.json 则对应金额低于 100 美元时直接自动报销的分支，两个文件放在一起可以对比「需要审批」与「自动通过」两种路径在事件流上的差异。这类 JSON 事件脚本可直接用于 ADK 的 Agent 测试工具（如adk test），验证 Agent 行为是否符合预期。

实践要点总结

1. 用LongRunningFunctionTool包装：任何需要外部异步确认的工具都必须用LongRunningFunctionTool包装，框架才能将其标记为长时运行（is_long_running = True），并在函数声明中提示模型不要重复调用。
2. 初始响应必须包含状态与追踪标识：立即返回pending状态与ticketId，让 Agent 先向用户交代清楚，也让应用侧有依据关联后续结果。
3. 回传时 id/name 必须一致：更新后的types.FunctionResponse必须复用原始FunctionCall的id与name，这是 Agent 将结果与调用关联起来的唯一凭据。
4. 回传消息 role 必须是user：携带function_responsePart 的types.Content必须以role="user"发送，通过runner.run_async(..., new_message=...)开启新一轮运行。
5. 借助event.long_running_tool_ids识别长时调用：在事件流中通过该字段判断哪些 FunctionCall 是长时运行调用，从而只对这些调用等待后续回传。

该模式是 ADK 中实现 Human-in-the-Loop（HITL）的核心手段之一：把「等待人类审批」建模为一次长时运行工具调用，既保持了 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