LangGraph部署实战：从脚本到微服务的三条演进路径

最近我一直在折腾一件事：把已经跑通的 LangGraph 应用从本地.py脚本，一步步拆成真正的 HTTP 服务，再变成能扔进容器里横向扩展的微服务组件。折腾完这轮之后，LangGraph 的三条部署路径算是彻底被我摸了一遍。这里说的三条路径，正好对应标题里"从脚本到服务"的演进过程：脚本化部署、单体服务封装、微服务化接入。这篇文章我会从具体代码出发，把每条路径的适用场景、实现方式、关键取舍讲透，也会把过程中踩过的坑整个倒出来。适合正在做 LangGraph 开发、并且对"本地能跑、线上不挂"有真实需求的同学。

先说结论：LangGraph 本身对部署形态非常中立，同一个StateGraph编译后的实例，既能被invoke()直接调用，也能交给 FastAPI 包装成 HTTP 接口，还能连上外部数据库做成多副本运行的有状态服务。难点从来不在框架，而在于你对不同阶段的工程要求有没有想清楚。

1. 为什么强调"从脚本到服务"：部署形态决定工程复杂度

1.1 脚本是起点，但脚本的边界很清晰

LangGraph 最吸引人的地方是它让 Agent 的逻辑变得可编排、可调试。你会很自然地先在 Notebook 或者.py文件里写好节点、边和状态，然后compile()出一个 graph 对象，再用几行invoke()验证效果。这个阶段非常爽，因为回环迭代成本极低，出问题直接改代码重跑就行。

但脚本作为部署形态，有四个绕不过去的硬伤：

• 没有 API 边界。业务系统、前端页面、其他服务想调你这个 Agent，总不能去执行一个 Python 脚本吧？标准 HTTP 接口才是沟通语言。
• 没有生命周期管理。脚本进程一退出，所有状态就没了。你要是用了MemorySaver做会话持久化，重启一次全部消失，没有任何恢复机制。
• 并发控制基本靠运气。多线程同时调用同一个 graph 实例，一旦涉及共享状态或者外部写操作，出问题非常隐蔽，进程直接挂掉也不罕见。
• 运维观测接近零。脚本打印的日志不像服务那样有结构、有级别、有链路 ID，出了线上故障连从哪查起都不知道。

所以我一直认为，脚本只适合两件事：开发期验证逻辑，以及跑那些没有对外访问需求的批处理任务。一旦这个 Agent 要被别人用，就必须往服务化走。

1.2 三条路径的边界：一张表看清差别

这里我把三条路径并行摆出来，帮助快速建立全局认识：

对比维度	路径一：脚本化部署	路径二：单体服务封装	路径三：微服务化接入
运行形态	Python 进程内直接跑	FastAPI / Flask 等单进程 HTTP 服务	容器化实例 + 外部状态存储
典型场景	本机调试、定时批处理、内部小工具	企业内部应用、中小流量线上服务	高并发、多团队、需要弹性伸缩
会话状态	进程内存，重启即失	可配 SQLite / 内存，单实例内有效	Postgres / Redis 等共享存储，多副本共享
并发扩展	基本不具备	单实例有限并发，可配合队列	水平扩缩容，按流量自动调整
运维复杂度	几乎为零	中等：部署、日志、健康检查	高：容器编排、服务发现、可观测性
上手成本	极低，几小时就能跑	中低，一天内能完成	高，需要团队基础设施配合

1.3 怎么选：问自己三个问题就够了

选型其实没那么玄，也不用把架构设计得天花乱坠。我在每次接手新项目时就问三个问题：

