DeerFlow 2.0 深度解析：从 Deep Research 到超级智能体运行时，52K Star 背后的架构设计与工程实践

字节跳动开源的 DeerFlow 2.0，30 天斩获 52K Star 登顶 GitHub Trending。这不是又一个套壳 ChatGPT，而是一套完整的 Agent 执行运行时——有沙箱、有文件系统、有记忆、有子智能体调度。本文从架构设计、核心模块源码级分析到实战部署，带你彻底搞懂它。

一、为什么需要 DeerFlow：AI Agent 的「最后一公里」

1.1 从 ChatGPT 到 Agent：范式转变已经发生

2023 年，所有人都在和 ChatGPT 聊天。2024 年，AutoGPT、LangChain、CrewAI 让 Agent 概念火了一把。2025 年，OpenAI 的 Deep Research 让人们第一次看到 AI 真正「干活」的可能。但问题也暴露了：

• 只会说，不会做：大多数 Agent 能生成代码，但没法真正执行
• 没有记忆：每次对话都从零开始，之前学的全忘了
• 长任务崩溃：超过 10 分钟的复杂任务，上下文就炸了
• 单线程思维：一个 Agent 串行处理所有子任务，效率低下

DeerFlow 2.0 直接瞄准了这四个痛点。它的定位不是「又一个聊天框架」，而是 super agent harness——一个让 Agent 真正能完成复杂工作的运行时基础设施。

1.2 DeerFlow 1.0 到 2.0：从研究助手到执行底座

DeerFlow 1.0 基于 LangGraph 构建了一个固定的 5 节点多智能体架构，定位是「深度研究助手」——帮你查资料、写报告。但它有几个硬伤：

一句话概括：1.0 是「帮你查资料写报告」，2.0 是「给你一台带 AI 大脑的电脑，让它自己干完整项目」。

而更关键的是——DeerFlow 2.0 是一次彻底重写，和 v1 没有共用任何代码。这不是渐进式迭代，这是推倒重来。

二、核心架构：Lead Agent + Middleware Chain + Sub-Agents

2.1 整体架构图

DeerFlow 2.0 的架构可以用一个公式概括：

Super Agent = Lead Agent + Middleware Chain + Dynamic Sub-Agents + Sandbox + Memory + Skills

更具体地说：

┌─────────────────────────────────────────────────────────┐
│                     Lead Agent                           │
│  ┌────────────────────────────────────────────────────┐ │
│  │              11-Layer Middleware Chain               │ │
│  │  ┌──────┐ ┌──────┐ ┌──────┐       ┌──────┐       │ │
│  │  │Skill │ │Plan  │ │Sandbox│  ...  │Memory│       │ │
│  │  │Loader│ │Parser│ │Manager│       │Manager│      │ │
│  │  └──────┘ └──────┘ └──────┘       └──────┘       │ │
│  └────────────────────────────────────────────────────┘ │
│                          │                               │
│              ┌───────────┼───────────┐                   │
│              ▼           ▼           ▼                   │
│        ┌──────────┐┌──────────┐┌──────────┐             │
│        │Sub-Agent ││Sub-Agent ││Sub-Agent │             │
│        │(Search)  ││(Analyze) ││(Generate)│             │
│        └──────────┘└──────────┘└──────────┘             │
│              │           │           │                   │
│              ▼           ▼           ▼                   │
│        ┌──────────────────────────────────────┐         │
│        │         Context Engine                 │         │
│        │  (上下文压缩 + 摘要 + 持久化)          │         │
│        └──────────────────────────────────────┘         │
│                          │                               │
│              ┌───────────┼───────────┐                   │
│              ▼           ▼           ▼                   │
│     ┌─────────────┐ ┌──────────┐ ┌──────────┐          │
│     │Docker       │ │File      │ │Long-term │          │
│     │Sandbox + FS │ │Operations│ │Memory    │          │
│     └─────────────┘ └──────────┘ └──────────┘          │
│                          │                               │
│              ┌───────────┼───────────┐                   │
│              ▼           ▼           ▼                   │
│     ┌─────────────┐ ┌──────────┐ ┌──────────┐          │
│     │Skills/Tools │ │MCP Server│ │IM Channel│          │
│     └─────────────┘ └──────────┘ └──────────┘          │
└─────────────────────────────────────────────────────────┘

2.2 Lead Agent：任务规划与调度的核心大脑

Lead Agent 是整个系统的调度中心。它不直接执行具体任务，而是做三件事：

1. 理解用户意图：解析用户的自然语言输入
2. 任务拆解：将复杂任务分解为可执行的子任务
3. 动态调度：根据子任务类型，决定是自行处理还是拉起 Sub-Agent

这种设计的精妙之处在于：新增能力时只需添加新 Skill，无需改动底层框架。2.0 把 1.0 的固定节点架构换成了 11 层中间件链，每一层处理不同维度的任务需求：

# 简化的中间件链概念（基于源码理解）
class MiddlewareChain:
    """11层中间件链，按顺序处理每个任务"""
    
    def __init__(self):
        self.middlewares = [
            SkillLoaderMiddleware(),      # 1. 按需加载技能
            PlanParserMiddleware(),        # 2. 解析任务规划
            SubAgentDispatcherMiddleware(), # 3. 子智能体调度
            SandboxManagerMiddleware(),    # 4. 沙箱环境管理
            MemoryMiddleware(),            # 5. 记忆读写
            ContextCompressorMiddleware(), # 6. 上下文压缩
            ToolExecutorMiddleware(),      # 7. 工具执行
            MCPRouterMiddleware(),         # 8. MCP 协议路由
            OutputFormatterMiddleware(),   # 9. 输出格式化
            SafetyCheckerMiddleware(),     # 10. 安全检查
            ResultAggregatorMiddleware(),  # 11. 结果汇总
        ]
    
    async def process(self, task: Task) -> Result:
        context = TaskContext(task)
        for middleware in self.middlewares:
            context = await middleware.process(context)
            if context.should_terminate:
                break
        return context.result

