ADK Workflow Triage 模式实战：用多智能体动态分流、并行执行并自动汇总结果

ADK Workflow Triage 模式实战用多智能体动态分流、并行执行并自动汇总结果【免费下载链接】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本文基于 ADKAgent Development Kit官方示例 workflow_triage讲解一种实用的多智能体工作流模式由一个“执行经理”智能体分析用户请求、动态挑选相关 Worker 智能体再由SequentialAgent协调ParallelAgent并行执行最后由汇总智能体根据实际激活的智能体动态生成总结。读完后你能掌握动态智能体选择Dynamic Agent Selection、基于回调的相关性过滤Relevance Filtering以及跨智能体状态传递这三项 ADK 核心机制的落地写法并将其迁移到自己的多域任务分流场景中。一、这个示例解决什么问题workflow_triage示例演示了如何构建一个智能分流triage工作流用户输入一个请求系统先判断这个请求与哪些专业智能体相关只激活相关的智能体并行执行最后把各智能体的输出汇总成一份总结。整个工作流由三个主要组件构成执行经理智能体agent.py——分析用户输入决定哪些执行智能体是相关的计划执行智能体plan_execution_agent——一个SequentialAgent负责协调执行与汇总两个阶段Worker 执行智能体execution_agent.py——执行具体任务的专用智能体可并行运行。这种“先分诊、再执行、后汇总”的结构本质上是把路由决策权交给 LLM经理智能体 把执行筛选交给确定性代码回调两条路径结合起来经理智能体负责“软”的判断用户到底想要什么before_agent_callback负责“硬”的兜底没被点名的智能体一律跳过两者配合避免了单个大提示词里塞多个角色导致的指令漂移。二、总体架构四智能体分工2.1 执行经理智能体execution_manager_agentroot_agent模型ADK 默认模型本示例中所有智能体都未显式设置model角色分析用户请求并更新执行计划工具update_execution_plan——决定应该激活哪些执行智能体子智能体把实际执行委托给plan_execution_agent澄清机制如果用户意图不清晰先向用户提问澄清再进入后续步骤。其提示词instruction明确了四条职责并给出了两条 NOTE 约束You are the Execution Manager Agent, responsible for setting up execution plan and delegate to plan_execution_agent for the actual plan execution. You ONLY have the following worker agents: code_agent, math_agent. You should do the following: 1. Analyze the user input and decide any worker agents that are relevant; 2. If none of the worker agents are relevant, you should explain to user that no relevant agents are available and ask for something else; 3. Update the execution plan with the relevant worker agents using update_execution_plan tool. 4. Transfer control to the plan_execution_agent for the actual plan execution. NOTE: * If you are not clear about users intent, you should ask for clarification first; * Only after youre clear about users intent, you can proceed to step #3.完整定义见 agent.pyroot_agent是一个Agent即 LLM 智能体sub_agents中挂载plan_execution_agenttools中注册update_execution_plan函数工具。当经理智能体完成任务后ADK 会把控制权转移transfer给子智能体plan_execution_agent继续执行。2.2 计划执行智能体plan_execution_agent类型SequentialAgent组成worker_parallel_agentParallelAgent——并行运行相关的 Worker 智能体execution_summary_agent——汇总执行结果。SequentialAgent保证了阶段顺序先并行执行再汇总二者不会交叉。2.3 Worker 智能体code_agent与math_agent系统包含两个并行运行的专业执行智能体代码智能体code_agent负责代码生成任务通过before_agent_callback_check_relevance回调在不相关时跳过执行输出写入 state 键code_agent_output数学智能体math_agent负责数学计算同样挂载before_agent_callback_check_relevance回调输出写入 state 键math_agent_output。两个 Worker 的提示词都刻意收窄了职责边界例如code_agent的 instruction 中写明 “You should only generate code and ignore other askings from the user.”防止并行分支互相抢答。2.4 执行汇总智能体execution_summary_agent模型ADK 默认模型同样未显式设置model角色汇总所有被激活智能体的输出动态指令根据本次实际激活了哪些智能体动态生成内容隔离include_contentsnone不携带会话历史专注于做总结。三、关键机制一用工具调用把“执行计划”写进状态经理智能体判断出相关智能体后并不是靠口头“通知”Worker而是通过一个函数工具把决定写入会话状态。agent.py 中的实现非常直接def update_execution_plan( execution_agents: list[str], tool_context: ToolContext ) - str: Updates the execution plan for the agents to run. tool_context.state[execution_agents] execution_agents return execution_agents updated.要点参数execution_agents: list[str]是要激活的智能体名列表取值只能来自code_agent、math_agent由经理提示词约束ToolContextADK 自动注入的工具上下文通过tool_context.state[...] ...写入的键会持久化到当前会话的 state 中供后续任意智能体包括后续轮次读取约定键名计划存在execution_agents各 Worker 的产出存在{agent_name}_output即code_agent_output/math_agent_output。这一命名约定同时被回调和汇总智能体依赖是整个模式的数据契约。四、关键机制二回调式相关性过滤Relevance Filtering这是本示例最核心的技巧。execution_agent.py 用一个回调工厂为每个 Worker 生成专属的before_agent_callbackdef before_agent_callback_check_relevance( agent_name: str, ) - BeforeAgentCallback: Callback to check if the state is relevant before executing the agent. def callback(callback_context: CallbackContext) - Optional[types.Content]: Check if the state is relevant. if agent_name not in callback_context.state[execution_agents]: return types.Content( parts[ types.Part( text( fSkipping execution agent {agent_name} as it is not relevant to the current state. ) ) ] ) return callback工厂参数agent_name通过闭包绑定到具体智能体两个 Worker 各自挂载code_agent Agent( namecode_agent, instruction..., before_agent_callbackbefore_agent_callback_check_relevance(code_agent), output_keycode_agent_output, ) math_agent Agent( namemath_agent, instruction..., before_agent_callbackbefore_agent_callback_check_relevance(math_agent), output_keymath_agent_output, )回调返回内容后的底层行为回调的判断逻辑只有两行当前智能体名不在state[execution_agents]里就返回一条“跳过”文本。这个“返回types.Content”的动作在框架层的含义可以从 base_agent.py 的_handle_before_agent_callback得到印证if before_agent_callback_content: ret_event Event( invocation_idctx.invocation_id, authorself.name, branchctx.branch, contentbefore_agent_callback_content, actionscallback_context._event_actions, ) ctx.end_invocation True return ret_event即当before_agent_callback返回了内容真值ADK 会直接以该内容生成一个事件并置位ctx.end_invocation True该智能体的 LLM 调用随即被跳过——不会发起模型请求、不消耗 token只在事件流中留下一条“Skipping execution agent xxx ...”的记录。这就是 README 所说 “Agents skip execution if theyre not relevant to the current state using callback mechanism” 的准确实现方式被跳过的 Worker 依然会产生一条可见事件便于调试与审计但不会执行任何模型逻辑。如果回调返回None智能体则按正常流程进入 LLM 调用。此外从该实现可以看到回调的触发顺序插件plugin先获得机会插件未提供覆盖内容时才执行智能体自身注册的canonical_before_agent_callbacks见 base_agent.py。五、关键机制三ParallelAgent并行执行与分支隔离两个 Worker 被挂载到一个ParallelAgent上worker_parallel_agent ParallelAgent( nameworker_parallel_agent, sub_agents[ code_agent, math_agent, ], )从 parallel_agent.py 的源码结构看ParallelAgent的行为有几个值得了解的特性分支隔离_run_async_impl会为每个子智能体调用_create_branch_ctx_for_sub_agentparallel_agent.py创建独立分支上下文。这意味着会话历史在各分支之间是隔离的——子智能体能看到分流发生前的事件与自己的事件但看不到兄弟分支的事件而会话 state 是所有分支共享的因此execution_agents、{agent_name}_output这些键可以被各分支安全地读取与写入本示例让两个 Worker 写不同的键正是为了避免共享 state 下的键冲突。事件交错合并Python 3.11 上使用asyncio.TaskGroup把各分支的事件流合并到一个队列里按序吐出_merge_agent_runparallel_agent.pyPython 3.10 上则使用等价的自定义任务调度实现两种实现保证事件按分支产出顺序被 Runner 消费。版本适用性提示源码中ParallelAgent带有deprecated标记说明其正被新的Workflow取代、将在未来版本移除且注明 “Workflow cannot yet be used as an LlmAgent sub-agent”parallel_agent.py。在当前仓库版本中ParallelAgent作为LlmAgent子智能体并行运行的方式仍然可用本示例即依赖此行为如果你在较新版本上迁移该模式需要留意该废弃说明对架构选型的影响。六、关键机制四动态指令驱动的汇总智能体execution_summary_agent没有写死提示词而是把instruction直接指向一个指令提供函数instruction provider。ADK 允许instruction为接收ReadonlyContext的函数在每次运行时动态求值。execution_agent.py 中的实现def instruction_provider_for_execution_summary_agent( readonly_context: ReadonlyContext, ) - str: Provides the instruction for the execution agent. activated_agents readonly_context.state[execution_agents] prompt f\ You are the Execution Summary Agent, responsible for summarizing the execution of the plan in the current invocation. In this invocation, the following agents were involved: {, .join(activated_agents)}. Below are their outputs: for agent_name in activated_agents: output readonly_context.state.get(f{agent_name}_output, ) prompt f\n\n{agent_name} output:\n{output} prompt ( \n\nPlease summarize the execution of the plan based on the above outputs. ) return prompt.strip() execution_summary_agent Agent( nameexecution_summary_agent, instructioninstruction_provider_for_execution_summary_agent, include_contentsnone, )这里体现了模式设计的三个细节只总结被激活的智能体提示词基于state[execution_agents]动态拼装只列出实际参与本次调用的 Worker 及其state.get(f{agent_name}_output, )输出未被激活的智能体不会出现在总结语境中注意state.get带空字符串默认值容忍个别输出缺失output_key闭环每个 Worker 通过output_key参数把最终回答写入约定 state 键如code_agent_output汇总函数按同一命名约定读取形成“写入—读取”闭环include_contentsnone汇总智能体不注入会话历史只依赖动态指令里内联的各 Worker 输出做总结上下文更干净、更聚焦。最后plan_execution_agent用SequentialAgent把两个阶段串起来plan_execution_agent SequentialAgent( nameplan_execution_agent, sub_agents[ worker_parallel_agent, execution_summary_agent, ], )七、完整执行流程与示例交互工作流遵循如下模式对应 README “Usage” 一节用户向根智能体execution_manager_agent输入请求经理智能体分析请求并识别相关智能体code_agent、math_agent如果用户意图不清晰经理智能体先请求澄清再往下走经理智能体调用update_execution_plan更新执行计划写入state[execution_agents]控制权转移给plan_execution_agentworker_parallel_agentParallelAgent根据更新后的计划只运行相关的 Worker不相关的 Worker 被回调跳过execution_summary_agent对所有被激活智能体的结果进行汇总。典型查询模糊请求触发澄清 hi Help me do this.根智能体execution_manager_agent会先问候用户并追问具体任务是什么澄清前不会更新执行计划。仅数学请求 Whats 11?只有math_agent执行code_agent被回调跳过事件流中会留下 “Skipping execution agent code_agent as it is not relevant to the current state.” 记录。跨域复合请求 Whats 111? Write a python function to verify it.code_agent与math_agent并行执行随后进入汇总阶段。八、可用执行智能体与扩展方式当前示例注册了两个 Workercode_agent—— 代码生成与编程任务math_agent—— 数学计算与分析。从源码结构看扩展一个新 Worker 的步骤是固定的在 execution_agent.py 中新增一个Agent挂载before_agent_callback_check_relevance(new_agent_name)、指定output_keynew_agent_name_output加入worker_parallel_agent的sub_agents同时在 agent.py 经理智能体的 instruction 中把新智能体名补进 “You ONLY have the following worker agents” 列表。汇总逻辑无需改动——指令提供函数会自动按execution_agents列表收集新 Worker 的输出。九、实现细节小结基于 Google ADK 智能体框架构建通过before_agent_callback_check_relevance实现基于回调的相关性检查未激活 Worker 以“返回内容 结束本次调用”的方式被跳过见 base_agent.py通过ToolContext与 state 键execution_agents、{agent_name}_output维持跨智能体状态使用ParallelAgent支持并行智能体执行分支历史隔离、state 共享见 parallel_agent.py使用SequentialAgent保证“并行执行 → 汇总”的协调顺序汇总智能体的指令基于被激活智能体动态生成配合include_contentsnone实现聚焦式总结。相关文件索引workflow_triage/README.md、workflow_triage/agent.py、workflow_triage/execution_agent.py、agents/base_agent.py、agents/parallel_agent.py。【免费下载链接】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），仅供参考