【DeerFlow 2.0】代码详解（二）：Lead Agent 与 Prompt 工程

拾-光

11人浏览 · 2026-05-06 10:45:00

拾-光 · 2026-05-06 10:45:00 发布

【DeerFlow 2.0】代码详解（二）：Lead Agent 与 Prompt 工程

在这里插入图片描述

系列导读：DeerFlow 2.0 是字节跳动开源的 SuperAgent 框架，基于 LangGraph + LangChain 构建。本系列共 5 篇，从架构总览到逐模块深入，带你彻底读懂每一行代码。

✅ 第一篇：架构总览与核心骨架

第二篇：Lead Agent 与 Prompt 工程（本文）

第三篇：SubAgent 并发执行引擎

第四篇：Sandbox 安全隔离与 Skills 技能系统

第五篇：Memory 记忆系统与 MCP/ACP 扩展

📑 本文目录

🧠 一、Lead Agent 创建流程：从请求到 Agent 实例
📝 二、System Prompt 解剖：10 层模板构建
🎭 三、SOUL.md 人格机制：让 Agent 有灵魂
🔧 四、Skills 技能注入：渐进式加载管线
🤝 五、SubAgent 调度 Prompt：DECOMPOSE → DELEGATE → SYNTHESIZE
💡 六、Clarification 澄清系统：5 种必须澄清的场景
🧩 七、TodoMiddleware：Plan Mode 的任务管理
🎯 八、总结：Prompt 即代码

🧠 一、Lead Agent 创建流程：从请求到 Agent 实例

Lead Agent 是 DeerFlow 2.0 的"大脑"——所有用户请求都经过它处理。它的创建过程是一个精密的工厂模式，每一步都有明确的职责。

1.1 入口函数：`make_lead_agent`

# backend/packages/harness/deerflow/agents/lead_agent/agent.py

def make_lead_agent(config: RunnableConfig):
    """LangGraph graph factory; keep the signature compatible with LangGraph Server."""
    runtime_config = _get_runtime_config(config)
    runtime_app_config = runtime_config.get("app_config")
    return _make_lead_agent(config, app_config=runtime_app_config or get_app_config())

这是 LangGraph Server 的标准入口。config 是 LangGraph 的运行时配置，包含 configurable 和 context 两个字典。

1.2 核心工厂：`_make_lead_agent`

def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
    cfg = _get_runtime_config(config)
    
    # 1️⃣ 解析运行时参数
    thinking_enabled = cfg.get("thinking_enabled", True)
    reasoning_effort = cfg.get("reasoning_effort", None)
    requested_model_name = cfg.get("model_name") or cfg.get("model")
    is_plan_mode = cfg.get("is_plan_mode", False)
    subagent_enabled = cfg.get("subagent_enabled", False)
    max_concurrent_subagents = cfg.get("max_concurrent_subagents", 3)
    agent_name = validate_agent_name(cfg.get("agent_name"))
    
    # 2️⃣ 加载 Agent 配置（如果有自定义 Agent）
    agent_config = load_agent_config(agent_name) if not is_bootstrap else None
    
    # 3️⃣ 解析模型名称：请求 → Agent配置 → 全局默认
    model_name = _resolve_model_name(
        requested_model_name or agent_model_name, 
        app_config=resolved_app_config
    )
    
    # 4️⃣ 创建 Agent
    return create_agent(
        model=create_chat_model(...),
        tools=get_available_tools(...) + extra_tools,
        middleware=_build_middlewares(...),
        system_prompt=apply_prompt_template(...),
        state_schema=ThreadState,
    )

关键设计：模型名称的三级回退机制——请求参数 > Agent 配置 > 全局默认。这保证了无论用户怎么配置，总能找到一个可用的模型。

1.3 Bootstrap Agent：特殊的初始化 Agent

if is_bootstrap:
    # 特殊的 bootstrap agent，用于初始自定义 Agent 创建流程
    return create_agent(
        model=...,
        tools=get_available_tools(...) + [setup_agent],  # 只有 bootstrap 有 setup_agent
        middleware=...,
        system_prompt=apply_prompt_template(
            available_skills=set(["bootstrap"]),  # 只加载 bootstrap 技能
        ),
        state_schema=ThreadState,
    )

