Researcher 执行摘要

任务：多Agent协作方案调研

搜索完成情况：
- 共完成 7 轮搜索，覆盖 35+ 来源
- 搜索方向：
1. LangGraph 多Agent架构与通信机制（官方博客、学术文章、文档）
2. AutoGen 任务分发与编排方式（Mixture of Agents、Group Chat、官方文档）
3. CrewAI Agent 协作模式（Processes、Flows、层级管理）
4. OpenAI Swarm / Agents SDK 设计（GitHub、官方博客、Cookbook）
5. 业界防止Agent卡死/超时的最佳实践（Galileo AI、LangGraph State Machines）
6. 任务状态追踪与恢复机制（Microsoft Agent Framework、AdaptOrch 论文）
7. OpenAI Agents SDK 补充（生产就绪特性、与 Swarm 对比）

关键发现：
- 主流框架通信机制：共享状态（LangGraph）、消息总线（AutoGen）、Handoff（Swarm/Agents SDK）
- 编排模式：Sequential、Hierarchical、Parallel、Hybrid
- 防卡死机制：Token 边界、超时、重试、断路器、Human-in-the-Loop
- 状态管理：Checkpointer（LangGraph）、Redis 持久化、Session 管理（Agents SDK）
- 业界数据：50% 错误率、30% 项目放弃率、72-86% token 重复浪费

原始搜索记录：已写入 search-log.md（7 个搜索轮次，每轮包含来源 URL、核心要点、关键数据）

下一步：移交给 reviewer 撰写调研报告

多Agent协作方案调研报告

1. 执行摘要

本次调研覆盖 LangGraph、AutoGen、CrewAI、OpenAI Swarm/Agents SDK 四大主流框架，分析了 35+ 来源（官方文档、学术论文、技术博客）。核心发现：

1. 通信机制：共享状态（LangGraph）、消息总线（AutoGen）、Handoff（Swarm/Agents SDK）三种主流模式，各有权衡
2. 编排模式：Sequential、Hierarchical、Parallel、Hybrid 四种拓扑，AdaptOrch 论文证明拓扑选择比模型选择更重要（性能影响超过 Ω(1/ε²) 倍）
3. 防卡死机制：Token 边界、超时、重试、断路器、Human-in-the-Loop 是业界标配，Galileo AI 数据显示 72-86% token 重复浪费
4. 状态管理：Checkpointer（LangGraph）、Redis 持久化、Session 管理（Agents SDK）提供中断恢复能力
5. 生产就绪度：OpenAI Agents SDK 是 Swarm 的演进版，新增 Tracing、Guardrails、Session 管理，适合生产环境

[reviewer] 章节已写完：执行摘要

2. 主流框架对比

2.1 LangGraph

定位：底层灵活的图编排框架，Agent 是图节点，边是状态转移

核心特性：
- 状态机模型：图状态（graph state）作为共享通信介质
- 三种协作模式：
- Multi-Agent Collaboration：共享 scratchpad，所有消息对所有 Agent 可见
- Agent Supervisor：各 Agent 独立 scratchpad，Supervisor 路由调度
- Hierarchical Agent Teams：支持嵌套图（子 Agent 本身也是 LangGraph 对象）
- Checkpointer 机制：每个 super-step 保存状态快照，支持中断/恢复、时间旅行、Human-in-the-Loop
- Thread 概念：唯一 thread_id 标识一系列运行的累积状态

适用场景：需要高度定制化控制流、复杂状态管理、嵌套层级结构的场景

权衡：灵活性高但学习曲线陡峭，需手动设计状态转移逻辑

2.2 AutoGen

定位：微软开源的多 Agent 对话框架，强调任务分发与编排

核心特性：
- Mixture of Agents 模式：模仿前馈神经网络，Worker 分多层，前一层输出拼接后发给下一层
- Group Chat 模式：一组 Agent 共享消息线程，Manager 维护轮次顺序（Round-robin 或 LLM-based selector）
- 消息协议：WorkerTask、WorkerTaskResult、UserTask、FinalResult 标准化消息类型
- 直接消息传递：send_message() API，便于添加任务取消、错误处理

