DeerFlow Harness 后端 Python 技术架构分析

1 系统架构总览

DeerFlow Harness 是基于 LangGraph/LangChain 构建的企业级 Agent 运行时框架，采用分层架构与中间件管线模式。

┌─────────────────────────────────────────────────────────────────┐ │ Gateway (FastAPI) │ │ ┌──────────┐ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │ │ │ Threads │ │ Runs │ │ Uploads │ │ MCP Config │ │ │ └──────────┘ └──────────┘ └───────────┘ └──────────────┘ │ └──────────────────────────┬──────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────┐ │ Lead Agent (LangGraph StateGraph) │ │ │ │ 14-Layer Middleware Pipeline │ │ ThreadData → Uploads → Sandbox → DanglingToolCall → │ │ Guardrail → ToolError → Summarization → Todo → Title → │ │ Memory → ViewImage → SubagentLimit → LoopDetection → │ │ Clarification │ │ │ │ ┌─────────────┐ ┌──────────────┐ ┌───────────────────┐ │ │ │ LLM 调用 │ │ 工具执行 │ │ 子代理派发 │ │ │ │ ModelFactory │ │ Sandbox/MCP │ │ SubagentExecutor │ │ │ └─────────────┘ └──────────────┘ └───────────────────┘ │ └─────────────────────────────────────────────────────────────────┘ │ ┌────────────────┼────────────────┐ ▼ ▼ ▼ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ Persistence │ │ Tracing │ │ Config │ │ SQLite/PG │ │ LangSmith │ │ YAML + Hot │ │ Checkpointer │ │ Langfuse │ │ Reload │ └──────────────┘ └──────────────┘ └──────────────┘

2 核心请求流程

从用户请求到响应的完整数据流，展示系统各组件如何协作。

请求入口 — Gateway / DeerFlowClient

FastAPI 路由或嵌入式 DeerFlowClient 接收用户消息，创建 Thread（如需）和 Run 记录

RunManager.create() DeerFlowClient.stream()

配置加载 — AppConfig

从 config.yaml 加载配置，解析环境变量引用（$VAR），检测 mtime 热重载，同步子配置单例

AppConfig.from_file() resolve_env_variables() mtime 热重载

Agent 构建 — make_lead_agent()

工厂函数组装 Lead Agent：创建模型、聚合工具、装配中间件链、注入 Checkpointer 和 Tracing

create_chat_model() get_available_tools() _build_middlewares()

中间件管线 — 14层顺序执行

请求依次穿过 14 层中间件：ThreadData 初始化 → Sandbox 获取 → Guardrail 审查 → Summarization 压缩 → LoopDetection 检测 → Clarification 拦截

wrap_model_call wrap_tool_call before_agent / after_agent

LLM 推理 — ModelFactory

通过 resolve_class() 动态加载模型类（ChatOpenAI/ChatAnthropic 等），处理 Thinking 模式切换，附加 Tracing 回调

resolve_class() thinking_enabled stream_usage

工具执行 — Sandbox / MCP / Subagent

LLM 生成 tool_calls 后，工具在沙箱内执行（bash/file ops）、通过 MCP 协议调用远程服务、或派发子代理并行处理子任务

Sandbox.execute_command() MCPSessionPool SubagentExecutor

流式输出 — Dual Streaming

双模式流式输出：values 模式（完整状态快照）和 messages 模式（增量消息），SSE 协议对齐，累积 Token 统计

astream_events() RunJournal Token 归因

状态持久化 — Checkpointer + ORM

LangGraph Checkpointer 持久化对话状态，SQLAlchemy ORM 存储线程元数据、运行记录、反馈数据

AsyncEngine ThreadMetaRow RunRecord

3 配置系统

基于 Pydantic 的类型安全配置，支持 YAML 加载、环境变量解析、热重载和运行时覆盖。

AppConfig — 核心配置模型

20+ 子配置：models / sandbox / tools / skills / extensions / memory / guardrails / loop_detection / summarization / subagents 等
from_file()：YAML 加载 → resolve_env_variables() → 版本检查 → model_validate()
热重载：get_app_config() 检测 config.yaml mtime 变更自动重载
ContextVar 栈式覆盖：push/pop_current_app_config() 支持嵌套作用域隔离

环境变量与单例同步

resolve_env_variables()：递归解析 $ENV_VAR 引用，支持默认值 $VAR:-default
_apply_singleton_configs()：加载后同步子配置单例（TitleConfig / LoopDetectionConfig 等）
_apply_database_defaults()：根据 checkpointer 类型自动推断数据库 URL
配置路径解析：参数 > 环境变量 > 项目根 > 遗留路径