1. 谁会调用它？如果只有我一个人在终端里手动调，脚本就行；如果前端页面、业务后端要对接，起码得是 HTTP 服务；如果多个不同团队的服务都要接入 Agent 能力，那就走微服务化。
2. 需要几个副本？单实例能扛住流量，单体服务封装足够；一旦需要多副本或者跨机房部署，MemorySaver 这种内存态方案就彻底失效了，必须上外部 Checkpointer。
3. 会话状态重要吗？用户只是问一次、不用上下文连续对话，那状态无所谓；但一旦涉及多轮对话、长流程任务、人工审批中断后恢复，checkpoint 的持久化能力就是核心需求。

这套判断逻辑看起来很朴素，但能避免大多数过度设计。很多项目根本不是技术上的问题，而是"明明一个脚本能解决的事，非得上微服务全家桶"。

2. 路径一：脚本化部署——在终端里把 Agent 当成命令行工具

2.1 脚本部署的定位：能跑起来也是生产力

脚本化部署很容易被低估，但我建议所有项目都从这一步开始。它最大的价值是让你在最短时间内验证"这个 Agent 到底有没有用"。比如我做内部数据清洗的 Agent，压根不需要 GUI 也不需要接口，一个交互式 CLI 脚本就是最好的形态：输入查询语句，回车，看返回的 SQL 和结果，调参也方便。

另外一个非常典型的场景是定时批处理。比如每天早上拉取一批工单，用 Agent 做自动分类和优先级打标，然后把结果写回数据库。这种任务用脚本+cron（或者 Windows 下的计划任务）就够了，做成服务反而是浪费。

2.2 从零搭一个 CLI 版 LangGraph Agent

我直接给一个最小但完整的例子。这个例子包含两个节点：用户入口节点和 Agent 节点，另外加了一个MemorySaver让多轮对话有记忆。

# main.py from dotenv import load_dotenv load_dotenv() from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from typing import TypedDict from langchain_core.messages import HumanMessage from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: list def agent_node(state: AgentState): return {"messages": [llm.invoke(state["messages"])]} llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_edge(START, "agent") graph.add_edge("agent", END) app = graph.compile(checkpointer=MemorySaver()) if __name__ == "__main__": config = {"configurable": {"thread_id": "cli-session"}} messages = [] print("LangGraph CLI Agent，输入 exit 退出") while True: user_input = input("你: ") if user_input.strip().lower() in ("exit", "quit"): break messages.append(HumanMessage(content=user_input)) result = app.invoke({"messages": messages}, config=config) print(f"Agent: {result['messages'][-1].content}")

这里有一个关键设计：我把messages整个作为状态维护，并且每次调用都带上同一个thread_id。原因是MemorySaver会把每一轮的对话状态落在这个 thread 名下，下一次invoke时 LangGraph 会自动从上一次的状态继续，而不是从空白开始。你如果只传当前用户输入，而不把历史消息传进去，多轮对话是割裂的。

运行方式也简单：

python main.py

这样你就有了一个能对话、有记忆、可随时改逻辑的 Agent。想接工具调用，就在节点里bind_tools，加上ToolNode，再加一条条件边判断是否进入工具循环。

2.3 脚本化最容易踩的坑

脚本阶段看似简单，坑其实不少。我单独拎出来说几个高频的，这些也是我和同事们实测过程中反复遇到过的。

坑一：Windows 双击脚本闪退。这个问题和 Python 没关系，纯粹是环境问题。双击.py或者.bat文件时，如果 Python 不在 PATH、或者.py关联了解释器但缺依赖，窗口会一闪而过，什么都看不到。我现在的处理习惯是：任何脚本入口的.bat文件第一行写chcp 65001保证编码，最后加一行pause，让窗口停下来显示真正的报错信息。这样排查问题效率高很多。

坑二：环境变量加载位置不对。load_dotenv()是相对路径的，它默认会找"当前工作目录"下的.env。你在项目根目录跑没问题，但一旦用 cron 定时任务跑，工作目录往往被切到别处，.env就加载不到，API Key 全部为空。建议的做法是在脚本里显式指定路径：

from pathlib import Path from dotenv import load_dotenv load_dotenv(Path(__file__).resolve().parent / ".env")

