一、问题的起点：传统多智能体系统的三大痛点

之前的一些多智能体框架（如MetaGPT、AutoGen、CrewAI 等），会发现它们普遍存在三个核心问题——这也正是字节在今年7月份的一篇 AIME 论文开篇就指出的Plan-and-Execute 框架的三大致命缺陷：

痛点1：僵化的执行计划（Rigid Plan Execution）

传统的「规划-执行」框架有个致命缺陷：计划一旦制定，就几乎不能改变。

用户请求 → 规划器制定计划 → 执行器逐个执行 → 返回结果
                ↓
        (规划器在首次执行完后就完全闲置，躺平摸鱼...)

这就像你用 GPS 导航，结果前方修路了，但 GPS 还在傻傻地让你前方 500 米右转...

真实场景：假设我要研究《2024年AI领域十大突破》，规划器分解出了5个搜索子任务。结果第一个搜索就发现，某个权威机构已经发布了完整的年度报告。此时，剩下的4个子任务其实都可以跳过了——但传统框架还是会机械地执行完。

痛点2：静态的智能体能力（Static Agent Capabilities）

大多数框架预定义了一组固定角色的智能体：研究员、程序员、评审员……

问题来了：如果遇到需要「金融分析师 + 法律顾问」组合能力的任务怎么办？

预定义的角色就像工厂流水线上的工人，每人只会拧一种螺丝。面对新型号的产品，整条产线可能都要停工改造。

痛点3：低效的信息传递（Inefficient Communication）

这是我认为多智能体协作中最大的坑。

当任务在多个智能体之间流转时，上下文信息会像传话游戏一样逐渐失真。智能体 A 的发现传递给智能体B，B又传给C...，层层消息传递中，关键细节可能被压缩、遗漏，甚至曲解。

更糟糕的是，由于缺乏全局状态管理，每个智能体都只能看到自己那一亩三分地—它不知道其他智能体做了什么，更不知道整体进度如何。

所以，我们必须设计更好更高效的信息传递机制（智能体间的协作机制）

---

二、AIME的多智能体方案解读

先看一下它的架构图：

简化版：

1. 分解任务 → 2. 分发任务 → 3. 子代理实例化 → 4. 执行子任务 → 5. 进度更新 → 6. 评估与迭代
                     ↑                                                    ↓
                     └────────────────────────────────────────────────────┘

详细说明：

步骤	名称	说明
1	分解任务	动态规划器（主智能体）将用户请求分解为层级子任务，在todo.md中初始化状态；关键点：规划是支持动态修改的；
2	分发任务（外包子任务）	动态规划器（主智能体进程）从全局任务列表中选择下一个可执行的子任务，发送到Actor Factory（即子智能体）
3	代理实例化	执行工厂根据子任务需求创建一个定制化动态执行者，赋予其特定身份（比如深度研究专家、代码编写专家、数据分析专家...）、工具集和任务背景（可详见下面我的案例）
4	执行子任务	子智能体subagent按照React范式（反复“推理-行动-观测”循环）通过多轮专属工具调用的方式推进完成子任务
5	进度更新	子智能体subagent在执行过程中（或结束后）通过Update_Progress工具向进度管理模块报告关键进度点或异常信息，保持全局状态与实际执行同步—即上下文信息共享和同步！
6	评估迭代	子任务完成后，执行者（subagent）生成包含状态更新和结果摘要的最终报告提交给动态规划器（或叫主智能体）。规划器（或叫主智能体）据此更新全局任务列表并循环回到步骤2，直到顶层任务完成

• 总结：

---

三、我的Multi-Agent实现：基于共享工作区的ReAct范式

我的实现方案和AIME非常相似，一句话总结的话就是：

为什么选择文件而不是message传递？

因为文件是当前最可靠的上下文压缩和传递媒介。

优势	说明
天然持久化	智能体崩溃或重启，上下文不会丢失
自然的边界	每个文件是独立的「信息包」
高效压缩	一个文件路径就代指了文件内的海量上下文，避免上下文爆炸，不需要在prompt中复制大量文本
人类可读	调试时可以直接查看文件内容
异步友好	不同智能体可以在不同时间读写同一文件