4 模型工厂

配置驱动的动态模型实例化，支持多供应商、Thinking 模式切换和 Tracing 注入。

create_chat_model() — 全局唯一模型工厂

def create_chat_model(name, thinking_enabled=False, app_config=None, attach_tracing=True, **kwargs):
    # 1. 从 AppConfig.models 查找 ModelConfig
    model_config = app_config.models[name]

    # 2. resolve_class() 动态加载模型类
    cls = resolve_class(model_config.use, BaseChatModel)
    # e.g. "langchain_openai:ChatOpenAI" → ChatOpenAI

    # 3. Thinking 模式参数合并
    if thinking_enabled:
        params = _deep_merge_dicts(params, model_config.when_thinking_enabled)
    else:
        params = _apply_thinking_disabled(model_config, cls)

    # 4. 自动启用 stream_usage (OpenAI 兼容)
    _enable_stream_usage_by_default(cls, params)

    # 5. 实例化 + 附加 Tracing
    model = cls(**params)
    if attach_tracing:
        model.callbacks = build_tracing_callbacks()
    return model

ModelConfig

use：类路径（如 langchain_openai:ChatOpenAI）
supports_thinking/vision/reasoning_effort：能力声明
when_thinking_enabled/disabled：条件参数注入
thinking：快捷方式字段

Thinking 模式

启用：合并 when_thinking_enabled + thinking 参数
禁用：依次尝试 when_thinking_disabled → OpenAI extra_body → vLLM chat_template → Anthropic thinking.type
Codex 模型：移除 max_tokens，映射 reasoning_effort

模型解析链

请求级 model 参数
Agent 配置 model
全局默认 model
未知名称 fallback

5 工具系统

多源工具聚合，支持配置工具、内置工具、MCP 工具和 ACP 工具的统一加载与去重。

get_available_tools() — 工具加载总入口

优先级合并：配置文件工具 > 内置工具 > MCP 工具 > ACP 工具
按名称去重：高优先级覆盖低优先级同名工具
安全过滤：LocalSandbox 下默认禁止 host bash
延迟加载：MCP 工具注册到 DeferredToolRegistry，通过 tool_search 按需提升

内置工具集

present_file_tool — 文件内容展示
ask_clarification_tool — 澄清请求
view_image_tool — 图片查看
task_tool — 子代理任务派发
skill_manage_tool — 技能管理
tool_search — 延迟工具搜索

异步工具同步包装

make_sync_tool_wrapper() 使用 ThreadPoolExecutor + contextvars.copy_context() 在独立线程中运行 asyncio.run()，确保 LangGraph 同步工具接口兼容异步实现。_ensure_sync_invocable_tool() 自动为仅有 coroutine 的工具补充 func 同步入口。

6 中间件管线

14 层固定顺序中间件管线，基于 RuntimeFeatures 声明式装配，支持 @Next/@Prev 锚点插入自定义中间件。

#	中间件	钩子	核心功能
0	ThreadDataMiddleware	before_agent	初始化线程工作区路径
1	UploadsMiddleware	before_agent	管理上传目录
2	SandboxMiddleware	before/after_agent	懒加载/释放沙箱实例
3	DanglingToolCallMiddleware	wrap_model_call	修补缺失 ToolMessage
4	GuardrailMiddleware	wrap_tool_call	工具调用前安全审查
5	ToolErrorHandlingMiddleware	wrap_tool_call	工具异常转 ToolMessage
6	SummarizationMiddleware	wrap_model_call	长对话压缩 + 记忆刷新
7	TodoMiddleware	wrap_model_call	计划模式任务追踪
8	TitleMiddleware	after_agent	自动生成线程标题
9	MemoryMiddleware	after_agent	异步记忆更新
10	ViewImageMiddleware	wrap_model_call	图片内容注入
11	SubagentLimitMiddleware	wrap_tool_call	并发子代理截断
12	LoopDetectionMiddleware	wrap_model_call	循环检测与中断
13	ClarificationMiddleware	wrap_model_call	澄清请求拦截（始终最后）

LoopDetectionMiddleware — 双层循环检测

哈希层：相同工具调用集合，warn=3 次，hard=5 次中断
频率层：单工具 warn=30 次，hard=50 次中断
滑动窗口：size=20，仅检测最近 N 轮
软警告：注入 HumanMessage(name="loop_warning")
硬限制：剥离 tool_calls，强制终止循环

RuntimeFeatures — 声明式特性标志

@dataclass
class RuntimeFeatures:
    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

每个字段可为 bool（使用默认实现）或直接传入自定义中间件实例，实现灵活替换。