2.3 Sub-Agents：并行执行，隔离上下文

这是 DeerFlow 2.0 最大的架构亮点。Lead Agent 可以按需动态拉起 Sub-Agents，每个 Sub-Agent 都有：

• 独立的上下文：看不到主 Agent 的上下文，也看不到其他 Sub-Agents 的上下文
• 独立的工具集：根据子任务类型加载对应的 Skills 和 Tools
• 独立的终止条件：完成子任务后自动退出

工作流如下：

用户指令 → 任务拆解 → Sub-Agent A (搜索) + Sub-Agent B (分析) + Sub-Agent C (生成)
                        ↓                    ↓                    ↓
                      结果汇总 → 最终输出

为什么要隔离上下文？这是 DeerFlow 团队在 1.0 中踩过的坑：当所有 Agent 共享上下文时，信息膨胀极快，一个长任务下来上下文窗口直接打爆。隔离后，每个 Sub-Agent 只聚焦当前任务，执行效率提升 3-5 倍。

# Sub-Agent 的动态创建与调度（概念代码）
class SubAgentDispatcher:
    """动态创建和调度子智能体"""
    
    def __init__(self, langgraph_client):
        self.client = langgraph_client
    
    async def dispatch(self, sub_tasks: list[SubTask]) -> list[SubResult]:
        # 并行创建独立的子智能体
        tasks = []
        for sub_task in sub_tasks:
            agent = await self.client.create_thread(
                assistant_id=sub_task.assistant_type,
                context=sub_task.isolated_context,  # 隔离上下文
                tools=sub_task.required_tools,       # 按需加载工具
                recursion_limit=sub_task.complexity   # 根据复杂度限制递归
            )
            tasks.append(agent.run(sub_task.prompt))
        
        # 并行执行，等待所有结果
        results = await asyncio.gather(*tasks)
        
        # 结构化汇总
        return [SubResult(task_id=t.id, output=r) for t, r in zip(sub_tasks, results)]

2.4 与 LangGraph 的深度整合

DeerFlow 2.0 基于 LangGraph 1.0 + LangChain 构建，但不是简单地在 LangGraph 上套了一层壳。它做了几件关键的事：

1. 自定义 Graph 拓扑：不是 LangGraph 默认的线性流，而是动态拓扑——根据任务类型在运行时决定图的形状
2. State 注入：把 Memory、Sandbox 状态、Skill 配置都注入到 LangGraph 的 State 中
3. Checkpoint 复用：利用 LangGraph 的 Checkpoint 机制实现任务断点续传
4. Recursion Control：动态调整 recursion_limit，避免简单任务浪费迭代、复杂任务提前截断

# LangGraph 状态定义（基于源码理解）
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph

class DeerFlowState(TypedDict):
    """DeerFlow 的 LangGraph 状态结构"""
    messages: list[BaseMessage]          # 对话历史
    current_plan: dict | None           # 当前任务规划
    sub_agent_results: list[dict]       # 子智能体结果
    sandbox_state: dict                 # 沙箱环境状态
    memory_snapshot: dict               # 记忆快照
    loaded_skills: list[str]            # 已加载的技能
    context_summary: str | None         # 上下文压缩摘要
    output_artifacts: list[str]         # 输出产物路径

# 构建动态图
def build_deerflow_graph(config: DeerFlowConfig) -> StateGraph:
    graph = StateGraph(DeerFlowState)
    
    # 核心节点
    graph.add_node("plan", plan_node)
    graph.add_node("execute", execute_node)
    graph.add_node("reflect", reflect_node)
    graph.add_node("summarize", summarize_node)
    
    # 动态边——根据任务类型决定流转
    graph.add_conditional_edges(
        "plan",
        should_dispatch_subagents,
        {
            "execute_directly": "execute",
            "dispatch_subagents": "execute",  # 内部会并行拉起
            "need_more_planning": "plan",
        }
    )
    
    return graph.compile(
        checkpointer=SqliteSaver(conn),  # Checkpoint 持久化
        interrupt_before=[],              # 支持人机协同断点
    )

三、五大核心能力深度解析

3.1 技能系统（Skills & Tools）：Markdown 即能力

Skills 是 DeerFlow 能做「几乎任何事」的关键。一个 Skill 本质上就是一个 结构化的 Markdown 文件，定义了工作流、最佳实践和参考资源。

这种设计有几个优势：

1. 人类可读：你不需要写代码就能理解一个 Skill 做什么
2. AI 友好：LLM 天然擅长处理 Markdown
3. 渐进加载：按需加载，不会一次塞满上下文
4. 组合复用：多个 Skill 可以组合成复合工作流

DeerFlow 内置的 Skill 覆盖了这些场景：

/mnt/skills/public/
├── research/SKILL.md              # 深度研究
├── report-generation/SKILL.md     # 报告生成
├── slide-creation/SKILL.md        # 演示文稿制作
├── web-page/SKILL.md              # 网页生成
└── image-generation/SKILL.md      # 图像生成

