LeadAgent 工厂模式

DeerFlow 的 agent 创建采用三层工厂架构，从低到高逐层封装：

create_deerflow_agent()     ◀── SDK 层：纯 Python 参数，零配置文件
        ▲
_make_lead_agent()          ◀── 应用层：config.yaml + agent 配置驱动
        ▲
make_lead_agent()           ◀── LangGraph Server 兼容入口层

一、三层工厂

第一层：create_deerflow_agent() — SDK 级 API

定义在 factory.py:61，这是最底层的纯函数工厂。不读任何配置文件，不依赖全局单例。

defcreate_deerflow_agent(
    model: BaseChatModel,
    tools: list[BaseTool] | None = None,
    *,
    system_prompt: str | None = None,
    middleware: list[AgentMiddleware] | None = None,
    features: RuntimeFeatures | None = None,
    extra_middleware: list[AgentMiddleware] | None = None,
    plan_mode: bool = False,
    state_schema: type | None = None,
    checkpointer: BaseCheckpointSaver | None = None,
    name: str = "default",
) -> CompiledStateGraph:

三种使用模式：

middleware 和 features 互斥 —— 给了 middleware 就不能给 features。

RuntimeFeatures — 声明式功能开关

@dataclass
classRuntimeFeatures:
    sandbox:         bool | AgentMiddleware = True# 沙箱（默认开）
    memory:          bool | AgentMiddleware = False
    summarization:   Literal[False] | AgentMiddleware = False# 无默认实现
    subagent:        bool | AgentMiddleware = False
    vision:          bool | AgentMiddleware = False
    auto_title:      bool | AgentMiddleware = False
    guardrail:       Literal[False] | AgentMiddleware = False# 无默认实现
    loop_detection:  bool | AgentMiddleware = True# 循环检测（默认开）
    token_budget:    bool | AgentMiddleware = False

每个字段三种取值：

• • False → 跳过
• • True → 使用内置默认中间件
• • AgentMiddleware 实例 → 使用自定义实现

_assemble_from_features() — 按 features 拼中间件链

# factory.py:155-308，固定顺序拼装
chain:
  [0]  ThreadDataMiddleware       # sandbox 基础设施
  [1]  UploadsMiddleware           # 文件上传处理
  [2]  SandboxMiddleware           # 沙箱管理
  [3]  DanglingToolCallMiddleware  # 总是开启
  [4]  GuardrailMiddleware         # guardrail feature（可选）
  [5]  ToolErrorHandlingMiddleware # 总是开启
  [6]  SummarizationMiddleware     # summarization feature（可选）
  [7]  TodoMiddleware              # plan_mode 参数
  [8]  TitleMiddleware             # auto_title feature（可选）
  [9]  MemoryMiddleware            # memory feature（可选）
  [10] ViewImageMiddleware         # vision feature（可选）
  [11] SubagentLimitMiddleware     # subagent feature（可选）
  [12] LoopDetectionMiddleware     # loop_detection feature（可选）
  [13] TokenBudgetMiddleware       # token_budget feature（可选）
  [14] ClarificationMiddleware     # 总是最后，保底

同时自动注入工具：vision 开启时注入 view_image_tool，subagent 开启时注入 task_tool，clarification 总是注入 ask_clarification_tool。用户提供的工具优先级更高（按名称去重）。

第二层：_make_lead_agent() — 应用级工厂

定义在 lead_agent/agent.py:423，这是 DeerFlow 服务端真正用的工厂。

def_make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):

跟 create_deerflow_agent() 的关键区别：

三路分支逻辑

_make_lead_agent(config)
  │
  ├─ is_bootstrap? → 最小化 agent，只暴露 bootstrap skill + setup_agent 工具
  │   用于首次自动创建自定义 agent
  │
  ├─ agent_name 指定? → custom agent
  │   加载 agent_config（模型、工具组、skills 等）
  │   注入 update_agent 工具（允许 agent 自己修改配置）
  │
  └─ 否则 → default agent
      使用全局默认模型和工具配置

模型解析三级链路

model_name = _resolve_model_name(
    requested_model_name  # 1. 用户请求中指定的模型
or agent_model_name   # 2. agent 配置中指定的模型
)
# 3. 如果上面都没有 → 取全局 config.yaml models[0]

每一级如果配置的模型名在全局配置中不存在 → 打 warning + fallback 到默认模型。

第三层：make_lead_agent() — LangGraph Server 兼容入口

# lead_agent/agent.py:416
defmake_lead_agent(config: RunnableConfig):
    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 要求 graph factory 的签名必须是 (config: RunnableConfig) -> CompiledStateGraph。make_lead_agent 就是这个兼容层 —— 它从 config 中提取 app_config，转发给 _make_lead_agent。

Gateway 中，resolve_agent_factory() 始终返回 make_lead_agent —— 不管 assistant_id 是什么，同一个工厂。区分不同 agent 的逻辑在工厂内部，而非选择不同的工厂函数。

_get_runtime_config() — 合并两个数据源

