LangGraph从脚本到服务的三种部署路径：FastAPI与微服务实战

写过 LangGraph 应用的同学应该都有过这种体验：本地把图搭好、脚本一跑，Agent 链路顺畅走通，感觉特别爽。但一旦要考虑“上线给人用”，各种问题马上冒出来——怎么把能力暴露成接口？并发请求来了扛不扛得住？用户多轮对话的状态放哪里？出了问题能不能快速定位？这一连串问题不解决，写好的 LangGraph 应用就只能停留在脚本阶段。

这篇文章我打算把 LangGraph 从“脚本”到“服务”的三条部署路径一次性讲透：脚本直跑、FastAPI 单体服务化、LangGraph Platform 与微服务架构。每条路径我都会结合自己的实际项目经验，说清楚核心思路、具体操作步骤、关键代码以及踩过的坑，方便你根据团队规模和业务场景做选型。

1. 先说结论：三条部署路径到底在解决什么问题

1.1 脚本模式解决了什么，又缺了什么

脚本模式是最原始、也是 LangGraph 最基本的使用方式。你在项目里定义一个 StateGraph，把节点函数、条件边配好，编译后在 Python 进程里直接invoke()一条消息进去，Agent 就开始按图执行。这种模式解决的核心问题是“快速验证图逻辑”：节点顺序对不对、状态字段有没有正确传递、工具调用是不是按预期触发。对原型验证和本地联调来说，脚本模式就是效率天花板。

但站在生产角度看，脚本模式天生缺三样东西。第一，没有对外接口。脚本的输出是给终端看的，程序没法通过 HTTP 调用它完成业务集成。第二，没有并发能力。一个 Python 进程跑一份 graph，最多写个 for 循环串行处理，完全支撑不了多个用户同时发起请求的场景。第三，没有状态持久化和可观测性。脚本进程结束，checkpoint 里的状态就丢了；链路出了问题只能靠 print 慢慢查，没有日志聚合、没有 trace、没有指标监控。

这就是你需要往“服务化”走的原因。服务化的本质，不是写几个 API 接口那么简单，而是把你的 Agent 应用当成一个独立运行的、可持续提供服务的后端系统来对待。

1.2 服务化要过的三关

无论选哪条部署路径，最终都要过这三关：

• 接口关：把invoke()封装成业务系统能调用的 HTTP 接口，处理好请求参数、响应格式、超时和异常。
• 状态关：Agent 的线程状态要能持久化。用户对话到一半断了，重新发起请求时还能接着上下文继续处理。
• 运维关：要有结构化的日志、健康检查、错误追踪和基本监控，线上出问题的时候不至于两眼一抹黑。

这三关对应三条路径的不同“解法”。脚本模式基本不管这些，FastAPI 单体服务化能解决前两关的大部分问题，而 LangGraph Platform 和微服务架构则把运维关也一起收编了。这就引出下面的选型问题。

2. 三条路径的整体认识与选型思路

2.1 三条路径速览

我先把三条路径的整体情况拉一张表，方便你第一时间建立印象：

在路径三里我还想多说一句：它并不一定非要直接上 LangGraph 官方平台。不少团队是用 LangGraph 搭多个 Agent，再自己拆分微服务，用消息队列和 API 网关把它们串起来。平台化部署和自研微服务这两条分支，本质上都是在回答同一个问题——“单机单进程跑不完的复杂 Agent 系统，该怎么组织”。这个等后面第 5 节再展开。

2.2 选型的关键判断标准

谈技术选型，我习惯先问三个问题，这比直接对比功能清单有用得多。

• 多少人用？如果只有你自己、或者最多几十个内部用户跑批处理任务，路径一完全够。一旦要对外提供 API 服务，至少得走路径二。
• Agent 有几个？单个 Agent 串联几个工具，路径二很稳；如果有多个 Agent 需要互相协调、共享状态、独立扩缩容，路径三的架构思路更合适。
• 团队有没有运维资源？路径二需要你自己管服务进程、数据库和部署脚本。路径三如果用官方平台，运维负担小，但有外部依赖和成本；如果自托管 Docker 镜像，又要维护集群。