而自定义 Skill 只需要在 /mnt/skills/custom/ 下新建一个文件夹：

<!-- /mnt/skills/custom/api-doc-generator/SKILL.md -->
# API Documentation Generator

## Purpose
Generate comprehensive API documentation from code annotations and OpenAPI specs.

## Workflow
1. Scan the target codebase for route handlers and model definitions
2. Extract type annotations, docstrings, and decorator metadata
3. Merge with any existing OpenAPI/Swagger specs
4. Generate structured Markdown documentation with:
   - Endpoint descriptions with request/response schemas
   - Authentication requirements
   - Rate limiting policies
   - Code examples in Python, JavaScript, and cURL
5. Write output to /mnt/user-data/outputs/api-docs.md

## Best Practices
- Always validate generated schemas against actual API responses
- Include error response documentation for each endpoint
- Add pagination details for list endpoints
- Cross-reference related endpoints

## Resources
- OpenAPI 3.0 Specification: https://swagger.io/specification/
- Internal API style guide: /mnt/user-data/uploads/api-style-guide.md

Skill 的加载是按需渐进的，核心逻辑如下：

class SkillLoader:
    """按需渐进加载技能"""
    
    def __init__(self, skills_dir: str = "/mnt/skills"):
        self.skills_dir = skills_dir
        self._loaded_skills: dict[str, Skill] = {}
        self._skill_registry: dict[str, SkillMeta] = self._scan_skills()
    
    def _scan_skills(self) -> dict[str, SkillMeta]:
        """扫描所有 Skill 元信息，但不加载内容"""
        registry = {}
        for skill_path in Path(self.skills_dir).rglob("SKILL.md"):
            # 只读取 frontmatter，不加载完整内容
            meta = self._parse_frontmatter(skill_path)
            registry[meta.name] = SkillMeta(
                name=meta.name,
                path=skill_path,
                description=meta.description,
                tags=meta.tags,
                content=None  # 延迟加载
            )
        return registry
    
    async def load_skill(self, skill_name: str) -> Skill:
        """按需加载 Skill 的完整内容"""
        if skill_name in self._loaded_skills:
            return self._loaded_skills[skill_name]
        
        meta = self._skill_registry.get(skill_name)
        if not meta:
            raise SkillNotFoundError(skill_name)
        
        # 只在需要时才读取完整 Markdown
        content = await asyncio.to_thread(meta.path.read_text, "utf-8")
        skill = Skill(name=meta.name, content=content, meta=meta)
        self._loaded_skills[skill_name] = skill
        return skill
    
    def get_relevant_skills(self, task_description: str) -> list[str]:
        """根据任务描述匹配相关技能（不加载内容）"""
        scored = []
        for name, meta in self._skill_registry.items():
            score = self._compute_relevance(task_description, meta)
            if score > 0.5:
                scored.append((name, score))
        scored.sort(key=lambda x: x[1], reverse=True)
        return [name for name, _ in scored[:3]]  # 最多加载3个

Tools 的设计也是同样的思路：内置网页搜索、网页抓取、文件操作、bash 执行等核心工具，同时支持通过 MCP Server 和 Python 函数扩展自定义工具。

3.2 Docker 沙箱：让 Agent 真正「动手」

这是 DeerFlow 区别于所有「聊天式 Agent」的核心能力。每个任务运行在隔离的 Docker 容器里，里面有完整的文件系统：

# sandbox 容器内的路径结构
/mnt/
├── skills/
│   ├── public/          # 内置技能
│   └── custom/          # 自定义技能
├── user-data/
│   ├── uploads/         # 用户上传的文件
│   ├── workspace/       # Agent 的工作目录
│   └── outputs/         # 最终交付物

DeerFlow 支持三种 Sandbox 模式：

1. 本地执行：直接在宿主机上运行 sandbox 代码（开发用，不安全）
2. Docker 执行：在隔离的 Docker 容器里运行（推荐）
3. Docker + Kubernetes 执行：通过 provisioner 服务在 K8s Pod 中运行（生产环境）

# config.yaml 中的 Sandbox 配置
sandbox:
  # 本地模式
  use: deerflow.community.local_sandbox:LocalSandboxProvider
  
  # Docker 模式（推荐）
  use: deerflow.community.docker_sandbox:DockerSandboxProvider
  
  # K8s 模式（生产级）
  use: deerflow.community.aio_sandbox:AioSandboxProvider
  provisioner_url: http://provisioner:8002

Sandbox 的隔离保证：

class SandboxExecutor:
    """沙箱代码执行器"""
    
    async def execute(
        self,
        code: str,
        language: str = "python",
        timeout: int = 300,
        resources: dict = None
    ) -> ExecutionResult:
        """在隔离环境中执行代码"""
        
        # 资源限制
        resource_limits = {
            "cpu": resources.get("cpu", 2),
            "memory": resources.get("memory", "4GB"),
            "disk": resources.get("disk", "10GB"),
            "network": resources.get("network", False),  # 默认禁止网络
        }
        
        # 创建隔离容器
        container = await self.docker_client.containers.run(
            image="deerflow-sandbox:latest",
            command=self._build_command(code, language),
            mem_limit=resource_limits["memory"],
            cpu_quota=resource_limits["cpu"] * 100000,
            network_disabled=not resource_limits["network"],
            volumes={
                self.workspace_path: {"bind": "/mnt/user-data/workspace", "mode": "rw"},
                self.uploads_path: {"bind": "/mnt/user-data/uploads", "mode": "ro"},
                self.outputs_path: {"bind": "/mnt/user-data/outputs", "mode": "rw"},
            },
            timeout=timeout,
            remove=True,  # 执行完自动清理
        )
        
        return ExecutionResult(
            exit_code=container.exit_code,
            stdout=container.stdout,
            stderr=container.stderr,
            artifacts=self._collect_artifacts(),  # 收集输出产物
        )