自定义中间件插入 — @Next / @Prev 锚点

自定义中间件可通过装饰器声明相对位置，_insert_extra() 迭代插入并保证 ClarificationMiddleware 始终在末尾。支持冲突检测和循环依赖预防。

@Next(LoopDetectionMiddleware)   # 插入到 LoopDetection 之后
class MyCustomMiddleware(AgentMiddleware): ...

@Prev(ClarificationMiddleware)  # 插入到 Clarification 之前
class AnotherMiddleware(AgentMiddleware): ...

7 沙箱执行环境

隔离的代码执行环境，支持本地和远程后端，虚拟路径映射和安全防护。

Sandbox ABC — 沙箱接口

execute_command — 命令执行
read_file / write_file — 文件读写
download_file — 文件下载
list_dir / glob / grep — 目录搜索
update_file — 文件更新

SandboxProvider ABC — 生命周期管理

acquire / acquire_async — 获取沙箱实例
get — 按 sandbox_id 获取
release — 释放沙箱资源
reset — 重置沙箱状态
get_sandbox_provider() — 模块级单例，resolve_class() 动态加载

安全机制

虚拟路径映射：/mnt/user-data/ 和 /mnt/skills/ 前缀映射到物理路径
路径遍历防护：validate_path_traversal() 阻止 ../ 攻击
Host Bash 门控：is_host_bash_allowed() 在 LocalSandboxProvider 下默认禁止
输出截断：超长输出自动截断，防止内存溢出
文件操作锁：并发安全的文件读写

8 MCP 集成

Model Context Protocol 外部工具集成，支持 stdio/sse/http 三种传输、OAuth 认证和持久会话池。

MCPSessionPool — 会话复用

LRU 淘汰：MAX_SESSIONS=256
隔离键：(server_name, scope_key) 按线程隔离
线程安全：threading.Lock 保护
事件循环感知：自动适配不同事件循环

OAuth 认证

OAuthTokenManager：Token 获取/缓存/刷新
授权类型：client_credentials + refresh_token
过期偏移：refresh_skew_seconds 提前刷新
工具拦截器：build_oauth_tool_interceptor() 注入 Authorization 头

工具缓存与刷新

模块级缓存 + extensions_config.json mtime 变更检测。缓存过期时 reset_mcp_tools_cache() 关闭会话池并重新初始化。工具加载流程：ExtensionsConfig.from_file() → build_servers_config() → 注入 OAuth 头 → MultiServerMCPClient → get_tools() → _make_session_pool_tool() 包装 → make_sync_tool_wrapper() 补充同步入口 → 缓存。

9 子代理系统

并行任务委派，独立 LangGraph Agent 执行，支持协作取消和超时控制。

SubagentExecutor — 执行引擎

同步执行：检测事件循环 → 隔离循环 / asyncio.run()
异步执行：_aexecute() → _build_initial_state() → _create_agent() → astream()
后台执行：_scheduler_pool 提交 → 隔离循环 → 超时取消
协作取消：cancel_event (threading.Event) 在 astream() 迭代边界检测
隔离事件循环：持久化守护线程事件循环，避免与父循环冲突

SubagentConfig — 配置模型

name / description / system_prompt：代理身份
tools / disallowed_tools：工具白/黑名单（默认禁止 task 工具递归）
skills：技能加载
model：模型配置（"inherit" 继承父代理）
max_turns：最大轮次（默认 50）
timeout_seconds：超时（默认 900s）

SubagentResult — 状态机

try_set_terminal() 原子转换到终态（COMPLETED/FAILED/CANCELLED/TIMED_OUT），防止重复状态转换。包含 token_usage_records 按 caller 分桶记录 Token 消耗。

10 技能系统

可插拔技能架构，支持安装、验证、LLM 安全扫描和技能注入。

技能加载

SkillStorage.load_skills() 遍历 PUBLIC + CUSTOM 目录
parse_skill_file() 解析 SKILL.md YAML frontmatter
过滤 enabled 技能
SkillCategory: PUBLIC（只读）/ CUSTOM（可编辑删除）

安全安装

safe_extract_skill_archive() 安全解压
拒绝绝对路径/遍历/符号链接
512MB zip bomb 防御
scan_skill_content() LLM 安全扫描
返回 allow/warn/block 决策

技能注入

SubagentExecutor._load_skills()
读取 SKILL.md 内容
包装为 SystemMessage
filter_tools_by_skill_allowed_tools() 过滤工具

11 安全护栏

Fail-closed 安全语义，工具调用前安全审查，异常时默认拒绝。

GuardrailMiddleware