适用场景：动态分解复杂任务、专业化 Agent 各司其职、需要层级汇总的场景

权衡：消息传递开销较大，需仔细设计消息协议防止乒乓传递

[reviewer] 章节已写完：主流框架对比（LangGraph、AutoGen）

2.3 CrewAI

定位：企业级 Agent 协作框架，模拟企业层级结构

核心特性：
- 三种流程类型：
- Sequential：任务按预定义顺序依次执行，前一任务输出作为下一任务上下文
- Hierarchical：Manager Agent 监督任务执行（规划、委托、验证），任务不预分配，根据 Agent 能力动态分配
- Consensual（计划中）：Agent 间协作决策任务执行
- Flows 机制：事件驱动架构，@start() 和 @listen(func) 装饰器链接任务
- 状态管理：每个 Flow 实例自动获得唯一 ID，self.state 字典贯穿整个流程

适用场景：企业级复杂工作流、需要层级管理和审批的场景、事件驱动任务链

权衡：抽象层次高，灵活性不如 LangGraph，但易用性更好

2.4 OpenAI Swarm / Agents SDK

定位：OpenAI 官方轻量级 Agent 编排框架

Swarm（教育性框架）：
- 核心抽象：Agent（指令 + 工具）+ Handoffs（Agent 间移交）
- 无状态设计：需手动传递 messages/context_variables 到下次调用
- client.run() 执行循环：获取 completion → 执行工具 → 切换 Agent → 更新上下文 → 重复
- 适用场景：原型开发、学习 Agent 编排概念

Agents SDK（生产就绪版本）：
- 新增特性：
- Session 管理：自动对话历史管理（跨 Agent 运行）
- Tracing：内置追踪 Agent 运行，查看/调试/优化工作流
- Guardrails：输入/输出验证的可配置安全检查
- Human-in-the-loop：内置人工介入机制
- Realtime Agents：语音 Agent 支持
- 与 Responses API 深度集成（Chat Completions 的超集）
- 支持 Redis Session 存储（可选）

适用场景：需要快速上手、轻量级编排、与 OpenAI API 深度集成的场景

权衡：功能相对简单，不适合超复杂状态管理和嵌套层级

[reviewer] 章节已写完：主流框架对比（CrewAI、Swarm/Agents SDK）

3. Agent 间通信机制分析

3.1 共享状态（Shared State）

代表框架：LangGraph

机制：
- Agent 通过读写共享图状态（graph state）通信
- 状态是结构化数据（字典/对象），包含所有 Agent 需要的上下文
- Agent 将结果添加到 state，而非直接调用其他 Agent

优点：
- 解耦 Agent 间依赖，Agent 只需关心状态读写
- 支持复杂状态管理（嵌套对象、类型检查）
- 易于实现中断/恢复（保存状态快照即可）

缺点：
- 状态膨胀风险（所有数据都在 state 里）
- 需仔细设计状态结构，避免冲突
- 调试困难（状态变化不直观）

3.2 消息总线（Message Bus）

代表框架：AutoGen、CrewAI（部分）

机制：
- Agent 通过发布/订阅消息总线通信
- 消息包含类型（WorkerTask、GroupChatMessage 等）和内容
- Manager/Orchestrator 负责路由消息到目标 Agent

优点：
- 显式消息流，易于追踪和调试
- 支持异步通信（Agent 不需等待响应）
- 易于添加任务取消、错误处理（消息协议扩展）

缺点：
- 消息传递开销（序列化/反序列化）
- 需设计消息协议，防止乒乓传递
- Manager 成为单点故障（需去中心化设计）

3.3 Handoff（移交）

代表框架：OpenAI Swarm/Agents SDK

机制：
- Agent 通过工具调用将对话移交给另一个 Agent
- 新 Agent 接收完整对话历史（messages）
- 无状态设计：需手动传递 context_variables