沙箱执行的安全边界：

• 文件系统隔离：不同 Session 之间的文件不会互相污染
• 网络隔离：默认禁止容器访问网络，需要时显式开启
• 资源限制：CPU、内存、磁盘都有上限
• 超时控制：代码执行超时自动终止
• 审计追踪：所有操作可审计

这就是「带工具的聊天机器人」和「真正有执行环境的 Agent」之间的根本差别。

3.3 上下文工程（Context Engineering）：长任务的命脉

上下文管理是 Agent 能否完成长任务的关键。DeerFlow 2.0 在这方面做了三件事：

策略一：Sub-Agent 上下文隔离

每个 Sub-Agent 在自己独立的上下文里运行。它看不到主 Agent 的上下文，也看不到其他 Sub-Agents 的上下文。这样做的目的很直接：让它只聚焦当前任务，不被无关信息干扰。

Lead Agent Context: [任务描述, 规划结果, 汇总指令]
    ├── Sub-Agent A Context: [子任务A描述, 搜索工具, 终止条件]
    ├── Sub-Agent B Context: [子任务B描述, 分析工具, 终止条件]
    └── Sub-Agent C Context: [子任务C描述, 生成工具, 终止条件]

策略二：摘要压缩

在单个 Session 内，DeerFlow 会积极地管理上下文：

1. 总结已完成的子任务：子任务完成后，用摘要替代完整对话
2. 中间结果转存文件系统：把中间数据写入 Sandbox，释放上下文空间
3. 压缩暂时不重要的信息：用 Context Compressor 压缩历史消息

class ContextCompressor:
    """上下文压缩器"""
    
    def __init__(self, llm: BaseChatModel, max_context_tokens: int = 128000):
        self.llm = llm
        self.max_tokens = max_context_tokens
    
    async def compress(self, messages: list[BaseMessage]) -> list[BaseMessage]:
        """压缩对话历史，保持关键信息"""
        
        # 计算当前 token 数
        current_tokens = self._count_tokens(messages)
        if current_tokens <= self.max_tokens * 0.7:  # 留30%给新内容
            return messages
        
        # 分段压缩：保留最近的消息，压缩更早的消息
        recent_messages = messages[-10:]  # 保留最近10条
        older_messages = messages[:-10]
        
        # 用 LLM 生成压缩摘要
        summary = await self._generate_summary(older_messages)
        
        # 替换：一条摘要消息替代所有旧消息
        compressed = [
            SystemMessage(content=f"[历史上下文摘要]\n{summary}"),
            *recent_messages
        ]
        
        return compressed
    
    async def _generate_summary(self, messages: list[BaseMessage]) -> str:
        """用 LLM 生成摘要"""
        prompt = f"""请将以下对话历史压缩为一段简洁的摘要，保留关键决策、结论和未完成的任务：

{self._format_messages(messages)}

摘要："""
        
        response = await self.llm.ainvoke(prompt)
        return response.content

策略三：中间结果持久化

对于耗时任务（可能运行数小时），DeerFlow 把中间结果转存到文件系统：

class IntermediateResultPersister:
    """中间结果持久化"""
    
    def save_intermediate(self, task_id: str, step: int, result: dict):
        """将子任务结果保存到文件"""
        path = f"/mnt/user-data/workspace/.intermediate/{task_id}/step_{step}.json"
        with open(path, "w") as f:
            json.dump(result, f, ensure_ascii=False, indent=2)
    
    def load_intermediate(self, task_id: str, step: int) -> dict:
        """从文件恢复子任务结果"""
        path = f"/mnt/user-data/workspace/.intermediate/{task_id}/step_{step}.json"
        if os.path.exists(path):
            with open(path) as f:
                return json.load(f)
        return None

3.4 长期记忆（Long-term Memory）：跨 Session 的知识积累

大多数 Agent 在对话结束后把一切都忘掉了。DeerFlow 不一样——它会在跨 Session 使用中逐步积累关于你的持久记忆：

• 个人偏好：你喜欢的输出格式、语言风格
• 知识背景：你的技术栈、专业领域
• 工作习惯：你常用的工具、重复出现的工作流

class MemoryManager:
    """三层记忆系统"""
    
    def __init__(self, user_id: str):
        self.user_id = user_id
        # 短期记忆：当前对话上下文
        self.short_term = ConversationBuffer()
        # 工作记忆：任务执行中的临时状态
        self.working = TaskStateBuffer()
        # 长期记忆：跨任务的持久化知识
        self.long_term = VectorStore(collection=f"memory_{user_id}")
    
    async def remember(self, key: str, value: str, scope: str = "long"):
        """存储记忆"""
        if scope == "short":
            self.short_term.add(key, value)
        elif scope == "working":
            self.working.set(key, value)
        else:
            await self.long_term.aadd_documents([
                Document(page_content=value, metadata={"key": key})
            ])
    
    async def recall(self, query: str, top_k: int = 5) -> list[str]:
        """检索相关记忆"""
        # 先查短期记忆
        short_results = self.short_term.search(query)
        if short_results:
            return short_results
        
        # 再查长期记忆
        long_results = await self.long_term.asimilarity_search(query, k=top_k)
        return [doc.page_content for doc in long_results]
    
    async def consolidate(self):
        """将短期/工作记忆沉淀到长期记忆"""
        # 提取关键信息
        working_items = self.working.get_all()
        for key, value in working_items.items():
            if self._is_worth_remembering(key, value):
                await self.remember(key, value, scope="long")
        
        self.working.clear()