Anthropic 的 Claude 团队在年中的一次访谈中也提到了类似的观点：

架构总览

┌─────────────────────────────────────────────────────────────────┐
│                      Multi-Agent                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────────┐    ┌─────────────────────────────────────┐     │
│  │   主智能体   │◄──►│           共享沙箱工作区             │     │
│  │  (Planner)  │    │  ┌─────────────────────────────┐    │     │
│  └──────┬──────┘    │  │  TODO.md (任务进度日志)      │    │     │
│         │           │  │  FILE_INDEX.json (文件索引)  │    │    │
│         │           │  │  outputs/ (交付物目录)       │    │    │
│         │           │  │  research/ (研究文档)        │    │    │
│         ▼           │  │  data/ (数据文件)            │    │    │
│  ┌──────────────┐   │  └─────────────────────────────┘    │    │
│  │ SubAgent Pool│◄─►│                                     │    │
│  │              │   └─────────────────────────────────────┘    │
│  │ • deep_researcher                                           │
│  │ • data_analyst                                              │
│  │ • fact_checker                                              │
│  │ • code_developer                                            │
│  │ • summarizer                                                │
│  │ • general_assistant                                         │
│  │ • planner                                                   │
│  └──────────────┘                                              │
│                                                                │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │                     基础工具层                           │   │
│  │  [web_search]  [open_link]  [code_interpreter]          │   │
│  └─────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────┘

核心工作范式：基于共享工作区的ReAct范式

可以概括为：思考-行动-观察-更新的循环，在共享文件系统中进行。

📍 关键架构特色：共享工作区 (Shared Workspace)

所有智能体（包括主智能体和所有子智能体）都共享同一个沙箱文件系统。这意味着：

• 文件是信息传递的主要桥梁，而不是在对话中复制大量文本
• 所有进度和产出都持久化保存在工作区文件中
• 子智能体能自动读取TODO.md了解当前进度，访问已生成的文件

🚀 标准工作流程（四步法）：

Step 1: 初始化工作区（必须首先执行）

收到任何任务后，主智能体立即创建标准化的工作区结构：

# 主智能体执行的第一段代码
import os
from datetime import datetime

os.makedirs("outputs", exist_ok=True)
os.makedirs("research", exist_ok=True)
os.makedirs("data", exist_ok=True)
os.makedirs("temp", exist_ok=True)

todo_content = f"""# 任务进度日志
## 原始任务
分析2024年全球AI投资趋势，生成可视化报告

## 任务计划
待规划...
"""
with open("TODO.md", "w") as f:
    f.write(todo_content)

/
├── TODO.md              # 任务进度日志（核心跟踪文件）
├── FILE_INDEX.json      # 文件索引
├── outputs/             # 最终交付物
├── research/            # 研究文档
├── data/                # 数据文件
└── temp/                # 临时文件

Step 2: 制定与规划

主智能体分析任务需求，制定详细执行计划，并将其写入TODO.md文件。

Step 3: 分步执行与委托

对于每个子任务：

• 简单任务：主智能体直接使用基础工具（搜索、开链接、写代码，即对应基础工具层的三个工具）
• 复杂任务：主智能体委托给合适的子智能体，它们(子智能体)被调用后会：
  1. 读取TODO.md了解上下文
  2. 访问工作区中的相关文件
  3. 执行任务并将结果保存到工作区
  4. 更新TODO.md进度（含更新TODO规划）

Step 4: 汇总与交付

主智能体从工作区文件中读取所有产出（研究报告、数据分析结果、代码等），整合成最终答案交付给用户。

---

📝 示例说明：如果用户问"分析2024年AI领域的主要趋势"

其工作流程如下：

1. 初始化：创建/TODO.md，记录原始任务
2. 规划：制定计划：
   • 1. 委派 deep_researcher 搜索权威投资数据来源
   • 2. 委派 data_analyst 清洗数据并生成可视化
   • 3. 委派 summarizer 生成执行摘要
   • 4. 整合所有产出，生成最终报告