这样不管在哪被启动，都按脚本文件所在目录去找配置文件。

坑三：asyncio 和 Jupyter 的事件循环冲突。如果你在 Notebook 里调试，直接写asyncio.run(main())大概率会报This event loop is already running。这是因为 Jupyter 本身就是跑在事件循环里的，你不能在里面再启动一个新的。常见处理方法是装一个nest_asyncio，或者在单独的线程里发起任务：

import nest_asyncio nest_asyncio.apply()

这个问题后续进入服务化阶段还会遇到另一个变体，后面我专门讲。

3. 路径二：单体服务封装——用 FastAPI 把 Agent 变成标准 HTTP 接口

3.1 为什么中间要加一层服务

脚本跑通之后，下一个自然动作就是把 Agent 暴露成 HTTP 接口。这一步的价值不是"看起来正规"，而是实实在在地解决协作问题：业务后端不需要知道你的 Python 项目结构，不需要装一堆依赖，它只需要按约定发一个 HTTP 请求，拿到一个结构化响应。

我最初做单体封装时也很犹豫，觉得直接用 Flask 包一层不就行了。后来发现真正要注意的其实是三件事：流式输出怎么做、鉴权怎么做、超时和并发怎么控制。这三件事在脚本阶段完全不需要考虑，一旦对外暴露服务就全是必答题。

3.2 同步接口与 SSE 流式输出

先看最简单的同步请求-响应接口。用 FastAPI 加 Pydantic 做参数校验，干净利落：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel from langchain_core.messages import HumanMessage app = FastAPI(title="LangGraph Agent Service") class ChatRequest(BaseModel): message: str thread_id: str = "default" class ChatResponse(BaseModel): reply: str thread_id: str @app.post("/agent/invoke") async def invoke(req: ChatRequest): config = {"configurable": {"thread_id": req.thread_id}} try: result = await agent_app.ainvoke( {"messages": [HumanMessage(content=req.message)]}, config=config, ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) return ChatResponse( reply=result["messages"][-1].content, thread_id=req.thread_id, )

注意一个细节：这里我用的是ainvoke（异步版本），而不是同步的invoke。原因很简单，FastAPI 的路径函数本身跑在事件循环里，如果调同步invoke，一旦 LLM 响应慢，整个进程的并发能力都会被阻塞。ainvoke会把调用挂起到后台，事件循环还能继续处理其他请求。

实际做对话体验时，同步接口体验不够，大部分产品都需要打字机式的流式输出。此时要上 SSE（Server-Sent Events）。关键区别在graph.stream和astream_events之间的选择：

• astream()/graph.stream()：返回的是节点级的状态流，粒度是"每一步图更新后的完整状态"，适合做步骤展示。
• astream_events(version="v2")：返回的是更细粒度的事件流，能拿到 LLM 吐出每个 token 的事件，这才是真正打字机效果。

如果要复刻 ChatGPT 那种逐字输出的体验，就要监听on_chat_model_stream事件：

from fastapi.responses import StreamingResponse import json @app.post("/agent/stream") async def stream(req: ChatRequest): config = {"configurable": {"thread_id": req.thread_id}} async def event_generator(): async for event in agent_app.astream_events( {"messages": [HumanMessage(content=req.message)]}, config=config, version="v2", ): # 只拿 LLM 输出 token 事件 if event["event"] == "on_chat_model_stream": token = event["data"]["chunk"].content if token: yield f"data: {json.dumps({'token': token}, ensure_ascii=False)}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(event_generator(), media_type="text/event-stream")

一个来自实践的忠告：不要把 token 事件和节点事件混在一起返回给前端，前端解析逻辑会乱成一团。建议后端先做好聚合，只推最外层需要的事件类型。

3.3 工程化细节：鉴权、超时、并发与优雅关闭

单体服务封装虽然简单，但该上的工程措施一个都不能少。