# lead_agent/agent.py:55
def_get_runtime_config(config: RunnableConfig) -> dict:
"""Merge legacy configurable options with LangGraph runtime context."""
    cfg = dict(config.get("configurable", {}) or {})
    context = config.get("context", {}) or {}
ifisinstance(context, dict):
        cfg.update(context)
return cfg

它将 RunnableConfig 中的两个字典合并为一个：

合并时 context 的值覆盖configurable 中的同名 key（cfg.update(context) 在后面执行）。

runtime_app_config — 从运行时上下文中提取 AppConfig

runtime_app_config = runtime_config.get("app_config")

app_config 作为 AppConfig 实例被注入到 config["context"] 中，有两个注入点：

注入点 1：Worker 层（主流程）

run_agent 启动时，build_runtime_context() 构建 ToolRuntime.context，将当前的 AppConfig 实例写入：

# runtime/runs/worker.py:67
if app_config isnotNone:
    runtime_ctx["app_config"] = app_config

然后 _install_runtime_context() 将其安装到 config["context"] 中：

# runtime/runs/worker.py:94
if"app_config"in runtime_context:
    existing_context["app_config"] = runtime_context["app_config"]

注入点 2：子智能体执行器（Subagent）

子智能体启动时将父级的 app_config 传入子运行的 context：

# subagents/executor.py:586
ifself.app_config isnotNone:
    context["app_config"] = self.app_config

这确保子智能体继承父级配置上下文，而非重新从文件加载。

优先级逻辑

app_config=runtime_app_config or get_app_config()

二、中间件拼装机制

原则 1：固定顺序

中间件链的顺序是固定的，两个工厂（SDK _assemble_from_features 和应用层 build_middlewares）都遵循相同的逻辑：

基础设施 → 上传 → 沙箱 → dangling → guardrail → 错误处理
→ 摘要 → todo → title → memory → vision → skill → 动态上下文
→ 延迟过滤 → 消息合并 → subagent → 循环检测 → token预算
→ 自定义中间件 → safety → clarification(最后)

原则 2：@Next / @Prev 定位

调用方如果想在固定链中插入自定义中间件，用装饰器声明插入位置：

@Next(SummarizationMiddleware)    # 插入到 SummarizationMiddleware 之后
classMyCustomMiddleware(AgentMiddleware):
    ...

# 然后：
create_deerflow_agent(
    features=RuntimeFeatures(...),
    extra_middleware=[MyCustomMiddleware()],
)

_insert_extra() 在 factory.py:316 处理插入逻辑：

def_insert_extra(chain, extras):
"""
    1. 验证：不能同时用 @Next 和 @Prev
    2. 冲突检测：两个 extras 锚定同一位置（同向或反向）→ 报错
    3. 无锚定 → 插入到 ClarificationMiddleware 之前
    4. 有锚定 → 迭代插入（支持外部中间件之间相互锚定）
    5. 锚定无法解析 → 报错
    """

约束规则：

• • 不能同时用 @Next 和 @Prev
• • 两个中间件不能锚定同一个位置（同一方向或相反方向）
• • 支持外部中间件之间相互锚定
• • ClarificationMiddleware 始终最后 —— 即使 @Next(ClarificationMiddleware) 推走了它，也自动归位

# factory.py:304
clar_idx = next(i for i, m inenumerate(chain) ifisinstance(m, ClarificationMiddleware))
if clar_idx != len(chain) - 1:
    chain.append(chain.pop(clar_idx))

原则 3：ClarificationMiddleware 必须最后

不管怎么插，ClarificationMiddleware 永远是链尾。它处理"模型叫你问用户"这种场景 —— 必须是最后一个看回复的中间件，否则它注入的提示消息可能被后续中间件误处理。

三、完整调用链路

一个用户请求从到底到上层：

1. LangGraph Server / Gateway
     │
     └─ resolve_agent_factory() → 返回 make_lead_agent
          │
2. make_lead_agent(config)
     │  提取 runtime context 中的 app_config
     │
     └─ _make_lead_agent(config, app_config=...)
          │
          ├─ 解析 runtime config:
          │    model_name, thinking_enabled, plan_mode, subagent, agent_name ...
          │
          ├─ 解析模型:
          │    _resolve_model_name(requested → agent config → global default)
          │
          ├─ 加载 agent 配置 (load_agent_config)
          │    ├─ bootstrap → 特殊路径
          │    ├─ custom agent → agent_config 中的 model/skills/tool_groups
          │    └─ default → 全局默认
          │
          ├─ 加载工具:
          │    ├─ get_available_tools() 从 MCP/Skill 注册表加载
          │    ├─ filter_tools_by_skill_allowed_tools() Skill 策略裁剪
          │    └─ assemble_deferred_tools() 分离 MCP deferred 工具
          │
          ├─ 构建中间件链:
          │    build_middlewares(config, model_name, agent_name, ...)
          │    ├─ build_lead_runtime_middlewares()  沙箱基础设施
          │    ├─ DynamicContextMiddleware          动态上下文
          │    ├─ SkillActivationMiddleware         /skill 激活
          │    ├─ SummarizationMiddleware           上下文摘要
          │    ├─ TodoMiddleware                    plan_mode 任务追踪
          │    ├─ TokenUsageMiddleware              令牌用量统计
          │    ├─ TitleMiddleware                   自动标题
          │    ├─ MemoryMiddleware                  记忆管理
          │    ├─ ViewImageMiddleware               视觉图片处理
          │    ├─ DeferredToolFilterMiddleware      延迟工具过滤
          │    ├─ SystemMessageCoalescingMiddleware  消息合并
          │    ├─ SubagentLimitMiddleware           子 agent 限制
          │    ├─ LoopDetectionMiddleware           循环检测
          │    ├─ TokenBudgetMiddleware             令牌预算
          │    ├─ SafetyFinishReasonMiddleware      安全终止
          │    └─ ClarificationMiddleware           澄清请求（最后）
          │
          ├─ 构建系统提示:
          │    apply_prompt_template(...)
          │
          ├─ 注入追踪回调:
          │    build_tracing_callbacks() → Langfuse/LangSmith
          │
          └─ 调用底层:
               create_agent(model, tools, middleware, system_prompt, ...)
               → CompiledStateGraph

