企业官网建设流程全解析

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

本文基于 ADK（Agent Development Kit）官方示例 workflow_triage，讲解一种实用的多智能体工作流模式：由一个“执行经理”智能体分析用户请求、动态挑选相关 Worker 智能体，再由SequentialAgent协调ParallelAgent并行执行，最后由汇总智能体根据实际激活的智能体动态生成总结。读完后，你能掌握动态智能体选择（Dynamic Agent Selection）、基于回调的相关性过滤（Relevance Filtering）以及跨智能体状态传递这三项 ADK 核心机制的落地写法，并将其迁移到自己的多域任务分流场景中。

一、这个示例解决什么问题

workflow_triage示例演示了如何构建一个智能分流（triage）工作流：用户输入一个请求，系统先判断这个请求与哪些专业智能体相关，只激活相关的智能体并行执行，最后把各智能体的输出汇总成一份总结。

整个工作流由三个主要组件构成：

1. 执行经理智能体（agent.py）——分析用户输入，决定哪些执行智能体是相关的；
2. 计划执行智能体（plan_execution_agent）——一个SequentialAgent，负责协调执行与汇总两个阶段；
3. Worker 执行智能体（execution_agent.py）——执行具体任务的专用智能体，可并行运行。

这种“先分诊、再执行、后汇总”的结构，本质上是把路由决策权交给 LLM（经理智能体）+ 把执行筛选交给确定性代码（回调）两条路径结合起来：经理智能体负责“软”的判断（用户到底想要什么），before_agent_callback负责“硬”的兜底（没被点名的智能体一律跳过），两者配合避免了单个大提示词里塞多个角色导致的指令漂移。

二、总体架构：四智能体分工

2.1 执行经理智能体execution_manager_agent（root_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 user's intent, you should ask for clarification first; * Only after you're clear about user's intent, you can proceed to step #3.

完整定义见 agent.py：root_agent是一个Agent（即 LLM 智能体），sub_agents中挂载plan_execution_agent，tools中注册update_execution_plan函数工具。当经理智能体完成任务后，ADK 会把控制权转移（transfer）给子智能体plan_execution_agent继续执行。

2.2 计划执行智能体plan_execution_agent

• 类型：SequentialAgent；
• 组成：
  • worker_parallel_agent（ParallelAgent）——并行运行相关的 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_contents="none"，不携带会话历史，专注于做总结。

三、关键机制一：用工具调用把“执行计划”写进状态

经理智能体判断出相关智能体后，并不是靠口头“通知”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，由经理提示词约束）；
• ToolContext：ADK 自动注入的工具上下文，通过tool_context.state[...] = ...写入的键会持久化到当前会话的 state 中，供后续任意智能体（包括后续轮次）读取；
• 约定键名：计划存在execution_agents，各 Worker 的产出存在{agent_name}_output（即code_agent_output/math_agent_output）。这一命名约定同时被回调和汇总智能体依赖，是整个模式的数据契约。

四、关键机制二：回调式相关性过滤（Relevance Filtering）

这是本示例最核心的技巧。execution_agent.py 用一个回调工厂为每个 Worker 生成专属的before_agent_callback：

def 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=( f"Skipping execution agent {agent_name} as it is not" " relevant to the current state." ) ) ] ) return callback

工厂参数agent_name通过闭包绑定到具体智能体，两个 Worker 各自挂载：

code_agent = Agent( name="code_agent", instruction="...", before_agent_callback=before_agent_callback_check_relevance("code_agent"), output_key="code_agent_output", ) math_agent = Agent( name="math_agent", instruction="...", before_agent_callback=before_agent_callback_check_relevance("math_agent"), output_key="math_agent_output", )

回调返回内容后的底层行为

回调的判断逻辑只有两行：当前智能体名不在state["execution_agents"]里，就返回一条“跳过”文本。这个“返回types.Content”的动作在框架层的含义，可以从 base_agent.py 的_handle_before_agent_callback得到印证：

