DeerFlow 2.0 深度拆解：当字节跳动决定重新定义「Super Agent 运行时」——从 LangGraph 子代理编排到 E2B 沙箱隔离的全链路架构革命

一年星破 5 万，上线即登顶 GitHub Trending。这不是营销噱头，而是字节跳动在 AI Agent 领域交出的一份硬核答卷。DeerFlow 2.0 不是简单的「Deep Research 工具」，而是一套完整的 Super Agent 运行时基础设施——从 Lead Agent + Sub-Agents 动态编排、Markdown 定义 Skills 工作流、E2B 沙箱隔离执行，到 Long/Short-term Memory 记忆系统、MCP 协议集成、Message Gateway 多渠道消息。本文将带你深入剖析这套架构的每一个技术细节。

一、背景：AI Agent 的「三座大山」与 Deep Research 的启示

1.1 现有 Agent 框架的结构性缺陷

如果你在过去两年尝试过 AutoGPT、MetaGPT、CrewAI 或 LangChain Agent，大概率遇到过以下问题：

执行能力薄弱：大多数 Agent 只能「说」不能「做」。让 AI 写个脚本？它能生成代码，但无法真正执行。需要跑个数据分析？它能给出 Python 片段，但没有运行环境。这种「纸上谈兵」的能力边界，让 Agent 在实际生产场景中大打折扣。

上下文管理混乱：长任务执行时，Agent 容易「遗忘」前期信息。一个持续 30 分钟的研究任务，到第 20 分钟时，Agent 可能已经忘记了最初的目标。这种「金鱼脑」现象，根源在于缺乏有效的记忆持久化机制。

扩展性不足：想给 Agent 加个新能力？要么改核心代码，要么写一堆胶水逻辑。没有标准化的能力封装方式，导致每个新功能都变成「侵入式改造」。

安全性缺失：让 Agent 执行代码？它可能在你的机器上跑 rm -rf /。访问网络？它可能泄露你的敏感数据。缺乏隔离执行环境，让 Agent 在生产环境中的风险不可控。

1.2 Deep Research 的市场启示

2025 年初，OpenAI 推出的 Deep Research 功能展示了 AI 在深度研究领域的巨大潜力：自动搜集信息、分析数据、生成报告，将原本需要数小时的研究工作压缩到几分钟。但这是闭源服务，需要订阅，且无法自定义和扩展。

这一市场空白催生了开源替代方案的需求。字节跳动技术团队敏锐捕捉到这一趋势，于 2025 年 5 月首次开源 DeerFlow 1.0，定位为「深度研究框架」。经过社区反馈和持续迭代，2026 年 2 月 28 日，DeerFlow 2.0 正式发布——这是一次彻底的重写，与 v1 版本没有共用代码，标志着项目从「研究工具」向「超级智能体执行底座」的战略转型。

1.3 DeerFlow 的定位差异

DeerFlow 2.0 的官方定义是 Super Agent Harness（超级 Agent 运行架构）。注意，不是「框架」，不是「工具链」，而是「Harness」——一个完整的运行时基础设施。

这个定位的差异体现在：

二、核心架构：Lead Agent + Sub-Agents 动态编排

2.1 架构总览

DeerFlow 2.0 的架构建立在 LangGraph 1.0 之上，采用模块化多代理设计。整体架构可以抽象为：

┌─────────────────────────────────────────────────────────────┐
│                      Lead Agent (主代理)                      │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐        │
│  │Planner  │→ │Coordinator│→ │Reporter │→ │Memory   │        │
│  └─────────┘  └─────────┘  └─────────┘  └─────────┘        │
└─────────────────────────────────────────────────────────────┘
                           ↓ 动态拉起
┌─────────────────────────────────────────────────────────────┐
│                    Sub-Agents Pool (子代理池)                 │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐        │
│  │Researcher│  │ Coder  │  │Analyst  │  │Custom...│        │
│  └─────────┘  └─────────┘  └─────────┘  └─────────┘        │
│       ↓           ↓           ↓           ↓                  │
│  ┌─────────────────────────────────────────────────────┐    │
│  │              Sandbox Environment (沙箱环境)           │    │
│  │  ┌───────┐  ┌───────┐  ┌───────┐  ┌───────┐        │    │
│  │  │File   │  │Shell  │  │Browser│  │MCP    │        │    │
│  │  │System │  │       │  │       │  │Server │        │    │
│  │  └───────┘  └───────┘  └───────┘  └───────┘        │    │
│  └─────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────┘
                           ↓ 持久化
┌─────────────────────────────────────────────────────────────┐
│                 Long-Term Memory (长期记忆)                   │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐                     │
│  │Vector DB│  │Relation │  │File     │                     │
│  │         │  │Graph    │  │Storage  │                     │
│  └─────────┘  └─────────┘  └─────────┘                     │
└─────────────────────────────────────────────────────────────┘

2.2 Lead Agent：任务编排的大脑

Lead Agent 是 DeerFlow 的核心调度器，负责：

1. 任务规划（Planner）：接收用户任务，拆解为多个子任务，生成执行计划
2. 协调调度（Coordinator）：根据计划动态拉起子代理，分配上下文和资源
3. 结果汇总（Reporter）：收集各子代理的输出，整合为最终结果
4. 记忆管理（Memory）：维护会话上下文，持久化关键信息

核心代码逻辑（简化版）：

# backend/agents/lead_agent.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Any

class AgentState(TypedDict):
    task: str
    plan: List[dict]
    sub_agent_results: dict
    final_report: str
    memory_context: dict