四、设计哲学

五、设计问答

Q1: 工厂函数每次请求都重新创建 Agent，会不会很慢？工具列表每次都要重新加载吗？

不会慢，有三个原因。

LangGraph 的 CompiledStateGraph 创建很轻量

create_agent() 做的事情是：构建一个 LangGraph 图的数据结构（节点 + 边 + channel 定义）。它不加载 ML 模型、不初始化推理引擎、不分配显存。本质上就是在内存中搭一个 DAG 结构：

create_agent(model, tools, middleware, ...)
  → 构建 StateGraph
  → 添加 model_node, tools_node
  → 添加边
  → compile() → CompiledStateGraph (纯数据结构)

这个过程的代价是：几十个 Python 对象的创建 + 图的拓扑排序。实际测量在几十毫秒以内。

LLM 模型不是每次都重新创建

# _make_lead_agent 内部
model = create_chat_model(name=model_name, ...)  # 创建的是 LangChain ChatModel 包装器

create_chat_model 创建的是 ChatOpenAI / ChatAnthropic 等 LangChain 包装器，不是加载实际的模型权重。API 调用是走到远程的，模型推理在服务端。真正耗时的网络 I/O 发生在 agent.astream() 调用时，不在工厂创建阶段。

工具列表确实每次都重新加载，但有模块缓存兜底

# tools/tools.py:67
tool_configs = [tool for tool in config.tools if groups isNoneor tool.group in groups]
loaded_tools_raw = [(cfg, resolve_variable(cfg.use, BaseTool)) for cfg in tool_configs]

resolve_variable(cfg.use, BaseTool) 每次都会做 import + 实例化。不过 Python 的 import 机制有模块级缓存（sys.modules），所以同一个工具类的模块代码只执行一次。重复的是实例化，而工具实例化通常很轻量（就是 __init__ 几个字段）。

# worker.py:252 —— 没有任何缓存
agent = agent_factory(config=runnable_config)  # 每次请求重建

设计者显然认为这个开销可以接受。如果未来 MCP 工具数量暴涨成为瓶颈，一层 @lru_cache 按 (model_name, agent_name, plan_mode, tool_groups_hash) 做 key 就能解决。

Q2: 如何创建一个只能搜索网络、不能写文件的 Agent？

改 agent 配置的 tool_groups，不写代码。

原理

config.yaml 中每个工具都标注了 group：

tools:
-name:web_search
group:web# ← 搜索工具
use:deerflow.community.ddg_search.tools:web_search_tool

-name:read_file
group:file:read# ← 读文件
use:deerflow.tools.builtins:read_file_tool

-name:write_file
group:file:write# ← 写文件
use:deerflow.tools.builtins:write_file_tool

-name:bash
group:bash# ← 执行命令
use:deerflow.sandbox.tools:bash_tool

Agent 配置中通过 tool_groups 白名单控制哪些工具可见：

agents:
-name:search-only-agent
model:gpt-4o
tool_groups:# 只加载这些 group 的工具
-web
skills: []

过滤链路

_make_lead_agent()
  │
  ├─ agent_config.tool_groups = ["web"]
  │
  └─ get_available_tools(groups=["web"])
       │
       ├─ tool_configs = [t for t in config.tools if t.group in ["web"]]
       │   → 只保留 group=web 的工具：web_search, web_fetch...
       │   → 排除 file:read, file:write, bash...
       │
       └─ resolve_variable → 实例化 → 返回工具列表

完整配置示例

# config.yaml
agents:
-name:search-only
description:"只能搜索网络，不能读写文件或执行命令"
model:gpt-4o
tool_groups:
-web
skills: []

创建后，这个 agent 只能看到 web_search、web_fetch 等 group: web 的工具。其他一切工具（读文件、写文件、bash、task_tool 等）都不在 groups=["web"] 白名单中，直接过滤掉了。

