ADK-Python 长时运行工具实战：用 LongRunningFunctionTool 构建异步数据导出 Agent

ADK-Python 长时运行工具实战用 LongRunningFunctionTool 构建异步数据导出 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 仓库中的long_running_functions示例为核心系统讲解如何用LongRunningFunctionTool为 Agent 挂载长时运行工具当函数需要启动后台任务、等待人工审批或结果无法立即返回时Agent 可以先向用户回复任务已开始再通过后续的 FunctionResponse 异步回传进度与最终结果。读完本文你将掌握长时工具的定义方式、Agent 注册方法、进度回传机制及其底层源码原理并能在 ADK Web UI 中完整跑通一次数据导出交互。什么是 Long Running Functions在标准的 Agent 工具调用中模型发起functionCall后框架会同步执行函数并把返回值作为functionResponse立即回传给模型模型随即基于结果继续推理。但当工具对应的是耗时操作如导出大批量数据、提交后台批处理任务、等待人工审批时同步等待既不现实也会阻塞 Agent 的响应节奏。adk-python 提供的LongRunningFunctionTool正是为这类场景设计的。当工具被标记为 long-running 后框架会理解函数返回的可能只是一个进行中pending状态最终结果将在稍后以异步方式提供。这样 Agent 就能立刻告知用户任务已启动需要一些时间而不用在工具内部阻塞等待。该能力适用于以下典型场景启动后台任务函数内部仅触发一个后台 job 并立即返回任务完成后由其他渠道回传结果请求人工审批工具先返回等待审批状态审批完成后通过后续响应继续推进任何结果无法立即获取的操作例如跨服务的数据导出、长查询、文件生成等。官方示例仓库中对应的示例输入包括Export my data to CSVStart a JSON data exportExport my data to both CSV and JSON simultaneously三步创建长时运行工具参照 long_running_functions 示例目录 中的 agent.py创建一个长时工具只需三步定义 Python 函数函数返回一个表示进行中状态的字典例如{status: in-progress}用LongRunningFunctionTool包装函数将包装后的工具传入Agent的tools参数。完整代码如下from google.adk import Agent from google.adk.tools.long_running_tool import LongRunningFunctionTool def export_data(export_type: str) - dict[str, str]: Exports user data. Args: export_type: The type of data to export (e.g., csv, json). Returns: A dict with the status. # In a real application, this would kick off a background job. # Here we just return a status. return { status: in-progress, progress: 0%, message: fExporting {export_type} data. This may take some time., } root_agent Agent( namelong_running_functions, instruction You are an assistant that can export user data. When the user asks to export data, call the export_data tool. , tools[LongRunningFunctionTool(funcexport_data)], )示例仓库中的 agent.py 在README.md简版代码的基础上额外补充了instruction提示词明确告诉模型当用户要求导出数据时调用export_data工具这能显著提升工具触发的准确率是实际项目中值得沿用的做法。几点关键设计说明返回字典即状态协议函数返回的 dict 是框架与模型约定的状态载体status字段用于标记in-progress进行中或completed已完成progress用于承载进度百分比message用于携带给人看的说明文字函数签名决定声明LongRunningFunctionTool继承自FunctionTool见 function_tool.py会通过CallableSpec内省函数签名自动生成 Function Declaration因此export_type: str这样的类型注解会成为模型可理解的标准参数参数必填校验FunctionTool.run_async在调用前会检查必填参数无默认值的参数见 function_tool.py若模型漏传参数会返回包含error的 dict 提示模型补全后重试长时工具同样继承这一行为。一次完整的长时交互流程README 用时序图描述了单次数据导出的交互过程原文为 Mermaid以下为等价流程示意对照仓库中的测试事件流 export_to_csv.json可以把完整流程展开为六个阶段用户发起请求用户发送export to csv模型发起工具调用模型事件事件e-2中携带functionCall参数为{export_type: csv}finishReason为STOP注意该事件带上了longRunningToolIds: [fc-1]字段标记这是一个先返回、后回传的长时调用工具返回进行中状态随后事件e-3以functionResponse形式回传{status: in-progress, progress: 0%, message: Exporting csv data. This may take some time.}Agent 向用户播报模型据此生成回复 Im exporting your data to CSV. This may take some time. I will let you know when its done.异步回传进度用户侧或后台任务侧再次以functionResponse回传{progress: 50%, status: in-progress}事件e-5这会触发新的一轮模型推理Agent 汇报进度模型基于 50% 进度生成新回复 Your data is 50% exported. The export is still in progress.。这一机制的核心价值在于每次追加的 FunctionResponse 都会触发一次新的模型 turn从而使 Agent 能把最新的进度状态持续同步给用户而不是发出任务后就沉默。底层原理LongRunningFunctionTool 源码剖析要理解这个机制为什么能生效需要看 long_running_tool.py 的实现它非常精炼class LongRunningFunctionTool(FunctionTool): def __init__(self, func: Callable[..., Any]): super().__init__(func) self.is_long_running True override def _get_declaration(self) - Optional[types.FunctionDeclaration]: declaration super()._get_declaration() if declaration: instruction ( \n\nNOTE: This is a long-running operation. Do not call this tool again if it has already returned some intermediate or pending status. ) if declaration.description: declaration.description instruction else: declaration.description instruction.lstrip() return declaration关键点有三is_long_running标记LongRunningFunctionTool在初始化时把is_long_running置为True基类 base_tool.py 中默认值为False。框架各环节如 llm_flows 的 _tool_caller.py会依据该标记走不同的调用与回传路径自动注入模型指令_get_declaration在生成 Function Declaration 时会向工具描述末尾追加一行NOTE: This is a long-running operation. Do not call this tool again if it has already returned some intermediate or pending status.。这行提示直接进入模型的提示上下文有效防止模型在拿到 in-progress 状态后重复调用同一个工具完整的 FunctionTool 能力继承参数内省、必填参数校验、tool_context自动注入、异步/同步函数调度_invoke_callable会自动识别 coroutine 函数见 function_tool.py等能力全部继承自父类长时工具与普通函数工具在使用方式上完全一致。从源码结构看is_long_running标记与event.long_running_tool_ids见 export_to_csv.json 中事件e-2的字段协同工作框架通过longRunningToolIds记录哪些 function call 尚未终结后续以相同function_call_id追加的 FunctionResponse 会被视为该调用的进度更新或最终结果而不是一个新任务。这也解释了为什么长时工具的每个响应都必须携带原始调用对应的id。进度更新与多工具并行从测试事件流看边界行为长时工具的能力边界在测试事件流中体现得非常清楚除了单次导出仓库还提供了同时导出 CSV 与 JSON的测试 export_to_csv_and_json.json它展示了两个值得注意的行为1. 一次模型 turn 发起多个长时调用事件e-2中模型同时发出两个functionCallfc-1导出 csv、fc-2导出 jsonlongRunningToolIds同时记录两个 id框架对每个调用独立维护其状态2. 各调用独立异步终结事件e-7先回传fc-2的{status: completed}事件e-9再回传fc-1的{status: completed}。每次回传都触发一轮模型推理Agent 的回复也会随之更新如 The JSON export is complete, and the CSV export is 50% complete.直到全部调用终结后才给出最终总结 I have exported the data to both CSV and JSON formats.。这说明在实际业务中一个 Agent 可以同时挂起多个长时任务并通过携带对应function_call_id的 FunctionResponse 分别推进每个任务的进度互不干扰。在 ADK Web UI 中回传进度在 ADK Web UI 中你不需要写任何额外代码即可模拟后台任务回传这一环节将鼠标悬停在函数响应function response按钮上从菜单中选择 Send another response即可发送额外的函数响应例如携带{status: in-progress, progress: 50%}或最终的{status: completed}的响应。每发送一次模型都会基于最新状态生成一轮新的回复从而完整体验从任务启动到进度汇报再到任务完成的完整交互闭环。测试与验证仓库为长时工具提供了两层测试保障单元测试test_long_running_tool.py 覆盖了is_long_running属性、声明中警告指令的注入含无 docstring 的边界情况、run_async正常调用、对FunctionTool的继承关系等可直接运行验证端到端测试事件流export_to_csv.json 与 export_to_csv_and_json.json 以完整事件序列的方式固化了交互行为既可用于回归测试也是理解框架对longRunningToolIds处理方式的绝佳参考。此外如果你需要构造更多长时场景可以参考 function_tool.py 中关于require_confirmation、tool_context注入的说明——这些能力对长时工具同样适用可与异步回传机制组合使用构建如人工审批 后台执行的更复杂流程。小结通过LongRunningFunctionTooladk-python 把工具返回 pending 状态、结果异步回传这一常见业务需求变成了框架内建的一等公民你只需要把普通函数用LongRunningFunctionTool(func...)包装并注册到 Agent即可获得先响应、后回传的异步交互能力再配合 ADK Web UI 的 Send another response 操作与测试事件流就能快速验证并交付具备长时任务处理能力的生产级 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考