class LeadAgent:
    def __init__(self, llm, memory, sandbox):
        self.llm = llm
        self.memory = memory
        self.sandbox = sandbox
        self.graph = self._build_graph()
    
    def _build_graph(self):
        """构建 LangGraph 工作流"""
        graph = StateGraph(AgentState)
        
        # 添加节点
        graph.add_node("planner", self._plan_task)
        graph.add_node("coordinator", self._coordinate_subagents)
        graph.add_node("aggregator", self._aggregate_results)
        graph.add_node("reporter", self._generate_report)
        
        # 定义边
        graph.set_entry_point("planner")
        graph.add_edge("planner", "coordinator")
        graph.add_edge("coordinator", "aggregator")
        graph.add_edge("aggregator", "reporter")
        graph.add_edge("reporter", END)
        
        return graph.compile()
    
    async def _plan_task(self, state: AgentState) -> dict:
        """任务规划：拆解为子任务"""
        task = state["task"]
        memory_context = await self.memory.retrieve_relevant(task)
        
        prompt = f"""
        用户任务: {task}
        
        相关历史上下文:
        {memory_context}
        
        请将此任务拆解为多个子任务，每个子任务应包含:
        - 子任务类型（research/coding/analysis/report）
        - 具体目标
        - 预期输出
        - 依赖关系
        """
        
        plan = await self.llm.ainvoke(prompt)
        return {"plan": plan["subtasks"]}
    
    async def _coordinate_subagents(self, state: AgentState) -> dict:
        """协调子代理执行"""
        plan = state["plan"]
        results = {}
        
        # 按依赖关系排序执行
        for subtask in self._topological_sort(plan):
            agent_type = subtask["type"]
            
            # 动态创建子代理
            sub_agent = await self._spawn_subagent(
                agent_type, 
                subtask["goal"],
                sandbox=self.sandbox
            )
            
            # 执行并收集结果
            result = await sub_agent.run()
            results[subtask["id"]] = result
            
            # 更新记忆
            await self.memory.store(subtask["id"], result)
        
        return {"sub_agent_results": results}

2.3 Sub-Agents：专业化执行单元

DeerFlow 的子代理系统是其核心创新之一。每个子代理是一个独立的专业化执行单元，拥有：

• 独立上下文：每个子代理有自己的会话上下文，不会相互干扰
• 专属沙箱：子代理在隔离的沙箱环境中执行代码
• 工具访问权限：根据任务类型动态分配工具权限

内置子代理类型：

子代理动态拉起的核心逻辑：

# backend/agents/sub_agent.py
class SubAgentFactory:
    """子代理工厂：根据类型动态创建"""
    
    AGENT_REGISTRY = {
        "researcher": ResearcherAgent,
        "coder": CoderAgent,
        "analyst": AnalystAgent,
        "reporter": ReporterAgent,
    }
    
    @classmethod
    async def create(
        cls, 
        agent_type: str,
        goal: str,
        sandbox: "Sandbox",
        tools: List[str] = None
    ) -> "SubAgent":
        """创建子代理实例"""
        
        agent_class = cls.AGENT_REGISTRY.get(agent_type)
        if not agent_class:
            raise ValueError(f"Unknown agent type: {agent_type}")
        
        # 创建专属沙箱环境
        agent_sandbox = await sandbox.create_isolated_environment()
        
        # 分配工具权限
        allowed_tools = tools or cls._get_default_tools(agent_type)
        
        return agent_class(
            goal=goal,
            sandbox=agent_sandbox,
            tools=allowed_tools
        )

class CoderAgent(SubAgent):
    """代码执行子代理"""
    
    async def run(self):
        """执行代码任务"""
        # 1. 生成代码
        code = await self._generate_code()
        
        # 2. 在沙箱中执行
        result = await self.sandbox.execute_python(code)
        
        # 3. 验证输出
        if result["exit_code"] != 0:
            # 错误处理：自动调试
            fixed_code = await self._debug_and_fix(code, result["stderr"])
            result = await self.sandbox.execute_python(fixed_code)
        
        return {
            "code": code,
            "output": result["stdout"],
            "files_created": result.get("files", [])
        }

2.4 并行调度与依赖管理

DeerFlow 的一个重要能力是并行子代理调度。对于没有依赖关系的子任务，系统会并行执行，显著缩短总任务时间。

依赖管理的实现：

# backend/agents/scheduler.py
import asyncio
from typing import List, Dict, Set
from collections import defaultdict

class SubAgentScheduler:
    """子代理调度器：处理依赖关系和并行执行"""
    
    def __init__(self, max_parallel: int = 4):
        self.max_parallel = max_parallel
    
    async def execute_plan(self, plan: List[dict]) -> Dict[str, Any]:
        """执行任务计划"""
        
        # 构建依赖图
        dependency_graph = self._build_dependency_graph(plan)
        
        # 拓扑排序
        execution_order = self._topological_sort(dependency_graph)
        
        results = {}
        running_tasks = []
        
        for level in execution_order:
            # 同一层级的任务可并行执行
            level_tasks = [
                self._execute_subtask(subtask, results)
                for subtask in level
            ]
            
            # 限制并发数
            while len(running_tasks) >= self.max_parallel:
                done, running_tasks = await asyncio.wait(
                    running_tasks,
                    return_when=asyncio.FIRST_COMPLETED
                )
                for task in done:
                    subtask_id, result = await task
                    results[subtask_id] = result
            
            # 启动新的并行任务
            for task in level_tasks:
                running_tasks.append(asyncio.create_task(task))
        
        # 等待所有任务完成
        if running_tasks:
            await asyncio.gather(*running_tasks)
        
        return results
    
    def _build_dependency_graph(self, plan: List[dict]) -> Dict[str, Set[str]]:
        """构建依赖关系图"""
        graph = defaultdict(set)
        for subtask in plan:
            task_id = subtask["id"]
            for dep_id in subtask.get("depends_on", []):
                graph[task_id].add(dep_id)
        return graph

性能实测：在一个典型的「市场研究 + 报告生成」任务中：