没有哪条路径是“最好”的，只有跟当前团队规模和业务阶段匹配的。我见过不少团队一上来就奔着微服务去，最后被运维和链路排查拖垮；也见过一直卡在脚本阶段不敢上线，耽误了业务验证。选型的关键是认清自己的真实需求。

3. 路径一：脚本直跑，最快的闭环

3.1 一个最小可跑的 LangGraph 脚本

先看一个最基础的例子，让你对整个执行机制有个具象认识：

# minimal_graph.py from typing import TypedDict, Annotated from langgraph.graph import StateGraph class AgentState(TypedDict): query: str steps: Annotated[list, lambda x, y: x + y] def node_input(state: AgentState): return {"steps": ["收到用户输入: " + state["query"]]} def node_process(state: AgentState): return {"steps": ["进入处理节点"]} def node_finish(state: AgentState): return {"steps": [f"最终处理完成: {state['query']}"]} builder = StateGraph(AgentState) builder.add_node("input", node_input) builder.add_node("process", node_process) builder.add_node("finish", node_finish) builder.add_edge("input", "process") builder.add_edge("process", "finish") builder.set_entry_point("input") builder.set_finish_point("finish") app = builder.compile() result = app.invoke({"query": "你好"}) print(result["steps"])

在终端里执行：

python minimal_graph.py

输出结果是：

['收到用户输入: 你好', '进入处理节点', '最终处理完成: 你好']

这个脚本里有几个值得注意的机制。AgentState定义了整张图的全局状态，每个节点函数的返回值是一个“部分状态”字典，LangGraph 会自动把它合并到全局状态里去。Annotated里第二个参数是 reducer 函数，我这里用lambda x, y: x + y表示“向列表追加”，如果不用这个注解，后面的节点返回同样 key 时会把前面的值直接覆盖掉。invoke()是同步阻塞的，逻辑简单直接，适合偏离线的场景。

3.2 脚本模式的适用场景与边界

脚本模式适合三类场景：离线批处理、命令行工具、联调测试。

拿离线批处理举例。我之前做过一个客服工单分类的 Agent 脚本，每天固定从数据库读取新增工单，经过 LangGraph 链路完成摘要提取、分类打标、紧急程度识别，最后把结果写回表里。整个流程没有实时交互需求，跑一次脚本半小时，非常适合脚本模式。命令行工具也很典型，比如面向开发者的“一键生成项目周报”“批量整理订单摘要”。联调测试就更常见了，在测试代码里直接调graph.invoke()断言输出，比写 HTTP 测试省事得多。

它的边界也同样明显：没有接口、没有并发、状态不持久化、没有监控。所以只要业务走到了“给别人用”这一步，就必须往服务化迁移。

3.3 从脚本到服务的过渡技巧

虽然脚本模式不需要服务化，但我在项目里有个习惯，就是从一开始就把“图构建”和“图执行”分离。具体来说，目录结构尽量这样组织：

agent_project/ ├── graph/ │ ├── state.py # AgentState 与字段定义 │ ├── nodes.py # 所有节点函数 │ └── build.py # 图组装与 compile ├── scripts/ │ └── run.py # 脚本入口，直接 invoke ├── main.py # 以后的服务入口，先预留 └── requirements.txt

这个结构看起来多花了一点功夫，但后面价值非常大。脚本模式里run.py调用build.py里导出的编译图来执行；迁移到 FastAPI 时，main.py只需要导入同一个编译图，再多包一层接口逻辑。你不需要改动任何节点和图的组装代码。我在实际项目里经历过几次重构，凡是当初“图构建与执行分离”做得好的，后面切服务几乎零成本；凡是把所有逻辑揉在一个文件里的，最后都要在接口层踩一圈坑才能把依赖理清。

4. 路径二：用 FastAPI 把 LangGraph 封装成服务

4.1 单体服务怎么搭

FastAPI 是 LangGraph 服务化最常见的搭档。它的生态成熟、支持异步、自动生成 OpenAPI 文档，拿来包装图非常顺手。我推荐的服务端基础结构如下：