记忆数据保存在本地，控制权始终在用户手里。你可以通过 IM 渠道发送 /memory 命令查看当前的记忆内容。

3.5 Claude Code 集成：终端里的 Agent 控制台

DeerFlow 提供了一个 claude-to-deerflow Skill，让你可以在 Claude Code 终端里直接和正在运行的 DeerFlow 实例交互：

# 安装 Skill
npx skills add https://github.com/bytedance/deer-flow --skill claude-to-deerflow

# 在 Claude Code 中使用
/claude-to-deerflow

支持的操作包括：

• 给 DeerFlow 发送消息，接收流式响应
• 选择执行模式：flash（快速）、standard、pro（规划模式）、ultra（Sub-Agents 模式）
• 检查健康状态，列出 Models / Skills / Agents
• 管理 Threads 和会话历史
• 上传文件做分析

四、从零开始：DeerFlow 2.0 完整部署实战

4.1 环境要求

4.2 一键安装（推荐 Docker 方式）

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

# 2. 生成配置文件
make config

# 3. 编辑配置（至少配置一个模型）
# 编辑 config.yaml，填入 API key

config.yaml 的模型配置示例：

models:
  - name: deepseek-v3
    display_name: DeepSeek V3.2
    use: langchain_openai:ChatOpenAI
    model: deepseek-chat
    api_key: $DEEPSEEK_API_KEY
    base_url: https://api.deepseek.com/v1
    max_tokens: 8192
    temperature: 0.7

  - name: gpt-4o
    display_name: GPT-4o
    use: langchain_openai:ChatOpenAI
    model: gpt-4o
    api_key: $OPENAI_API_KEY
    max_tokens: 4096
    temperature: 0.7

环境变量配置（.env 文件）：

# 必填
DEEPSEEK_API_KEY=sk-xxxxxxxxxxxxxxxx
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxx
TAVILY_API_KEY=tvly-xxxxxxxxxxxxxxxx

# 可选：IM 渠道
TELEGRAM_BOT_TOKEN=123456789:ABCdefGHIjklMNOpqrSTUvwxYZ
SLACK_BOT_TOKEN=xoxb-...
SLACK_APP_TOKEN=xapp-...
FEISHU_APP_ID=cli_xxxx
FEISHU_APP_SECRET=your_app_secret

# 可选：LangSmith 追踪
LANGSMITH_TRACING=true
LANGSMITH_API_KEY=lsv2_pt_xxxxxxxxxxxxxxxx
LANGSMITH_PROJECT=deerflow

启动服务：

# Docker 方式（推荐）
make docker-init      # 拉取 sandbox 镜像（首次执行）
make docker-start     # 启动所有服务

# 或者本地开发方式
make check            # 校验依赖
make install          # 安装依赖
make dev              # 启动服务

访问 http://localhost:2026 即可看到 DeerFlow 的 Web 界面。

4.3 生产部署

对于长期运行的服务，推荐使用 Linux + Docker 生产模式：

# 构建并启动生产环境
make up

# 停止服务
make down

资源规划建议：

4.4 IM 渠道配置：让 Agent 像同事一样工作

DeerFlow 支持 Telegram、Slack、飞书、企业微信四种 IM 渠道。配置完成后，你可以直接在聊天窗口里和 DeerFlow 交互：

# config.yaml 中的渠道配置
channels:
  langgraph_url: http://localhost:2024
  gateway_url: http://localhost:8001
  
  feishu:
    enabled: true
    app_id: $FEISHU_APP_ID
    app_secret: $FEISHU_APP_SECRET

  telegram:
    enabled: true
    bot_token: $TELEGRAM_BOT_TOKEN
    allowed_users: []  # 留空表示允许所有人

渠道支持的命令：

4.5 MCP Server 扩展

DeerFlow 支持通过 MCP Server 扩展能力，还支持 OAuth token 流程：

# config.yaml 中的 MCP Server 配置
mcp_servers:
  - name: my-tool-server
    url: http://localhost:3000/sse
    # 可选：OAuth 认证
    auth:
      type: client_credentials
      token_url: https://auth.example.com/oauth/token
      client_id: $MCP_CLIENT_ID
      client_secret: $MCP_CLIENT_SECRET

五、性能优化与生产调优

5.1 模型选型策略

DeerFlow 对模型没有强绑定，只要实现了 OpenAI 兼容 API 的 LLM 都可以接入。但以下能力的模型更适合 DeerFlow：

• 长上下文窗口（100k+ tokens）：适合深度研究和多步骤任务
• 推理能力：适合自适应规划和复杂拆解
• 工具调用：支持 function calling 是基本要求
• 代码生成：Sandbox 里要执行代码，代码质量直接影响成功率

官方推荐模型：

5.2 Sub-Agent 并发控制

Sub-Agent 并行执行是性能提升的关键，但也需要控制并发度：

# 配置建议
class SubAgentConfig:
    max_concurrent_subagents: int = 5      # 最大并发子智能体数
    subagent_timeout: int = 600             # 单个子智能体超时（秒）
    recursion_limit: int = 100              # LangGraph 递归深度限制
    
    # 根据任务复杂度动态调整
    complexity_map = {
        "simple": {"max_concurrent": 2, "timeout": 300, "recursion": 50},
        "medium": {"max_concurrent": 5, "timeout": 600, "recursion": 100},
        "complex": {"max_concurrent": 8, "timeout": 1800, "recursion": 150},
    }