• 串行执行：约 12 分钟
• 并行调度（4 个子代理）：约 3.5 分钟
• 效率提升：约 3.4 倍

三、Skills 系统：用 Markdown 定义工作流

3.1 为什么是 Markdown？

DeerFlow 的 Skills 系统是其最具争议也最具创新性的设计之一。传统的 Agent 框架通常用 Python 代码定义能力，而 DeerFlow 选择用 Markdown 文件定义工作流。

这个设计背后的理念：

1. 可读性优先：Markdown 天然可读，便于人类审查和修改
2. 版本友好：纯文本格式，天然支持 Git 版本管理
3. 渐进加载：按需加载 Skills，不占用上下文窗口
4. 跨平台复用：同一份 Markdown 可以在不同 Agent 实例间共享

3.2 Skills 文件结构

一个典型的 Skill 定义文件（SKILL.md）：

---
name: deep-research
version: 2.0
description: 深度研究能力，支持多源搜索、信息整合和报告生成
author: deerflow-team
tags: [research, web-search, report]
tools: [web_search, browser, file_write, memory]
---

# Deep Research Skill

## 目标
对给定主题进行深度研究，生成结构化报告。

## 工作流程

### Phase 1: 信息搜集
1. 使用 `web_search` 工具搜索相关信息
2. 使用 `browser` 工具访问关键页面
3. 提取并结构化关键信息

### Phase 2: 信息整合
1. 使用 `memory` 工具存储关键发现
2. 识别信息间的关联和矛盾
3. 构建知识图谱

### Phase 3: 报告生成
1. 根据模板生成结构化报告
2. 包含引用来源
3. 导出为 Markdown 格式

## 最佳实践

- 每个搜索关键词不超过 5 个单词
- 优先访问权威来源（官方文档、学术论文）
- 记录所有引用来源，便于追溯

## 示例

用户输入: "分析 Rust 在系统编程中的优势"

预期输出:
- Rust vs C/C++ 性能对比
- 内存安全机制分析
- 社区生态现状
- 企业采用案例

3.3 Skills 加载机制

DeerFlow 的 Skills 按需渐进加载，不会一次性占用大量上下文。核心加载逻辑：

# backend/skills/loader.py
import yaml
from pathlib import Path
from typing import Dict, Any, Optional

class SkillLoader:
    """Skills 加载器：按需加载和缓存"""
    
    def __init__(self, skills_dir: Path):
        self.skills_dir = skills_dir
        self.cache: Dict[str, dict] = {}
        self.loaded_skills: set = set()
    
    async def load_skill(self, skill_name: str) -> dict:
        """加载指定 Skill"""
        
        # 缓存命中
        if skill_name in self.cache:
            return self.cache[skill_name]
        
        # 读取 Skill 文件
        skill_path = self.skills_dir / skill_name / "SKILL.md"
        if not skill_path.exists():
            raise FileNotFoundError(f"Skill not found: {skill_name}")
        
        # 解析 YAML front matter + Markdown body
        skill_data = await self._parse_skill_file(skill_path)
        
        # 缓存
        self.cache[skill_name] = skill_data
        self.loaded_skills.add(skill_name)
        
        return skill_data
    
    async def _parse_skill_file(self, path: Path) -> dict:
        """解析 Skill 文件"""
        content = await path.read_text()
        
        # 提取 YAML front matter
        if content.startswith("---"):
            parts = content.split("---", 2)
            if len(parts) >= 3:
                metadata = yaml.safe_load(parts[1])
                body = parts[2].strip()
                
                return {
                    "metadata": metadata,
                    "content": body,
                    "tools": metadata.get("tools", []),
                    "workflow": self._extract_workflow(body)
                }
        
        return {"content": content}
    
    def _extract_workflow(self, markdown: str) -> list:
        """从 Markdown 中提取工作流步骤"""
        steps = []
        lines = markdown.split("\n")
        
        current_phase = None
        for line in lines:
            # 识别 Phase 标题
            if line.startswith("### Phase"):
                current_phase = line.replace("###", "").strip()
                steps.append({"phase": current_phase, "actions": []})
            # 识别数字列表项
            elif line.strip().startswith(("1.", "2.", "3.", "4.", "5.")):
                if steps:
                    action = line.strip().split(".", 1)[1].strip()
                    steps[-1]["actions"].append(action)
        
        return steps

3.4 内置 Skills 详解

DeerFlow 2.0 内置以下 Skills：

skills/public/
├── research/SKILL.md           # 深度研究
├── report-generation/SKILL.md  # 报告生成
├── slide-creation/SKILL.md     # PPT 制作
├── web-page/SKILL.md           # 网页生成
└── image-generation/SKILL.md   # 图片/视频生成

research/SKILL.md 核心片段：

# 示例：如何在代码中使用 Skill
async def execute_research_skill(agent, topic: str):
    """执行深度研究 Skill"""
    
    # 1. 加载 Skill
    skill = await agent.skill_loader.load_skill("research")
    
    # 2. 解析工作流
    workflow = skill["workflow"]
    
    # 3. 按阶段执行
    results = {}
    for phase in workflow:
        phase_name = phase["phase"]
        phase_results = []
        
        for action in phase["actions"]:
            # 解析动作中的工具调用
            tool_calls = parse_tool_calls(action)
            
            # 执行工具
            for tool_name, params in tool_calls:
                result = await agent.tools[tool_name].execute(params)
                phase_results.append(result)
        
        results[phase_name] = phase_results
    
    return results

3.5 自定义 Skill 示例

假设你需要创建一个「技术文档翻译」Skill：

---
name: tech-translate
version: 1.0
description: 技术文档翻译，保持专业术语一致性
tools: [file_read, file_write, llm, terminology_db]
---

# Technical Document Translation

## 目标
翻译技术文档，确保专业术语一致性。