wrap_tool_call / awrap_tool_call：拦截每次工具调用
GuardrailProvider.evaluate()：评估工具调用安全性
Fail-closed：provider 异常时默认拒绝（allow=False）
保留 GraphBubbleUp：不拦截 LangGraph 内部信号

SafetyFinishReasonMiddleware

检测 provider 安全终止（content_filter / refusal / SAFETY）
剥离 tool_calls，阻止不安全操作
发射审计事件
AllowlistProvider：内置简单允许/拒绝列表

12 持久化层

SQLAlchemy 异步 ORM，支持 memory/sqlite/postgres 三种后端，自动建表。

引擎初始化

init_engine()：创建 AsyncEngine
SQLite：WAL + NORMAL 同步 + 外键
PostgreSQL：自动建库
Base.metadata.create_all() 自动建表
init_engine_from_config() 从 DatabaseConfig 便捷初始化

核心 ORM 模型

ThreadMetaRow：thread_id / assistant_id / user_id / display_name / status / metadata_json / created_at / updated_at
RunRecord：run_id / thread_id / status / token 统计（lead_agent/subagent/middleware 分类）
Base.to_dict()：基于 sa_inspect() 遍历列属性的通用序列化

13 可观测性

LangSmith / Langfuse 双追踪后端，RunJournal 事件捕获和 Token 归因。

build_tracing_callbacks()

读取 get_enabled_tracing_providers()
_create_langsmith_tracer()：LangChainTracer(project_name=...)
_create_langfuse_handler()：LangfuseCallbackHandler，先初始化 Langfuse 客户端单例
注入点：make_lead_agent() 图调用根 + create_chat_model(attach_tracing=True)

RunJournal — 事件捕获

LangChain BaseCallbackHandler 实现
on_chat_model_start：记录延迟起点
on_llm_end：累计 token，按 caller 分桶
on_tool_end：记录工具结果
缓冲区满时异步 flush 到 RunEventStore

14 文件上传

安全文件写入，防符号链接、路径遍历和文件名碰撞。

安全上传流程

normalize_filename()：提取 basename、拒绝反斜杠、长度限制 255 字节
validate_path_traversal()：路径遍历检测
open_upload_file_no_symlink()：POSIX O_NOFOLLOW / Windows 双重 lstat+fstat 验证
claim_unique_filename()：碰撞时追加 _N 后缀
路径映射：物理路径 → 虚拟路径（/mnt/user-data/uploads/）→ artifact URL

15 动态反射

系统的"反射脊梁"，贯穿全系统的配置驱动类加载机制。

resolve_variable() / resolve_class()

# 解析流程
"langchain_openai:ChatOpenAI"
    → rsplit(":", 1)
    → import_module("langchain_openai")
    → getattr(module, "ChatOpenAI")
    → isinstance() 类型校验

被几乎所有需要动态加载的模块依赖：models/（模型类）、tools/（工具类）、sandbox/（提供者类）、mcp/（拦截器）、guardrails/（provider）。_build_missing_dependency_hint() 为已知包生成安装提示。

16 核心设计模式

DeerFlow Harness 后端架构中反复出现的设计模式总结。

🔄 动态解析模式

resolve_variable() / resolve_class() 贯穿全系统，实现配置驱动的类加载。所有可插拔组件（模型、工具、沙箱、护栏）均通过字符串路径动态实例化。

🔗 中间件管线模式

14 层固定顺序中间件 + @Next/@Prev 锚点插入。ClarificationMiddleware 始终末尾保证拦截优先级。RuntimeFeatures 声明式控制装配。

📦 单例 + 热重载模式

AppConfig / SandboxProvider / MCPSessionPool / MCP 工具缓存均使用模块级单例 + mtime 检测热重载，平衡性能与配置实时性。

📚 ContextVar 栈式覆盖

push/pop_current_app_config() 支持运行时配置隔离，嵌套作用域内可安全覆盖全局配置而不影响其他请求。

🔌 协议驱动扩展

GuardrailProvider(Protocol) / SandboxProvider(ABC) / Sandbox(ABC) 等抽象允许插件化实现，符合依赖倒置原则。

🛑 协作取消模式

子代理通过 threading.Event 实现协作取消，在 astream() 迭代边界检测取消信号，避免强制中断导致状态不一致。

🔁 隔离事件循环

子代理使用持久化守护线程事件循环，避免与父循环冲突。_get_isolated_subagent_loop() 确保长生命周期异步操作安全执行。

🔒 Fail-closed 安全语义

Guardrail / 安全扫描在异常时默认拒绝而非放行。安全关键路径不依赖"默认允许"策略，确保系统在异常情况下仍保持安全。

DeerFlow Harness

目录