5.3 上下文窗口优化

对于使用 128K 上下文窗口的模型：

# 上下文预算分配策略
class ContextBudget:
    total_budget = 128000  # 总 token 预算
    
    # 分配比例
    system_prompt = 0.05      # 5% - 系统提示
    skill_content = 0.10      # 10% - 技能内容
    conversation = 0.35       # 35% - 对话历史
    subagent_results = 0.30   # 30% - 子智能体结果
    workspace_state = 0.10    # 10% - 工作区状态
    reserve = 0.10            # 10% - 预留空间给模型输出

当对话历史超过预算时，触发压缩策略：

1. 先压缩最早的消息为摘要
2. 如果还不够，把中间结果转存到文件系统
3. 最后限制每个 Sub-Agent 的结果长度

5.4 Sandbox 资源隔离

生产环境中，合理的资源隔离配置：

# 生产环境 Sandbox 配置
sandbox:
  mode: docker+k8s
  provisioner_url: http://provisioner:8002
  
  # 默认资源限制
  default_resources:
    cpu: 2
    memory: "4GB"
    disk: "10GB"
    timeout: 600
    
  # 按任务类型差异化配置
  task_profiles:
    research:
      cpu: 2
      memory: "4GB"
      timeout: 1800
      network: true    # 研究任务需要网络访问
    code_execution:
      cpu: 4
      memory: "8GB"
      timeout: 600
      network: false   # 代码执行禁止网络
    report_generation:
      cpu: 2
      memory: "4GB"
      timeout: 300
      network: false

5.5 记忆系统的性能考量

长期记忆基于向量数据库，在大规模使用时需要关注：

# 记忆检索的性能优化
class MemoryOptimization:
    # 1. 索引优化：使用 HNSW 索引加速近似搜索
    index_type = "hnsw"
    hnsw_params = {
        "M": 16,           # 连接数
        "ef_construction": 200,  # 构建时搜索深度
        "ef_search": 50,   # 查询时搜索深度
    }
    
    # 2. 记忆衰减：时间越久的记忆权重越低
    def compute_relevance(self, doc, query_similarity, age_days):
        recency_weight = math.exp(-0.1 * age_days)  # 指数衰减
        return query_similarity * (0.7 + 0.3 * recency_weight)
    
    # 3. 记忆去重：相似度过高的记忆合并
    def deduplicate(self, memories, threshold=0.95):
        unique = []
        for mem in memories:
            if not any(self.similarity(mem, u) > threshold for u in unique):
                unique.append(mem)
            else:
                # 合并：保留更完整的那个
                pass
        return unique

六、实战案例：用 DeerFlow 构建自动化研报系统

6.1 需求场景

假设我们需要构建一个自动化研报系统，用户只需要输入一个研究主题，系统就能：

1. 自动搜索相关资料
2. 分析数据趋势
3. 生成结构化报告
4. 制作出演示文稿

6.2 自定义 Skill

<!-- /mnt/skills/custom/market-research/SKILL.md -->
# Market Research Skill

## Purpose
Automated market research and analysis with structured report generation.

## Workflow
1. Decompose research topic into sub-areas
2. For each sub-area:
   a. Search for latest industry reports and news
   b. Extract key data points and trends
   c. Analyze competitive landscape
3. Synthesize findings into structured report
4. Generate executive summary and key takeaways
5. Create presentation slides if requested

## Report Structure
- Executive Summary
- Market Overview
- Key Trends & Drivers
- Competitive Analysis
- Data & Forecasts
- Risk Factors
- Recommendations

## Data Sources Priority
1. Official company filings and earnings reports
2. Government statistics and policy documents
3. Industry analyst reports (Gartner, IDC, etc.)
4. News articles from reputable sources
5. Academic papers and conference proceedings

## Output Format
- Primary: Markdown report saved to /mnt/user-data/outputs/
- Secondary: PDF if pandoc is available
- Tertiary: Presentation slides if explicitly requested

6.3 配置与执行

# 通过 DeerFlow API 提交研究任务
import httpx

async def submit_research(topic: str):
    """提交一个研究任务到 DeerFlow"""
    
    async with httpx.AsyncClient() as client:
        # 创建新 Thread
        thread_resp = await client.post(
            "http://localhost:2024/threads",
            json={}
        )
        thread_id = thread_resp.json()["thread_id"]
        
        # 提交研究任务
        run_resp = await client.post(
            f"http://localhost:2024/threads/{thread_id}/runs",
            json={
                "assistant_id": "lead_agent",
                "input": {
                    "messages": [{
                        "role": "user",
                        "content": f"""请对以下主题进行深度研究并生成结构化报告：
                        
{topic}

要求：
1. 使用 market-research 技能
2. 报告至少包含5个章节
3. 每个章节需要引用至少3个数据来源
4. 最终报告保存到 /mnt/user-data/outputs/ 目录
5. 同时生成一份10页以内的演示文稿"""
                    }]
                },
                "config": {
                    "configurable": {
                        "subagent_enabled": True,  # 启用 Sub-Agent
                        "thinking_enabled": True,   # 启用深度思考
                    }
                }
            }
        )
        
        return run_resp.json()

6.4 执行流程追踪

当 DeerFlow 执行这个任务时，内部流程大致如下：