如果还要禁止 present_files（因为它也会"写出"文件），它在 BUILTIN_TOOLS 中默认属于特定 group，groups 过滤会自动排除它。

扩展：如果需要更细粒度的控制

如果 tool_groups 不够精确，可以在 _make_lead_agent 的工具加载后做额外过滤，或者通过 Skill 的 allowed_tools 机制做进一步裁剪（filter_tools_by_skill_allowed_tools）。

Q3: lazy_init=True 指什么？为什么需要它？

lazy_init=True 同时传给两个中间件：

为什么需要

减少无意义的资源开销。 大部分对话不涉及文件操作和代码执行。

场景：用户问 "Python 的 list 和 tuple 有什么区别？"

eager (lazy_init=False):
  before_agent → 创建沙箱容器（启动 Docker/Podman 容器）→ 50ms-2s
  before_agent → 创建目录 /threads/xxx/workspace /uploads /outputs
  模型回复 → 不需要任何工具 → 沙箱白创建了

lazy (lazy_init=True):
  before_agent → 只算路径（纯字符串拼接，纳秒级）
  模型回复 → 不需要任何工具 → 零额外开销

ThreadDataMiddleware 的懒初始化

# thread_data_middleware.py:94
ifself._lazy_init:
    paths = self._get_thread_paths(thread_id)  # 纯计算，不 mkdir
else:
    paths = self._get_thread_paths(thread_id)
for p in paths.values():
        os.makedirs(p, exist_ok=True)  # 立即创建目录

懒模式下，目录在使用时由相应的工具自行创建（如 write_file 打开文件前 os.makedirs）。

SandboxMiddleware 的懒初始化

# sandbox/middleware.py:69
defbefore_agent(self, state, runtime):
ifself._lazy_init:
returnsuper().before_agent(state, runtime)  # 跳过沙箱创建
# eager: 立即创建沙箱

# wrap_tool_call 中：
# 检测到 sandbox 为 None → 首次工具调用时惰性创建
# 创建后将 sandbox_id 写入 Command(update={"sandbox": {...}})

关键逻辑在 wrap_tool_call：工具执行前检查 state["sandbox"] 是否为 None，如果是则调用 _acquire_sandbox() 创建，并通过 merge_sandbox reducer 写入图状态。

性能对比

对于 web 搜索类 agent（只用 API 不用沙箱），lazy_init=True 意味着沙箱永远不会被创建——这是巨大的性能节省。

为什么还要保留 eager 模式？

lazy_init=False 主要用于：

• • 测试环境：确定性行为，方便断言沙箱状态
• • 安全审计：确保每个请求都有沙箱记录，即使模型没有显式调用工具
• • 特定部署场景：某些沙箱 provider 的创建有冷启动延迟，eager 创建让延迟在 before_agent 阶段发生，用户感知的"第一个工具调用"更快

总结

lazy_init=True 的本质是 "按需分配"——不在 before_agent 时预分配资源，而是在真正需要的那一刻才创建。这对 DeerFlow 的混合负载（对话 + 工具执行）是最优策略：大部分请求根本不需要沙箱和文件目录。

六、config、agent_config、app_config 三级配置体系

_make_lead_agent 内部实际涉及三层配置，从上到下越来越"稳定"：

config (RunnableConfig)        ← 请求级："用户这次想要什么"
    │
    ├─ agent_name ──→ agent_config (AgentConfig)  ← Agent 级："这个 Agent 的特化设置"
    │
    └─ app_config (AppConfig)                      ← 系统级："服务器全局基础设施"

三者对比

三者的"管辖范围"

┌─────────────────────────────────────────────────────┐
│  app_config（系统级）                                  │
│  "服务器有什么"                                        │
│  ─ models 池、全量工具定义、沙箱、中间件参数              │
│  ─ 管理员通过 config.yaml 控制                         │
│    ┌───────────────────────────────────────────┐     │
│    │  agent_config（Agent 级）                    │     │
│    │  "这个家伙能干什么"                           │     │
│    │  ─ 从 app_config 的全量池中做减法              │     │
│    │  ─ tool_groups 白名单、skills 白名单          │     │
│    │    ┌─────────────────────────────────┐     │     │
│    │    │  config（请求级）                  │     │     │
│    │    │  "用户这次要什么"                   │     │     │
│    │    │  ─ model_name 临时覆盖             │     │     │
│    │    │  ─ thinking/plan 开关             │     │     │
│    │    │  ─ 选择哪个 agent                 │     │     │
│    │    └─────────────────────────────────┘     │     │
│    └───────────────────────────────────────────┘     │
└─────────────────────────────────────────────────────┘

核心关系：config 指定"用哪个 agent"，agent_config 从 app_config 的全局资源池中做减法（限定工具和技能），config 再做临时覆盖（如换模型）。app_config 是所有校验的最终裁决者 —— 用户和 agent 都不能突破系统能力的边界。

在 _make_lead_agent 中的具体交互

# 1. 从 config 中提取 agent_name（用户这次想用哪个 agent）
cfg = _get_runtime_config(config)          # ← 来自请求
agent_name = validate_agent_name(cfg.get("agent_name"))