## 工作流程

### Phase 1: 文档分析
1. 使用 `file_read` 读取源文档
2. 识别文档类型（API 文档、教程、规范）
3. 提取专业术语列表

### Phase 2: 术语映射
1. 查询 `terminology_db` 获取标准翻译
2. 对于未记录术语，使用 `llm` 生成建议翻译
3. 构建术语映射表

### Phase 3: 翻译执行
1. 按段落翻译正文
2. 保持代码块不变
3. 更新内部链接

### Phase 4: 质量检查
1. 检查术语一致性
2. 验证链接有效性
3. 生成翻译报告

## 术语处理规则

- API → 保持原样
- Endpoint → 端点
- Request → 请求
- Response → 响应
- Webhook → Webhook（不翻译）

四、沙箱执行环境：三级隔离的安全护城河

4.1 为什么沙箱至关重要？

让 AI Agent 执行代码，最大的风险是什么？答案显而易见：不可控的代码执行。

假设 Agent 生成了这样的代码：

import os
# 删除所有文件
for root, dirs, files in os.walk("/"):
    for f in files:
        os.remove(os.path.join(root, f))

如果没有隔离环境，这段代码会直接在你的机器上执行，后果不堪设想。

DeerFlow 的解决方案是原生 Docker 沙箱，采用三级隔离模型：

1. 容器级隔离：每个子代理在独立的 Docker 容器中执行
2. 用户级隔离：容器内使用非特权用户运行
3. 网络级隔离：限制网络访问，仅允许出站 HTTP/HTTPS

4.2 容器级隔离：物理边界的划定

DeerFlow 的整个服务，包括编码员运行环境，都部署在独立的 Docker 容器中。这个容器经过定制加固：

# docker-compose.yml
services:
  deerflow-app:
    image: deerflow:latest
    container_name: deerflow-app
    security_opt:
      - no-new-privileges:true
    cap_drop:
      - ALL
    cap_add:
      - CHOWN
      - SETUID
      - SETGID
    read_only: true  # 只读根文件系统
    tmpfs:
      - /tmp:size=100M,mode=1777
      - /run:size=10M,mode=1777
    volumes:
      - ./workspace:/workspace:rw  # 唯一可写区域
      - ./config:/config:ro
    networks:
      - deerflow-internal
    environment:
      - SANDBOX_MODE=strict

networks:
  deerflow-internal:
    internal: true  # 禁止外部访问

安全配置验证命令：

# 检查容器安全配置
docker inspect deerflow-app | jq '.[0].HostConfig | {
  ReadOnlyRootfs,
  CapDrop,
  CapAdd,
  SecurityOpt,
  NetworkMode
}'

4.3 E2B 集成：企业级沙箱方案

DeerFlow 2.0 集成了 E2B（企业级沙箱服务），提供更强的隔离能力：

# backend/sandbox/e2b_sandbox.py
from e2b import Sandbox

class E2BSandbox:
    """E2B 沙箱集成"""
    
    def __init__(self, api_key: str):
        self.client = Sandbox(api_key=api_key)
    
    async def create_environment(self) -> str:
        """创建隔离执行环境"""
        sandbox = await self.client.create(
            template="base",
            timeout=300,  # 5 分钟超时
            metadata={"session_id": self.session_id}
        )
        return sandbox.id
    
    async def execute_python(
        self, 
        code: str, 
        timeout: int = 60
    ) -> dict:
        """执行 Python 代码"""
        result = await self.client.run_code(
            language="python",
            code=code,
            timeout=timeout
        )
        
        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "exit_code": result.exit_code,
            "files": result.files
        }
    
    async def execute_shell(
        self, 
        command: str, 
        timeout: int = 30
    ) -> dict:
        """执行 Shell 命令"""
        result = await self.client.run_command(
            command=command,
            timeout=timeout
        )
        
        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "exit_code": result.exit_code
        }

E2B vs 本地 Docker 沙箱对比：

4.4 Python 代码执行的沙箱实现

# backend/sandbox/python_sandbox.py
import ast
import subprocess
import tempfile
from pathlib import Path
from typing import Optional, List

class PythonSandbox:
    """Python 代码沙箱执行器"""
    
    # 允许的标准库模块
    ALLOWED_MODULES = {
        'os', 'sys', 'json', 're', 'math', 'datetime',
        'collections', 'itertools', 'functools', 'typing',
        'pandas', 'numpy', 'requests', 'beautifulsoup4'
    }
    
    # 禁止的危险函数
    FORBIDDEN_FUNCTIONS = {
        'eval', 'exec', 'compile', 'open',
        'os.system', 'subprocess.call', 'subprocess.run'
    }
    
    def __init__(self, sandbox_dir: Path, timeout: int = 60):
        self.sandbox_dir = sandbox_dir
        self.timeout = timeout
    
    async def execute(
        self, 
        code: str, 
        files: dict = None
    ) -> dict:
        """执行 Python 代码"""
        
        # 1. AST 安全检查
        violations = self._check_security(code)
        if violations:
            return {
                "success": False,
                "error": f"Security violation: {violations}",
                "stdout": "",
                "stderr": ""
            }
        
        # 2. 创建临时执行目录
        exec_dir = self._create_exec_environment(files)
        
        # 3. 写入代码文件
        code_file = exec_dir / "main.py"
        code_file.write_text(code)
        
        # 4. 在隔离环境中执行
        try:
            result = subprocess.run(
                ["python3", str(code_file)],
                cwd=str(exec_dir),
                capture_output=True,
                timeout=self.timeout,
                env={
                    "PYTHONPATH": str(exec_dir),
                    "PYTHONIOENCODING": "utf-8"
                }
            )
            
            return {
                "success": result.returncode == 0,
                "stdout": result.stdout.decode("utf-8"),
                "stderr": result.stderr.decode("utf-8"),
                "exit_code": result.returncode,
                "output_files": self._collect_output_files(exec_dir)
            }
        
        except subprocess.TimeoutExpired:
            return {
                "success": False,
                "error": f"Execution timeout ({self.timeout}s)",
                "stdout": "",
                "stderr": ""
            }
        
        finally:
            # 清理临时目录
            self._cleanup(exec_dir)
    
    def _check_security(self, code: str) -> List[str]:
        """AST 安全检查"""
        violations = []
        
        try:
            tree = ast.parse(code)
            
            for node in ast.walk(tree):
                # 检查 import 语句
                if isinstance(node, ast.Import):
                    for alias in node.names:
                        if alias.name not in self.ALLOWED_MODULES:
                            violations.append(f"Forbidden import: {alias.name}")
                
                # 检查 from ... import 语句
                elif isinstance(node, ast.ImportFrom):
                    if node.module not in self.ALLOWED_MODULES:
                        violations.append(f"Forbidden import from: {node.module}")
                
                # 检查危险函数调用
                elif isinstance(node, ast.Call):
                    func_name = self._get_func_name(node.func)
                    if func_name in self.FORBIDDEN_FUNCTIONS:
                        violations.append(f"Forbidden function: {func_name}")
        
        except SyntaxError as e:
            violations.append(f"Syntax error: {e}")
        
        return violations
    
    def _get_func_name(self, node) -> str:
        """提取函数名"""
        if isinstance(node, ast.Name):
            return node.id
        elif isinstance(node, ast.Attribute):
            return f"{self._get_func_name(node.value)}.{node.attr}"
        return ""

