服务化部署

把 LangGraph 图写成 Python 脚本只能在本地跑，要给业务方用，得变成一个 HTTP 服务。本篇对比三种主流部署方式，并重点手把手教你用 FastAPI 自建一个生产可用的 LangGraph 服务——这种方式对栈的掌控最完整，也最容易理解官方 Server 内部在做什么。

一、三种部署方式对比

方式	优点	缺点	适用
① LangGraph Platform（Cloud/Self-hosted）	官方维护、内置持久化/流式/Studio/多租户	Self-hosted 需 Docker+Postgres；Cloud 受地域限制	想省事、要 Studio
② 自建 FastAPI 包装	完全掌控、轻量、易嵌入现有系统	持久化/流式/监控得自己接	已有 Python 后端、私有化
③ Serverless（Vercel/Lambda/Cloud Run）	按量付费、自动伸缩	长连接/流式受限、冷启动	流量小、低延迟场景

选型建议

中小项目、已有 FastAPI 后端 → 选 ②
重度依赖 Studio 调试、多租户 → 选 ①
偶发型 webhook 触发 → 选 ③，但流式输出会受限

本篇重点讲 ②。① 的本地体验见 LangGraph 平台与 Studio。

二、自建 FastAPI：整体架构

mermaid

flowchart TB
    Client[客户端 curl/前端] -->|HTTP| API[FastAPI]
    API -->|invoke / stream| Graph[编译好的图]
    Graph --> LLM[ChatOpenAI]
    Graph --> Tools[工具]
    Graph -->|checkpointer| PG[(Postgres)]
    API -->|结构化日志| Log[文件/Loki]
    API -->|trace| Smith[LangSmith]

要点：

图在应用启动时编译一次，全局复用（编译开销大）
checkpointer 用 PostgresSaver，跨进程、跨重启保留线程状态
thread_id 由客户端传入，决定会话隔离
异步端点调用图的 ainvoke / astream，不阻塞 worker

三、最小可运行 FastAPI 代码

1. 目录结构

text

fastapi_demo/
├── app.py              # FastAPI 主文件
├── graph.py            # 图定义
├── requirements.txt
├── Dockerfile
└── .env

2. graph.py

python

from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

@tool
def add(a: float, b: float) -> float:
    """两数相加"""
    return a + b

tools = [add]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)

class State(TypedDict):
    messages: Annotated[list, add_messages]

def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

builder = StateGraph(State)
builder.add_node("chatbot", chatbot)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "chatbot")
builder.add_conditional_edges("chatbot", tools_condition, "tools")
builder.add_edge("tools", "chatbot")
builder.add_edge("chatbot", END)

3. app.py（核心）

python

import os
import uuid
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.graph.state import CompiledStateGraph
from psycopg_pool import AsyncConnectionPool

import graph as graph_mod

# 编译图（不带 checkpointer，启动后再替换）
builder = graph_mod.builder

# 请求体模型
class InvokeRequest(BaseModel):
    message: str
    thread_id: str | None = None  # 不传则新建线程