# 2. 用 agent_name 加载 agent_config（该 agent 的特化设置）
agent_config = load_agent_config(agent_name) ifnot is_bootstrap elseNone
#  从文件系统 {base_dir}/users/{uid}/agents/{name}/config.yaml 读取

# 3. 三级模型解析：config → agent_config → app_config
requested_model_name = cfg.get("model_name")          # ① 用户本次直接指定
agent_model_name = agent_config.model if agent_config elseNone# ② agent 配置的默认模型
model_name = _resolve_model_name(                     # ③ app_config.models[0] 兜底
    requested_model_name or agent_model_name,
    app_config=resolved_app_config
)

# 4. 验证：用户意图 vs 系统能力
model_config = resolved_app_config.get_model_config(model_name)
if thinking_enabled andnot model_config.supports_thinking:  # app_config 说了算
    thinking_enabled = False# 降级

# 5. 工具组：agent_config.tool_groups 过滤 app_config.tools
raw_tools = get_available_tools(
    groups=agent_config.tool_groups if agent_config elseNone,  # agent 级白名单
    app_config=resolved_app_config                             # 全局工具池
)

# 6. Skills：agent_config.skills 过滤全局 skill 列表
available_skills = _available_skill_names(agent_config, is_bootstrap)
# None → 全部启用, [] → 全部禁用, ["a","b"] → 仅启用指定

实例场景

假设一个搜索专用 agent，用户请求时临时指定了 deepseek-v3 模型：

config = {
    "model_name": "deepseek-v3",   # ← 用户这次临时指定
    "agent_name": "search-bot",    # ← 路由到 search-bot
    "thinking_enabled": False,
}
    │
    ├─ agent_name="search-bot" ──→ load_agent_config("search-bot")
    │      agent_config = {
    │          name: "search-bot",
    │          model: "gpt-4o",               # search-bot 默认模型
    │          tool_groups: ["web"],           # 只能用搜索工具
    │          skills: ["web-research"],       # 只加载研究技能
    │      }
    │
    └─ app_config = {
           models: [{name:"deepseek-v3", ...}, {name:"gpt-4o", ...}],
           tools: [web_search, read_file, write_file, bash, ...],
           skills: [web-research, code-review, ...],
       }

最终生效：
  模型: deepseek-v3           ← config 优先于 agent_config.model (gpt-4o)
  工具: [web_search]           ← agent_config.tool_groups 从 app_config.tools 筛选
  技能: [web-research]         ← agent_config.skills 从 app_config 全局技能筛选
  thinking: False              ← config 指定，app_config 验证 deepseek-v3 不支持

在 _make_lead_agent 中的实际分工

_make_lead_agent(config, app_config)
  │
  ├─ cfg = _get_runtime_config(config)   ← 提取用户意图
  │    ├─ thinking_enabled  ── 用户要不要深度思考
  │    ├─ model_name        ── 用户指定了哪个模型
  │    ├─ is_plan_mode      ── 是否开启计划模式
  │    ├─ subagent_enabled  ── 是否允许子智能体
  │    ├─ agent_name        ── 使用哪个自定义智能体
  │    └─ is_bootstrap      ── 是否为 bootstrap 流程
  │
  ├─ resolved_app_config = app_config    ← 静态基础设施
  │    ├─ .models            ── 验证模型是否存在、是否支持vision/thinking
  │    ├─ .tools             ── 可用工具列表
  │    ├─ .tool_search       ── 是否启用延迟工具加载
  │    ├─ .memory.enabled    ── 记忆中间件是否启用
  │    ├─ .summarization     ── 摘要中间件配置
  │    ├─ .loop_detection    ── 循环检测参数
  │    ├─ .token_budget      ── Token预算限制
  │    ├─ .safety_finish_reason ── 安全拦截配置
  │    └─ .sandbox           ── 沙箱路径/提供者
  │
  └─ 交叉验证
       ├─ model_name 来自 cfg → app_config.get_model_config() 验证存在性
       ├─ thinking_enabled 来自 cfg → model_config.supports_thinking 验证可行性
       └─ agent_name 来自 cfg → load_agent_config() + app_config 加载 agent 配置

关键使用场景对照：

为什么这样设计

          ┌─────────────────────────────────┐
          │          config                  │
          │  "用户这次想要什么"                 │
          │  ─ 每次请求不同                     │
          │  ─ 用户可控（通过API参数）            │
          │  ─ 生命周期 = 一次 run              │
          │  ─ 可被前端/IM/子智能体任意覆盖        │
          └─────────────────────────────────┘
                         │
                         ▼
          ┌─────────────────────────────────┐
          │        app_config                │
          │  "系统有什么能力"                   │
          │  ─ 服务器启动时加载                 │
          │  ─ 管理员可控（通过config.yaml）      │
          │  ─ 生命周期 = 进程级别               │
          │  ─ 运行时不可由用户修改               │
          └─────────────────────────────────┘

