DeerFlow Gateway 与 LangGraph 运行时：一个 FastAPI 的 Agent 宇宙

这是「DeerFlow 源码全景解析」专题的第 02 篇。在本系列中，我们将从 Gateway 的 FastAPI 骨架出发，逐层深入 LangGraph 兼容运行时、SSE 流式协议、RunManager 状态机与 Worker 异步模型，还原一个工业级 AI Agent 后端的工程全貌。

故事场景：那个「503 Service Unavailable」的下午

小陈——刚加入 AI Infra 团队的后端工程师——正盯着 Postman 里飘红的响应发呆。

"我已经按照 LangGraph 文档配好了 RemoteGraph 的 URL，为什么一跑就 503？"他在飞书群里问。

老王——负责 DeerFlow 后端架构的资深工程师——回了一条消息："你看一下 Nginx 配置，我们的 Gateway 不是独立的 LangGraph Server，LangGraph 运行时嵌入在 FastAPI 里，走的是 /api/langgraph/* 这个代理路径。"

小陈打开 nginx.conf，看到那行关键的 rewrite 规则：location /api/langgraph/ { rewrite ^/api/langgraph/(.*)$ /api/$1 break; proxy_pass http://gateway:8001; }，恍然大悟：

"原来不是端口直连，"小陈喃喃道，"整个 Agent 宇宙都藏在 2026 这个统一入口后面。"

老王发了一个微笑表情："DeerFlow 的 Gateway 是一个嵌入式的 LangGraph 运行时，不是独立进程。这是理解整个架构的第一把钥匙。"

一、Gateway 的定位：为什么不是独立的 LangGraph Server？

1.1 嵌入式设计：一把钥匙开一扇门

在 LangGraph 官方架构中，LangGraph Server 通常是一个独立进程，通过 HTTP 或 gRPC 暴露图运行能力。而 DeerFlow 选择了一条不同的路——将 LangGraph 运行时嵌入 FastAPI Gateway 的同一个进程。

这个决策背后有三个核心考量：

考量维度	独立 LangGraph Server	DeerFlow 嵌入式 Gateway
部署复杂度	需维护两个服务、两套配置、两个健康检查端点	单一进程，一条 `docker-compose up` 启动
状态一致性	跨进程通信（HTTP/gRPC）引入序列化开销和延迟	内存共享：RunManager、StreamBridge、Checkpointer 直接引用
运维心智负担	需协调两个服务的版本兼容、重启顺序、连接池	统一生命周期：Gateway 启停即 LangGraph 启停
LangGraph Studio 兼容	原生支持	通过 `langgraph.json` 保留兼容，但默认走嵌入式运行时

老王曾在一个技术分享中打过一个比方："独立 Server 像是两个相邻的厨房，每次传菜都要经过一扇门；嵌入式 Gateway 则是同一个开放式厨房，厨师和传菜员共享同一个操作台。"

1.2 Nginx 反向代理：统一入口的三叉路口

DeerFlow 的所有外部流量，都经由 Nginx（端口 2026） 这一道门。Nginx 的配置逻辑清晰地将 URL 空间划分为三个区域：

flowchart LR
    A[Client Browser] -->|任意请求| B[Nginx
Port 2026]
    B -->|/api/langgraph/*| C[rewrite → /api/*
Gateway Port 8001]
    B -->|/api/*| D[Gateway Port 8001]
    B -->|/*| E[Frontend Port 3000]

    style B fill:#f9f,stroke:#333,stroke-width:2px
    style C fill:#bbf,stroke:#333,stroke-width:2px
    style D fill:#bbf,stroke:#333,stroke-width:2px

这里的关键设计是 /api/langgraph/* 的 rewrite 规则。外部客户端（包括 LangGraph SDK 和 LangGraph Studio）看到的是标准的 LangGraph Platform 路径，但内部 Gateway 的 FastAPI 路由注册在 /api/* 前缀下。Nginx 的 rewrite 抹平了这一差异，让 DeerFlow 无需在代码层面维护两套路由表。

1.3 端口 2026：一个数字的隐喻

2026 是 DeerFlow 的默认对外端口。为什么选择这个数字？答案藏在项目名里——DeerFlow 的发音与 "Dear Flow" 相近，而 2026 是项目正式启动的年份。但更重要的是，单一端口意味着单一认知边界：开发者只需要记住一个地址，Ops 只需要暴露一个端口，防火墙只需要开放一条规则。

二、FastAPI Gateway 架构

2.1 入口文件：`app/gateway/app.py`

Gateway 的 FastAPI 实例在 create_app() 函数中组装完成，而 lifespan 则掌管 Gateway 的「出生」与「死亡」。这是一个典型的「洋葱模型」：从外到内依次是 CORS 层 → CSRF 层 → Auth 层 → 业务 Router 层。

# app/gateway/app.py（精简）
@asynccontextmanager
async def lifespan(app: FastAPI):
    startup_config = get_app_config()
    apply_logging_level(startup_config.log_level)
    if startup_config.memory.token_counting != "char":
        await asyncio.wait_for(
            asyncio.to_thread(warm_tiktoken_cache), timeout=5
        )
    async with langgraph_runtime(app, startup_config):
        await _ensure_admin_user(app)
        yield  # 启动 IM 通道服务
    await auth.close_oidc_service()

def create_app() -> FastAPI:
    app = FastAPI(
        title="DeerFlow API Gateway",
        lifespan=lifespan,
        docs_url="/docs" if config.enable_docs else None,
    )
    app.add_middleware(AuthMiddleware)       # fail-closed 安全网
    app.add_middleware(CSRFMiddleware)       # Double Submit Cookie
    if cors_origins:
        app.add_middleware(CORSMiddleware, ...)  # 显式白名单
    app.include_router(models.router)      # /api/models
    app.include_router(mcp.router)         # /api/mcp
    app.include_router(skills.router)      # /api/skills
    app.include_router(thread_runs.router) # /api/threads/{id}/runs
    app.include_router(runs.router)        # /api/runs
    # ... 更多 routers
    return app

这段代码展现了两个严谨的工程习惯。其一，中间件的顺序就是安全边界的顺序：AuthMiddleware 在最外层，意味着任何请求都要先过身份认证；CSRF 紧随其后，保护状态变更请求免受跨站攻击；CORS 在最内层，确保只有明确允许的源可以访问。其二，lifespan 故意不将 startup_config 缓存到 app.state——get_app_config() 每次请求时都会重新解析 config.yaml，编辑文件后无需重启进程。这是 DeerFlow 在工程便利性与运维灵活性之间做出的取舍。

2.3 Router 全景：十二金刚

Gateway 注册了 14+ 个 Router，覆盖从模型管理到 IM 集成的完整功能矩阵。以下是核心 Router 的职责速查表：

Router	路径前缀	核心职责
`models`	`/api/models`	列出可用模型及其配置
`thread_runs`	`/api/threads/{id}/runs`	LangGraph 兼容的运行创建、流式输出、等待、取消、消息查询
`runs`	`/api/runs`	无状态运行（自动创建临时线程），`stream`/`wait` 端点
`mcp`	`/api/mcp`	MCP Server 配置管理、缓存刷新
`skills`	`/api/skills`	Skill 加载、启用/禁用、元数据查询
`uploads`	`/api/threads/{id}/uploads`	文件上传（文档自动转 Markdown）
`artifacts`	`/api/threads/{id}/artifacts`	产物文件服务（生成的图表、代码文件等）
`threads`	`/api/threads/{id}`	本地线程数据清理（文件系统级）
`suggestions`	`/api/threads/{id}/suggestions`	基于对话历史的后续问题生成
`auth`	`/api/v1/auth`	本地认证、OIDC、JWT、登出
`feedback`	`/api/threads/{id}/runs/{rid}/feedback`	运行级反馈（点赞/点踩）
`agents`	`/api/agents`	自定义 Agent 配置与提示词管理
`channels`	`/api/channels`	飞书/Slack/Telegram 连接配置
`assistants_compat`	`/api/assistants`	LangGraph Platform 兼容的 assistants 接口（stub）

2.4 认证体系：从「无门」到「多门」

DeerFlow 的认证体系设计了一个渐进式的安全路径：

无认证模式：首次启动，无 admin 用户。Gateway 打印提示信息，引导用户访问 /setup 创建首个管理员。
本地认证：创建 admin 后，启用基于 SQLAlchemy 的本地用户/密码体系，支持基于 session 的浏览器登录。
OIDC / SSO：通过 config.yaml 配置外部身份提供商（如 Okta、Keycloak），实现企业级单点登录。
JWT：API 调用（如 LangGraph SDK）使用 Bearer Token 认证。
CSRF 保护：浏览器端的非安全请求（POST/PUT/DELETE）必须携带 Double Submit Cookie，防止跨站伪造。

一个值得注意的细节是 AuthMiddleware 的 fail-closed 设计：它默认拒绝所有未经认证的请求，只有明确标记为 public 的路径（如 /health、/setup）才能放行。这比「fail-open」（默认放行，显式拦截）要安全得多，因为新增路由不会意外暴露在公网。

三、LangGraph 兼容运行时

3.1 `langgraph.json`：一个备用入口

在 DeerFlow 仓库根目录，有一个 langgraph.json 文件，内容只有短短几行——agent.type 为 "agent"，agent.path 指向 deerflow.agents:make_lead_agent。如果你愿意，可以用 langgraph dev 或 langgraph up 启动一个标准的 LangGraph Server，它会读取这个配置，通过 make_lead_agent 构造图实例。

但这只是「备用方案」。 DeerFlow 的默认部署路径是嵌入式 Gateway 运行时。Docker 和源码启动脚本都直接运行 uvicorn app.gateway.app:app，而不是 langgraph CLI。

这种「双轨制」设计的妙处在于：开发阶段可以用 LangGraph Studio 可视化调试图结构，生产阶段则享受嵌入式 Gateway 的部署便利和性能优势。

3.2 `make_lead_agent`：图的构造与注册

make_lead_agent 是 Agent 运行时的心脏。它构建了一个 LangGraph 的 StateGraph，并挂载了一条精心设计的 Middleware Chain：

flowchart TD
    A[make_lead_agent
config] --> B[ThreadDataMiddleware
初始化 workspace/uploads/outputs]
    B --> C[UploadsMiddleware
注入已上传文件列表]
    C --> D[SandboxMiddleware
获取沙箱环境]
    D --> E[SummarizationMiddleware
上下文压缩]
    E --> F[TitleMiddleware
自动生成标题]
    F --> G[TodoListMiddleware
任务追踪 plan_mode]
    G --> H[ViewImageMiddleware
视觉模型支持]
    H --> I[ClarificationMiddleware
处理澄清请求]
    I --> J[Agent Core
Model + Tools + System Prompt]

这条链的每个节点都是状态转换器（state transformer），它们接收 ThreadState（继承自 LangGraph 的 AgentState），添加 DeerFlow 特有的扩展字段——如 sandbox（沙箱环境信息）、artifacts（生成的文件路径列表）、thread_data（workspace/uploads/outputs 路径映射）、title（自动生成的对话标题）、todos（plan mode 下的任务追踪列表）和 viewed_images（视觉模型图像数据）——然后传递给下一个节点。

Middleware Chain 的设计体现了 DeerFlow 的核心哲学：不对 LangGraph 的底层做侵入式修改，而是在图的入口和出口插入可插拔的「信封」。每个 Middleware 都可以独立开关，配置项在 config.yaml 的 middleware 章节中统一控制。

四、请求生命周期详解

4.1 从 POST 到 SSE：一次对话的完整旅程

让我们追踪一个最典型的场景：用户在 Web UI 中发送一条消息，Agent 开始思考并流式返回 token。这条请求的完整生命周期如下：

sequenceDiagram
    participant C as Client Browser
    participant N as Nginx :2026
    participant G as Gateway :8001
    participant R as RunManager
    participant W as Worker
(asyncio.Task)
    participant B as StreamBridge
    participant L as LangGraph
Graph.astream

    C->>N: POST /api/langgraph/threads/{id}/runs
    N->>G: rewrite → /api/threads/{id}/runs
    G->>R: start_run(body, thread_id)
    R->>R: create_or_reject() 检查并发策略
    R->>R: create() 生成 run_id
    R->>W: asyncio.create_task(run_agent(...))
    W->>L: graph.astream(input, config)
    L-->>W: 实时流式 chunk
    W->>B: bridge.publish(run_id, event, data)
    G->>B: sse_consumer(bridge, record)
    B-->>G: yield 格式化 SSE 事件
    G-->>N: StreamingResponse
    N-->>C: text/event-stream

从代码层面看，这个流式入口由 thread_runs.py 的 stream_run 端点实现。它通过 start_run 创建运行记录，然后将 StreamBridge 的消费器包装为 StreamingResponse 返回给客户端：

# app/gateway/routers/thread_runs.py（精简）
@router.post("/{thread_id}/runs/stream")
async def stream_run(
    thread_id: str, body: RunCreateRequest, request: Request
) -> StreamingResponse:
    bridge = get_stream_bridge(request)
    run_mgr = get_run_manager(request)
    record = await start_run(body, thread_id, request)
    return StreamingResponse(
        sse_consumer(bridge, record, request, run_mgr),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
            "Content-Location": f"/api/threads/{thread_id}/runs/{record.run_id}",
        },
    )

注意 Content-Location 响应头——它是 LangGraph Platform 协议的约定，SDK 的 useStream 通过贪婪正则从中提取 run_id，用于后续的流重连或取消操作。

4.2 RunManager：运行状态的「户籍科」

RunManager 是 Gateway 中负责运行状态管理的核心组件。它是一个内存注册表（in-memory registry），可选地以 SQLite 或 PostgreSQL 作为持久化后端。

RunManager 管理的状态转换可以用一个状态机来描述：

stateDiagram-v2
    [*] --> pending: create()
    pending --> running: set_status(running)
    running --> success: 正常完成
    running --> error: 异常 / LLM 失败
    running --> interrupted: cancel(action=interrupt)
    running --> error: cancel(action=rollback) + 回滚失败
    running --> error: 捕获 rollback 回滚成功

    interrupted --> [*]
    success --> [*]
    error --> [*]

这里有一个关键的工程细节：RunManager 的 create_or_reject 方法在 asyncio.Lock 的保护下原子性地检查并发策略。如果线程上已有 inflight 运行，根据 multitask_strategy 的值：

reject：抛出 ConflictError，返回 409
interrupt / rollback：取消现有运行，然后创建新运行

这消除了「检查 → 等待 → 创建」之间的 TOCTOU（Time-of-Check to Time-of-Use）竞态条件。

4.3 Worker 模型：异步执行与取消机制

每个运行都在一个独立的 asyncio.Task 中执行。run_agent 函数（位于 `worker.py）是 Worker 的核心逻辑：

# worker.py（run_agent 核心流式循环，精简）
async def run_agent(
    bridge: StreamBridge,
    run_manager: RunManager,
    record: RunRecord,
    *,
    agent_factory: Any,
    graph_input: dict,
    config: dict,
    stream_modes: list[str] | None = None,
    ...
) -> None:
    # 1. 标记运行中
    await run_manager.set_status(run_id, RunStatus.running)

    # 2. 注入 runtime context（thread_id, run_id, app_config）
    runtime_ctx = _build_runtime_context(thread_id, run_id, ...)
    _install_runtime_context(config, runtime_ctx)

    # 3. 构建 Agent 实例
    agent = agent_factory(config=runnable_config)

    # 4. 捕获运行前 checkpoint，用于 rollback
    ckpt_tuple = await checkpointer.aget_tuple(config_for_check)
    pre_run_checkpoint_id = ckpt_config.get("checkpoint_id")

    # 5. 流式执行
    async for chunk in agent.astream(graph_input, config=runnable_config, stream_mode=single_mode):
        if record.abort_event.is_set():
            break  # 用户请求取消
        await bridge.publish(run_id, sse_event, serialize(chunk, mode=single_mode))

    # 6. 最终状态与清理
    await run_manager.set_status(run_id, final_status)
    await bridge.publish_end(run_id)

取消机制的设计非常巧妙。RunRecord 包含一个 asyncio.Event（abort_event）和一个 asyncio.Task（task）。当用户点击「停止」按钮时，Gateway 调用 RunManager.cancel()，该方法会：

设置 abort_event，让 Worker 的流式循环在下一次迭代时优雅退出
调用 task.cancel()，触发 asyncio.CancelledError
根据 action（interrupt 或 rollback）决定最终状态

如果是 rollback，Worker 会在 finally 块中执行回滚逻辑：将线程状态恢复到运行前的 checkpoint。这得益于步骤 4 中预先捕获的 pre_run_checkpoint_id。

4.4 StreamBridge：跨线程的 SSE 流桥

StreamBridge 是 Gateway 架构中最容易被低估的组件。它解决了这样一个问题：Worker 的 asyncio.Task 与 FastAPI 的 StreamingResponse 可能运行在不同的事件循环线程中，如何安全地传递流式事件？

StreamBridge 的实现本质上是一个多生产者-单消费者的内存队列：publish(run_id, event, data) 由 Worker 调用，将事件写入对应 run_id 的队列；sse_consumer(bridge, record, request, run_mgr) 由 FastAPI 端点调用，从队列中读取事件并格式化为 SSE 协议。当客户端通过 POST /api/threads/{id}/runs/stream 发起流式请求时，Gateway 返回的 StreamingResponse 会携带 Content-Location 响应头，其值为 /api/threads/{thread_id}/runs/{record.run_id}。这个 URL 扮演了「流身份证」的角色——LangGraph SDK 的 useStream 使用贪婪正则从中提取 run_id，用于后续的 join 或 cancel 操作。

五、SSE Streaming 与事件协议

5.1 LangGraph SSE 协议：事件类型速查

DeerFlow 的 SSE 流与 LangGraph Platform 的事件协议对齐，确保 @langchain/langgraph-sdk/react 的 useStream Hook 可以无修改地工作。以下是核心事件类型：

SSE 事件名	来源	内容说明	前端表现
`metadata`	Worker	`run_id`, `thread_id`	React Hook 初始化状态
`values`	`graph.astream(stream_mode="values")`	图的完整状态快照	更新消息列表
`messages`	`graph.astream(stream_mode="messages")`	`(chunk, metadata)` 元组	逐 token 渲染
`messages-tuple`	同上	原始 message 元组	用于调试或自定义解析
`updates`	`graph.astream(stream_mode="updates")`	`{node: writes}`	显示节点级更新
`checkpoints`	`graph.astream(stream_mode="checkpoints")`	Checkpoint 快照	显示恢复点
`end`	Worker `finally`	运行终止信号	关闭流，更新 UI 状态
`error`	Worker except	`{message, name}`	显示错误提示

5.2 实时 Token 流与工具调用事件

当 Agent 调用工具时，前端观察到的 SSE 流不是简单的「文本 → 文本」。以一次「搜索 → 总结」的流为例，客户端会依次收到 metadata 事件（携带 run_id 和 thread_id）、values 事件（包含图的完整状态快照，如 messages 数组）、messages 事件（逐 token 的 (chunk, metadata) 元组）、updates 事件（{node: writes} 节点级更新）、checkpoints 事件（Checkpoint 快照）和最终的 end 事件。values 事件携带的是图的完整状态快照，因此每次事件都包含完整的 messages 列表。前端框架（如 React）通常会对消息数组进行去重和差异渲染，只更新新增或变化的部分。

5.3 流的重连与恢复

Web 连接不是永久的。当用户刷新页面、切换网络或服务器重启时，SSE 流会中断。DeerFlow 提供了两个机制来恢复流：

GET /api/threads/{id}/runs/{run_id}/join：加入一个正在运行的 SSE 流。如果运行仍在进行，新连接会接入 StreamBridge 的队列，继续接收后续事件。
POST /api/threads/{id}/runs/{run_id}/stream：LangGraph SDK 的「停止按钮」实际发送的是 POST 请求，携带 action=interrupt 或 action=rollback，先取消运行，然后继续流式输出剩余缓冲事件，确保客户端观察到干净的终止状态。

LangGraph SDK 的 useStream 使用贪婪正则从 Content-Location 响应头中提取 run_id，用于后续的 join 或 cancel 操作。

六、配置与扩展

6.1 `config.yaml`：热重载的运行时配置

DeerFlow 的配置体系分为两层：

config.yaml：核心运行配置，包括模型列表、工具列表、沙箱设置、摘要策略、中间件开关等。Gateway 不缓存这份配置——每次请求时通过 get_app_config() 重新读取，实现运行时热重载。
extensions_config.json：扩展配置，主要管理 MCP Server 和 Skills 状态。通过 /api/mcp 和 /api/skills 端点可在线修改，修改后立即生效。

这种设计的代价是每次请求都有一次磁盘 I/O（或文件系统缓存命中），但在配置变更不频繁的场景下，现代操作系统的页缓存几乎可以抹平这一开销。换来的好处是：

添加新模型无需重启 Gateway
调整 MCP Server 参数无需重启 Gateway
修改中间件行为无需重启 Gateway

6.2 MCP 与 Skills 的动态扩展

extensions_config.json 是 DeerFlow 的「能力插件目录」。一个典型的配置包含 mcpServers 字段（如 github 使用 stdio 模式运行 npx 命令，brave-search 使用 SSE 模式连接本地端口）和 skills 字段（如 pdf-processing 和 frontend-design 的启用状态）。当用户通过 UI 启用一个 MCP Server 时，Gateway 会更新 extensions_config.json、重置 MCP tools 缓存、关闭已有会话连接，并在下次 Agent 运行时重新初始化 MultiServerMCPClient。

总结

DeerFlow 的 Gateway 不是传统意义上的「API 网关」——它更像是一个嵌入在 FastAPI 中的 Agent 宇宙。从 Nginx 的统一入口，到 FastAPI 的洋葱安全模型；从 RunManager 的原子并发控制，到 Worker 的异步生命周期管理；从 StreamBridge 的跨线程流桥接，到与 LangGraph Platform 对齐的 SSE 协议——每一个设计决策都体现了「降低部署复杂度，提升运行期灵活性」的工程哲学。

理解 Gateway 的架构，就是理解 DeerFlow 如何在「LangGraph 生态兼容」与「独立产品体验」之间走钢丝。它既保留了 LangGraph Studio 的调试能力，又提供了单一进程的生产部署便利；既支持标准的 LangGraph SDK 客户端，又扩展了 DeerFlow 特有的模型管理、MCP 配置、Skills 系统和 IM 通道集成。

下一篇，我们将从 Gateway 的 HTTP 层下沉到 Agent 核心，拆解 make_lead_agent 的 Middleware Chain、Tool 系统编排和 Sandbox 执行模型——那里藏着 Agent 真正「思考」和「行动」的秘密。

参考链接：

DeerFlow 官方仓库：https://github.com/bytedance/deer-flow
LangGraph 文档：https://langchain-ai.github.io/langgraph/
LangGraph Platform API 规范：https://langchain-ai.github.io/langgraph/cloud/reference/api/api/
FastAPI 官方文档：https://fastapi.tiangolo.com/
SSE 规范（WHATWG）：https://html.spec.whatwg.org/multipage/server-sent-events.html
本系列上一篇：DeerFlow 全景概览与架构总览

本文基于 DeerFlow 开源仓库源码分析撰写，部分内部实现细节可能随版本迭代而调整。