4.5 沙箱执行的最佳实践

在生产环境中使用 DeerFlow 沙箱，需要注意：

1. 资源限制：为每个沙箱设置 CPU/内存上限
2. 超时控制：所有操作必须有超时限制
3. 日志审计：记录所有沙箱内的操作
4. 文件隔离：不同会话的文件存储在不同目录

# config.yaml
sandbox:
  mode: "strict"  # strict | permissive
  timeout:
    default: 60
    python: 300
    shell: 30
  resources:
    cpu_limit: "2.0"
    memory_limit: "4Gi"
  network:
    allow_outbound: true
    allowed_domains:
      - "api.openai.com"
      - "api.anthropic.com"
    deny_private_ips: true
  filesystem:
    root_readonly: true
    writable_dirs:
      - "/workspace"
      - "/tmp"

五、记忆系统：Long/Short-term Memory 的工程实现

5.1 为什么 Agent 需要「记忆」？

传统的 AI Agent 有一个致命缺陷：上下文窗口限制。一旦对话超过模型支持的 Token 数量，早期的信息就会被截断。这导致：

• 任务执行到一半，「忘记」了最初的目标
• 用户偏好无法持久化，每次都要重新说明
• 跨会话的知识无法复用

DeerFlow 2.0 引入了完整的记忆系统，分为三层：

1. Working Memory（工作记忆）：当前会话的短期记忆，存储在上下文中
2. Session Memory（会话记忆）：单次任务执行过程中的记忆，存储在数据库
3. Long-term Memory（长期记忆）：持久化的知识存储，支持跨会话检索

5.2 记忆存储架构

┌─────────────────────────────────────────────────────────┐
│                   Memory Manager                         │
├─────────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐     │
│  │Working      │  │Session      │  │Long-term    │     │
│  │Memory       │  │Memory       │  │Memory       │     │
│  │(Context)    │  │(SQLite)     │  │(Vector DB)  │     │
│  └─────────────┘  └─────────────┘  └─────────────┘     │
│         ↓                ↓                ↓              │
│  ┌─────────────────────────────────────────────────┐   │
│  │              Memory Retrieval API                │   │
│  │  retrieve(query, k=5, scope="all")              │   │
│  │  store(key, value, scope="long_term")           │   │
│  │  forget(key, scope="session")                   │   │
│  └─────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────┘

5.3 长期记忆的实现

# backend/memory/long_term_memory.py
from typing import List, Dict, Any, Optional
import chromadb
from chromadb.config import Settings

class LongTermMemory:
    """长期记忆：向量数据库实现"""
    
    def __init__(self, persist_dir: str, embedding_model: str):
        self.client = chromadb.PersistentClient(
            path=persist_dir,
            settings=Settings(
                anonymized_telemetry=False,
                allow_reset=True
            )
        )
        
        self.collection = self.client.get_or_create_collection(
            name="deerflow_memory",
            metadata={"hnsw:space": "cosine"}
        )
        
        self.embedding_model = embedding_model
    
    async def store(
        self, 
        content: str, 
        metadata: dict = None,
        memory_id: str = None
    ) -> str:
        """存储记忆"""
        
        # 生成 embedding
        embedding = await self._get_embedding(content)
        
        # 生成唯一 ID
        if not memory_id:
            memory_id = self._generate_id(content)
        
        # 存入向量数据库
        self.collection.add(
            ids=[memory_id],
            embeddings=[embedding],
            documents=[content],
            metadatas=[metadata or {}]
        )
        
        return memory_id
    
    async def retrieve(
        self, 
        query: str, 
        k: int = 5,
        filter: dict = None
    ) -> List[Dict[str, Any]]:
        """检索相关记忆"""
        
        # 生成查询 embedding
        query_embedding = await self._get_embedding(query)
        
        # 向量检索
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=k,
            where=filter
        )
        
        # 格式化结果
        memories = []
        for i, doc in enumerate(results["documents"][0]):
            memories.append({
                "id": results["ids"][0][i],
                "content": doc,
                "metadata": results["metadatas"][0][i],
                "distance": results["distances"][0][i]
            })
        
        return memories
    
    async def forget(self, memory_id: str) -> bool:
        """删除记忆"""
        try:
            self.collection.delete(ids=[memory_id])
            return True
        except Exception as e:
            print(f"Failed to forget memory: {e}")
            return False
    
    async def _get_embedding(self, text: str) -> List[float]:
        """获取文本 embedding"""
        # 实际实现中调用 embedding API
        # 这里简化处理
        from openai import AsyncOpenAI
        client = AsyncOpenAI()
        
        response = await client.embeddings.create(
            model=self.embedding_model,
            input=text
        )
        
        return response.data[0].embedding