1. 安全性 — 用户不能通过 API 参数修改服务器基础设施。模型 API Key、沙箱路径、数据库连接等只存在于 app_config，从未暴露给 config。

2. 权限校验 — 用户通过 config 传入 model_name: "gpt-5"，但如果 app_config.models 中没有该模型，get_model_config() 返回 None，直接降级到默认模型并打 warning：

# agent.py:447-453
model_name = _resolve_model_name(requested_model_name or agent_model_name, app_config=resolved_app_config)
model_config = resolved_app_config.get_model_config(model_name)
if thinking_enabled andnot model_config.supports_thinking:
    logger.warning(f"Thinking mode is enabled but model '{model_name}' does not support it")
    thinking_enabled = False# 用户意图 vs 系统能力 → 降级

3. 子智能体继承 — 父运行通过 context["app_config"] 将配置传给子智能体，子智能体的 make_lead_agent 读取 runtime_app_config，使用传入的实例而非重新调用 get_app_config()：

父 run:
  config["context"]["app_config"] = AppConfig实例  ← worker 注入

  子 agent 工厂:
    make_lead_agent(config)
      → runtime_config["app_config"]  ← 读到父级传入的实例
      → _make_lead_agent(config, app_config=父级实例)  ← 跳过 get_app_config()

4. 可测试性 — _make_lead_agent 的 app_config 是显式参数，测试中可以直接传入 mock/构造的 AppConfig，无需依赖文件系统上的 config.yaml。

5. config 被可变修改有原因 — _make_lead_agent 向 config["metadata"] 写入追踪标签、向 config["callbacks"] 注入 Langfuse/LangSmith 回调。这些信息需要在 LangGraph 运行的整个生命周期内被下游节点和回调处理器读取，所以必须写回 config 对象本身。

6. 单一入口，内部路由 — 不管是 custom agent、bootstrap 还是 default，Gateway 都调用同一个 make_lead_agent(config)。路由信息（agent_name、is_bootstrap）通过 config["configurable"] 传入，在工厂内部读取后分流。这满足了 LangGraph Server 只接受一个工厂函数签名的约束。

七、SOUL.md 与 Agent 人格系统

SOUL.md 定义了 agent 的人格、价值观和行为约束，通过 apply_prompt_template() 注入到 system prompt 的 <soul> 块中。

文件位置

关键代码路径

# prompt.py:699-704 — 注入到 system prompt
defget_agent_soul(agent_name: str | None) -> str:
    soul = load_agent_soul(agent_name)  # → agents_config.py
if soul:
returnf"<soul>\n{soul}\n</soul>\n"
return""# 文件不存在 → 空字符串，prompt 中无 soul 块

# agents_config.py:131-153 — 文件读取
defload_agent_soul(agent_name, ...):
if agent_name:
        agent_dir = resolve_agent_dir(agent_name, ...)  # per-user 目录
else:
        agent_dir = get_paths().base_dir                   # ← 全局 base_dir
    soul_path = agent_dir / "SOUL.md"
ifnot soul_path.exists():
returnNone
return soul_path.read_text(...).strip() orNone

默认 agent：如果 {base_dir}/SOUL.md 不存在，prompt 中就没有 soul 块；有则注入。系统中没有预置的默认 SOUL.md，默认 agent 的身份仅由 system prompt 模板中的 {agent_name} 决定（默认值 "DeerFlow 2.0"）。

三种生成途径

途径 1：setup_agent 工具 — Bootstrap 首次创建

Bootstrap 模式下（is_bootstrap=True），LLM 持有 setup_agent 工具，与用户对话后自动创建 agent：

# setup_agent_tool.py:80-81
soul_file = agent_dir / "SOUL.md"
soul_file.write_text(soul, encoding="utf-8")  # LLM 生成的 soul 内容

# 同时生成 config.yaml
config_file = agent_dir / "config.yaml"
yaml.dump({"name": agent_name, "description": ..., "skills": ...}, ...)

如果有 agent_name，写入 {base_dir}/users/{uid}/agents/{name}/；如果无（默认 agent），写入 {base_dir}/。

途径 2：update_agent 工具 — 运行时自我更新

自定义 agent 运行时可调用 update_agent 修改自己的 SOUL.md。采用原子写入策略（先写 temp 文件，再 os.replace），防止写入中途崩溃导致文件损坏：

# update_agent_tool.py:213-217
if soul isnotNone:
    soul_target = agent_dir / "SOUL.md"
    soul_tmp = _stage_temp(soul_target, soul)  # 先写临时文件
    pending.append((soul_tmp, soul_target))    # 再原子 rename

update_agent 的绑定条件：只有自定义 agent（agent_name 非空）才能看到该工具。

# agent.py:529
extra_tools = [update_agent] if agent_name else []

途径 3：手动创建 / HTTP API

通过 PUT /api/agents/{name} API（需 agents_api.enabled=true）或直接在文件系统中创建。

生命周期示意