3. 执行（串行）：
   • 1）deep_researcher子智能体搜索并整理生成AI趋势报告 -> 报告文件存储到research目录下 -> 更新TODO.md
   • 2）data_analyst子智能体分析相关数据并制作图表 -> 文件存储到合适目录下 -> 更新TODO.md
   • 3）summarizer子智能体生成执行摘要 -> 报告文件存储到research目录下 -> 更新TODO.md
4. 汇总：最后主智能体读取前序所有文件产出后先创建汇总报告，再把最终报告导入到/outputs/目录下 -> 更新TODO.md -> 最后响应回答给用户

整个过程中，TODO.md都会实时更新：

## 执行进度
| # | 步骤 | 执行者 | 状态 | 产出文件 |
|---|------|--------|------|----------|
| 1 | 搜索AI趋势报告 | deep_researcher | ✅完成 | research/ai_trends_2024.md |
| 2 | 分析关键数据 | data_analyst | ✅完成 | data/trend_analysis.csv, outputs/charts.png |
| 3 | 生成执行摘要 | summarizer | 🔄进行中 | |

与 AIME 的架构映射和对比

AIME 组件	我的Multi-Agent 对应	核心职责
Dynamic Planner	主智能体	持续监控、动态调整策略
Actor Factory	SubAgentDefinition + execute_subagent()	按需实例化专业智能体
Dynamic Actor	SubAgent 执行循环（ReAct 范式）	思考-行动-观察的迭代循环
Progress Management Module	TODO.md + FILE_INDEX.json	全局状态的单一事实来源

---

🔧 四、核心组件详解

组件一：全局动态规划器（即主智能体）

主智能体不是一个静态的「计划制定者」，而是一个 持续监控、动态调整 的「指挥官」。

关键点：

• 实时反馈循环：每执行完一个步骤，主智能体都会根据最新的上下文重新评估全局状态
• 动态计划调整：如果发现有更优路径，可以随时修改 TODO.md 中的计划
• 异常处理能力：子任务失败时，可以立即制定备选方案

组件二：SubAgent 工厂

通过 SubAgent 预定义角色模板 ，然后结合具体任务，实现按需实例化。

首先，定义n个专业的子智能体角色：

@dataclass
class SubAgentDefinition:
    name: str                    # 子智能体名称
    description: str             # 能力描述（用于主智能体选择，且子智能体的描述部分要注入主智能体的系统提示词）
    system_prompt: str           # 不同角色的子智能体的系统提示词（定义角色和行为）
    max_tool_calls: int          # 最大工具调用次数
    available_tools: List[str]   # 可用工具集
    auto_read_todo: bool = True  # 是否自动读取 TODO 日志
    auto_update_todo: bool = True # 是否自动更新 TODO 日志

子智能体（随意diy设计的）	专长	配备工具
deep_researcher	深度网络调研，多源信息整合	web_search, open_link, code_interpreter
data_analyst	数据清洗、统计分析、可视化	code_interpreter
fact_checker	事实核查与可信度评估	web_search, open_link, code_interpreter
code_developer	代码编写、调试与优化	code_interpreter
summarizer	长文档阅读与精炼摘要	open_link, code_interpreter
general_assistant	通用任务处理	web_search, open_link, code_interpreter
planner	复杂任务分解与规划	code_interpreter

SubAgent实例化时的上下文注入

每个子智能体在实例化时，除了角色定义模板外，还会自动注入当前工作区的状态信息和具体任务信息的提示：

# 伪代码参考
async def execute_subagent(subagent_name, task, shared_context, step_number):
    subagent_def = SUBAGENT_DEFINITIONS[subagent_name]
    
    # 🔑 关键：自动注入工作区上下文
    if shared_context and subagent_def.auto_read_todo:
        todo_content = await shared_context.read_todo()
        file_list = await shared_context.get_file_list()
        
        workspace_context = f"""
        ## 📋 当前工作区状态
        ### TODO日志摘要:
        {todo_content[:2000]}
        
        ### 已有文件列表:
        {json.dumps(file_list, indent=2)}
        
        **重要**：请先阅读上述工作区状态，了解前序步骤的产出。
        """
    
    # 基于 角色模块+工作区状态+具体任务等，构建完整的系统提示词
    full_prompt = subagent_def.system_prompt + workspace_context + task +  step_number
    # ... 执行子智能体
    ...