if before_agent_callback_content: ret_event = Event( invocation_id=ctx.invocation_id, author=self.name, branch=ctx.branch, content=before_agent_callback_content, actions=callback_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 they're 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( name="worker_parallel_agent", sub_agents=[ code_agent, math_agent, ], )

从 parallel_agent.py 的源码结构看，ParallelAgent的行为有几个值得了解的特性：

1. 分支隔离：_run_async_impl会为每个子智能体调用_create_branch_ctx_for_sub_agent（parallel_agent.py）创建独立分支上下文。这意味着会话历史在各分支之间是隔离的——子智能体能看到分流发生前的事件与自己的事件，但看不到兄弟分支的事件；而会话 state 是所有分支共享的，因此execution_agents、{agent_name}_output这些键可以被各分支安全地读取与写入（本示例让两个 Worker 写不同的键，正是为了避免共享 state 下的键冲突）。
2. 事件交错合并：Python 3.11+ 上使用asyncio.TaskGroup把各分支的事件流合并到一个队列里按序吐出（_merge_agent_run，parallel_agent.py），Python 3.10 上则使用等价的自定义任务调度实现，两种实现保证事件按分支产出顺序被 Runner 消费。
3. 版本适用性提示：源码中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( name="execution_summary_agent", instruction=instruction_provider_for_execution_summary_agent, include_contents="none", )

这里体现了模式设计的三个细节：

• 只总结被激活的智能体：提示词基于state["execution_agents"]动态拼装，只列出实际参与本次调用的 Worker 及其state.get(f"{agent_name}_output", "")输出，未被激活的智能体不会出现在总结语境中（注意state.get带空字符串默认值，容忍个别输出缺失）；
• output_key闭环：每个 Worker 通过output_key参数把最终回答写入约定 state 键（如code_agent_output），汇总函数按同一命名约定读取，形成“写入—读取”闭环；
• include_contents="none"：汇总智能体不注入会话历史，只依赖动态指令里内联的各 Worker 输出做总结，上下文更干净、更聚焦。

最后，plan_execution_agent用SequentialAgent把两个阶段串起来：

plan_execution_agent = SequentialAgent( name="plan_execution_agent", sub_agents=[ worker_parallel_agent, execution_summary_agent, ], )

七、完整执行流程与示例交互

工作流遵循如下模式（对应 README “Usage” 一节）：

1. 用户向根智能体execution_manager_agent输入请求；
2. 经理智能体分析请求并识别相关智能体（code_agent、math_agent）；
3. 如果用户意图不清晰，经理智能体先请求澄清再往下走；
4. 经理智能体调用update_execution_plan更新执行计划（写入state["execution_agents"]）；
5. 控制权转移给plan_execution_agent；
6. worker_parallel_agent（ParallelAgent）根据更新后的计划，只运行相关的 Worker（不相关的 Worker 被回调跳过）；
7. execution_summary_agent对所有被激活智能体的结果进行汇总。

典型查询

模糊请求（触发澄清）：

> hi > Help me do this.

根智能体（execution_manager_agent）会先问候用户，并追问具体任务是什么，澄清前不会更新执行计划。

仅数学请求：

> What's 1+1?

只有math_agent执行，code_agent被回调跳过（事件流中会留下 “Skipping execution agent code_agent as it is not relevant to the current state.” 记录）。

跨域复合请求：

> What's 1+11? Write a python function to verify it.

code_agent与math_agent并行执行，随后进入汇总阶段。

八、可用执行智能体与扩展方式

当前示例注册了两个 Worker：

• code_agent—— 代码生成与编程任务；
• math_agent—— 数学计算与分析。

从源码结构看，扩展一个新 Worker 的步骤是固定的：在 execution_agent.py 中新增一个Agent，挂载before_agent_callback_check_relevance("<new_agent_name>")、指定output_key="<new_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_contents="none"实现聚焦式总结。

相关文件索引：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