鉴权。至少加一个 API Key 中间件。我的做法是在 FastAPI 依赖里校验请求头：

from fastapi import Depends, Header def verify_api_key(x_api_key: str = Header(...)): if x_api_key != API_KEY: raise HTTPException(status_code=401, detail="Invalid API Key") @app.post("/agent/invoke", dependencies=[Depends(verify_api_key)]) async def invoke(req: ChatRequest): ...

这样内网或者 API 网关先拦住一层，服务自身也有一层兜底。

超时控制。LLM 服务经常是延迟不可控的，所以一定要给外部调用套上超时：

import asyncio try: result = await asyncio.wait_for( agent_app.ainvoke(...), timeout=60, ) except asyncio.TimeoutError: raise HTTPException(status_code=504, detail="Agent timeout")

不要觉得 60 秒很长，在很多 Agent 工具调用场景里，一次完整流程可能真的会拖到一分钟以上。设超时不是为了打断正常请求，而是防止死等一个无响应的模型调用。

并发控制。同一个 graph 实例被大量并发调用时，MemorySaver 虽然内部有锁保护，但高并发下还是容易出现状态串线的风险。如果同一个thread_id同时被两个请求写入，甚至可能产生竞态。我的方案是两层：外部用asyncio.Semaphore限制服务最大并发数保护下游，同时约定前端对同一个thread_id必须串行请求，避免一个会话多路并发。

优雅关闭。用 FastAPI 的 lifespan 来管理初始化与关闭逻辑。如果用了 SQLite 或 Postgres 版的 Checkpointer，进程退出时一定要关闭连接，否则小文件/连接池容易出问题：

from contextlib import asynccontextmanager @asynccontextmanager async def lifespan(app): # 启动时加载模型、初始化空间 yield # 关闭时释放资源 if hasattr(agent_app, "checkpointer"): agent_app.checkpointer.close()

4. 路径三：微服务化接入——把 LangGraph 放进真实的生产架构

4.1 单体服务的上限在哪里

单体服务封装到一定程度，你一定会撞到两个天花板。

第一个是有状态扩展的天花板。MemorySaver把状态放在进程内存里，这决定了你没法随便水平扩副本。你想多加两个实例扛流量，但请求落在不同副本上时，会话状态互相不感知，用户忽而连不上之前的上下文。所以只要是多副本架构，进程内 Checkpointer 就必须换成外部存储。

第二个是架构边界的天花板。单体服务把 Graph 的所有逻辑塞进一个进程，代码一多，不同团队的分工就乱了。有的团队管对话逻辑，有的团队管工具调用，有的团队管审批流，全揉在一起，发版就是一个大爆炸。微服务化的本质是让每个服务只管一件事，LangGraph 服务在这里也只是一个独立组件。

4.2 Checkpointer 升级：从 MemorySaver 到 PostgresSaver

这是微服务化最关键的一步。把状态从内存搬到统一的数据库之后，多副本共享会话就是一个配置问题。

用 LangGraph 官方提供的PostgresSaver非常简单：

from langgraph.checkpoint.postgres import PostgresSaver conn_string = "postgresql://user:pass@host:5432/langgraph" with PostgresSaver.from_conn_string(conn_string) as checkpointer: checkpointer.setup() # 创建必要的数据表，只执行一次 agent_app = graph.compile(checkpointer=checkpointer)

要注意的是这里接口变成上下文管理器了，数据库连接会占用一个连接池。所以更常见的生产模式是先启动连接池，再把它注入到 Checkpointer，最后交给任务常驻进程：

from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool pool = ConnectionPool(conninfo=conn_string, max_size=20) checkpointer = PostgresSaver(pool) checkpointer.setup() agent_app = graph.compile(checkpointer=checkpointer)

配置好之后，任何一个副本上发起带thread_id的请求，状态都会读写到统一的 Postgres。副本 A 发起对话，请求落到副本 B，照样能正确续上文。这也是"有状态服务水平扩展"的标准解法。