5.4 记忆管理策略

DeerFlow 的记忆管理策略包括：

1. 重要性评估：根据信息的重要性决定存储层级
2. 时效性衰减：旧记忆的权重随时间降低
3. 主题聚类：相关记忆自动聚类，便于检索
4. 冲突检测：检测记忆中的矛盾信息

# backend/memory/manager.py
from datetime import datetime
from typing import List, Dict, Any

class MemoryManager:
    """记忆管理器"""
    
    def __init__(
        self, 
        working_memory: dict,
        session_memory: "SessionMemory",
        long_term_memory: "LongTermMemory"
    ):
        self.working_memory = working_memory
        self.session_memory = session_memory
        self.long_term_memory = long_term_memory
    
    async def store_fact(
        self, 
        fact: str, 
        importance: float = 0.5,
        metadata: dict = None
    ):
        """存储事实信息"""
        
        # 根据重要性决定存储层级
        if importance >= 0.8:
            # 高重要性：长期记忆
            await self.long_term_memory.store(
                content=fact,
                metadata={
                    **(metadata or {}),
                    "importance": importance,
                    "timestamp": datetime.now().isoformat()
                }
            )
        
        elif importance >= 0.5:
            # 中等重要性：会话记忆
            await self.session_memory.store(fact, metadata)
        
        else:
            # 低重要性：工作记忆
            self.working_memory["facts"].append(fact)
    
    async def retrieve_relevant(
        self, 
        query: str, 
        scope: str = "all",
        max_tokens: int = 4000
    ) -> str:
        """检索相关记忆"""
        
        memories = []
        total_tokens = 0
        
        # 1. 从长期记忆检索
        if scope in ["all", "long_term"]:
            long_term = await self.long_term_memory.retrieve(
                query, k=5
            )
            memories.extend(long_term)
        
        # 2. 从会话记忆检索
        if scope in ["all", "session"]:
            session = await self.session_memory.retrieve(
                query, k=10
            )
            memories.extend(session)
        
        # 3. 从工作记忆检索
        if scope in ["all", "working"]:
            working = self._search_working_memory(query)
            memories.extend(working)
        
        # 4. 去重和排序
        memories = self._deduplicate(memories)
        memories = self._rank_by_relevance(memories, query)
        
        # 5. 截断到 token 限制
        result = []
        for mem in memories:
            mem_tokens = len(mem["content"]) // 4  # 粗略估计
            if total_tokens + mem_tokens > max_tokens:
                break
            result.append(mem)
            total_tokens += mem_tokens
        
        return self._format_memories(result)
    
    def _rank_by_relevance(
        self, 
        memories: List[dict], 
        query: str
    ) -> List[dict]:
        """按相关性排序"""
        # 综合考虑：
        # 1. 向量相似度
        # 2. 时间衰减
        # 3. 重要性权重
        
        now = datetime.now()
        
        def score(mem):
            # 向量距离（越小越好）
            distance_score = 1 - mem.get("distance", 0.5)
            
            # 时间衰减
            timestamp = mem.get("metadata", {}).get("timestamp")
            if timestamp:
                age_hours = (now - datetime.fromisoformat(timestamp)).total_seconds() / 3600
                time_score = 1 / (1 + age_hours / 24)  # 每天衰减一半
            else:
                time_score = 0.5
            
            # 重要性
            importance = mem.get("metadata", {}).get("importance", 0.5)
            
            # 综合得分
            return distance_score * 0.5 + time_score * 0.3 + importance * 0.2
        
        return sorted(memories, key=score, reverse=True)

六、Message Gateway：多渠道消息集成

6.1 为什么需要 Message Gateway？

在实际生产场景中，用户可能从不同渠道发起任务：

• 在 Telegram 中发送研究请求
• 在 Slack 中触发自动化任务
• 在飞书中接收报告推送

DeerFlow 的 Message Gateway 提供统一的消息接入层，支持多种 IM 平台。

6.2 支持的消息渠道

6.3 Message Gateway 实现

# backend/channels/gateway.py
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
import asyncio

class MessageChannel(ABC):
    """消息渠道抽象基类"""
    
    @abstractmethod
    async def send_message(self, target: str, content: str, **kwargs):
        """发送消息"""
        pass
    
    @abstractmethod
    async def receive_message(self) -> Dict[str, Any]:
        """接收消息"""
        pass

class TelegramChannel(MessageChannel):
    """Telegram 渠道"""
    
    def __init__(self, bot_token: str):
        self.bot_token = bot_token
        self.base_url = f"https://api.telegram.org/bot{bot_token}"
    
    async def send_message(
        self, 
        chat_id: str, 
        content: str,
        parse_mode: str = "Markdown",
        **kwargs
    ):
        """发送消息到 Telegram"""
        import aiohttp
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/sendMessage",
                json={
                    "chat_id": chat_id,
                    "text": content,
                    "parse_mode": parse_mode,
                    **kwargs
                }
            ) as resp:
                return await resp.json()
    
    async def receive_message(self) -> Dict[str, Any]:
        """通过 Webhook 或长轮询接收消息"""
        # 实现 webhook 或 polling 逻辑
        pass