用户输入: "分析2026年全球AI芯片市场竞争格局"
    │
    ▼
Lead Agent: 解析意图 → 识别需要 market-research 技能
    │
    ▼
加载 Skill: market-research/SKILL.md → 生成任务规划
    │
    ├── Sub-Agent 1: 搜索NVIDIA、AMD、Intel最新财报和产品发布
    ├── Sub-Agent 2: 搜索AI芯片市场份额数据（Gartner/IDC）
    ├── Sub-Agent 3: 搜索中国芯片厂商动态（华为昇腾、寒武纪等）
    │
    ▼
Lead Agent: 汇总三个Sub-Agent结果 → 生成报告框架
    │
    ├── Sub-Agent 4: 撰写"市场概览"和"竞争格局"章节
    ├── Sub-Agent 5: 撰写"技术趋势"和"风险因素"章节
    │
    ▼
Lead Agent: 合并报告 → 生成摘要 → 制作案板
    │
    ▼
Sandbox: 将报告和演示文稿保存到 /mnt/user-data/outputs/
    │
    ▼
输出: 报告链接 + 演示文稿链接

七、与竞品对比：为什么选 DeerFlow

7.1 Agent 框架横向对比

7.2 核心差异化

DeerFlow 的差异化可以用一句话概括：它是目前唯一一个「开箱即用、同时具备沙箱执行+长期记忆+子智能体调度」的 Agent 运行时。

其他框架要么只是 SDK（LangChain），要么只有概念没有执行环境（AutoGPT），要么专注协作但没有沙箱（CrewAI）。DeerFlow 是第一个真正把「Agent 执行全链路」串起来的开源项目。

7.3 适用场景

适合用 DeerFlow 的场景：

• 深度研究任务（需要搜索+分析+生成报告）
• 自动化内容创作（从选题到成品）
• 数据分析流水线（从采集到可视化）
• 长耗时任务（数分钟到数小时）
• 需要跨 Session 知识积累的场景
• 企业级部署（需要沙箱隔离、IM 集成）

不太适合的场景：

• 简单的问答/聊天（杀鸡用牛刀）
• 实时性要求极高的场景（Sub-Agent 调度有延迟）
• 纯代码生成（Cursor/Copilot 更轻量）
• 移动端场景（资源消耗大）

八、踩坑经验与最佳实践

8.1 常见部署问题

问题一：Sandbox 镜像拉取失败

# 解决方案：手动拉取并打标签
docker pull deerflow/sandbox:latest
docker tag deerflow/sandbox:latest deerflow-sandbox:latest

# 或者配置 Docker 镜像加速
# 编辑 /etc/docker/daemon.json
{
  "registry-mirrors": ["https://mirror.ccs.tencentyun.com"]
}

问题二：Windows 上 make 命令不可用

DeerFlow 的开发流程基于 bash 脚本，Windows 用户需要使用 Git Bash，不要用 cmd.exe 或 PowerShell。WSL 也不保证可用，因为部分脚本依赖 Git for Windows 的 cygpath 等工具。

问题三：内存不足导致 OOM

# 检查 Docker 资源限制
docker stats

# 调整 Docker Desktop 资源分配
# Settings → Resources → Memory: 至少 8GB

# 或降低并发
# config.yaml 中减少 max_concurrent_subagents

8.2 模型配置最佳实践

# 推荐的模型配置组合
models:
  # 主力模型：处理规划、摘要、报告生成
  - name: deepseek-v3
    display_name: DeepSeek V3.2
    use: langchain_openai:ChatOpenAI
    model: deepseek-chat
    api_key: $DEEPSEEK_API_KEY
    base_url: https://api.deepseek.com/v1
    max_tokens: 8192
    temperature: 0.7

  # 推理模型：处理复杂分析
  - name: deepseek-r1
    display_name: DeepSeek R1
    use: langchain_openai:ChatOpenAI
    model: deepseek-reasoner
    api_key: $DEEPSEEK_API_KEY
    base_url: https://api.deepseek.com/v1
    max_tokens: 16384
    temperature: 0.3

  # 快速模型：处理简单子任务
  - name: gpt-4o-mini
    display_name: GPT-4o Mini
    use: langchain_openai:ChatOpenAI
    model: gpt-4o-mini
    api_key: $OPENAI_API_KEY
    max_tokens: 4096
    temperature: 0.7

8.3 自定义 Skill 开发规范

1. Skill 名称要有辨识度：不要用 my-skill，用 api-doc-generator 这种具体名称
2. Workflow 要具体：不要写「分析数据」，写「读取 CSV → 统计缺失值 → 生成分布图 → 输出 Markdown 报告」
3. 指定输出路径：所有输出写到 /mnt/user-data/outputs/
4. 包含错误处理：Skill 中要描述异常情况的处理方式
5. 声明资源需求：是否需要网络访问、需要多少内存等

九、源码级架构分析

9.1 项目结构