# main.py from fastapi import FastAPI from pydantic import BaseModel from graph.build import app as graph_app # 复用编译好的图 api = FastAPI(title="LangGraph Agent API") class AgentRequest(BaseModel): user_id: str query: str thread_id: str = "default" class AgentResponse(BaseModel): answer: str thread_id: str steps: list @api.post("/agent", response_model=AgentResponse) async def run_agent(req: AgentRequest): result = await graph_app.ainvoke( {"query": req.query}, config={"configurable": {"thread_id": req.thread_id}} ) return AgentResponse( answer=str(result.get("answers", "")), thread_id=req.thread_id, steps=result.get("steps", []) )

这里有个很核心的细节：我用的是ainvoke()，而不是invoke()。FastAPI 的async def接口跑在事件循环上，如果里面调用同步阻塞的invoke()，整个事件循环都会被卡住，所有并发请求都要排队。LangGraph 从早期版本开始就支持异步调用，在 async 接口里直接await graph_app.ainvoke(...)是标准做法。这一点我会在后面“踩过的坑”里再展开。

thread_id是 LangGraph 的线程机制。它实现的是“会话隔离”：每个thread_id对应一条独立的会话上下文。业务上可以把用户的登录态 ID 或会话 ID 映射成 thread_id，这样不同用户之间的 Agent 状态不会互相干扰。在请求模型里把thread_id设置成可选，带一个默认值，是为了兼容单轮查询的场景。如果你做的是多轮对话，一定要让前端每次请求带上同一个 thread_id。

4.2 状态持久化：从 MemorySaver 到 Postgres

默认编译出来的图，每次invoke()都是独立执行的，状态不会延续。要实现跨请求记住对话内容，必须配置 checkpointer，也就是检查点机制。

最低成本的方案是内存检查点：

from langgraph.checkpoint.memory import MemorySaver app = builder.compile(checkpointer=MemorySaver())

这个方案用起来非常方便，但有两个明显问题：服务进程重启后所有状态清空，且所有会话历史都堆在内存里，会话多了内存占用会直线飙升。它适合开发调试，不适合生产环境。

一旦有线上需求，我建议换成 PostgresSaver。它把检查点写入 PostgreSQL 数据库，进程重启、多副本部署都能共享状态：

from langgraph.checkpoint.postgres import PostgresSaver conn_string = "postgresql://agent_user:agent_password@localhost:5432/agent_db" with PostgresSaver.from_conn_string(conn_string) as checkpointer: app = builder.compile(checkpointer=checkpointer)

这里有个实际体验要提醒你：PostgresSaver.from_conn_string()在 LangGraph 较新的版本里是作为上下文管理器使用的。你需要在with块内完成图的编译，这样才能保证数据库连接关闭前图对象已经被正确初始化。我第一次用的时候踩过这个坑，把图编译写在with外面，结果连接断开后图无法执行。

4.3 同步阻塞与异步执行的坑

这节是我最想分享的实战经验，因为踩坑的人实在太多了。LangGraph 图里的节点函数可以定义为同步函数，也可以定义为async def异步函数。如果你的图里全是同步节点，那么invoke()和ainvoke()都可以用；但如果图里有一个节点是 async 函数，整张图就只能用ainvoke()，用invoke()会直接报错。

在 FastAPI 集成时，最安全的组合是：接口用async def，调用用ainvoke()。我看过不少人的代码，接口定义成def（同步），里面直接调同步的invoke()，有时也能跑通，但并发一大就开始卡。原因是 FastAPI 对def接口是放到线程池里执行的，线程池默认大小有限，Agent 长时间执行时线程被占满，新的请求只能排队等待。所以除非你的接口逻辑确实不需要异步，否则尽量统一用async def + ainvoke。

另一个常见的坑是节点风格不统一。我建议在一个项目里明确约定：要么所有节点都写同步函数，要么都写异步函数。混着写虽然 LangGraph 支持，但排查问题时心智负担明显加大。我自己的习惯是，涉及外部 HTTP 调用的节点写成 async 函数（比如调大模型 API、调第三方查询接口），纯内部计算的节点写同步函数。即便如此，我也会尽量避免在同一个图里混用。