# 全局持有编译好的图与连接池
compiled: CompiledStateGraph
pool: AsyncConnectionPool

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期：启动建池、关闭销毁"""
    global compiled, pool
    dsn = os.getenv("PG_DSN", "postgresql://postgres:postgres@localhost:5432/langgraph")
    # 异步连接池，生产建议 min_size=5 max_size=20
    pool = AsyncConnectionPool(conninfo=dsn, min_size=2, max_size=10, open=False)
    await pool.open()
    # Postgres 检查点保存器
    saver = AsyncPostgresSaver(pool)
    await saver.setup()  # 首次启动建表
    compiled = builder.compile(checkpointer=saver)
    yield
    await pool.close()

app = FastAPI(title="LangGraph Service", lifespan=lifespan)

@app.post("/invoke")
async def invoke(req: InvokeRequest):
    """同步调用：等图跑完一次性返回"""
    thread_id = req.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}
    # 用 HumanMessage 包装用户输入
    from langchain_core.messages import HumanMessage
    result = await compiled.ainvoke(
        {"messages": [HumanMessage(content=req.message)]},
        config=config,
    )
    # 取最后一条 AI 消息返回
    last = result["messages"][-1]
    return {"thread_id": thread_id, "reply": last.content}

@app.get("/history/{thread_id}")
async def history(thread_id: str):
    """按线程取历史消息"""
    config = {"configurable": {"thread_id": thread_id}}
    snapshot = await compiled.aget_state(config)
    messages = snapshot.values.get("messages", []) if snapshot else []
    # 序列化：取 content 与 type
    return {"messages": [{"type": m.type, "content": m.content} for m in messages]}

@app.get("/health")
async def health():
    return {"status": "ok"}

4. requirements.txt

text

fastapi>=0.110
uvicorn[standard]>=0.27
langgraph>=0.2
langgraph-checkpoint-postgres>=2.0
langchain-openai>=0.1
langchain-core>=0.2
psycopg[binary,pool]>=3.1
python-dotenv>=1.0

5. 启动

bash

# 先装依赖（国内源）
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

# 确保 Postgres 跑起来（docker 一行起）
docker run -d --name pg -p 5432:5432 -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=langgraph postgres:16

# 启服务
# Windows / macOS / Linux 通用
uvicorn app:app --host 0.0.0.0 --port 8000 --reload

Windows 注意

psycopg 在 Windows 上装 psycopg[binary] 即可，无需装 PostgreSQL 客户端库。若报 DLL load failed，确认 Python 是 64 位。

四、用 curl / requests 调用

bash

# 第一次对话（不传 thread_id，服务端生成）
curl -X POST http://localhost:8000/invoke \
  -H "Content-Type: application/json" \
  -d '{"message": "把 3 和 5 加起来"}'

# 返回示例：{"thread_id":"abc-123","reply":"3 + 5 = 8"}

python

# Python 客户端
import requests
# 第一轮
r = requests.post("http://localhost:8000/invoke", json={"message": "把 3 和 5 加起来"})
data = r.json()
print(data["reply"])           # 3 + 5 = 8
tid = data["thread_id"]
# 第二轮，复用 thread_id 让模型记住上文
r2 = requests.post("http://localhost:8000/invoke", json={"message": "再乘以 2", "thread_id": tid})
print(r2.json()["reply"])      # 8 * 2 = 16

五、线程管理（thread_id）

thread_id 是 LangGraph 会话隔离的核心：

同一 thread_id = 同一会话，共享 state 历史（消息记忆）
不同 thread_id = 完全独立会话，互不影响
客户端通常用「用户 ID + 会话 ID」组合生成，例如 f"{user_id}-{session_id}"

mermaid

flowchart LR
    U1[用户A] -->|thread_id=A-1| S[FastAPI]
    U1 -->|thread_id=A-2| S
    U2[用户B] -->|thread_id=B-1| S
    S --> PG[(Postgres)]
    PG --> T1[线程A-1的state]
    PG --> T2[线程A-2的state]
    PG --> T3[线程B-1的state]

安全

不要直接用前端传来的 user_id 当 thread_id，否则用户能伪造别人会话。务必在服务端从鉴权 token 解析用户身份后再拼装。

六、Dockerfile 示例

dockerfile

# 构建阶段
FROM python:3.11-slim AS builder
WORKDIR /app
# 国内 pip 源
RUN pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

# 运行阶段
FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY . .
ENV PATH=/root/.local/bin:$PATH
ENV PYTHONUNBUFFERED=1
EXPOSE 8000
# 生产用多 worker，注意：每 worker 各自连池，按机器规格调
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

构建运行：

bash

docker build -t langgraph-svc .
docker run -d -p 8000:8000 --env-file .env --name svc langgraph-svc

七、与 LangGraph4j / Spring Boot 部署对比

维度	LangGraph + FastAPI	LangGraph4j + Spring Boot
Web 框架	FastAPI（异步 ASGI）	Spring WebFlux/MVC
包管理	pip + venv/uv	Maven/Gradle
持久化	PostgresSaver（psycopg）	自实现 JDBC Checkpointer
部署单元	uvicorn 进程 / Docker	jar / Docker
多 worker	`--workers N`（多进程）	内置线程池/反应式
配置	`.env` + `os.getenv`	`application.yml`

Java 同学迁移要点：

FastAPI 的 lifespan ≈ Spring 的 @PostConstruct/@PreDestroy
全局 compiled 图对象 ≈ Spring @Bean 单例
async/await ≈ WebFlux 的 Mono/Flux，但 Python 协程更轻

八、常见踩坑

1. 并发时 state 串了 原因：多个请求用了同一个 thread_id，Postgres 同一行的 checkpointer 会互相覆盖。解决：每个会话严格用唯一 thread_id；同一 thread 高并发写入要加业务锁。

2. 状态隔离不彻底 图里有模块级可变全局变量（如缓存 dict），多 worker 各有一份不共享。要么挪进 state，要么用 Redis 等外部存储。

3. 序列化错误 state 里塞了 ChatOpenAI 实例、httpx.Client 等不可序列化对象，PostgresSaver 落库报错。原则：state 只放数据，不放行为对象。

4. 请求超时 LLM 调用慢导致 uvicorn 默认 keep-alive 超时。加 --timeout-keep-alive 120，并在 ChatOpenAI 上设 timeout=60、max_retries=2。

5. 多 worker 下 checkpointer 连接数爆炸 4 worker × 10 连接 = 40 连接，Postgres 默认 100 连接很快撑爆。生产建议加 PgBouncer 做连接池前置。

九、小结

三种部署里，自建 FastAPI 最灵活：图编译一次全局复用，ainvoke 暴露成 HTTP
生产持久化用 AsyncPostgresSaver，thread_id 做会话隔离
Dockerfile 分阶段构建，多 worker 部署注意连接数
Java 同学可把 FastAPI 类比为 Spring Boot 的轻量替代，概念能直接迁移

下一篇我们解决「流式输出怎么对接前端」——把 astream 包成 SSE。

服务化部署 ​

一、三种部署方式对比 ​

二、自建 FastAPI：整体架构 ​

三、最小可运行 FastAPI 代码 ​

1. 目录结构 ​

2. graph.py ​

3. app.py（核心） ​

4. requirements.txt ​

5. 启动 ​

四、用 curl / requests 调用 ​

五、线程管理（thread_id） ​

六、Dockerfile 示例 ​

七、与 LangGraph4j / Spring Boot 部署对比 ​

八、常见踩坑 ​

九、小结 ​

服务化部署

一、三种部署方式对比

二、自建 FastAPI：整体架构

三、最小可运行 FastAPI 代码

1. 目录结构

2. graph.py

3. app.py（核心）

4. requirements.txt

5. 启动

四、用 curl / requests 调用

五、线程管理（thread_id）

六、Dockerfile 示例

七、与 LangGraph4j / Spring Boot 部署对比

八、常见踩坑

九、小结