deer-flow/
├── backend/                  # Python 后端
│   ├── pyproject.toml        # 依赖管理（uv）
│   ├── src/
│   │   └── deerflow/
│   │       ├── agents/       # 智能体定义
│   │       │   ├── lead_agent.py      # 主智能体
│   │       │   └── sub_agents/        # 子智能体
│   │       ├── core/         # 核心模块
│   │       │   ├── sandbox/           # 沙箱管理
│   │       │   ├── memory/            # 记忆系统
│   │       │   ├── skills/            # 技能系统
│   │       │   └── context/           # 上下文工程
│   │       ├── gateway/      # API 网关
│   │       ├── channels/     # IM 渠道
│   │       │   ├── telegram/
│   │       │   ├── slack/
│   │       │   ├── feishu/
│   │       │   └── wecom/
│   │       └── tools/        # 工具集
│   └── tests/
├── frontend/                 # Next.js 前端
│   ├── package.json
│   └── src/
├── skills/                   # 内置技能
│   └── public/
│       ├── research/SKILL.md
│       ├── report-generation/SKILL.md
│       ├── slide-creation/SKILL.md
│       ├── web-page/SKILL.md
│       └── image-generation/SKILL.md
├── config.yaml               # 配置文件
├── Makefile                  # 构建脚本
└── Dockerfile                # 容器构建

9.2 关键模块交互流程

用户请求
  │
  ▼
Gateway (FastAPI, port 8001)
  │
  ├── 身份验证 & 会话管理
  │
  ▼
LangGraph Agent Server (port 2024)
  │
  ├── Lead Agent 接收消息
  │   ├── Middleware Chain 处理
  │   ├── Skill Loader 加载相关技能
  │   ├── Plan Parser 生成任务规划
  │   │
  │   ├── 是否需要 Sub-Agent?
  │   │   ├── 是 → Sub-Agent Dispatcher → 并行执行
  │   │   └── 否 → 直接执行
  │   │
  │   ├── Context Compressor 压缩上下文
  │   ├── Memory Manager 读写记忆
  │   └── Result Aggregator 汇总结果
  │
  ▼
Sandbox Manager
  │
  ├── 创建/复用 Docker 容器
  ├── 执行代码和命令
  ├── 管理文件系统
  └── 收集输出产物
  │
  ▼
Response → Gateway → 用户

9.3 线程模型与并发

DeerFlow 的并发模型基于 Python asyncio，核心线程模型：

1. Gateway 线程：FastAPI + uvicorn，处理 HTTP 请求
2. LangGraph 线程：每个用户 Session 对应一个独立的 LangGraph Thread
3. Sandbox 线程：Docker API 调用是异步的，不阻塞主线程
4. Sub-Agent 线程：每个 Sub-Agent 在独立的 asyncio Task 中运行

# 简化的并发模型
async def handle_user_message(message: UserMessage):
    """处理用户消息的入口"""
    
    # 获取或创建 LangGraph Thread
    thread = await get_or_create_thread(message.user_id)
    
    # 在独立 Task 中运行 Agent
    run_task = asyncio.create_task(
        run_agent(thread, message)
    )
    
    # 流式返回结果
    async for event in run_task:
        if event.type == "token":
            yield SSEEvent(data=event.token)
        elif event.type == "subagent_start":
            yield SSEEvent(data=f"[Sub-Agent {event.agent_name} 启动]")
        elif event.type == "subagent_complete":
            yield SSEEvent(data=f"[Sub-Agent {event.agent_name} 完成]")
    
    # 等待最终结果
    result = await run_task
    yield SSEEvent(data=result.final_output)

十、未来展望与生态

10.1 Agent 生态的演进方向

DeerFlow 代表了 Agent 从「聊天工具」向「执行系统」转变的趋势。接下来的演进方向可能是：

1. Agent-to-Agent 协议：不同框架的 Agent 能互相发现和协作（类似 MCP 的 Agent 版）
2. 更精细的权限模型：基于 RBAC 的 Agent 权限控制，企业级安全
3. Agent 市场：Skill 和 Agent 模板可以像 App Store 一样分发
4. 边缘部署：轻量级 Sandbox 适配树莓派等边缘设备
5. 多模态 Sandbox：不仅执行代码，还能渲染 UI、播放音视频

10.2 DeerFlow 的局限与挑战

客观地说，DeerFlow 目前也有不足：

1. 资源消耗大：Docker Sandbox 至少需要 8GB 内存，个人开发者门槛不低
2. 冷启动慢：Sandbox 容器创建需要几秒钟，频繁创建影响体验
3. 中文生态偏弱：虽然官方推荐了豆包模型，但 Skill 和文档以英文为主
4. 调试困难：Sub-Agent 嵌套 + LangGraph 状态机，出问题时排查链路很长
5. 生产验证不足：社区以体验和 Demo 为主，大规模生产部署的案例还不多

10.3 谁应该关注 DeerFlow

• 想构建 AI Agent 应用的开发者：DeerFlow 是目前最完整的开源 Agent 运行时
• 需要自动化研究和内容生成的团队：开箱即用的 Deep Research 能力
• 企业 AI 基础设施团队：Sandbox 隔离 + IM 集成 + 长期记忆，适合内部工具
• 对 Agent 架构感兴趣的技术人：源码质量高，是学习 Agent 系统设计的优秀样本

结语

DeerFlow 2.0 不是一个简单的 AI 聊天框架，它是第一个真正把「Agent 执行全链路」串起来的开源项目。从任务规划到子智能体调度，从沙箱执行到长期记忆，从技能扩展到 IM 集成，它覆盖了一个生产级 Agent 系统需要的所有能力。

52K Star 不是因为噱头，而是因为它解决了一个真实的问题：让 Agent 不只是能说，而是能做。

如果你正在寻找一个 Agent 运行时来构建你的 AI 应用，DeerFlow 值得认真研究。即使你最终不用它，它的架构设计——特别是 Middleware Chain、Sub-Agent 隔离、Context Engineering 这几个模块——也值得借鉴。

项目地址：https://github.com/bytedance/deer-flow
官网：https://deerflow.tech
许可证：MIT