Bootstrap 流程（首次创建 agent）:
  用户说"帮我创建一个代码审查 agent"
    → is_bootstrap=True
    → LLM 持有 setup_agent 工具
    → LLM 调用 setup_agent(soul="你是专业的代码审查助手...", ...)
    → 写入 /users/{uid}/agents/code-reviewer/SOUL.md
    → 写入 /users/{uid}/agents/code-reviewer/config.yaml

运行时更新:
  用户对 code-reviewer agent 说"以后用中文回答"
    → agent 调用 update_agent(soul="你是专业的代码审查助手，用中文回答...")
    → 原子写入新 SOUL.md（先 temp 后 rename）
    → 下一次请求 rebuild agent 时生效

默认 agent:
  agent_name=None
    → 读取 {base_dir}/SOUL.md
    → 不存在？prompt 中无 soul 块，仅以 "DeerFlow 2.0" 身份出现
    → 存在？prompt 中注入 <soul>...</soul>

八、get_app_config() — 全局配置单例详解

get_app_config()（app_config.py:497）是 deerflow 获取全局配置的唯一入口，实现了缓存、热重载和运行时覆盖三层机制。

模块级缓存变量

# app_config.py:448-454
_app_config: AppConfig | None = None# 缓存的 AppConfig 实例
_app_config_path: Path | None = None# 上次加载的源文件路径
_app_config_mtime: float | None = None# 上次加载时的文件 mtime
_app_config_signature: tuple | None = None# (mtime, size, sha256) 三元组签名
_app_config_is_custom = False# True = 手动注入，不检测文件变更
_current_app_config: ContextVar[...] = ...        # 运行时覆盖（ContextVar，协程安全）

两个辅助函数判断是否需要重载：

• • _get_config_mtime(path) — 返回文件 st_mtime，文件不存在返回 None
• • _get_config_signature(path) — 返回 (mtime, size, sha256)。用 SHA256 内容摘要而非仅 mtime，因为某些部署环境（容器挂载、CI）的 mtime 可能不准确

三级优先级

get_app_config()
  │
  ├─ ① ContextVar 运行时覆盖 (_current_app_config)
  │     └─ 存在？直接返回 → 跳过所有文件和缓存检查
  │     └─ 用途：子智能体注入 (push/pop_current_app_config)
  │
  ├─ ② 自定义注入实例 (_app_config_is_custom=True)
  │     └─ 存在？直接返回 → 不检测文件变更
  │     └─ 用途：测试环境 set_app_config(mock_config)
  │
  └─ ③ 文件缓存 + 热重载
        ├─ 首次调用 → _load_and_cache_app_config()
        ├─ 文件路径变了？ → 重载
        ├─ 文件内容签名变了？ → 重载
        └─ 都没变？ → 直接返回缓存 _app_config (零开销)

核心源码

defget_app_config() -> AppConfig:
global _app_config, _app_config_path, _app_config_mtime, _app_config_signature

# ① 运行时 ContextVar 覆盖 — 优先级最高
    runtime_override = _current_app_config.get()
if runtime_override isnotNone:
return runtime_override

# ② 注入的自定义实例 — 不检测文件变更
if _app_config isnotNoneand _app_config_is_custom:
return _app_config

# ③ 文件缓存 + 热重载
    resolved_path = AppConfig.resolve_config_path()
    current_mtime = _get_config_mtime(resolved_path)
    current_signature = _get_config_signature(resolved_path)

    should_reload = (
        _app_config isNone# 首次加载
or _app_config_path != resolved_path   # 路径变了
or _app_config_signature != current_signature  # 文件内容变了
    )
if should_reload:
if _app_config_path == resolved_path and _app_config_mtime != current_mtime:
            logger.info("Config file has been modified (mtime: %s -> %s), reloading", ...)
elif _app_config_path == resolved_path and _app_config_signature != current_signature:
            logger.info("Config file content signature changed, reloading AppConfig")
        _load_and_cache_app_config(str(resolved_path))

return _app_config

_load_and_cache_app_config() — 加载与缓存刷新

def_load_and_cache_app_config(config_path=None) -> AppConfig:
    resolved_path = AppConfig.resolve_config_path(config_path)  # 解析路径
    _app_config = AppConfig.from_file(str(resolved_path))       # YAML → Pydantic
    _app_config_path = resolved_path                            # 记录路径
    _app_config_mtime = _get_config_mtime(resolved_path)        # 记录 mtime
    _app_config_signature = _get_config_signature(resolved_path)# 记录签名
    _app_config_is_custom = False# 标记为文件来源
return _app_config

路径解析优先级：config_path 参数 → DEER_FLOW_CONFIG_PATH 环境变量 → 项目根目录 config.yaml → 旧版兼容路径。

配套 API

push/pop 实现堆栈语义，支持嵌套：

# 子智能体中临时覆盖
push_current_app_config(sub_config)   # 压入
# ... 子任务执行 ...
pop_current_app_config()              # 恢复父级

与工厂的关联

make_lead_agent(config)
  │
  ├─ runtime_config["app_config"]  ← 优先取 Worker/Subagent 运行时注入的实例
  │
  └─ or get_app_config()          ← 回退到全局单例
       │
       ├─ ContextVar 覆盖？    → 直接返回（子智能体继承场景）
       ├─ 自定义注入实例？    → 直接返回（测试场景）
       └─ 文件缓存 → 签名比较 → 热重载或返回缓存（生产常态）

