Agent 观察与执行分离：思考不该等工具返回才继续 Agent 观察与执行分离思考不该等工具返回才继续一、同步等待工具返回是 Agent 效率的隐形杀手当前主流的 Agent 框架LangChain、AutoGPT 等大多采用同步交互模式Agent 发出工具调用请求等待工具返回结果再根据结果进行下一轮思考。这符合直觉——先拿到结果再决定下一步。但在工程实践中这种串行等待模式造成显著的效率损失。考虑一个典型场景Agent 需要同时查询三个不同数据源的 API 来回答问题。同步模式下每个 API 调用耗时 12 秒串行执行就需要 36 秒。但这个等待过程中Agent 完全可以并行发出多个工具调用并在思考如何组合这些结果。更深层的问题是思考与执行不应是同一线程的串行过程。人类在等待外卖时不会干坐着会同时做其他事情。Agent 也是一样——模型在等待工具返回的间隙可以提前规划下一步的策略或者重新审视当前任务是否偏离了用户意图。见证奇迹的时刻是当把观察Observation和执行Action解耦到不同的异步通道后一个原本需要 15 轮同步交互的任务可以压缩到 5 轮异步调度内完成。二、观察与执行分离的架构设计分离的核心思想是将 Agent 的决策循环拆分为独立的思考者Thinker和执行者Executor两者通过消息队列异步通信。sequenceDiagram participant User as 用户 participant Thinker as 思考者br/(LLM Core) participant Queue as 消息队列br/(Task Bus) participant Executor as 执行者br/(Tool Runner) participant Env as 外部环境br/(APIs/Tools) User-Thinker: 发起任务 Thinker-Thinker: 初步策略分析 Thinker-Queue: 发布并行任务 [T1, T2, T3] Note over Thinker,Queue: 思考者不等待继续规划 Queue-Executor: 分发 T1 Queue-Executor: 分发 T2 Queue-Executor: 分发 T3 par 并行执行 Executor-Env: 调用 API 1 Env--Executor: 返回结果 R1 and Executor-Env: 调用 API 2 Env--Executor: 返回结果 R2 and Executor-Env: 调用 API 3 Env--Executor: 返回结果 R3 end Executor-Queue: 聚合结果 [R1, R2, R3] Queue-Thinker: 推送聚合结果 Thinker-Thinker: 基于完整结果推理 Thinker-User: 返回最终回答这个架构中思考者不直接与外部环境交互。它负责两件事解析用户意图制定执行计划收到聚合结果后进行整合推理。执行者负责实际调用工具、处理异常、聚合多个工具的返回结果。消息队列充当异步缓冲层支持任务优先级、超时重试和并发控制。关键的工程考量思考者发出的工具调用描述中需要声明前置条件和可并行度。如果工具 B 的输出依赖于工具 A 的结果则标注为串行依赖如果没有依赖关系则可以进行并行调度。三、异步调度器的核心实现以下代码展示了观察-执行分离框架中调度器的实现。import asyncio from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional, Set from enum import Enum class TaskStatus(Enum): PENDING pending RUNNING running COMPLETED completed FAILED failed dataclass class ToolTask: 工具执行任务 设计原因每个任务包含工具定义、参数和依赖关系 调度器根据此信息决定执行顺序和并行度。 task_id: str tool_name: str params: Dict[str, Any] depends_on: List[str] field(default_factorylist) status: TaskStatus TaskStatus.PENDING result: Optional[Any] None error: Optional[str] None class AsyncScheduler: 异步任务调度器 设计原因根据任务间的依赖关系构建 DAG 无依赖的任务并行执行有依赖的等前置完成后执行。 def __init__(self, executor: Callable, max_concurrency: int 5): self.executor executor # 工具执行函数 self.max_concurrency max_concurrency self.semaphore asyncio.Semaphore(max_concurrency) def _build_dag(self, tasks: List[ToolTask]) - Dict[str, Set[str]]: 构建任务的依赖图 设计原因使用邻接表表示 DAG用于拓扑排序和并行调度。 依赖关系由思考者在制定计划时显式声明。 dag: Dict[str, Set[str]] {} for task in tasks: dag[task.task_id] set(task.depends_on) return dag async def _execute_task(self, task: ToolTask, results: Dict[str, Any]) - None: 执行单个任务处理依赖注入 async with self.semaphore: # 控制并发数 task.status TaskStatus.RUNNING try: # 注入前置任务的输出作为参数 resolved_params {} for k, v in task.params.items(): if isinstance(v, str) and v.startswith($): dep_id v[1:] # 引用前置结果 resolved_params[k] results.get(dep_id) else: resolved_params[k] v task.result await self.executor( task.tool_name, resolved_params ) task.status TaskStatus.COMPLETED results[task.task_id] task.result except Exception as e: task.status TaskStatus.FAILED task.error str(e) async def schedule(self, tasks: List[ToolTask]) - Dict[str, Any]: 调度所有任务 设计原因使用多轮扫描调度每轮找出所有就绪的任务 所有依赖已完成并并行执行。循环直到全部完成或超时。 results: Dict[str, Any] {} completed: Set[str] set() failed: Set[str] set() all_tasks {t.task_id: t for t in tasks} dag self._build_dag(tasks) while len(completed) len(failed) len(tasks): ready [] for task_id, task in all_tasks.items(): if task_id in completed or task_id in failed: continue if dag.get(task_id, set()).issubset(completed): ready.append(task) if not ready: # 存在循环依赖或全部任务已处理 break await asyncio.gather(*[ self._execute_task(task, results) for task in ready ]) for task in ready: if task.status TaskStatus.COMPLETED: completed.add(task.task_id) elif task.status TaskStatus.FAILED: failed.add(task.task_id) return results上述调度器的核心逻辑是根据依赖关系构建 DAG每轮扫描找出所有就绪任务依赖已全部完成并行执行。max_concurrency参数控制最大并行数防止对外部 API 造成过大的压力。四、同步与异步 Agent 架构的权衡分析观察-执行分离不是银弹两者在不同的任务特征下有各自的优势。同步模式的适用场景工具调用间存在强依赖关系如必须先用搜索工具找到文档再根据文档内容提取答案工具调用数量少≤3总执行时间在可接受范围内需要精确控制每一步的决策因为每步决策都可能根据上一步结果调整异步模式的适用场景工具调用可以独立执行没有数据依赖工具调用数量多或单个调用耗时长需要批量处理同类任务维度同步模式异步模式实现复杂度低逻辑线性高需处理并发和状态管理延迟等于各步延迟之和近似最慢步骤的延迟容错性某步失败需从头重试可以单独重试失败任务资源消耗单线程占用少多任务并发资源占用高调试难度容易执行顺序确定较难执行顺序不确定见证奇迹的时刻不在于选择了哪种模式而在于能够根据任务依赖图的形状动态决定哪些步骤可以并行。批量信息收集应该异步并行相关性推理应该同步串行。混合调度是最佳实践。五、总结Agent 的观察与执行分离架构通过将 LLM 推理和工具调用解耦实现了并行执行和异步调度。核心设计包括任务依赖图的构建与拓扑排序、并行执行的并发控制、以及失败任务的隔离重试。同步模式适合强依赖的任务链异步模式适合可独立执行的批量任务。在实际工程中根据任务依赖图的形状动态选择调度策略——可并行的步骤异步执行有依赖的步骤串行执行——是最优的工程方案。