4.4 再配一点工程化细节

单体服务说起来简单，真正能上线的服务还要补一些细节。

图对象是重量级的，启动时只编译一次，不要在请求处理函数里重新 compile。我见过有人把builder.compile()写在请求函数内部，每来一个请求就重新构建一次图，性能直接退化成玩具级。

日志要做好结构化。请求进来时记录一条包含 thread_id 和 query 摘要的日志，响应出去时记录一条结果日志。Agent 内部每个节点的重要流转也要适当埋点。这样线上排查问题时，你可以按 thread_id 把一次完整请求的日志串起来。

给接口设计统一错误码。Agent 内部异常（比如模型调用失败、工具超时）不要直接把堆栈抛给客户端，而是包一层业务错误码，返回 HTTP 200 但不能算成功的业务响应，或者返回带错误信息的 500 响应体。我习惯的做法是同时支持：真实异常记录在服务端日志，客户端收到的是可读的错误信息。

对并发也不能掉以轻心。Agent 应用和普通 CRUD 接口不一样，一次请求可能持续几十秒甚至几分钟，如果不做限流，几十个请求就能把模型 API 限额打爆。我建议用asyncio.Semaphore做一个简单的并发控制，或者在网关层配置速率限制。这个防护措施在真实线上场景里非常管用。

5. 路径三：LangGraph Platform 与微服务架构

5.1 LangGraph Platform 到底给了什么

LangGraph Platform 是 LangChain 官方的部署方案。它把路径二里很多需要自己实现的事情直接托管了：部署编排、状态持久化、监控告警、版本管理。对团队来说，这省下了不少运维成本。

它支持的部署形态大体分三种：云托管、自托管 Docker、本地开发模式。云托管最省心，git push 之后平台自动构建镜像并部署；自托管是在你自己的服务器上跑官方 Docker 镜像，数据存在自己的数据库里，适合对数据主权有要求的团队；本地开发模式则是用 CLI 工具帮你在开发机上跑一个起步环境。

CLI 工具的体验在开发阶段确实很顺，大致流程是这样：新建项目、本地调试、构建镜像、部署到目标环境。一天之内从零到线上其实是可达成的。但它的代价主要是学习成本和平台绑定：CLI 和平台的有很多既定约定，你需要按它的目录结构和配置规范来组织项目。如果你们团队已经有成熟的 CI/CD 体系，也可以跳过官方平台，只用 LangGraph 的库能力，自己搭建微服务架构。

5.2 微服务拆分的实际场景

很多团队选择路径三，并不是上了 LangGraph Platform，而是基于 LangGraph 做服务拆分。这要从实际场景说起，我梳理了三种典型的微服务化动因。

第一种是多 Agent 协作。比如一个客服系统里有意图识别 Agent、商品推荐 Agent、售后处理 Agent。这三个 Agent 的功能边界清晰，谁挂了都不应该影响其他模块。把它们拆成独立服务，各自扩缩容、各自发布，比塞在单个单体应用里更利于长期迭代。LangGraph 负责每个 Agent 内部的编排，服务之间的通信则用 HTTP 或消息队列。多 Agent 编排如果放在一个图里，状态会高度耦合、热路径集中在单一执行引擎上，很快就变成巨型图，老板看着都头大。

第二种是长任务处理。Agent 执行一个复杂任务可能要几分钟，同步 API 根本扛不住。这种情况下，把任务投递到消息队列，消费端启动 LangGraph 工作流执行，执行完成后通过 Webhook 或者轮询接口向调用方通知结果。LangGraph 在这里负责的是任务编排，消息队列负责解耦和削峰。

第三种是事件驱动架构。有些业务天然是事件流驱动的，比如订单状态变化后自动触发售后 Agent、库存 Agent、物流 Agent 的联动。上游业务产生事件，消息中间件广播给多个 Agent 服务，每个 Agent 服务独立消费事件、独立处理。LangGraph 在图内部处理 Agent 的状态流转，对外只需暴露“接收事件、产出处理结果”这两个接口。

