本文基于 DeerFlow 开源项目bytedance/deer-flow的最新架构变更分析其 Gateway 如何从依赖外部 langgraph-cli 进程演进为内置 LangGraph Runtime 的单进程架构并实现 LangGraph Platform 兼容 API。问题三进程部署的痛在 DeerFlow 的早期架构中后端由三个进程组成┌──────────┐ ┌──────────────────┐ ┌──────────────┐ │ Nginx │────▶│ LangGraph Server │ │ Gateway API │ │ :2026 │ │ (langgraph-cli) │ │ (FastAPI) │ │ │────▶│ :2024 │ │ :8001 │ │ │────▶│ │ │ │ └──────────┘ └──────────────────┘ └───────────────┘Nginx统一入口按路径分发LangGraph Serverlanggraph-cliagent 运行时处理/api/langgraph/*请求Gateway APIFastAPI处理模型管理、MCP 配置、文件上传等非 agent 操作这个架构有几个实际问题运维复杂度三个进程要分别启动、监控、重启。开发环境需要同时跑三个终端。进程间通信开销Gateway 需要通过 HTTP 调用 LangGraph Server 来操作 thread 和 checkpoint多了一跳网络延迟。状态不共享Gateway 无法直接访问 LangGraph 的 checkpointer 和 store只能通过 API 间接操作。langgraph-cli 的版本耦合langgraph-cli 有自己的版本节奏升级可能引入不兼容变更。解法Gateway 内置 LangGraph RuntimeDeerFlow 的新架构把 LangGraph 的核心运行时组件直接嵌入 Gateway 进程┌──────────┐ ┌─────────────────────────────────────────┐ │ Nginx │────▶│ Gateway API (FastAPI) │ │ :2026 │ │ │ │ │ │ ┌─────────────────────────────────────┐ │ │ │ │ │ LangGraph Runtime (内置) │ │ │ │ │ │ StreamBridge · RunManager │ │ │ │ │ │ Checkpointer · Store │ │ │ │ │ └─────────────────────────────────────┘ │ │ │ │ │ │ │ │ Routers: runs · thread_runs · threads │ │ │ │ models · memory · skills · uploads ... │ │ │ └─────────────────────────────────────────┘ └──────────┘部署从三进程简化为两进程Nginx Gateway甚至在开发环境可以只跑 Gateway 一个进程。实现细节deps.py运行时组件的生命周期管理app/gateway/deps.py是这次变更的核心文件它负责初始化和管理四个 LangGraph 运行时单例fromdeerflow.runtimeimportRunManager,StreamBridgeasynccontextmanagerasyncdeflanggraph_runtime(app:FastAPI)-AsyncGenerator[None,None]:Bootstrap and tear down all LangGraph runtime singletons.fromdeerflow.agents.checkpointer.async_providerimportmake_checkpointerfromdeerflow.runtimeimportmake_store,make_stream_bridgeasyncwithAsyncExitStack()asstack:app.state.stream_bridgeawaitstack.enter_async_context(make_stream_bridge())app.state.checkpointerawaitstack.enter_async_context(make_checkpointer())app.state.storeawaitstack.enter_async_context(make_store())app.state.run_managerRunManager()yield四个组件各司其职组件职责StreamBridgeSSE 流式推送桥接管理 agent 输出到客户端的实时流RunManager管理 agent run 的生命周期创建、取消、状态查询Checkpointer持久化 thread 状态对话历史、agent state支持 sqliteStore键值存储用于 thread 元数据、快速列表查询这四个组件都来自deerflow.runtimeharness 层Gatewayapp 层只负责组装和暴露。app.py通过 lifespan 启动asynccontextmanagerasyncdeflifespan(app:FastAPI)-AsyncGenerator[None,None]:# 加载配置get_app_config()# 初始化 LangGraph Runtimeasyncwithlanggraph_runtime(app):logger.info(LangGraph runtime initialised)# 启动 IM 渠道服务如果配置了try:channel_serviceawaitstart_channel_service()exceptException:logger.exception(No IM channels configured)yield# 应用运行中# 关闭 IM 渠道服务awaitstop_channel_service()AsyncExitStack确保了所有运行时组件在应用关闭时被正确清理——checkpointer 关闭数据库连接store 刷新缓存stream bridge 断开所有活跃连接。Getter 函数路由中的依赖注入路由通过 getter 函数获取运行时组件组件不可用时返回 503defget_stream_bridge(request:Request)-StreamBridge:bridgegetattr(request.app.state,stream_bridge,None)ifbridgeisNone:raiseHTTPException(status_code503,detailStream bridge not available)returnbridgedefget_run_manager(request:Request)-RunManager:mgrgetattr(request.app.state,run_manager,None)ifmgrisNone:raiseHTTPException(status_code503,detailRun manager not available)returnmgrdefget_checkpointer(request:Request):cpgetattr(request.app.state,checkpointer,None)ifcpisNone:raiseHTTPException(status_code503,detailCheckpointer not available)returncpdefget_store(request:Request):Store 可能为 None未配置时。returngetattr(request.app.state,store,None)注意get_store的特殊处理——store 是可选的未配置时返回None而不是 503。这让 Gateway 在最小配置下也能启动。LangGraph Platform 兼容 API内置 runtime 后DeerFlow 实现了一组与 LangGraph Platform 兼容的 REST API可以直接替代 langgraph-cli 提供的接口# app/gateway/routers/ 下的新路由app.include_router(runs.router)# /api/runs — run 生命周期app.include_router(thread_runs.router)# /api/threads/{id}/runs — thread 级别的 runapp.include_router(threads.router)# /api/threads — thread CRUD 搜索app.include_router(assistants_compat.router)# /api/assistants — 兼容 stub这意味着前端的useStreamReact hook 和 LangGraph SDK 客户端不需要任何修改——它们看到的 API 接口和之前调用 langgraph-cli 时完全一致。Thread 管理的统一之前 thread 的管理分散在两个进程中LangGraph Server 管理 thread state对话历史、checkpointGateway 管理 thread 的文件系统数据workspace、uploads、outputs删除一个对话需要两步先调 LangGraph Server 删 thread state再调 Gateway 删文件系统数据。现在 Gateway 直接持有 checkpointer 和 storethread 的创建、查询、删除都在一个进程内完成router.delete(/{thread_id})asyncdefdelete_thread_data(thread_id:str,request:Request):# 1. 删除本地文件系统数据response_delete_thread_data(thread_id)# 2. 删除 Store 记录storeget_store(request)ifstoreisnotNone:awaitstore.adelete(THREADS_NS,thread_id)# 3. 删除 Checkpointcheckpointerget_checkpointer(request)ifhasattr(checkpointer,adelete_thread):awaitcheckpointer.adelete_thread(thread_id)returnresponseThread 搜索与分页threads.router实现了完整的 thread 搜索支持按 metadata 过滤比如按user_id过滤当前用户的对话router.post(/search)asyncdefsearch_threads(request:Request,body:ThreadSearchRequest):storeget_store(request)checkpointerget_checkpointer(request)# 优先从 Store 查询快速路径ifstoreisnotNone:itemsawaitstore.asearch(THREADS_NS,filterbody.metadata,...)# 回退到 Checkpointer 遍历兼容路径asyncforcheckpoint_tupleincheckpointer.alist(None):# ... 构建 thread 列表Store 提供了快速的元数据查询能力而 Checkpointer 作为兜底确保即使 Store 数据不完整也能返回结果。Checkpointer 配置DeerFlow 通过config.yaml配置 checkpointer 类型checkpointer:type:sqliteconnection_string:checkpoints.dbmake_checkpointer()根据配置创建对应的 checkpointer 实例。目前支持 sqlite开发/小规模部署和 postgres生产环境。checkpointer 作为 async context manager在 Gateway 启动时初始化连接关闭时释放资源。对前端的影响前端几乎不需要改动。useStreamhook 之前调用/api/langgraph/threads/{id}/runs现在调用/api/threads/{id}/runs——只是 URL 前缀变了Nginx 配置相应调整即可。更重要的是前端现在可以通过同一个 Gateway 完成所有操作// 之前thread 列表要调 LangGraph ServerconstthreadsawaitlangGraphClient.threads.search({metadata:{user_id}});// 现在Gateway 直接提供同一个 base URLconstthreadsawaitapiClient.threads.search({metadata:{user_id}});这对多用户场景特别有价值——Gateway 可以在 thread 搜索时直接按user_id过滤不需要额外的 BFF 层。与 Harness/App 拆分的配合这次 runtime 内置和上一篇讲的 Harness/App 拆分是配套的Harness 提供运行时组件deerflow.runtime导出StreamBridge、RunManager、make_store、make_stream_bridgedeerflow.agents.checkpointer导出make_checkpointer。App 负责组装deps.py把这些组件初始化并挂到app.state上路由通过 getter 函数获取。依赖方向清晰App 调用 Harness 的工厂函数创建组件Harness 不知道 App 的存在。如果未来有人想用 DeerFlow 的 agent 能力但不需要 Web 接口比如嵌入式场景可以直接用DeerFlowClient它内部也是调用同样的 harness 组件只是不经过 FastAPI。迁移注意事项如果你也在用 LangGraph 构建产品想从 langgraph-cli 迁移到内置 runtime以下是几个关键点langgraph-runtime-inmem 和 langgraph-api 的版本要锁定。DeerFlow 锁定了langgraph-api0.7.0,0.8.0和langgraph-runtime-inmem0.22.1这些包的 API 还在快速变化。Checkpointer 要用 async 版本。FastAPI 是异步框架checkpointer 必须支持aput/aget/alist等异步方法。langgraph-checkpoint-sqlite提供了 async sqlite checkpointer。Store 是可选的但强烈推荐。没有 Store 时 thread 列表只能遍历 checkpointer性能差很多。SSE 流式推送需要 StreamBridge。这是 LangGraph 的流式输出到 HTTP SSE 的桥接层不能省略。Nginx 路由规则要更新。之前/api/langgraph/*转发到 langgraph-cli 的端口现在统一转发到 Gateway。总结DeerFlow 的 Gateway 内置 LangGraph Runtime 是一个务实的架构决策减少了一个进程、消除了进程间通信开销、统一了 thread 管理、简化了部署。它不是重新实现 LangGraph而是把 LangGraph 的运行时组件checkpointer、store、stream bridge、run manager作为库来使用嵌入到自己的 FastAPI 应用中。这个模式适合已经有自己 Web 框架的项目——与其让 langgraph-cli 作为独立进程运行不如把它的核心能力内化。前提是你愿意承担版本锁定和 API 变化的维护成本。