优点：
- 极简设计，易于理解和实现
- 灵活性高（Agent 可动态决定移交目标）
- 与 LLM 工具调用机制天然契合

缺点：
- 无状态设计需手动管理历史（Agents SDK 通过 Session 解决）
- 不适合复杂状态管理
- 移交链过长时性能下降

3.4 混合模式

代表框架：CrewAI Flows、LangGraph Hierarchical

机制：
- 结合共享状态和消息传递
- 例如：CrewAI Flows 使用 self.state 共享状态 + @listen(func) 事件驱动

优点：
- 兼顾灵活性和易用性
- 适合复杂工作流

缺点：
- 学习曲线陡峭
- 需仔细设计避免状态和消息冲突

[reviewer] 章节已写完：Agent 间通信机制分析

4. 任务分发与编排模式

4.1 Sequential（顺序）

机制：任务按预定义顺序依次执行，前一任务输出作为下一任务输入

适用场景：
- 任务间有明确依赖关系（如数据处理管道）
- 需要严格顺序保证（如审批流程）

实现框架：CrewAI Sequential、LangGraph 线性图

优点：简单直观，易于调试
缺点：无法并行，性能受限于最慢任务

4.2 Parallel（并行）

机制：独立子任务同时执行，最后汇总结果

适用场景：
- 子任务间无依赖（如多源数据采集）
- 需要加速执行（如并行搜索）

实现框架：AutoGen Mixture of Agents（同层 Worker 并行）、AdaptOrch Parallel 拓扑

优点：性能最优，充分利用并发
缺点：需处理结果汇总和冲突解决

4.3 Hierarchical（层级）

机制：树形结构，Manager 分解任务并分配给 Worker，Worker 执行后返回结果，Manager 汇总

适用场景：
- 复杂大型任务需要分解（如软件开发项目）
- 需要层级审批和质量控制

实现框架：CrewAI Hierarchical、AutoGen Group Chat（Manager 选择发言者）、LangGraph Agent Supervisor

关键设计：
- Manager Agent 能力：规划、委托、验证、评估完成度
- 任务不预分配，根据 Agent 能力动态分配
- 支持嵌套层级（子 Manager 管理子团队）

优点：适合复杂任务，易于扩展
缺点：Manager 成为瓶颈，需仔细设计避免单点故障

4.4 Hybrid（混合）

机制：根据任务依赖 DAG 的结构特性（并行宽度、关键路径深度、子任务耦合度）动态选择拓扑

代表研究：AdaptOrch 论文（NeurIPS 2024）

核心洞察：
- 当 LLM 能力收敛（GPT-4o/Claude/Gemini 在标准基准相差 2-5%），拓扑选择对性能的影响超过模型选择 Ω(1/ε²) 倍
- 拓扑路由算法：基于 DAG 结构特性选择最优拓扑，O(|V|+|E|) 时间复杂度
- 实证效果：拓扑感知编排比静态单拓扑基准提升 12-23%（SWE-bench/GPQA/RAG 任务）

适用场景：任务结构动态变化、需要最优性能的生产环境

优点：性能最优，自适应
缺点：实现复杂，需额外计算开销

4.5 编排模式选择指南

[reviewer] 章节已写完：任务分发与编排模式

5. 防卡死/超时机制

5.1 业界现状数据（Galileo AI 研究）

多 Agent 系统风险：
- 50% 错误率，30% 项目在 PoC 后被放弃（Gartner 预测）
- Token 重复浪费 53-86% 计算资源（MetaGPT 72%、CAMEL 86%、AgentVerse 53%）
- OWASP 将 prompt injection 列为 #1 安全漏洞（46% 成功率）
- 首个综合失败分类法（MAST）记录 1600+ 失败轨迹

成本影响：
- Agent 完成任务后陷入无意义辩论，燃烧计算
- GPT-4o 成本从 $225/月增至 $387/月（100万 token/天，72-86% 重复率）

5.2 Token 边界与超时（Token Boundaries & Timeouts）