组件三：共享工作区（核心设计）

这是整个系统的中枢神经。所有智能体（主智能体和子智能体）共享同一个沙箱文件系统。

我的工作区结构：

/
├── TODO.md              # 📋 任务进度日志（核心！）
├── FILE_INDEX.json      # 📑 文件索引
├── outputs/             # 📦 最终交付物
├── research/            # 📚 研究文档
├── data/                # 📊 数据文件
└── temp/                # 🗑️ 临时文件

TODO.md：不只是任务清单，更是信息协作的参照物

TODO.md 是整个系统的核心，或者说是信息协作的「第一事实来源」（Single Source of Truth），它的结构经过精心设计，参考demo如下：

# 🎯 任务进度日志 (Task Journal)

> 创建时间: 2025-12-18 10:30:00

---

## 📋 原始任务 (Original Task)

分析2024年AI领域的主要趋势并生成可视化报告

---

## 📝 任务计划 (Task Plan)

1. 搜索权威AI年度报告
2. 提取关键趋势数据
3. 进行数据可视化
4. 生成最终报告

---

## ⏳ 执行进度 (Execution Progress)

| # | 步骤描述 | 执行者 | 状态 | 产出文件 |
|---|----------|--------|------|----------|
| 1 | 搜索AI趋势报告 | deep_researcher | ✅ 完成 | research/ai_trends_raw.md |
| 2 | 数据提取与清洗 | data_analyst | ✅ 完成 | data/trends_data.csv |
| 3 | 生成可视化图表 | data_analyst | 🔄 进行中 | - |

---

## 📁 已生成文件索引 (Generated Files Index)

| 文件名 | 类型 | 创建者 | 描述 |
|--------|------|--------|------|
| research/ai_trends_raw.md | 文档 | deep_researcher | 原始调研报告 |
| data/trends_data.csv | 数据 | data_analyst | 清洗后的趋势数据 |

---

## 💡 关键发现摘要 (Key Findings Summary)

-  发现斯坦福HAI发布的2024年度AI指数报告
-  确认生成式AI投资同比增长超200%

组件四：工作区管理器（WorkspaceManager）

这是一个纯 Python 类，封装了所有工作区操作的代码操作逻辑：

# 伪代码参考
class WorkspaceManager:
    @staticmethod
    def generate_init_workspace_code(task_description: str) -> str:
        """生成初始化工作区的Python代码"""
        return f'''
import os
os.makedirs("outputs", exist_ok=True)
os.makedirs("research", exist_ok=True)
os.makedirs("data", exist_ok=True)

todo_content = """# 任务进度日志
## 原始任务
{task_description}
..."""

with open("TODO.md", "w") as f:
    f.write(todo_content)
'''
    
    @staticmethod
    def generate_update_progress_code(step_number, description, executor, status, output_files) -> str:
        """生成更新进度的Python代码"""
        # ... 生成更新 TODO.md 的代码
    
    @staticmethod
    def generate_register_file_code(filename, file_type, creator, description) -> str:
        """生成注册新文件到索引的Python代码"""
        # ... 生成更新 FILE_INDEX.json 的代码

组件五：基础工具集（3个）

• web_search：执行网络搜索，获取最新信息（基于Tavily API）
• open_link：打开特定URL，提取详细页面内容（基于Tavily API）
• code_interpreter：在e2b沙箱环境中执行Python代码，处理数据、创建文件等等

---

五、关键设计细节

1. 增量写入策略

针对长文档生成，我设计了「分块写入」机制：

写入超长文件时，让大模型按分多次增量写入的方式执行，这缓解了 LLM 单次输出长度限制的问题，避免了「写到一半被截断」的尴尬。