Bootstrap Agent 是一个精简版的 Lead Agent，专门用于引导用户创建自定义 Agent。它只加载 bootstrap 技能和 setup_agent 工具，避免在初始化阶段加载过多无关内容。

📝 二、System Prompt 解剖：10 层模板构建

在这里插入图片描述

DeerFlow 2.0 的 System Prompt 不是一段静态文本，而是一个动态构建的 10 层模板。每一层都有独立的生成逻辑和注入时机。

2.1 模板骨架：`SYSTEM_PROMPT_TEMPLATE`

# prompt.py 中的模板骨架（简化版）
SYSTEM_PROMPT_TEMPLATE = """
<role>
You are {agent_name}, an open-source super agent that researches, codes, and creates.
</role>

{soul}

{self_update_section}

<memory_context>
{memory_context}
</memory_context>

<thinking_style>
...先思考后行动 / 优先澄清 / 不在思考中写答案...
</thinking_style>

<clarification_system>
...5 种必须澄清的场景...
</clarification_system>

{skills_section}

{subagent_section}

<working_directory>
.../mnt/user-data/ 三级目录结构...
</working_directory>

<citations>
...[citation:Title](URL) 格式...
</citations>

{deferred_tools_section}

{acp_section}

<critical_reminders>
...关键提醒...
</critical_reminders>
"""

2.2 `apply_prompt_template`：动态组装函数

def apply_prompt_template(
    subagent_enabled: bool = False,
    max_concurrent_subagents: int = 3,
    agent_name: str | None = None,
    available_skills: set[str] | None = None,
    app_config: AppConfig | None = None,
) -> str:
    # 1. 获取记忆上下文
    memory_context = get_memory_context(agent_name)
    
    # 2. 构建 SubAgent 段落
    subagent_section, subagent_reminder, subagent_thinking = _build_subagent_sections(
        subagent_enabled, max_concurrent_subagents
    )
    
    # 3. 获取技能段落
    skills_section = get_skills_prompt_section(available_skills, app_config=app_config)
    
    # 4. 获取延迟工具段落
    deferred_tools_section = get_deferred_tools_prompt_section(app_config=app_config)
    
    # 5. 构建 ACP 段落
    acp_section = _build_acp_section(app_config=app_config)
    
    # 6. 格式化模板
    prompt = SYSTEM_PROMPT_TEMPLATE.format(
        agent_name=agent_name or "DeerFlow 2.0",
        soul=get_agent_soul(agent_name),
        self_update_section=_build_self_update_section(agent_name),
        skills_section=skills_section,
        deferred_tools_section=deferred_tools_section,
        memory_context=memory_context,
        subagent_section=subagent_section,
        subagent_reminder=subagent_reminder,
        subagent_thinking=subagent_thinking,
        acp_section=acp_and_mounts_section,
    )
    
    # 7. 追加当前日期
    return prompt + f"\n<current_date>{datetime.now().strftime('%Y-%m-%d, %A')}</current_date>"

关键洞察：Prompt 的构建是声明式的——你不需要手动拼接字符串，只需要提供参数，模板引擎自动组装。这让 Prompt 的维护和扩展变得非常清晰。

🎭 三、SOUL.md 人格机制：让 Agent 有灵魂

在这里插入图片描述

SOUL.md 是 DeerFlow 2.0 最优雅的设计之一——它让每个 Agent 都有独特的"人格"，而不仅仅是不同的工具集。

3.1 文件定位逻辑

# agents_config.py

def load_agent_soul(agent_name: str | None, *, user_id: str | None = None) -> str | None:
    """读取 SOUL.md 文件，定义 Agent 的性格、价值观、行为准则。"""
    if agent_name:
        # 自定义 Agent: agents/{name}/SOUL.md
        agent_dir = resolve_agent_dir(agent_name, user_id=user_id)
    else:
        # 默认 Agent: data/SOUL.md
        agent_dir = get_paths().base_dir
    
    soul_path = agent_dir / "SOUL.md"
    if not soul_path.exists():
        return None
    return soul_path.read_text(encoding="utf-8").strip() or None

3.2 用户级覆盖

