最近我一直在折腾一件事把已经跑通的 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 怎么选问自己三个问题就够了选型其实没那么玄也不用把架构设计得天花乱坠。我在每次接手新项目时就问三个问题谁会调用它如果只有我一个人在终端里手动调脚本就行如果前端页面、业务后端要对接起码得是 HTTP 服务如果多个不同团队的服务都要接入 Agent 能力那就走微服务化。需要几个副本单实例能扛住流量单体服务封装足够一旦需要多副本或者跨机房部署MemorySaver 这种内存态方案就彻底失效了必须上外部 Checkpointer。会话状态重要吗用户只是问一次、不用上下文连续对话那状态无所谓但一旦涉及多轮对话、长流程任务、人工审批中断后恢复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(modelgpt-4o-mini, temperature0) graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_edge(START, agent) graph.add_edge(agent, END) app graph.compile(checkpointerMemorySaver()) 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(contentuser_input)) result app.invoke({messages: messages}, configconfig) print(fAgent: {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(titleLangGraph 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(contentreq.message)]}, configconfig, ) except Exception as e: raise HTTPException(status_code500, detailstr(e)) return ChatResponse( replyresult[messages][-1].content, thread_idreq.thread_id, )注意一个细节这里我用的是ainvoke异步版本而不是同步的invoke。原因很简单FastAPI 的路径函数本身跑在事件循环里如果调同步invoke一旦 LLM 响应慢整个进程的并发能力都会被阻塞。ainvoke会把调用挂起到后台事件循环还能继续处理其他请求。实际做对话体验时同步接口体验不够大部分产品都需要打字机式的流式输出。此时要上 SSEServer-Sent Events。关键区别在graph.stream和astream_events之间的选择astream()/graph.stream()返回的是节点级的状态流粒度是每一步图更新后的完整状态适合做步骤展示。astream_events(versionv2)返回的是更细粒度的事件流能拿到 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(contentreq.message)]}, configconfig, versionv2, ): # 只拿 LLM 输出 token 事件 if event[event] on_chat_model_stream: token event[data][chunk].content if token: yield fdata: {json.dumps({token: token}, ensure_asciiFalse)}\n\n yield data: [DONE]\n\n return StreamingResponse(event_generator(), media_typetext/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_code401, detailInvalid 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(...), timeout60, ) except asyncio.TimeoutError: raise HTTPException(status_code504, detailAgent 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:passhost:5432/langgraph with PostgresSaver.from_conn_string(conn_string) as checkpointer: checkpointer.setup() # 创建必要的数据表只执行一次 agent_app graph.compile(checkpointercheckpointer)要注意的是这里接口变成上下文管理器了数据库连接会占用一个连接池。所以更常见的生产模式是先启动连接池再把它注入到 Checkpointer最后交给任务常驻进程from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool pool ConnectionPool(conninfoconn_string, max_size20) checkpointer PostgresSaver(pool) checkpointer.setup() agent_app graph.compile(checkpointercheckpointer)配置好之后任何一个副本上发起带thread_id的请求状态都会读写到统一的 Postgres。副本 A 发起对话请求落到副本 B照样能正确续上文。这也是有状态服务水平扩展的标准解法。Redis 也有相应的官方 Checkpointer 实现适合对低延迟要求更高的场景但需要额外处理快照序列化与版本兼容问题。没有特殊需求时Postgres 会更皮实排查问题也直观。4.3 容器化与编排要点微服务化的标配是容器化。语言本身不要求你必须用 Docker但我们团队实测下来容器化把依赖地狱问题直接消灭了模型 SDK 版本、LangGraph 版本、Python 小版本不一致导致的诡异问题全都隔离在镜像里。一个能直接用的 DockerfileFROM python:3.11-slim WORKDIR /app ENV PYTHONDONTWRITEBYTECODE1 \ PYTHONUNBUFFERED1 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 可观测性与团队协作微服务化之后链路变得复杂可观测性就是保命工程。我务实的建议是三层结构化日志。别再用print()直接上结构化日志。日志里必须带request_id和thread_id这样整条 Agent 运行链路才能串起来查。链路追踪。LangGraph 官方生态里的 LangSmith 天然支持 graph 运行追踪上线前务必接上。如果不想依赖外部平台可以用 OpenTelemetry 打 span至少把模型调用、工具调用、节点耗时几个关键 span 记录在案。指标监控。请求量、延迟分位数、工具调用成功率、checkpointer 读写耗时这几个指标要提前埋好。工具调用失败率往往比模型报错更早暴露问题。团队协作方面接口契约建议从一开始就版本化。/v1/agent/invoke、/v2/agent/stream改动不破坏旧客户端。OpenAPI 文档直接由 FastAPI 自动生成让下游团队拿到 yaml 就能生成客户端省去大量联调扯皮。5. 常见问题与排查技巧实录5.1 服务起不来闪退、PATH 与端口占用脚本阶段最常见的问题上面提到了 Windows 下的闪退。服务阶段最常见的则是端口占用和启动路径错误。排查时先看这几点端口被占用Linux 下用ss -lntp | grep 8000Windows 下用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 很精简。要在配置里显式environmentPATH...注入。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( modelgpt-4o-mini, request_timeout60, max_retries2, timeout_seconds30, )另外要注意区分模型生成超时和工具调用返回超时。如果 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、容器编排、消息队列和链路追踪。这三条路径不是选择题更像是升级打怪的自然顺序。业务和代码可以一路演进但每一层的基础设施都必须站稳了再往上走。