性能特性：正常运行时签名不变，get_app_config() 只走 return _app_config，零文件 IO，零解析开销。仅在文件实际变更时才触发 YAML 解析 + Pydantic 验证。

九、apply_prompt_template() — System Prompt 组装工厂

apply_prompt_template()（prompt.py:779）是 prompt 的组装入口，将多个独立的提示片段拼装成完整的 system prompt 字符串，传给 create_agent()。

核心设计：全静态 prompt

# prompt.py:821-824
# Memory and current date are injected per-turn via DynamicContextMiddleware
# as a <system-reminder> in the first HumanMessage, keeping this prompt
# identical across users and sessions for maximum prefix-cache reuse.

函数生成的 prompt 完全不变。动态内容（记忆、当前日期）由 DynamicContextMiddleware 在 before_agent 阶段注入到 HumanMessage 中，而非修改 system prompt。这最大化 LLM 前缀缓存（prefix cache）命中率——所有用户、所有会话共享同一份系统提示，缓存永不失效。

函数签名

defapply_prompt_template(
    subagent_enabled: bool = False,           # 是否启用子智能体（来自 cfg）
    max_concurrent_subagents: int = 3,        # 每轮最大并发数（来自 cfg）
    *,
    agent_name: str | None = None,            # agent 名（来自 cfg，None=默认）
    available_skills: set[str] | None = None, # 可用技能白名单（来自 agent_config）
    app_config: AppConfig | None = None,      # 全局配置（来自 _make_lead_agent）
    deferred_names: frozenset[str] = ...,     # 延迟工具名集合（来自 assemble_deferred_tools）
) -> str:

模板占位符与组装逻辑

模板 SYSTEM_PROMPT_TEMPLATE 包含 {placeholder} 占位符，函数逐段计算后统一填充：

SYSTEM_PROMPT_TEMPLATE = """
<role>            {agent_name}
<soul>            {soul}                  ← get_agent_soul(agent_name)
<self_update>     {self_update_section}   ← _build_self_update_section(agent_name)
<thinking_style>  {subagent_thinking}     ← 子智能体条件注入
<clarification_system>                   ← 完全静态
{skills_section}                          ← get_skills_prompt_section()
{deferred_tools_section}                  ← get_deferred_tools_prompt_section()
{subagent_section}                        ← _build_subagent_section() 或 ""
<working_directory>  {acp_section}        ← ACP agent + 自定义挂载
<response_style>                          ← 完全静态
<citations>                               ← 完全静态
<critical_reminders>  {subagent_reminder} ← 子智能体条件注入
"""

return SYSTEM_PROMPT_TEMPLATE.format(
    agent_name=agent_name or"DeerFlow 2.0",       # 默认身份
    soul=get_agent_soul(agent_name),                # 从 SOUL.md 读取
    self_update_section=_build_self_update_section(agent_name),  # 仅自定义 agent
    skills_section=skills_section,                  # 技能列表（带 LRU 缓存）
    deferred_tools_section=deferred_tools_section,  # 延迟工具使用说明
    subagent_section=subagent_section,              # 条件注入
    subagent_reminder=subagent_reminder,            # 条件注入
    subagent_thinking=subagent_thinking,            # 条件注入
    acp_section=acp_and_mounts_section,             # ACP + 挂载
)

各子函数职责

子智能体三段条件注入

子智能体启用时，三个位置同时注入：

subagent_enabled=True:
  subagent_section   → 完整 <subagent_system> 块
  subagent_reminder  → "HARD LIMIT: max N task calls per response"
  subagent_thinking  → "DECOMPOSITION CHECK: Can this task be broken..."

subagent_enabled=False:
  上述三个全部为空字符串 ""

Skills 的 LRU 缓存

get_skills_prompt_section() → _get_cached_skills_prompt_section() 使用 @lru_cache(maxsize=32)，缓存键是技能列表的内容签名（名称+描述+类别+路径），不同 agent 共用相同技能集时直接命中。技能文件更新时通过 _invalidate_enabled_skills_cache() 失效。

Memory 和 Date 不在这个函数中

动态上下文通过 DynamicContextMiddleware 在 before_agent 阶段注入到第一条 HumanMessage 前：

<system-reminder>
<memory>用户的个性化记忆内容...</memory>
<current_date>2026-07-01, Tuesday</current_date>
</system-reminder>

与工厂的关联

# agent.py:545-551 — 传入的参数来源
system_prompt=apply_prompt_template(
    subagent_enabled=subagent_enabled,           # ← cfg
    max_concurrent_subagents=max_concurrent_subagents,  # ← cfg
    agent_name=agent_name,                       # ← cfg → agent_config
    available_skills=available_skills,            # ← agent_config → app_config
    app_config=resolved_app_config,               # ← _make_lead_agent 接收
    deferred_names=setup.deferred_names,          # ← assemble_deferred_tools()
)