def resolve_agent_dir(name: str, *, user_id: str | None = None) -> Path:
    """Agent 目录解析，优先用户级，回退共享级。"""
    paths = get_paths()
    effective_user = user_id or get_effective_user_id()
    
    # 1. 用户级: {base_dir}/users/{user_id}/agents/{name}/
    user_path = paths.user_agent_dir(effective_user, name)
    if user_path.exists():
        return user_path
    
    # 2. 共享级: {base_dir}/agents/{name}/
    legacy_path = paths.agent_dir(name)
    if legacy_path.exists():
        return legacy_path
    
    # 3. 都不存在，返回用户级路径（用于创建）
    return user_path

设计亮点：两级目录结构实现了"共享模板 + 用户定制"的完美平衡——管理员可以创建共享的 Agent 模板，用户可以在自己的目录下覆盖 SOUL.md 实现个性化。

3.3 SOUL.md 示例

# 数据分析师 Agent

你是一个严谨的数据分析师，专注于提供准确、有洞察力的数据分析。

## 核心原则
- 回答前必须验证数据来源的可靠性
- 偏好使用 Python + pandas 进行数据处理
- 永远不要编造数据，如果数据不足，明确告知用户

## 沟通风格
- 使用简洁的表格和图表呈现结果
- 在给出结论前，先展示数据支撑
- 对不确定的结论标注置信度

3.4 注入方式

SOUL.md 的内容被包裹在 <soul> 标签中，注入到 System Prompt 的第二层：

<soul>
# 数据分析师 Agent
你是一个严谨的数据分析师...
</soul>

这个位置非常关键——它在角色定义之后、工具说明之前，确保 Agent 的"人格"在所有行为之前就被确立。

🔧 四、Skills 技能注入：渐进式加载管线

在这里插入图片描述

Skills 是 DeerFlow 2.0 的"可插拔能力模块"——每个 Skill 是一个 SKILL.md 文件，定义了 Agent 在特定领域的行为规范。

4.1 五阶段注入管线

# 阶段 1: SkillStorage.load_skills() - 扫描目录，加载元数据
skills = get_or_new_skill_storage().load_skills(enabled_only=True)

# 阶段 2: _get_enabled_skills_for_config() - 按 app_config 过滤
if agent_config and agent_config.skills is not None:
    if len(agent_config.skills) == 0:
        return []  # 显式空列表 = 禁用所有技能
    return [s for s in skills if s.name in set(agent_config.skills)]

# 阶段 3: _get_cached_skills_prompt_section() - LRU Cache + signature 去重
skill_signature = frozenset(s.name for s in sorted_skills)
# 如果 signature 没变，直接返回缓存的 Prompt 段落

# 阶段 4: get_skills_prompt_section() - 构建 XML 块
# 阶段 5: apply_prompt_template() - 注入模板

4.2 生成的 XML 结构

<skill_system>
  你有访问技能的权限。技能是 Markdown 文件，包含特定领域的专业知识。
  使用渐进式加载模式：先读 SKILL.md 摘要，需要时再读完整内容。

  <available_skills>
    <skill>
      <name>deep-research</name>
      <description>深度研究技能 [built-in]</description>
      <location>/mnt/skills/public/deep-research/SKILL.md</location>
    </skill>
    <skill>
      <name>my-custom-skill</name>
      <description>自定义技能 [custom, editable]</description>
      <location>/mnt/skills/custom/my-custom-skill/SKILL.md</location>
    </skill>
  </available_skills>

  ## Skill Self-Evolution
  如果启用，Agent 可以自动创建和改进技能...
</skill_system>

关键设计：渐进式加载——Prompt 中只注入技能的名称、描述和路径，不注入完整内容。Agent 需要时才通过 read_file 读取 SKILL.md 的完整内容。这大大节省了 Token 消耗。

4.3 技能过滤的三种模式

# AgentConfig.skills 字段的三种语义:
skills: list[str] | None = None

# None (默认): 加载所有已启用的技能
# [] (空列表): 禁用所有技能
# ["skill1", "skill2"]: 只加载指定技能

这个三态设计非常精妙——None 和 [] 是不同的语义，前者是"不限制"，后者是"明确禁用"。

🤝 五、SubAgent 调度 Prompt：DECOMPOSE → DELEGATE → SYNTHESIZE

在这里插入图片描述