2. 递归文件变化检测与云存储上传

# 伪代码参考
async def code_interpreter_async(sbx, code: str) -> Dict[str, Any]:
    # 执行前：递归扫描所有文件及修改时间
    file_dict_original = await scan_all_files(sbx)
    
    # 执行用户代码
    results = await execute(code)
    
    # 执行后：再次扫描，检测变化
    file_dict_after = await scan_all_files(sbx)
    
    # 新增或修改的文件自动上传到 OSS 云存储
    for changed_file in detect_changes(file_dict_original, file_dict_after):
        if is_image(changed_file):
            url = await pic2cloud_async(changed_file)
        else:
            url = await file2cloud_async(changed_file)
        results['generated_files_urls'][changed_file] = url

递归扫描代码支持检测所有子目录：

# 伪代码参考
RECURSIVE_FILE_SCAN_CODE = '''
def get_all_files_with_mtime(root_dir="."):
    """递归获取目录下所有文件及其修改时间"""
    exclude_dirs = {'.git', '__pycache__', 'node_modules', '.cache'}
    file_dict = {}
    
    for dirpath, dirnames, filenames in os.walk(root_dir):
        dirnames[:] = [d for d in dirnames if d not in exclude_dirs]
        for filename in filenames:
            rel_path = os.path.relpath(os.path.join(dirpath, filename), root_dir)
            file_dict[rel_path] = os.path.getmtime(full_path)
    
    return file_dict
'''

这让产出物可以通过 OSS 后的 URL 直接分享给用户，而不是困在沙箱里。

3. XML 格式的工具调用

与常见的 JSON 格式不同，我选择了 XML 格式进行工具调用：

<tool_call>
<function>code_interpreter</function>
<code>
import pandas as pd
df = pd.read_csv("data/trends.csv")
print(df.head())
</code>
</tool_call>

之所以不用json格式或者说选择了json -> xml的工具调用格式风格切换，可以参阅XML Tool Calls: Beyond JSON Constraints。

---

六、当前局限与未来方向

1. 串行执行的效率瓶颈

当前所有SubAgent串行执行。对于相互独立的子任务场景（比如"搜索A"和"搜索B"这类独立任务），理论上可以并行，但需要解决：

• 文件锁问题
• TODO.md并发写入冲突
• 错误传播和回滚机制

# 伪代码参考
async def parallel_execute(tasks: List[SubTask]):
    independent_groups = analyze_dependencies(tasks)
    for group in independent_groups:
        await asyncio.gather(*[execute_subagent(t) for t in group])

2. 自动错误恢复

当 SubAgent 执行失败时，自动尝试修复：

# 伪代码参考
async def execute_with_retry(subagent_name, task, max_retries=3):
    for attempt in range(max_retries):
        try:
            result = await execute_subagent(subagent_name, task)
            if result["status"] == "success":
                return result
        except Exception as e:
            task = refine_task_based_on_error(task, e)
    return {"status": "failed", "reason": "Max retries exceeded"}

3、缺乏跨会话的记忆

当前每个会话的工作区是独立的，会话结束后沙箱被销毁。未来可以考虑：

• 持久化关键产出到外部存储
• 沙箱重加载

---

七、案例展示：

我用这套系统完成了一个端到端的复杂任务：让智能体写出一整本《RAG课程的教学课件》

详细执行过程和结果：Multi-Agent协作_测试任务_端到端输出RAG教学书籍

---

📚 参考资料

1. Aime: Towards Fully-Autonomous Multi-Agent Framework - 字节跳动, 2025
2. ReAct: Synergizing Reasoning and Acting in Language Models - Yao et al., 2023
3. MetaGPT: Meta Programming for Multi-Agent Collaborative Framework - Hong et al., 2024
4. Why do multi-agent LLM systems fail? - Cemri et al., 2025
5. Claude Code 最佳实践-访谈
6. XML Tool Calls: Beyond JSON Constraints
7. e2b - Code Interpreter Sandbox

---

如果你也在做多智能体相关的工作，欢迎交流讨论！