Redis 也有相应的官方 Checkpointer 实现，适合对低延迟要求更高的场景，但需要额外处理快照序列化与版本兼容问题。没有特殊需求时，Postgres 会更皮实，排查问题也直观。

4.3 容器化与编排要点

微服务化的标配是容器化。语言本身不要求你必须用 Docker，但我们团队实测下来，容器化把"依赖地狱"问题直接消灭了：模型 SDK 版本、LangGraph 版本、Python 小版本不一致导致的诡异问题，全都隔离在镜像里。

一个能直接用的 Dockerfile：

FROM python:3.11-slim WORKDIR /app ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

编排层面有几个细节，踩过坑才知道：

• 千万不要只靠进程内队列做扩展。每副本各自维护的内存队列在扩容时完全是垃圾，要么用外部消息队列，要么让无状态请求直接打到任意副本。
• 健康检查必须和业务联通。/healthz不要只返回 200，最好顺带验证一下模型服务是否可达、数据库连接池是否正常。否则副本明明已经不可用了，网关还在往里转发流量。
• 资源限制要给足。LangGraph 的 Agent 节点往往要同时缓存不小的状态对象，Python 进程的内存占比波动很大。K8s 里给 Pod 设置 requests/limits 时，建议给 ratio 留出余量，实测一般 requests 设 256MiB、limits 设 1GiB 起步比较稳妥，具体取决于你的状态大小。

服务注册与发现这一层，如果已经在用 Kubernetes，那直接用 Service 就够了；如果是传统的 CentOS + Supervisor 环境，就接一个 Consul/Nacos 之类的注册中心，让网关动态感知后端实例。

4.4 异步任务与消息队列的集成

有些 Agent 任务特别长，比如"帮我分析这个季度的所有工单数据并生成报告"，一次跑下来几分钟甚至几十分钟。这类任务如果走同步 HTTP，客户端会等到怀疑人生，网关的超时也会把你干掉。

正确的思路是异步任务化：

• 客户端提交任务，API 网关返回一个task_id；
• 后端把任务写入消息队列（RabbitMQ、Kafka、Redis Stream 都行）；
• 消费端跑 LangGraph，跑到中途可以更新任务状态表；
• 跑完后把结果写回存储，通过 WebSocket 推送或者让客户端轮询GET /task/{task_id}获取结果。

这个模式解决的不只是长任务问题，它还给 LangGraph 服务提供了一层天然的削峰缓冲。流量突然暴涨时，队列堆积带来的压力远小于同步服务被拖垮。我参与的一个客服工单分类系统就是这个结构，实测下来即使在高峰期，Agent 节点挂掉重启都只会让任务排队，而不会造成用户请求直接失败。

4.5 可观测性与团队协作

微服务化之后，链路变得复杂，可观测性就是保命工程。我务实的建议是三层：

1. 结构化日志。别再用print()，直接上结构化日志。日志里必须带request_id和thread_id，这样整条 Agent 运行链路才能串起来查。
2. 链路追踪。LangGraph 官方生态里的 LangSmith 天然支持 graph 运行追踪，上线前务必接上。如果不想依赖外部平台，可以用 OpenTelemetry 打 span，至少把模型调用、工具调用、节点耗时几个关键 span 记录在案。
3. 指标监控。请求量、延迟分位数、工具调用成功率、checkpointer 读写耗时，这几个指标要提前埋好。工具调用失败率往往比模型报错更早暴露问题。

团队协作方面，接口契约建议从一开始就版本化。/v1/agent/invoke、/v2/agent/stream，改动不破坏旧客户端。OpenAPI 文档直接由 FastAPI 自动生成，让下游团队拿到 yaml 就能生成客户端，省去大量联调扯皮。

5. 常见问题与排查技巧实录

5.1 服务起不来：闪退、PATH 与端口占用