class MessageGateway:
    """消息网关：统一管理多个渠道"""
    
    def __init__(self):
        self.channels: Dict[str, MessageChannel] = {}
        self.message_handlers = []
    
    def register_channel(self, name: str, channel: MessageChannel):
        """注册消息渠道"""
        self.channels[name] = channel
    
    async def start(self):
        """启动所有渠道的监听"""
        tasks = []
        for name, channel in self.channels.items():
            task = asyncio.create_task(
                self._listen_channel(name, channel)
            )
            tasks.append(task)
        
        await asyncio.gather(*tasks)
    
    async def _listen_channel(self, name: str, channel: MessageChannel):
        """监听单个渠道的消息"""
        while True:
            try:
                message = await channel.receive_message()
                
                # 标准化消息格式
                normalized = self._normalize_message(name, message)
                
                # 分发到处理器
                for handler in self.message_handlers:
                    await handler(normalized)
            
            except Exception as e:
                print(f"Error in channel {name}: {e}")
                await asyncio.sleep(5)
    
    def _normalize_message(
        self, 
        channel_name: str, 
        raw_message: dict
    ) -> dict:
        """标准化消息格式"""
        # 不同渠道的消息格式不同，需要统一
        if channel_name == "telegram":
            return {
                "channel": channel_name,
                "user_id": str(raw_message["from"]["id"]),
                "chat_id": str(raw_message["chat"]["id"]),
                "content": raw_message.get("text", ""),
                "timestamp": raw_message["date"],
                "raw": raw_message
            }
        
        elif channel_name == "slack":
            return {
                "channel": channel_name,
                "user_id": raw_message["user"],
                "chat_id": raw_message["channel"],
                "content": raw_message.get("text", ""),
                "timestamp": raw_message["ts"],
                "raw": raw_message
            }
        
        # 其他渠道...
        return raw_message

# 使用示例
async def setup_gateway():
    gateway = MessageGateway()
    
    # 注册 Telegram 渠道
    telegram = TelegramChannel(bot_token="YOUR_BOT_TOKEN")
    gateway.register_channel("telegram", telegram)
    
    # 注册消息处理器
    async def handle_message(msg):
        print(f"Received from {msg['channel']}: {msg['content']}")
        # 分发给 Lead Agent 处理
        # ...
    
    gateway.message_handlers.append(handle_message)
    
    await gateway.start()

七、MCP 协议集成：标准化的工具生态

7.1 什么是 MCP？

MCP（Model Context Protocol）是由 Anthropic 推出的标准化协议，用于连接 AI 模型与外部工具。DeerFlow 2.0 完整支持 MCP 协议，这意味着：

• 可以接入任何 MCP 兼容的工具
• 可以复用现有的 MCP 生态系统
• 可以自定义 MCP 服务器扩展能力

7.2 MCP 集成示例

# backend/mcp/client.py
from typing import List, Dict, Any
import json

class MCPClient:
    """MCP 客户端"""
    
    def __init__(self, server_config: dict):
        self.server_config = server_config
        self.tools = {}
    
    async def connect(self, server_name: str):
        """连接到 MCP 服务器"""
        config = self.server_config[server_name]
        
        # 根据 transport 类型初始化连接
        if config["transport"] == "stdio":
            await self._connect_stdio(config)
        elif config["transport"] == "http":
            await self._connect_http(config)
        
        # 获取可用工具列表
        tools = await self._list_tools()
        self.tools[server_name] = tools
    
    async def call_tool(
        self, 
        server_name: str, 
        tool_name: str, 
        arguments: dict
    ) -> Any:
        """调用 MCP 工具"""
        # 构建请求
        request = {
            "jsonrpc": "2.0",
            "method": "tools/call",
            "params": {
                "name": tool_name,
                "arguments": arguments
            },
            "id": self._generate_id()
        }
        
        # 发送请求
        response = await self._send_request(server_name, request)
        
        return response.get("result")
    
    async def _list_tools(self) -> List[dict]:
        """获取工具列表"""
        request = {
            "jsonrpc": "2.0",
            "method": "tools/list",
            "params": {},
            "id": self._generate_id()
        }
        
        response = await self._send_request(request)
        return response.get("result", {}).get("tools", [])

# 配置示例
mcp_config = {
    "filesystem": {
        "transport": "stdio",
        "command": "mcp-filesystem",
        "args": ["/workspace"]
    },
    "brave-search": {
        "transport": "http",
        "url": "https://api.brave.com/mcp",
        "headers": {
            "Authorization": "Bearer YOUR_API_KEY"
        }
    }
}

八、实战：从零构建一个 DeerFlow 应用

8.1 环境准备

# 克隆仓库
git clone https://github.com/bytedance/deer-flow.git
cd deer-flow

# 安装依赖（推荐使用 uv）
uv sync

# 复制配置文件
cp .env.example .env
cp config.example.yaml config.yaml

# 编辑配置
vim config.yaml

8.2 配置 LLM

# config.yaml
models:
  - name: deepseek-v3
    display_name: DeepSeek V3
    use: langchain_openai:ChatOpenAI
    model: deepseek-chat
    api_key: $DEEPSEEK_API_KEY
    base_url: https://api.deepseek.com/v1
  
  - name: doubao-seed
    display_name: Doubao Seed 2.0
    use: langchain_openai:ChatOpenAI
    model: doubao-seed-2.0
    api_key: $DOUBAO_API_KEY
    base_url: https://api.doubao.com/v1

# 默认模型
default_model: deepseek-v3

8.3 配置搜索服务

# config.yaml
search:
  provider: "tavily"  # tavily | brave | duckduckgo | infoquest
  tavily_api_key: $TAVILY_API_KEY
  
  # 备选搜索
  fallback:
    provider: "duckduckgo"

8.4 配置沙箱

# config.yaml
sandbox:
  mode: "docker"  # docker | e2b | local
  
  docker:
    image: "deerflow/sandbox:latest"
    cpu_limit: "2.0"
    memory_limit: "4Gi"
    timeout: 300
  
  e2b:
    api_key: $E2B_API_KEY
    template: "base"

