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 推理和工具调用解耦实现了并行执行和异步调度。核心设计包括任务依赖图的构建与拓扑排序、并行执行的并发控制、以及失败任务的隔离重试。同步模式适合强依赖的任务链异步模式适合可独立执行的批量任务。在实际工程中根据任务依赖图的形状动态选择调度策略——可并行的步骤异步执行有依赖的步骤串行执行——是最优的工程方案。

Agent 观察与执行分离：思考不该等工具返回才继续

相关推荐

文本去重的工程方法：MinHash 到 SimHash，在精度与速度间选择

Git 标签实战：从 v1.0 创建到 origin 推送的 3 步完整流程

Git 标签 vs 分支：3 个核心差异与 4 种典型应用场景解析

更多资讯

东莞光明复审电工证怎么考？学历不高也能过！考前押题帮你上岸

初中应届考电工证多久拿证？刚毕业想找工作的你必须知道！

永兴考电工证在哪良心建议：别被中介坑了，自己搞懂流程更靠谱

襄汾电工证培训：上岗必备，先考证再找工作

卢氏哪里能办电工证多久拿证？急需特种作业证才能上岗找工作

工地太忙根本没时间复习考试？北京低压电工证刷题千万别踩坑

今日更新

河南特种工八大员含金量高吗？别被中介坑了，这些关键点你要知道

电工证ps软件多少钱？工地太忙根本没时间复习考试怎么办

牡丹江报考高压电工证在哪考？新手避坑指南

本周精选

安监局电工证分数查询怎么查才能知道含金量高不高

新疆电工作业干活累吗良心建议助你避开考不过的坑

中牟县电工证怎么报名外省证能转吗

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？