当 subagent_enabled=True 时，Prompt 中会注入一段详细的 SubAgent 调度指令。这段指令的核心逻辑是 DECOMPOSE → DELEGATE → SYNTHESIZE 三步法。

5.1 调度指令的核心规则

<subagent_system>
  你可以将复杂任务分解为并行子任务，使用 task 工具委派给子 Agent。

  ## 硬性限制
  每个响应最多 {max_concurrent_subagents} 个 task 调用。
  超出部分会被静默丢弃！

  ## 调度流程
  1. DECOMPOSE: 在 thinking 中显式计数子任务数量
  2. COUNT: 确认子任务数 ≤ {max_concurrent_subagents}
  3. PLAN BATCHES: 如果超过限制，规划批次
  4. DELEGATE: 只启动当前批次的子任务
  5. WAIT: 等待当前批次完成
  6. REPEAT: 启动下一批次
  7. SYNTHESIZE: 综合所有结果为最终答案
</subagent_system>

5.2 批次规划示例

用户: "帮我调研 AI 编程助手的现状，包括 Cursor、Windsurf、Claude Code、Aider"

Agent thinking:
  这个任务可以分解为 4 个并行子任务:
  1. 调研 Cursor
  2. 调研 Windsurf
  3. 调研 Claude Code
  4. 调研 Aider

  max_concurrent = 3，需要分 2 批:
  Batch 1: Cursor, Windsurf, Claude Code
  Batch 2: Aider

  先启动 Batch 1...

Agent action:
  task("调研 Cursor AI 编程助手", ...)
  task("调研 Windsurf AI 编程助手", ...)
  task("调研 Claude Code", ...)

  [等待 3 个子任务完成...]

  task("调研 Aider", ...)

  [等待完成...]

  SYNTHESIZE: 综合所有调研结果...

5.3 SubagentLimitMiddleware：硬性保障

Prompt 中的限制只是"软约束"——LLM 可能不遵守。SubagentLimitMiddleware 是"硬约束"：

# subagent_limit_middleware.py
class SubagentLimitMiddleware(AgentMiddleware):
    """截断超出的并行 task 调用。"""
    
    def __init__(self, max_concurrent: int = 3):
        self.max_concurrent = max_concurrent
    
    def after_agent(self, state, response, *, config=None):
        # 检查 AI 响应中的 tool_calls
        tool_calls = response.tool_calls if hasattr(response, 'tool_calls') else []
        task_calls = [tc for tc in tool_calls if tc.get('name') == 'task']
        
        if len(task_calls) > self.max_concurrent:
            # 静默丢弃超出的 task 调用
            logger.warning(f"Truncating {len(task_calls)} task calls to {self.max_concurrent}")
            # ...保留前 N 个，丢弃其余

双重保障：Prompt 告诉 LLM “不要超过 N”，Middleware 强制执行"超过 N 就截断"。这种"软硬结合"的设计在 Agent 系统中非常常见。

💡 六、Clarification 澄清系统：5 种必须澄清的场景

在这里插入图片描述

DeerFlow 2.0 的 Clarification 系统是它区别于其他 Agent 框架的关键特性——它不是"能做就做"，而是"不确定就问"。

6.1 五种必须澄清的场景

类型	场景	示例
信息缺失	缺少关键参数	“创建爬虫” 但没指定目标网站
需求模糊	多种理解方式	“优化代码” → 性能? 可读性? 内存?
方案选择	多种技术路线	“添加认证” → JWT? OAuth? Session?
风险确认	危险操作	删除文件 / 修改生产配置 / 数据库操作
建议审批	Agent 的建议需要用户确认	“我建议重构这段代码，是否继续？”

6.2 强制工作流

CLARIFY → PLAN → ACT
   ↑         |
   |_________|  (如果执行中发现新的不清楚之处，回到 CLARIFY)

这个工作流被硬编码在 Prompt 中：

<clarification_system>
  ## 强制规则
  在以下场景中，你必须使用 ask_clarification 工具向用户确认:
  1. 信息缺失: 缺少执行所需的关键参数
  2. 需求模糊: 用户的请求有多种理解方式
  3. 方案选择: 存在多种技术路线，需要用户选择
  4. 风险确认: 操作可能导致数据丢失或不可逆变更
  5. 建议审批: 你的建议需要用户明确同意

  ## 工作流
  CLARIFY → PLAN → ACT
  永远不要在不确定的情况下猜测用户意图。