8.5 运行

# 方式一：控制台模式
uv run main.py

# 方式二：Web UI
./bootstrap.sh -d
# 访问 http://localhost:3000

8.6 自定义 Skill 实战

创建一个「技术调研报告生成」Skill：

mkdir -p skills/public/tech-research

<!-- skills/public/tech-research/SKILL.md -->
---
name: tech-research
version: 1.0
description: 技术调研报告生成，包含技术背景、实现方案、对比分析
tools: [web_search, browser, python_executor, file_write, memory]
---

# Technical Research Report Generator

## 目标
对给定技术主题进行深度调研，生成结构化技术报告。

## 工作流程

### Phase 1: 信息搜集
1. 使用 `web_search` 搜索技术背景和原理
2. 使用 `browser` 访问官方文档和 GitHub 仓库
3. 提取关键特性、架构设计、性能数据

### Phase 2: 竞品对比
1. 识别主要竞品/替代方案
2. 从以下维度对比：
   - 核心能力
   - 性能指标
   - 社区活跃度
   - 企业采用情况
3. 生成对比表格

### Phase 3: 实践验证
1. 使用 `python_executor` 运行示例代码
2. 记录环境配置、运行结果、遇到的问题
3. 截图关键步骤

### Phase 4: 报告生成
1. 使用 `file_write` 生成 Markdown 报告
2. 包含：
   - 技术概述
   - 核心原理
   - 快速上手
   - 最佳实践
   - 踩坑记录
   - 参考资料
3. 使用 `memory` 存储关键发现

## 报告模板

\`\`\`markdown
# {技术名称} 技术调研报告

## 一、技术概述

### 1.1 什么是 {技术名称}
### 1.2 核心特性
### 1.3 适用场景

## 二、核心原理

### 2.1 架构设计
### 2.2 关键技术
### 2.3 性能特点

## 三、快速上手

### 3.1 环境准备
### 3.2 Hello World
### 3.3 进阶示例

## 四、竞品对比

| 维度 | {技术} | 竞品A | 竞品B |
|------|--------|-------|-------|
| ... | ... | ... | ... |

## 五、最佳实践

## 六、踩坑记录

## 七、总结

## 八、参考资料
\`\`\`

九、生产踩坑清单

基于社区反馈和实践经验，整理以下踩坑清单：

9.1 环境配置

• Python 版本必须 >= 3.12（使用了新的语法特性）
• Node.js 版本必须 >= 22（Web UI 依赖）
• Docker 版本必须 >= 24.0（支持新的安全特性）
• 配置文件中的 API Key 使用环境变量，不要硬编码

9.2 沙箱安全

• 生产环境必须使用 sandbox.mode: docker 或 e2b
• 设置 read_only: true 防止容器内文件被篡改
• 配置 network.internal: true 禁止容器访问外网
• 定期更新沙箱镜像，修复安全漏洞

9.3 性能优化

• 使用 uv 管理 Python 依赖，比 pip 快 10 倍
• 配置 max_parallel 限制并发子代理数量（建议 4-8）
• 启用记忆压缩，减少上下文占用
• 使用 light_context 模式减少 token 消耗

9.4 错误处理

• 所有工具调用都要有 timeout
• 子代理失败时，要有重试机制（最多 3 次）
• 记录完整的执行日志，便于调试
• 使用 make support-bundle 生成诊断包

9.5 成本控制

• 配置模型定价，实时监控成本
• 使用 subagents.max_total_per_run 限制子代理数量
• 启用 token 预算，超支自动终止
• 优先使用性价比高的模型（DeepSeek、Doubao）

十、与主流框架对比

10.1 DeerFlow vs OpenClaw vs AutoGen vs CrewAI

10.2 选型建议

• 选择 DeerFlow：如果你需要长时程任务、动态子代理编排、原生沙箱隔离
• 选择 OpenClaw：如果你偏向个人 AI 工作站体验、快速上手
• 选择 AutoGen：如果你需要围绕多智能体对话做深度开发
• 选择 CrewAI：如果你希望快速搭建角色化流程

十一、总结与展望

11.1 DeerFlow 的核心价值

DeerFlow 2.0 不是一个简单的「Deep Research 工具」，而是一套完整的 Super Agent 运行时基础设施。它的核心价值在于：

1. 架构创新：Lead Agent + Sub-Agents 动态编排，实现复杂任务的自动化分解和并行执行
2. 安全可信：原生 Docker 沙箱 + E2B 集成，提供企业级的执行隔离
3. 易于扩展：Markdown 定义 Skills，降低能力扩展门槛
4. 记忆持久：Long/Short-term Memory 系统，解决上下文丢失问题
5. 生态开放：MCP 协议集成，复用现有工具生态

11.2 未来展望

DeerFlow 团队在路线图中提到了以下方向：

• 更强的推理能力：集成更强的推理模型，支持复杂决策
• 更丰富的 Skills 生态：社区贡献更多垂直领域 Skills
• 更完善的企业特性：权限管理、审计日志、成本控制
• 更友好的开发者体验：可视化编排、调试工具、性能分析

11.3 写在最后

如果你正在评估 AI Agent 框架，或者想要构建自己的 Super Agent 应用，DeerFlow 2.0 是一个值得深入研究的对象。它不仅有字节跳动的大厂背书，更重要的是，它展示了一套完整的、可落地的 Super Agent 架构范式。

从 Deep Research 工具到 Super Agent 运行时，DeerFlow 2.0 完成了一次质的飞跃。而这，可能只是开始。

参考资料

• DeerFlow GitHub 仓库
• DeerFlow 官网
• DeerFlow 中文站
• LangGraph 文档
• MCP 协议规范
• E2B 文档