机制：
- 显式 token 预算作为断路器，超出则强制 Agent 结束或让步
- 时间超时：asyncio.wait_for / 独立超时 goroutine

实现示例：

# OpenAI Swarm
client.run(agent=agent, messages=messages, max_turns=10)  # 限制最大轮次

# LangGraph
graph.compile(checkpointer=checkpointer, recursion_limit=50)  # 防止无限循环

# AutoGen
agent.max_consecutive_auto_reply = 5  # 限制连续自动回复次数

最佳实践：
- 生产环境必须设置 max_turns/recursion_limit（默认 inf 危险）
- Token 预算根据任务复杂度动态调整（简单任务 1000 token，复杂任务 10000 token）
- 超时时间建议：单 Agent 调用 30s，整体工作流 5min

5.3 状态持久化与恢复（State Persistence）

机制：
- Redis/数据库保存状态快照
- 会话过期时间（如 1 小时）
- 支持从最后稳定状态恢复

实现示例：

class StateManager:
    def save_state(self, session_id, state):
        redis_client.set(f"state:{session_id}", json.dumps(state), ex=3600)

    def load_state(self, session_id):
        return json.loads(redis_client.get(f"state:{session_id}"))

LangGraph Checkpointer：
- 每个 super-step 自动保存状态快照
- Checkpoint 内容：所有 executor 状态、待处理消息、共享状态
- 支持时间旅行（回溯到历史状态重新执行）

Microsoft Agent Framework Checkpoints：
- 适用场景：长时间运行工作流、需要暂停/恢复、定期状态保存（审计/合规）
- 可从特定 checkpoint 直接恢复同一 run

5.4 错误恢复机制（Error Recovery）

机制：
- 最大重试次数（如 3 次）
- 重试间隔递增（exponential backoff）
- 最终错误处理：保存错误状态、回滚到最后稳定状态

实现示例：

class ErrorHandler:
    max_retries = 3

    async def with_retry(self, func, state):
        retries = 0
        while retries < max_retries:
            try:
                return await func(state)
            except Exception as e:
                retries += 1
                if retries == max_retries:
                    return handle_final_error(e, state)
                await asyncio.sleep(2 ** retries)  # exponential backoff
                await handle_retry(e, state, retries)

最佳实践：
- 区分可重试错误（网络超时、API 限流）和不可重试错误（参数错误、权限不足）
- 重试前记录错误日志，便于调试
- 最终错误处理：通知用户、保存错误状态、触发告警

5.5 Human-in-the-Loop（人工介入）

机制：
- LangGraph：interrupt_before/interrupt_after，执行前/后暂停等待人工确认
- OpenAI Agents SDK：内置 Human-in-the-loop 机制
- 防止 Agent 错误决策链锁定

适用场景：
- 高风险操作（删除数据、发送邮件、支付）
- 需要人工审批的流程（财务审批、代码合并）
- Agent 不确定时请求人工指导

实现示例：

# LangGraph
graph = StateGraph(...)
graph.add_node("risky_operation", risky_operation_node)
graph.compile(checkpointer=checkpointer, interrupt_before=["risky_operation"])

# 执行时会在 risky_operation 前暂停，等待人工确认
result = graph.invoke(input, config={"configurable": {"thread_id": "1"}})
# 人工确认后继续
result = graph.invoke(None, config={"configurable": {"thread_id": "1"}})

5.6 确定性任务分配（Deterministic Task Allocation）

问题：Agent 乒乓传递同一任务，重复规划，浪费计算

解决方案（Galileo AI 研究）：
- 两层去中心化架构 + 本地投票协议（LVP），消除单点故障
- 唯一任务 ID、日志记录分配 Agent、拒绝重新分配（除非显式释放）

实现要点：
- 任务分配前检查任务 ID 是否已分配
- 分配后记录到共享日志（Redis/数据库）
- Agent 完成任务后显式释放任务 ID

5.7 监控与可观测性（Monitoring & Observability）

工具：
- Langfuse/LangSmith：追踪 Agent 执行轨迹
- OpenAI Agents SDK：内置 Tracing
- 每个步骤记录输入/输出/token 消耗/时延