</clarification_system>

6.3 ClarificationMiddleware：最后一道防线

# clarification_middleware.py
class ClarificationMiddleware(AgentMiddleware):
    """拦截 clarification 请求，在模型调用之后。"""
    
    # 这个 Middleware 被放在 Middleware Pipeline 的最后
    # 确保所有其他 Middleware 处理完毕后，再检查是否需要澄清

它被放在 Pipeline 的最后一位，这意味着所有其他 Middleware（LoopDetection、Memory、Title 等）都处理完毕后，才检查是否需要澄清。

🧩 七、TodoMiddleware：Plan Mode 的任务管理

当 is_plan_mode=True 时，DeerFlow 2.0 会注入一个额外的 TodoMiddleware，让 Agent 具备任务管理能力。

7.1 何时启用

# agent.py
cfg = _get_runtime_config(config)
is_plan_mode = cfg.get("is_plan_mode", False)
todo_list_middleware = _create_todo_list_middleware(is_plan_mode)
if todo_list_middleware is not None:
    middlewares.append(todo_list_middleware)

7.2 TodoMiddleware 的 System Prompt

<todo_list_system>
You have access to the `write_todos` tool to help you manage and track 
complex multi-step objectives.

**CRITICAL RULES:**
- Mark todos as completed IMMEDIATELY after finishing each step
- Keep EXACTLY ONE task as `in_progress` at any time
- Update the todo list in REAL-TIME as you work
- DO NOT use this tool for simple tasks (< 3 steps)

**When to Use:**
- Complex multi-step tasks requiring 3+ distinct steps
- User explicitly requests a todo list
- The plan may need revisions based on intermediate results

**When NOT to Use:**
- Single, straightforward tasks
- Trivial tasks (< 3 steps)
- Purely conversational requests
</todo_list_system>

设计哲学：Todo 不是给简单任务用的——少于 3 步的任务直接做，不需要管理。只有复杂任务才需要 Todo 来追踪进度。

🎯 八、总结：Prompt 即代码

8.1 DeerFlow Prompt 工程的三个核心原则

原则	体现
声明式构建	`apply_prompt_template()` 不是拼接字符串，而是声明式地组装模板
渐进式加载	Skills 只注入摘要，需要时才读完整内容，节省 Token
软硬结合	Prompt 中的规则是"软约束"，Middleware 是"硬约束"，双重保障

8.2 Prompt 层次结构速查

层次	标签	来源	动态性
1	`<role>`	硬编码	agent_name 可变
2	`<soul>`	SOUL.md 文件	每个 Agent 不同
3	`<self_update>`	条件生成	仅自定义 Agent
4	`<memory_context>`	记忆系统	每次对话不同
5	`<thinking_style>`	硬编码	固定
6	`<clarification_system>`	硬编码	固定
7	`<skill_system>`	Skills 存储	技能变更时刷新
8	`<subagent_system>`	条件生成	subagent_enabled 时
9	`<working_directory>`	硬编码	固定
10	`<citations>`	硬编码	固定

8.3 一句话总结

DeerFlow 2.0 的 Prompt 工程把"写 Prompt"变成了"组装 Prompt"——10 层模板、5 阶段管线、3 级缓存，每一层都有独立的生成逻辑和注入时机。SOUL.md 让 Agent 有人格，Skills 让 Agent 有能力，Clarification 让 Agent 有分寸，SubAgent 让 Agent 有帮手。Prompt 不再是一段静态文本，而是一个动态构建的"Agent 操作系统"。

下一篇预告：《DeerFlow 2.0 代码详解（三）：SubAgent 并发执行引擎》——深入拆解 executor.py 的 ThreadPool 并发模型、异步任务状态机、超时取消机制、以及 Lead Agent 与 SubAgent 的通信协议。

作者简介：小李同学_LSH，CSDN博主，专注AI前沿技术解读与开发实战，持续分享LLM应用、Agent开发、深度学习等领域的深度内容。

如果觉得有帮助，欢迎点赞、收藏、关注！你的支持是我持续创作的动力！ 🚀