ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

LangGraph 部署实战:从脚本到服务,三条路径与避坑指南

LangGraph 部署实战:从脚本到服务,三条路径与避坑指南 1. 从脚本到服务LangGraph 部署到底在解决什么问题先说个大实话LangGraph 在本地跑通一个 demo 和真正把它放到线上让人用中间隔着一条很宽的河。我自己最早接触 LangGraph 的时候是在 Jupyter Notebook 里写了个简单的 StateGraph用graph.invoke()跑通了一个带工具调用的 Agent当时觉得这玩意儿真香——状态流转清晰图结构一目了然节点之间的传递逻辑全在图上画明白了。然后我就想这要是能让同事通过 API 调一下甚至做成一个微服务丢到生产环境里那不就完美了吗结果一做才发现脚本里跑得欢快的图一到服务化部署就各种翻车。这个标题里有个关键词从脚本到服务。这其实是一个很典型的演进路径一开始大家做 AI Agent 都是写脚本试验证思路、调 prompt、测工具调用等到思路验证完了就面临一个非常现实的问题——怎么把这张图变成一个稳定的、可以被外部调用的服务。LangGraph 本身是一个编排框架它不是部署框架所以把图部署成服务这件事需要你自己去选择路径、设计方案、处理状态和并发。这篇文章我就从实际踩坑的角度把三条我验证过的部署路径掰开揉碎讲清楚包括各自的适用场景、操作细节、坑点以及我个人的推荐结论。先交代一下背景我做的项目是一个基于 FastAPI LangChain LangGraph 的 AI Agent 系统核心功能是让 Agent 能够调用内部工具比如查订单、算报价、调数据库然后通过标准 HTTP 接口对外提供服务。整体架构上采用了微服务拆分思路把 Agent 服务独立成一个模块不跟业务系统耦合在一起。这个项目经历了从纯脚本验证到服务化的完整过程所以下面讲的东西都是实测过的不是纸上谈兵。2. 三条部署路径的整体设计与选型逻辑2.1 路径一脚本直跑适合验证和内部自动化第一条路径最简单也是大多数人起步的方式把编译好的 LangGraph 图写成一个 Python 脚本通过命令行或者定时任务直接运行。这条路径连 Web 服务都不用起只需要保证 Python 环境、依赖包、API Key 配置好就行了。举个例子我用 LangGraph 做了一个定时抓取数据的 Agent逻辑是每天凌晨两点自动跑一次抓取外部数据源经过清洗和处理后写入数据库。这个场景根本不需要一个常驻服务——任务结束进程就退了第二天再拉起来跑一次就行。部署方式就是写一个main.py里面定义好 StateGraph、编译、调用然后用系统的cron或者systemd timer定时触发。这条路径的优点是显而易见的零额外组件、调试方便、出问题直接看日志。但它有一个很严重的限制无状态。你每次运行脚本都是一个全新的过程如果这个 Agent 需要多轮对话、需要在多步之间维护上下文脚本直跑就非常别扭。虽然 LangGraph 有 checkpointer 机制可以持久化状态但脚本场景下通常不会去配置一个专门的存储所以这更适合跑完即走的批处理任务。我后来在工作中发现很多所谓Agent 服务化的需求本质上就是让我能通过 HTTP 调用一下这个脚本。这种情况下其实没必要上全套微服务框架先评估一下这个任务是不是一次性的需不需要维护跨请求的状态需不需要并发处理如果答案都是否那脚本直跑反而是最优解别为了技术追求过度设计。2.2 路径二自建 HTTP 服务用 FastAPI 把图包成 API第二条路径也是我个人最推荐、并且目前项目里正在用的方式用 FastAPI 写一个独立的服务把 LangGraph 编译好的图封装成 REST API。这是从脚本迈向服务最关键的一步。FastAPI 在这个场景下几乎是天然适配的异步支持好、类型校验全、自动生成 OpenAPI 文档而且和 Pydantic 的集成极其顺畅。LangGraph 的输入输出正好可以基于 Pydantic 模型做约束这样整个 API 层的参数校验、错误反馈都能统一处理。在这一步设计上有个核心问题要想清楚一张图对应一个服务还是一个服务承载多张图我在实际项目中选择了后者——一个 Agent 服务暴露多个端点不同的图对应不同的业务场景。原因很简单这些图共享了底层的大模型配置、共享了工具注册表、共享了日志和监控体系拆成多个服务反而增加了链路复杂度。但如果你团队的规模和并发量很大比如不同图之间的调用频次差异巨大那拆开部署会更合理。之所以强调自建是因为它不需要依赖任何外部平台代码完全掌控在自己手里不管是部署到容器、K8s 还是裸机都能灵活调整。代价就是所有服务化的细节——并发控制、超时管理、状态持久化、日志采集——都得自己处理。这正好是这篇文章想重点展开的部分。2.3 路径三上 LangGraph Platform用托管服务换省心第三条路径是把图部署到 LangGraph Platform以及自托管版也就是用官方托管的平台来运行你的图。这条路径的优点是省心平台帮你处理了任务队列、持久化、可观测性、自动伸缩甚至带了一个能直接对话的调试前端。你只需要上传代码、绑定配置、获得一个 API URL然后用 SDK 或者直接 HTTP 调它就行。但我得说句实话这条路不是所有人都适合走。原因有几个。第一它引入了对平台的强依赖图代码和运行时要和平台版本对齐升级时会有迁移成本。第二如果业务场景本身对 AI Agent 的调用量不大为托管服务付出的成本可能不划算。第三在企业内网环境里数据合规和网络策略往往不允许你把 Agent 逻辑放到外部平台自托管版倒是可以解决这个问题但这又等于回到了自己运维一套平台的命题上。从架构角度讲LangGraph Platform 和自建服务本质上都是在解决同一个问题如何把一张图变成一个可以被调用的、有状态的服务。区别在于自建是把所有组件自己拼装托管是把组件外包出去。我的建议是如果你的项目还处于早期探索阶段不要急着上平台先把自建服务跑通搞清楚 LangGraph 服务的核心机制之后再判断是否需要平台的托管能力。至少我自己在评估完之后决定继续用自建方案。3. 实操过程三条路径的落地细节与避坑指南3.1 脚本直跑的完整配置我们先来看脚本直跑的具体操作。为了演示我创建了一个带工具调用的 Agent 脚本核心逻辑是用一个 ReAct 风格的图循环执行思考-调用工具-再思考这个过程。脚本的第一步永远是加载环境变量。我的习惯是老老实实在脚本开头写上from dotenv import load_dotenv load_dotenv()这里有个很多人忽略的细节如果你在脚本里配置了 LangSmith 的追踪LANGSMITH_TRACINGtrue那环境变量的加载顺序很关键必须在创建任何 LangChain 对象之前执行否则追踪可能不会生效。接着定义图from langgraph.graph import StateGraph, START, END from typing import TypedDict, Annotated from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: Annotated[list, add_messages] next_step: str def agent_node(state: AgentState): llm ChatOpenAI(modelgpt-4o-mini, temperature0) response llm.invoke(state[messages]) return {messages: [response]} graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_edge(START, agent) graph.add_edge(agent, END) app graph.compile()脚本的入口部分我强烈建议加上if __name__ __main__的隔离逻辑if __name__ __main__: result app.invoke({messages: [{role: user, content: 你好帮我查一下今天的订单量}]}) print(result[messages][-1].content)这个脚本在本地跑没问题但一旦你把它丢到服务器上用 cron 调度就会遇到一个很典型的坑环境变量丢失。因为 cron 的环境和你的 shell 登录环境不一样很多变量比如 API Key根本不会传给脚本。所以用 cron 跑这类脚本要么在 crontab 里显式声明环境变量要么在脚本里不依赖用户级环境变量用一个固定的配置文件加载。还有一个经验脚本直跑时不要忘记给 LLM 调用设置超时和重试。别以为本地跑得好好的服务器上网络环境可能有差异一次超时整个任务就断了。在 LangChain 里可以通过模型的timeout参数和max_retries参数控制ChatOpenAI(modelgpt-4o-mini, timeout30, max_retries2)另外如果你需要让脚本支持多轮记忆可以在编译图的时候传入一个checkpointer。脚本场景下最简单的是用MemorySaver但注意它只存在内存里进程退出就没了。如果要持久化到磁盘可以用SqliteSaverfrom langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string(checkpoints.sqlite) as checkpointer: app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: job-001}} result app.invoke(..., configconfig)这样即使脚本跑到一半中断了重新运行时只要用同一个thread_id还能接着之前的上下文继续。3.2 自建 FastAPI 服务的关键步骤接下来是重头戏把图封装成 FastAPI 服务。这一步做得好不好直接决定了你的 Agent 服务能不能扛住真实的调用压力。项目结构上我建议分这么几层agent_service/ ├── main.py # FastAPI 入口 ├── graph/ │ ├── state.py # 状态定义 │ ├── nodes.py # 节点逻辑 │ ├── tools.py # 工具注册 │ └── builder.py # 图构建与编译 ├── schemas.py # 请求/响应模型 └── config.py # 配置管理核心的builder.py里会构建好图并编译但我有一个很重要的经验教训不要在每次请求时编译图。图的编译过程是有开销的如果每次请求都graph.compile()高并发下性能会非常难看。正确做法是模块加载时只编译一次把编译好的CompiledStateGraph实例放到全局或应用状态里复用。然后定义请求和响应的 Pydantic 模型from pydantic import BaseModel class ChatRequest(BaseModel): thread_id: str message: str temperature: float 0.7 class ChatResponse(BaseModel): response: str thread_id: str接口层可以这样写from fastapi import FastAPI, HTTPException from graph.builder import get_compiled_graph app FastAPI(titleAgent Service) app.post(/chat, response_modelChatResponse) async def chat(req: ChatRequest): compiled_graph get_compiled_graph() config {configurable: {thread_id: req.thread_id}} try: result await compiled_graph.ainvoke( {messages: [{role: user, content: req.message}]}, configconfig ) return ChatResponse( responseresult[messages][-1].content, thread_idreq.thread_id ) except Exception as e: raise HTTPException(status_code500, detailstr(e))这里有个关键点用ainvoke而不是invoke。LangGraph 的ainvoke是异步版本FastAPI 的 async 端点里调用它才能避免阻塞事件循环。如果你在 async 函数里用了同步的invoke那整个服务在 Agent 思考的时间段内都会卡住并发一上来就崩。同理如果你的节点函数是同步的建议也别在 async 上下文里硬包装一层asyncio.to_thread除非你明确知道节点逻辑里有阻塞性 I/O。状态持久化是自建服务绕不开的问题。本地测试时可以用SqliteSaver但生产环境我更推荐PostgresSaver——因为数据库连接本身就是 Python 生态里最成熟的基础设施团队对它最熟悉也不引入新的存储组件from langgraph.checkpoint.postgres import PostgresSaver DB_URI postgresql://user:passhost:5432/agent_db with PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup() app graph.compile(checkpointercheckpointer)这里特别提醒PostgresSaver的setup()方法需要执行一次建表语句必须在首次部署时手动调用或者通过初始化逻辑确保执行。我当时第一次部署就是因为忘了调用setup()结果一接请求就报错排查了半天才发现是表不存在。工具调用的配置也值得单独说。我的 Agent 会用到大量业务工具FastAPI 服务启动时我建议把工具提前序列化并注册好而不是在每轮对话里动态生成。这是因为工具列表会塞进 LLM 的 prompt 里如果每次请求都重新构造工具描述token 消耗会成倍增加。正确的做法是在服务启动时构建工具列表编译进图里请求期间只传递用户消息和必要的 thread_id。3.3 流式输出与超时管理如果你的 Agent 服务要集成到聊天 UI 或类似的前端场景那么流式输出几乎是必须的。FastAPI 天然支持StreamingResponse配合 LangGraph 的流式接口可以做到打字机效果。LangGraph 提供了不同的流式模式。最简单的是在ainvoke里加上stream_modemessages但更细粒度的方式是直接用astream_events逐帧读取 LLM 输出。我在项目里是用astream遍历每个节点的输出然后通过队列传给前端from fastapi.responses import StreamingResponse import json app.post(/chat/stream) async def chat_stream(req: ChatRequest): compiled_graph get_compiled_graph() config {configurable: {thread_id: req.thread_id}} async def event_generator(): async for event in compiled_graph.astream_events( {messages: [{role: user, content: req.message}]}, configconfig, versionv2 ): if event[event] on_chat_model_stream: chunk event[data][chunk] if chunk.content: yield json.dumps({delta: chunk.content}, ensure_asciiFalse) \n return StreamingResponse(event_generator(), media_typeapplication/x-ndjson)这里有两个坑。第一前端解析流式响应时最好用ndjson这种每行一个 JSON 的格式而不是全部挤在一起否则解析会有各种兼容问题。第二流式接口必须设置合理的timeout——我见过不少人的 Agent 服务在流式模式下因为 HTTP 连接超时而断流这往往不是你的代码问题而是前端服务器的代理超时配置太短了。我的建议是Agent 场景下代理超时至少给到 120 秒以上。超时管理还涉及一个容易被忽略的层面单次请求内 LLM 调用的总时长。如果你的 Agent 图里有多个节点每个节点都可能调用一次 LLM那总耗时不是一次调用耗时而是串行节点耗时的累加。比如一个三节点的图每步 5 秒总共可能 15 秒甚至更多。所以在设计 API 超时时要把图的整体执行时间算进去不能只按单次 LLM 调用来估算。3.4 并发与性能调优经验自建服务一旦上线并发问题就来了。我在这方面的第一个教训是千万别让所有人都共享一个 thread_id。如果多人同时调用同一个 thread_id状态就会被互相覆盖对话上下文直接乱套。我在 API 层对thread_id做了强制性校验前端每次会话必须生成一个新的 UUID同一个会话内才复用。第二个教训是要控制并发时对 LLM API 的连接数。OpenAI 这类 API 都有速率限制你的服务并发一高后端就会开始报 429。解决思路是加一个信号量限流import asyncio llm_semaphore asyncio.Semaphore(10) async def limited_ainvoke(graph, state, config): async with llm_semaphore: return await graph.ainvoke(state, config)这样能保证同时最多 10 个请求去调用底层模型其他请求排队等待。实测下来这种方式在控制成本、减少 429 错误上效果明显。如果你用的是企业级模型网关还可以把限流粒度做细一点比如按模型分别设置信号量。还有一个经验是关于 FastAPI 的 worker 数量。很多人在部署 FastAPI 时直接用uvicorn main:app跑单进程这在开发环境没问题但生产环境单进程扛不住并发。我的做法是用gunicorn配合uvicorn workergunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app --bind 0.0.0.0:8000但注意多 worker 下thread_id的状态管理会跨进程失效。如果你用的是PostgresSaver这个问题天然不存在因为状态存在数据库里。这也是我为啥坚持生产环境用PostgresSaver的另一个原因——它不仅持久化还能在多 worker 间共享状态。3.5 接入 LangGraph Platform 的适配思路最后说说第三条路径的实操层面。如果你确定要对接 LangGraph Platform流程上其实不太复杂把图代码打包上传平台会自动构建运行环境然后给你一个 API 端点。调用方式可以用langgraph-sdkfrom langgraph_sdk import get_client client get_client(urlhttp://localhost:8123, api_keyyour_key) result client.invoke( agent_graph, input{messages: [{role: user, content: 你好}]}, config{configurable: {thread_id: test-1}} )但我要泼一盆冷水如果你们团队没有专门的平台运维能力直接上托管版会把很多业务逻辑和平台绑定得很深。比如你的工具调用里如果有关键业务函数这些函数需要能上传到平台运行——在企业内网环境里这个限制往往是致命的。自托管版倒是可以解决但你需要自己去维护一套平台的部署这和你自建一个 FastAPI 服务的人力成本其实是差不多的。所以我的结论是如果你已经有能力自建服务第三条路的价值主要体现在多快好省地跑 POC而不是生产环境的长期方案。4. 常见问题与排查技巧实录4.1 状态不持久化重启后对话上下文丢失这是自建服务最常遇见的坑。现象是服务正常运行但每次重启之后同一个thread_id的新请求就像第一次对话一样完全没有之前的上下文。排查思路先看 checkpointer 到底有没有生效。最简单的方式是检查数据库里有没有 checkpoint 表的数据。如果用的是SqliteSaver可以打开 sqlite 文件看checkpoints表如果用PostgresSaver查checkpoints表。如果表里有数据但对话依然丢失多半是请求时传入的config里的thread_id和保存时用的不一致。另一个隐蔽的问题图的结构变化会导致 checkpoint 反序列化失败。比如你给AgentState增加了一个字段旧的 checkpoint 里没有这个字段加载时就会报错或者丢弃旧状态。解决方法是给状态类字段设置默认值并对版本迁移做好兼容。4.2 工具调用报错工具不存在或参数不匹配LangGraph 的 Agent 节点在调用工具时如果工具函数本身报错调试起来很费劲。我遇到过的情况是工具函数明明写对了但 LLM 生成的工具参数不对比如把字符串传给了 int 参数导致工具内部TypeError。LangChain 的create_react_agent在工具调用上有一个容错设计——如果工具执行抛出异常它会默认捕获并把错误信息包装成一条消息返回给 LLM让 LLM 自行纠错。这在开发期是个好特性但在生产环境错误信息可能会暴露内部细节。我建议在工具层做一层统一封装的异常处理把错误信息做脱敏后再抛给模型。还有一点工具描述一定要写清楚。我在项目里反复调整过工具的描述文本发现描述含糊的后果是 LLM 根本不知道该在什么时候调用这个工具或者把参数填得乱七八糟。但是反之我也提醒别在描述里写太多限制条件否则 LLM 会过于保守而不去调用工具——这两个方向都试过描述上千篇一律只有当用户明确要求时才调用反而让 Agent 变得极其被动。4.3 服务启动闪退或无法启动如果你把 LangGraph 服务部署到新的环境遇到服务启动闪退的问题大概率是依赖冲突。LangChain 生态的依赖更新频率很高经常出现langchain-core和langgraph版本不匹配的情况。网上搜到无法将pnpm项识别claude 无法识别这类问题的场景虽然不同但本质都一样环境变量没配好或者可执行文件没有加入 PATH。对 LangGraph 来说最直接的排查方式是看启动日志确认是 import 阶段报错还是运行时才报错。import 阶段报错通常是依赖问题运行时报错通常是配置问题。还有一个小技巧在启动脚本里加上python -c from graph.builder import get_compiled_graph; print(ok)来快速验证能不能正常构建图比直接起服务快得多。4.4 长耗时任务导致请求超时如果你的 Agent 图特别复杂比如有十几步循环单次请求可能要跑几分钟这时候用同步 HTTP 请求会非常难受——客户端很可能等不到响应就超时断开了。我在项目里针对这个场景做了改造把长耗时任务拆成提交任务 异步轮询两步。提交任务时生成一个task_id后台用 Celery 或简单的asyncio.Task去跑图把结果写到 Redis前端通过轮询接口获取状态和最终结果。虽然实现上多了一些代码但用户体验好了很多也避开了 HTTP 超时问题。如果你不想引入 Celery 那么重的组件用 FastAPI 的BackgroundTasks配合 Redis 存储结果也是一种轻量替代方案。我自己在实测中还发现长任务的中途失败恢复是个很容易被忽略的坑。Agent 跑到第三步挂了前面的上下文就丢了。这时候PostgresSaver的价值再次体现——它可以基于已有的 checkpoint 继续跑而不是从头再来。配合thread_id做重试可以做到失败续跑的效果这对生产环境特别关键。4.5 排查问题时的黄金思路聊了这么多具体问题分享一套我实战中总结的排查思路。遇到 LangGraph 服务的问题我会按照这个顺序排查先看 LangSmith 追踪。如果你开了LANGSMITH_TRACING追踪面板会记录每一步的输入输出、耗时、token 消耗90% 的问题从这里一眼就能看出来。再看 checkpointer 配置。状态相关的问题重点检查thread_id是否一致、数据库连接是否正常。隔离工具层排查。如果怀疑是工具调用的问题把一个模拟输入直接喂给工具函数测试确认工具本身没问题再去看 LLM 生成的参数。最后查模型配置。比如 temperature 设置是否有问题、模型名称是否存在、API Key 是否有效。这套排查顺序看起来简单但在实际项目中真的帮我省了大量时间。很多问题不是出在最表面那层而是底层基础设施的配置不对直接查上层代码往往事倍功半。5. 我对 LangGraph 部署现状的一些体会最后说说我对这三条路径的整体感受。LangGraph 的出现确实让我写 Agent 的方式从堆 if-else变成了画状态图这种思维转变在复杂 Agent 场景里的收益非常明显。但框架只解决了编排的问题部署还需要自己踩坑。从脚本到服务不只是把app.invoke变成app.ainvoke的过程而是整个思维方式的转变脚本只需要考虑逻辑正确服务需要考虑并发、状态、超时、安全、可观测性。这三条路径没有绝对的优劣只有适不适合当前的阶段。项目处于早期验证阶段脚本直跑最合适项目要真正服务用户自建 FastAPI 服务大概率是最可控的选择如果你不想运维任何东西、且能接受平台绑定LangGraph Platform 值得一试。我自己目前的实践组合是开发调试阶段用脚本直跑 LangSmith 追踪生产环境用自建 FastAPI 服务 PostgresSaver gunicorn 多 worker在评估 LangGraph Platform 作为未来演进方向。这套组合经历了几轮迭代稳定性已经验证过了希望这篇文章里踩过的坑能帮你少走一些同样的弯路。
返回列表