脚本阶段最常见的问题，上面提到了 Windows 下的闪退。服务阶段最常见的则是端口占用和启动路径错误。排查时先看这几点：

• 端口被占用：Linux 下用ss -lntp | grep 8000，Windows 下用netstat -ano | findstr :8000，先把进程杀掉或者换端口。
• 环境变量没进去：export之后启动 uvicorn 发现 key 为空，八成是.env没加载或者加载顺序错了。
• 依赖不完整：pip install -r requirements.txt之后还报ModuleNotFoundError，检查是不是装到了别的 Python 环境里。用pip list | grep langgraph先确认。

还有一个非常隐蔽的问题：如果在控制台直接执行python main.py没问题，但通过 systemd/supervisor 启动就报错，多半是环境变量 PATH 里少了 Python 路径，supervisor 默认 PATH 很精简。要在配置里显式environment=PATH=...注入。

5.2 asyncio 事件循环冲突与线程安全

Jupyter 里的This event loop is already running上面提过。服务化之后更常见的问题是在 FastAPI 同步接口里调asyncio.run()，直接报RuntimeError: asyncio.run() cannot be called from a running event loop。解决方式就一句话：在异步接口里不要手动创建事件循环，配合await去调ainvoke/astream。

线程安全这条线，我的经验是"单个thread_id必须串行访问"。并发写同一个会话时，即使有 Checkpointer 的锁保护，状态恢复也容易出怪问题。一定要在服务入口做拦截，或者引导前端给每个 thread 加锁。

5.3 模型限流、超时与重试

生产环境里，模型限流太常见了。OpenAI 等模型服务返回RateLimitError时，LangChain 的ChatOpenAI默认会重试若干次，但默认策略在高并发下并不优雅。我建议显式配置超时和重试：

llm = ChatOpenAI( model="gpt-4o-mini", request_timeout=60, max_retries=2, timeout_seconds=30, )

另外要注意区分"模型生成超时"和"工具调用返回超时"。如果 Agent 的工具节点是个 HTTP 请求，它同样需要自己的超时时间，不能依赖模型超时去兜底。

5.4 长时运行的内存和 Checkpoint 清理

单体服务长时间不重启，最典型的问题是内存缓慢增长。这通常来自两个地方：日志对象累积、以及模型缓存对象没有被释放。排查时可以用tracemalloc看内存分配热点，也可以用psutil定时打印 RSS 做趋势观察。如果确认是 Checkpointer 累积太多旧会话，可以定时清理超过 N 天的 thread 数据：

DELETE FROM checkpoints WHERE created_at < NOW() - INTERVAL '30 days';

这条 SQL 虽然粗暴，但能解决无限增长问题。欠优雅，胜在有效。

5.5 排查工具箱

最后给一个我每次排查 LangGraph 服务问题都会用的命令清单：

• 单步测试图逻辑：python -m pdb main.py或者直接写一个测试脚本调graph.stream，逐事件观察输出。
• 查看链路上所有调用：接好 LangSmith 之后，直接在 web 端按 thread_id 搜索整条 trace。
• 本地压测流式接口：用curl -N -X POST http://localhost:8000/agent/stream看事件是否按预期到达。
• 验证 Checkpointer 是否生效：起两个进程，用同一个thread_id连续请求，看第二个进程能否接上第一个进程的上下文。

这套工具箱解决问题的速度，远高于瞎猜和打印日志。

我个人在实际操作中的最深体会是：部署形态不该被"技术潮流"推着走，而应该跟业务阶段匹配。先用脚本验证 Agent 真正有用，然后快速用 FastAPI 把它变成能被别人调用的服务，等流量和协作复杂度上来了，再逐步引入外部 Checkpointer、容器编排、消息队列和链路追踪。这三条路径不是选择题，更像是升级打怪的自然顺序。业务和代码可以一路演进，但每一层的基础设施都必须站稳了再往上走。