在本文主要讨论两个问题：
1、像Manus这样的多Agent系统，可以有哪些实现方式
2、多Agent系统中，如何实现信息的管理和共享

我在用LangGraph开发一个多Agent系统，LangGraph基于Langchain开发，是一个基于图的控制流框架，它可以更好的实现动态循环调用工作流。

LangChain的优点我们就不赘述，可以说是早期的完善的对开发者非常友好的框架。但LangGraph的控制流是静态的，它需要预先定义好所有可能的节点和边。而我们用的所有的AI产品都希望可以动态执行工作流，这就是我们的核心问题：
如何用LangGraph实现动态控制流？

OpenAI 目前尚未公开其应用产品中多智能体（multi-agent）系统的具体架构细节（例如 ChatGPT 内部是否使用多个 Agent、是否使用静态图 LangGraph 或动态控制流等），但我们可以从 已公开的系统设计模式、相关研究趋势，以及 OpenAI 生态中类似产品（如 Manuscripts、AutoGPT、LangGraph 等）来合理推测其可能的实现方式。以下是系统性回答：

多 Agent 系统的两种主流结构

A. 静态图结构（LangGraph、AutoGen）

核心机制：任务和 Agent 之间的流程是预定义好的（如图结构），控制流是**“可变但边界清晰”**的。

代表性框架：LangGraph、AutoGen、CrewAI

优势：可视化强、易于维护、debug-friendly。

限制：灵活性不足，不擅长处理复杂的、嵌套的动态决策链。

B. 动态执行控制（BOSS Agent / ReAct 式 Planner）

核心机制：一个主控 Agent（通常称为 BOSS、Manager、Planner）基于输入和上下文动态决定调用哪些工具或子 Agent。

代表性范式：ReAct、Toolformer、AutoGPT/Manuscripts 风格系统。

优势：灵活应对复杂任务，自我调整路径。

限制：调试成本高，稳定性依赖于 Prompt 工程和记忆能力。

Manus 的控制流推测（或类似系统）：基于 BOSS Agent 的动态任务调度

根据 Manus 等项目的描述，其内部系统大概率采用如下结构：

     ┌─────────────┐
     │ User Input  │
     └─────┬───────┘
           ↓
     ┌─────────────┐
     │  Boss Agent │  ← 推理/规划中心，基于上下文决定下一步
     └─────┬───────┘
           ↓
  ┌───────────────────┐
  │ Tool/Agent Routing│ ← 动态分配给专门 Agent（如Writer, Coder, Searcher等）
  └────┬──────────────┘
       ↓

┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Agent: Writer│ │ Agent: Coder │ │ Agent: Critic│ ← 子Agent，工具化
└──────────────┘ └──────────────┘ └──────────────┘

控制任务流的方式

Boss Agent 拥有上下文窗口与记忆能力，能观察输入 + 当前任务状态。

基于策略（Prompt 编程或微调），它会选择：

  • 调用哪个子 Agent；

  • 传什么内容；

  • 是否 loop 回来做 refinement；

  • 是否中止 / 输出最终结果。

Agent 之间可能通过一种「任务容器」交换信息，例如：

  • 中间 JSON 格式（task + result + metadata）

  • 或封装在 LLM 内部 Token 流中（隐式表示）

这类似于 OpenAI GPT 工具调用中使用 function_calling + tool_choice。

目前动态路由方案并不好用，我们可以使用其他类似的方式

方案1: 使用调度器节点模式，然后在工作流定义中使用条件路由。 方案2: 状态驱动执行计划

可能问题

  1. 可能会产生节点路由错误，为什么会产生这个错误？

StateGraph 的执行模型一次“执行”或“调度步”（step）内，StateGraph 会并发调用所有可执行的节点（Node）。

每个节点执行完毕后，都会返回一个字典，表示它要写入状态（state）里的哪些键（key）和对应的值（value）。

单值字段的保护策略

默认情况下，每个状态键（例如 mcp_results、messages、response）都被视为“单值”字段。

如果多个节点在同一个 step 返回了同一个键的不同值，StateGraph 无法判断应该采用哪一个，也不能安全地自动合并，就会认为这是并发更新冲突

这个错误本质上是 LangGraph（LangChain 的 StateGraph 引擎）为了保证状态的一致性，对同一个状态字段在同一步（step）里只允许一个节点写入而抛出的保护性警告。

“Use an Annotated key to handle multiple values”

这是多Agent在冲突时会报的错误，意味着你可以用 Python 的 typing.Annotated + Pydantic 的 Field(default_factory=…) 来声明这个字段是“可合并”的（通常是列表类型），让框架知道它可以接收来自多个节点的值，并以“追加”的方式合并它们。

什么时候会触发

并行拆分路径：当你在工作流图里，用多个并行分支（比如并行查询不同 API、并行调用不同检索 Agent）都输出到同一个字段。

广播型任务：多个 Agent 需要把自己的结果写到同一个“收集器”字段里，比如不同检索源的结果、不同专家 Agent 的反馈等。

为什么 Annotated 合并有效？

Pydantic 元信息：LangGraph 会读取 Pydantic 模型中 Field(default_factory=list) 的元信息，识别这是个“可合并”字段。

合并策略：当检测到多个返回值，它就知道要对列表进行拼接，而不是覆盖。

无需手动聚合：你不用在每个节点里先读旧值再写新值，框架自动帮你在 step 结束后统一合并。

2、 即便使用了Annotated，还是会报错，为什么？

我就遇到了这个问题，困扰了我很久，最终我找到了问题来源。StateGraph 会自动查找并行节点，即使你使用了Annotated，静态节点还是会导致他报错。

修改方式：在使用Annotated的前提下，修改LangGraph的节点逻辑，用循环来处理多个节点的排序。我想我已经讲的很清楚了。

1
2
3
4
5
6
7
8
9
10
state = TravelState(user_intent=..., messages=[...])
# 1) 让 BOSS 生成执行计划
boss_output = BossAgent().run_step(state)
state = state.copy(update=boss_output)    # state.execution_plan 里有 ["searcher","summarizer",...]

# 2) 按顺序执行
for agent_name in state.execution_plan:
    agent = agent_factory[agent_name]()
    update = agent.run_step(state)        # 只写它负责的字段
    state = state.copy(update=update)     # 立即合并结果

它的核心要点是：

-静态图并行 → 可能同时触发多个节点，导致多个写入冲突。

-流式（顺序）执行 → 每次只跑一个 Agent，写入立刻合并，天然无并发。

-BOSS Agent 负责“生成”顺序，把原本图上的并行分支转化成一条条可控的序列。

这样既保留了“BOSS 动态决策”能力，又避开了 LangGraph 静态并行下的写入冲突。