AI创想

标题: 告别 langgraph-cli：在 FastAPI 中内置 LangGraph Runtime [打印本页]

---

作者: 创想小编    时间: 2026-7-7 07:15
标题: 告别 langgraph-cli：在 FastAPI 中内置 LangGraph Runtime
作者：CSDN博客

本文基于 DeerFlow 开源项目（bytedance/deer-flow）的最新架构变更，分析其 Gateway 如何从依赖外部 langgraph-cli 进程，演进为内置 LangGraph Runtime 的单进程架构，并实现 LangGraph Platform 兼容 API。

问题：三进程部署的痛

在 DeerFlow 的早期架构中，后端由三个进程组成：

1. ┌──────────┐     ┌──────────────────┐     ┌──────────────┐
2. │  Nginx   │────▶│ LangGraph Server │     │  Gateway API  │
3. │  :2026   │     │  (langgraph-cli) │     │  (FastAPI)    │
4. │          │────▶│  :2024           │     │  :8001        │
5. │          │────▶│                  │     │               │
6. └──────────┘     └──────────────────┘     └───────────────┘

复制代码

Nginx：统一入口，按路径分发LangGraph Server（langgraph-cli）：agent 运行时，处理 /api/langgraph/* 请求Gateway API（FastAPI）：处理模型管理、MCP 配置、文件上传等非 agent 操作

这个架构有几个实际问题：

运维复杂度：三个进程要分别启动、监控、重启。开发环境需要同时跑三个终端。进程间通信开销：Gateway 需要通过 HTTP 调用 LangGraph Server 来操作 thread 和 checkpoint，多了一跳网络延迟。状态不共享：Gateway 无法直接访问 LangGraph 的 checkpointer 和 store，只能通过 API 间接操作。langgraph-cli 的版本耦合：langgraph-cli 有自己的版本节奏，升级可能引入不兼容变更。

解法：Gateway 内置 LangGraph Runtime

DeerFlow 的新架构把 LangGraph 的核心运行时组件直接嵌入 Gateway 进程：

1. ┌──────────┐     ┌─────────────────────────────────────────┐
2. │  Nginx   │────▶│           Gateway API (FastAPI)          │
3. │  :2026   │     │                                         │
4. │          │     │  ┌─────────────────────────────────────┐ │
5. │          │     │  │      LangGraph Runtime (内置)        │ │
6. │          │     │  │  StreamBridge · RunManager           │ │
7. │          │     │  │  Checkpointer · Store                │ │
8. │          │     │  └─────────────────────────────────────┘ │
9. │          │     │                                         │
10. │          │     │  Routers: runs · thread_runs · threads  │
11. │          │     │  models · memory · skills · uploads ... │
12. │          │     └─────────────────────────────────────────┘
13. └──────────┘

复制代码

部署从三进程简化为两进程（Nginx + Gateway），甚至在开发环境可以只跑 Gateway 一个进程。
实现细节

deps.py：运行时组件的生命周期管理

app/gateway/deps.py 是这次变更的核心文件，它负责初始化和管理四个 LangGraph 运行时单例：

1. from deerflow.runtime import RunManager, StreamBridge
3. @asynccontextmanagerasyncdeflanggraph_runtime(app: FastAPI)-> AsyncGenerator[None,None]:"""Bootstrap and tear down all LangGraph runtime singletons."""from deerflow.agents.checkpointer.async_provider import make_checkpointer
4.     from deerflow.runtime import make_store, make_stream_bridge
6.     asyncwith AsyncExitStack()as stack:
7.         app.state.stream_bridge =await stack.enter_async_context(make_stream_bridge())
8.         app.state.checkpointer =await stack.enter_async_context(make_checkpointer())
9.         app.state.store =await stack.enter_async_context(make_store())
10.         app.state.run_manager = RunManager()yield

复制代码

四个组件各司其职：

组件	职责
StreamBridge	SSE 流式推送桥接，管理 agent 输出到客户端的实时流
RunManager	管理 agent run 的生命周期（创建、取消、状态查询）
Checkpointer	持久化 thread 状态（对话历史、agent state），支持 sqlite
Store	键值存储，用于 thread 元数据、快速列表查询

这四个组件都来自 deerflow.runtime（harness 层），Gateway（app 层）只负责组装和暴露。
app.py：通过 lifespan 启动

1. @asynccontextmanagerasyncdeflifespan(app: FastAPI)-> AsyncGenerator[None,None]:# 加载配置
2.     get_app_config()# 初始化 LangGraph Runtimeasyncwith langgraph_runtime(app):
3.         logger.info("LangGraph runtime initialised")# 启动 IM 渠道服务（如果配置了）try:
4.             channel_service =await start_channel_service()except Exception:
5.             logger.exception("No IM channels configured")yield# 应用运行中# 关闭 IM 渠道服务await stop_channel_service()

复制代码

AsyncExitStack 确保了所有运行时组件在应用关闭时被正确清理——checkpointer 关闭数据库连接，store 刷新缓存，stream bridge 断开所有活跃连接。
Getter 函数：路由中的依赖注入

路由通过 getter 函数获取运行时组件，组件不可用时返回 503：

1. defget_stream_bridge(request: Request)-> StreamBridge:
2.     bridge =getattr(request.app.state,"stream_bridge",None)if bridge isNone:raise HTTPException(status_code=503, detail="Stream bridge not available")return bridge
4. defget_run_manager(request: Request)-> RunManager:
5.     mgr =getattr(request.app.state,"run_manager",None)if mgr isNone:raise HTTPException(status_code=503, detail="Run manager not available")return mgr
7. defget_checkpointer(request: Request):
8.     cp =getattr(request.app.state,"checkpointer",None)if cp isNone:raise HTTPException(status_code=503, detail="Checkpointer not available")return cp
10. defget_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 提供的接口：

1. # app/gateway/routers/ 下的新路由
2. app.include_router(runs.router)# /api/runs — run 生命周期
3. app.include_router(thread_runs.router)# /api/threads/{id}/runs — thread 级别的 run
4. app.include_router(threads.router)# /api/threads — thread CRUD + 搜索
5. app.include_router(assistants_compat.router)# /api/assistants — 兼容 stub

复制代码

这意味着前端的 useStream React hook 和 LangGraph SDK 客户端不需要任何修改——它们看到的 API 接口和之前调用 langgraph-cli 时完全一致。
Thread 管理的统一

之前 thread 的管理分散在两个进程中：

LangGraph Server 管理 thread state（对话历史、checkpoint）Gateway 管理 thread 的文件系统数据（workspace、uploads、outputs）

删除一个对话需要两步：先调 LangGraph Server 删 thread state，再调 Gateway 删文件系统数据。
现在 Gateway 直接持有 checkpointer 和 store，thread 的创建、查询、删除都在一个进程内完成：

1. @router.delete("/{thread_id}")asyncdefdelete_thread_data(thread_id:str, request: Request):# 1. 删除本地文件系统数据
2.     response = _delete_thread_data(thread_id)# 2. 删除 Store 记录
3.     store = get_store(request)if store isnotNone:await store.adelete(THREADS_NS, thread_id)# 3. 删除 Checkpoint
4.     checkpointer = get_checkpointer(request)ifhasattr(checkpointer,"adelete_thread"):await checkpointer.adelete_thread(thread_id)return response

复制代码

Thread 搜索与分页

threads.router 实现了完整的 thread 搜索，支持按 metadata 过滤（比如按 user_id 过滤当前用户的对话）：

1. @router.post("/search")asyncdefsearch_threads(request: Request, body: ThreadSearchRequest):
2.     store = get_store(request)
3.     checkpointer = get_checkpointer(request)# 优先从 Store 查询（快速路径）if store isnotNone:
4.         items =await store.asearch(THREADS_NS,filter=body.metadata,...)# 回退到 Checkpointer 遍历（兼容路径）asyncfor checkpoint_tuple in checkpointer.alist(None):# ... 构建 thread 列表

复制代码

Store 提供了快速的元数据查询能力，而 Checkpointer 作为兜底确保即使 Store 数据不完整也能返回结果。
Checkpointer 配置

DeerFlow 通过 config.yaml 配置 checkpointer 类型：

1. checkpointer:type: sqlite
2.   connection_string: checkpoints.db

复制代码

make_checkpointer() 根据配置创建对应的 checkpointer 实例。目前支持 sqlite（开发/小规模部署）和 postgres（生产环境）。checkpointer 作为 async context manager，在 Gateway 启动时初始化连接，关闭时释放资源。
对前端的影响

前端几乎不需要改动。useStream hook 之前调用 /api/langgraph/threads/{id}/runs，现在调用 /api/threads/{id}/runs——只是 URL 前缀变了，Nginx 配置相应调整即可。
更重要的是，前端现在可以通过同一个 Gateway 完成所有操作：

1. // 之前：thread 列表要调 LangGraph Serverconst threads =await langGraphClient.threads.search({ metadata:{ user_id }});// 现在：Gateway 直接提供，同一个 base URLconst threads =await apiClient.threads.search({ metadata:{ user_id }});

复制代码

这对多用户场景特别有价值——Gateway 可以在 thread 搜索时直接按 user_id 过滤，不需要额外的 BFF 层。
与 Harness/App 拆分的配合

这次 runtime 内置和上一篇讲的 Harness/App 拆分是配套的：

Harness 提供运行时组件：deerflow.runtime 导出 StreamBridge、RunManager、make_store、make_stream_bridge；deerflow.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-api>=0.7.0,<0.8.0 和 langgraph-runtime-inmem>=0.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 变化的维护成本。




原文地址：https://blog.csdn.net/qhvssonic/article/details/160411376

---

欢迎光临 AI创想 (https://llms-ai.com/)

Powered by Discuz! X3.4