我个人的看法是：如果单体服务几百并发能扛住，很多拆分其实是不必要的。微服务拆分的前提是出现了真实的压力，比如多个团队协作频繁冲突、模块间资源竞争导致故障互相扩散、发布时长和风险已经难以接受。如果这些压力都不存在，聚合反而是更合理的选择。技术方案选型是为业务服务的，不要为了架构“高级”去做微服务。

5.3 与前端和调用方的交互方式

路径三对外暴露的交互方式跟路径二有明显区别。LangGraph Platform 提供了 SDK 和 REST API，它的交互模型是“线程 + 消息”。调用方先创建一个 thread，然后把用户消息推到 thread 里，通过轮询或事件流接收 Agent 的产出。这个模型和聊天应用天然匹配，每个 thread 就是一次完整的会话。

当你在 LangGraph Platform 的配置里部署好服务后，调用方通常不需要直接感知 LangGraph 的细节。只要约定好输入消息格式、初始化参数和回调 URL，就能实现集成。这种抽象对前端团队尤其友好，他们看到的只是一个有状态会话接口，而不是一张复杂的图。

如果你是自己拆分微服务，交互方式就更自由。内部服务之间可以用 HTTP REST，也可以用 gRPC，或者干脆走消息队列异步通信。关键是把 LangGraph 边界隔离开，不要让图的内部状态溢出到服务调用方，这样才能保留后续演进的空间。

6. 常见问题排查与实操建议

6.1 高频问题速查表

这里把我运维 LangGraph 服务时经常遇到的问题整理成一张速查表，非常实用：

第 1 条和第 2 条是我遇到频率最高的。前者是性能和并发问题，后者是功能性缺陷，都不难查，但代码里很容易漏。

6.2 部署前的检查清单

我每次部署 LangGraph 服务前，都会过一遍下面这份清单，你直接拿来就能用：

• 是否锁定了所有依赖版本？LangGraph 迭代速度很快，接口有变动，生产环境必须锁版本。
• 环境变量是否与代码分离？数据库连接串、模型 API Key、服务端口一律走环境变量，不要硬编码。
• 是否配置了 checkpointer？没有检查点，多轮对话就是摆设。
• 是否配置了结构化日志和错误告警？线上出问题的时候，这是第一道排查依据。
• 是否做过并发自测？用 wrk 或 locust 压一下，看并发到多少时 P99 开始恶化。
• 是否处理了 Agent 执行超时和模型接口抖动？外部依赖不稳定时，要有重试和降级策略。

清单本身不复杂，但每一条背后都是真实线上故障换来的教训。

6.3 我的几个实战体会

最后分享几条自己在实操中慢慢沉淀出来的体会。

第一，路径一和路径二之间没有鸿沟。只要你从一开始就把“图构建”和“图执行”分离，从脚本切到 FastAPI 服务只是加一层接口的事。我见过太多人把节点函数和invoke()写在同一个文件里，最后切服务时代价非常大。

第二，LangGraph 的版本演进很激进。生产环境一定要锁定版本，并且关注 release note。有些版本升级会改 checkpointer 的 API、改节点执行顺序的语义，没有读过变更说明就贸然升级，很容易在不知不觉中踩雷。

第三，Agent 服务不是普通 CRUD 服务。一次请求的链路时间可能长达几十秒，调用方会超时，模型会抖动，工具会抽风。所以超时控制和重试机制不是可选项，而是必选项。我的习惯是把外部调用统一封装成带超时和重试的工具函数，内部再设置一个整链路超时兜底。

第四，状态持久化选型不要图省事。MemorySaver 调试归调试，生产环境尽早切到 PostgresSaver。这个切换越早做，后面要迁移的历史数据就越少。

我自己的习惯是：只要 Agent 逻辑超过三个节点，就先按“图构建/执行”分层写；只要服务要上线，就先配好 PostgresSaver、结构化日志和超时机制。这些功夫放在前面，后面无论走哪条路径都很顺。希望这篇总结能让你少踩一些我踩过的坑。