关键指标：
- Token 消耗率（token/min）
- Agent 切换次数
- 任务完成时间
- 错误率和重试次数

告警规则：
- Token 消耗超过预算 80% → 警告
- 单 Agent 执行超过 1min → 警告
- 错误率超过 10% → 告警

[reviewer] 章节已写完：防卡死/超时机制

6. 对我们当前系统的启发与建议

6.1 当前系统问题诊断

基于本次调研和业界数据，我们当前系统可能存在以下风险：

1. Token 重复浪费：业界 72-86% 重复率，我们的 researcher → reviewer → main 链可能也存在类似问题
2. 缺乏超时机制：task.md 中要求 timeoutSeconds=0（禁止超时），这在生产环境是危险的
3. 状态管理不足：session.md 只追加日志，缺乏结构化状态快照和恢复机制
4. 防卡死机制缺失：没有 token 边界、重试、断路器等保护措施
5. 可观测性不足：缺乏 Agent 执行轨迹追踪、token 消耗监控

6.2 短期改进建议（1-2 周内可实施）

6.2.1 添加超时保护

问题：timeoutSeconds=0 禁止超时，Agent 可能无限等待

建议：
- 改为 timeoutSeconds=300（5 分钟），防止 Agent 卡死
- 超时后自动重试（最多 3 次）
- 最终超时后通知主人，请求人工介入

实现：

# 修改 task.md 模板
sessions_send 通知 reviewer（`agent:reviewer:main`，timeoutSeconds=300）

6.2.2 添加 Token 边界

问题：researcher 搜索可能无限扩展，reviewer 报告可能过长

建议：
- researcher：每轮搜索限制 5000 token，总计不超过 50000 token
- reviewer：每章节限制 3000 token，总计不超过 30000 token
- 超出预算时强制结束当前章节，追加 [Token 预算耗尽] 标记

实现：

# 在 AGENTS.md 中添加 token 预算配置
researcher_token_budget=50000
reviewer_token_budget=30000

6.2.3 改进状态管理

问题：session.md 只追加日志，缺乏结构化状态

建议：
- 添加 state.json：记录当前执行阶段、已完成章节、token 消耗、错误次数
- 每个 Agent 完成后更新 state.json
- 失败重试时从 state.json 恢复，跳过已完成章节

实现：

{
  "task_id": "task-multiagent-collab-c89114",
  "current_agent": "reviewer",
  "completed_sections": ["执行摘要", "主流框架对比"],
  "token_consumed": 15000,
  "retry_count": 0,
  "last_update": "2026-03-02T23:30:00Z"
}

6.2.4 添加错误恢复机制

问题：rejectCount 机制只支持 1 次重试，且需要人工裁决

建议：
- 自动重试（最多 3 次），每次重试间隔递增（1min、2min、4min）
- 重试前分析失败原因，调整执行策略（如减少搜索轮次、简化报告结构）
- 3 次重试仍失败后才通知主人

实现：

# 在 task.md 中添加重试逻辑
if [ $retry_count -lt 3 ]; then
  sleep $((2 ** retry_count * 60))
  retry_count=$((retry_count + 1))
  # 调整执行策略
  if [ $retry_count -eq 2 ]; then
    # 第二次重试：减少搜索轮次
    search_rounds=3
  fi
  # 重新执行
  sessions_send ...
else
  # 通知主人
  message ...
fi

6.3 中期改进建议（1-2 月内可实施）

6.3.1 引入 Checkpointer 机制

参考：LangGraph Checkpointer、Microsoft Agent Framework Checkpoints

建议：
- 每个 Agent 完成后保存 checkpoint（包含状态、消息历史、token 消耗）
- 支持从任意 checkpoint 恢复（时间旅行）
- 支持人工介入（interrupt_before/interrupt_after）

实现：

# 保存 checkpoint
checkpoint_id=$(uuidgen)
echo "{\"checkpoint_id\": \"$checkpoint_id\", \"state\": ...}" > /root/.openclaw/workspace/tasks/$task_id/checkpoints/$checkpoint_id.json

# 恢复 checkpoint
state=$(cat /root/.openclaw/workspace/tasks/$task_id/checkpoints/$checkpoint_id.json)

6.3.2 实现拓扑感知编排

参考：AdaptOrch 论文

建议：
- 分析任务依赖 DAG，自动选择最优拓扑（Sequential/Parallel/Hierarchical/Hybrid）
- 例如：researcher 搜索可并行（5 个方向同时搜），reviewer 报告可顺序写

实现：

# 分析任务 DAG
task_dag=$(analyze_task_dag task.md)

# 选择拓扑
if [ "$task_dag" == "parallel" ]; then
  # 并行执行
  for direction in "${directions[@]}"; do
    sessions_send researcher "搜索方向: $direction" &
  done
  wait
else
  # 顺序执行
  for direction in "${directions[@]}"; do
    sessions_send researcher "搜索方向: $direction"
  done
fi

6.3.3 添加监控与可观测性

参考：Langfuse/LangSmith、OpenAI Agents SDK Tracing

建议：
- 记录每个 Agent 的输入/输出/token 消耗/时延
- 可视化 Agent 执行轨迹（类似 Gantt 图）
- 实时监控 token 消耗率、错误率

实现：

# 记录 Agent 执行
echo "{\"agent\": \"researcher\", \"start\": \"$start_time\", \"end\": \"$end_time\", \"token\": $token_consumed}" >> /root/.openclaw/workspace/tasks/$task_id/trace.jsonl

# 可视化（使用 Python 脚本）
python /root/.openclaw/workspace/scripts/visualize-trace.py /root/.openclaw/workspace/tasks/$task_id/trace.jsonl

6.4 长期改进建议（3-6 月内可实施）

6.4.1 迁移到主流框架

建议：
- 评估 LangGraph、AutoGen、OpenAI Agents SDK 三个框架
- 选择标准：
- LangGraph：需要高度定制化控制流、复杂状态管理
- AutoGen：需要层级汇总、动态任务分解
- OpenAI Agents SDK：需要快速上手、轻量级编排、与 OpenAI API 深度集成

迁移路径：
1. 先用 OpenAI Agents SDK 重写简单任务（如单 Agent 搜索）
2. 再用 LangGraph 重写复杂任务（如多 Agent 协作）
3. 最后评估性能和易用性，选择最终方案

6.4.2 实现 Agent 专业化

参考：CrewAI Hierarchical、AutoGen Group Chat

建议：
- 细化 Agent 职责：researcher（搜索）、analyzer（分析）、writer（写作）、reviewer（审查）
- 每个 Agent 只做一件事，避免职责重叠
- 使用 Manager Agent 协调任务分配

实现：

# Manager Agent 分析任务
task_type=$(analyze_task task.md)

# 分配给专业 Agent
if [ "$task_type" == "research" ]; then
  sessions_send researcher ...
elif [ "$task_type" == "analysis" ]; then
  sessions_send analyzer ...
elif [ "$task_type" == "writing" ]; then
  sessions_send writer ...
fi

6.4.3 实现自适应编排

参考：AdaptOrch Hybrid 拓扑

建议：
- 根据任务复杂度动态调整编排策略
- 简单任务：Sequential（快速完成）
- 中等任务：Hierarchical（层级分解）
- 复杂任务：Hybrid（自适应拓扑）

实现：

# 评估任务复杂度
complexity=$(evaluate_task_complexity task.md)

# 选择编排策略
if [ "$complexity" -lt 5 ]; then
  strategy="sequential"
elif [ "$complexity" -lt 10 ]; then
  strategy="hierarchical"
else
  strategy="hybrid"
fi

# 执行
execute_with_strategy $strategy task.md

6.5 优先级排序

建议优先实施 P0 和 P1 项，这些改进成本低、收益高，可立即提升系统稳定性和成本效率。

[reviewer] 章节已写完：对我们当前系统的启发与建议