前面4篇讲了LangGraph怎么用，这篇讲怎么从本地开发推到生产部署。本地跑通了不代表上线能用——Checkpointer从MemorySaver换成PostgreSQL数据丢了、FastAPI没配thread_id并发全乱、Docker里SQLite单写锁10个用户排队3秒、LangSmith采样率没配API费翻了5倍。

和前4篇LangGraph系列的差异化

主题 角度
为什么用 Agent不够用+4踩坑
5大API Send/Command/Interrupt/Subgraph/Streaming
4大能力 Memory+Tool+HITL+Persistence
多Agent Supervisor调度
部署上线 从开发→生产4踩坑

前4篇讲"怎么写代码"，这篇讲"代码写好了怎么上线"。

坑1：Checkpointer从MemorySaver换到PostgreSQL——状态数据格式不兼容

翻车现场

# 开发环境：MemorySaver，一切正常
from langgraph.checkpoint.memory import MemorySaver
app = graph.compile(checkpointer=MemorySaver())

# 本地跑通了！100次测试没问题！

# 上线时换成PostgreSQL——直接换配置就行？
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://...")
app = graph.compile(checkpointer=checkpointer)

# ❌ 报错：State的字段类型不匹配！
# MemorySaver存的checkpoint是Python pickle格式
# PostgreSQL存的是JSON格式
# pickle和JSON对TypedDict的序列化方式不同：
# - pickle保留Python类型（list、dict、bool）
# - JSON全转成字符串（"True" 而不是 True）

# 结果：review_passed字段从bool变成了str "True"
# 条件边 if state["review_passed"] 判断str而不是bool
# "True" 作为str永远不等于True → 条件边永远走错分支！

# 就像Java里从内存缓存换成Redis，JSON反序列化把Boolean.TRUE变成了"true"字符串

修复：Pydantic State + 数据迁移

# ✅ 修复方式1：用Pydantic BaseModel定义State（强类型，序列化一致）
from pydantic import BaseModel, Field
from typing import Annotated

class RAGState(BaseModel):
    """✅ Pydantic强类型——JSON序列化后类型不变"""
    messages: Annotated[list, add_messages] = Field(default_factory=list)
    question: str = ""
    context: str = ""
    answer: str = ""
    review_passed: bool = False   # ✅ Pydantic保证JSON里也是bool
    retry_count: int = 0

# ✅ 修复方式2：PostgreSQL连接池配置（生产必备）
from psycopg_pool import ConnectionPool

pool = ConnectionPool(
    "postgresql://langgraph:password@localhost:5432/lg_db",
    min_size=5,    # ✅ 最小连接数（Java: minIdle=5）
    max_size=20,   # ✅ 最大连接数（Java: maxActive=20）
    open=True
)
checkpointer = PostgresSaver(pool)

# ✅ 修复方式3：数据迁移脚本
# 如果之前用MemorySaver或SQLite存了数据，迁移到PG时要类型转换
import json

def migrate_sqlite_to_postgres(sqlite_path: str, pg_pool):
    """从SQLite迁移到PostgreSQL，修复类型问题"""
    # 读取SQLite中的checkpoint
    from langgraph.checkpoint.sqlite import SqliteSaver
    sqlite_saver = SqliteSaver.from_conn_string(sqlite_path)

    # ✅ 遍历所有checkpoint，修复类型后写入PG
    with pg_pool.connection() as conn:
        for checkpoint in sqlite_saver.list(config={"configurable": {"thread_id": "*"}}):
            # ✅ Pydantic自动修复类型（str "True" → bool True）
            fixed_state = RAGState(**checkpoint.data).model_dump()
            # 写入PG
            conn.execute("INSERT INTO checkpoints ...", (json.dumps(fixed_state),))

# ✅ TypedDict vs Pydantic对比（Java: Map vs DTO）
方式 序列化 类型安全 JSON兼容 Java对照 推荐
TypedDict pickle（Python专有） ❌ 运行时才检查 ❌ 跨后端不一致 HashMap 不推荐
Pydantic BaseModel JSON（通用） ✅ 定义时校验 ✅ 跨后端一致 @Data DTO ✅ 推荐

坑2：FastAPI没配thread_id——10个并发用户状态全乱了

翻车现场

from fastapi import FastAPI
from pydantic import BaseModel

app_api = FastAPI()

class QueryRequest(BaseModel):
    query: str

@app_api.post("/agent/query")
async def agent_query(request: QueryRequest):
    # ❌ 每次请求都不传thread_id！
    # LangGraph默认用None作为thread_id
    # 10个用户同时请求，全写到同一个State里
    # 用户A的对话出现在用户B的回复中！
    result = await app.ainvoke({"question": request.query})
    return {"answer": result["answer"]}

# 更惨的：Checkpointer用的是PostgreSQL
# 10个用户共享一个thread → 并发写入 → 数据互相覆盖
# 就像Java里10个线程写同一个HttpSession → 数据错乱

修复：每个请求绑定userId + thread_id

from fastapi import FastAPI, Header, Depends
from pydantic import BaseModel
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

app_api = FastAPI(title="LangGraph RAG API", version="1.0")

class QueryRequest(BaseModel):
    query: str
    user_id: str = "anonymous"  # ✅ 必须传userId

# ✅ 连接池（生产级）
pool = ConnectionPool(
    "postgresql://langgraph:password@localhost:5432/lg_db",
    min_size=5, max_size=20, open=True
)
checkpointer = PostgresSaver(pool)

# ✅ 编译时注入checkpointer
compiled_app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]  # ✅ 审核前可中断
)

@app_api.post("/agent/query")
async def agent_query(request: QueryRequest):
    """✅ 每个请求绑定thread_id（Java: sessionId）"""
    # ✅ thread_id = userId-sessionId，确保每个用户独立
    config = {
        "configurable": {
            "thread_id": f"{request.user_id}-{uuid4()}"  # ✅ 每个请求独立
        }
    }

    result = await compiled_app.ainvoke(
        {"question": request.query, "user_id": request.user_id},
        config=config  # ✅ 绑定thread_id
    )
    return {
        "answer": result.get("answer"),
        "thread_id": config["configurable"]["thread_id"]  # ✅ 返回给前端
    }

# ✅ 流式API（前端逐token展示）
from fastapi.responses import StreamingResponse

@app_api.post("/agent/stream")
async def agent_stream(request: QueryRequest):
    """✅ 流式输出——前端SSE逐token接收"""
    config = {
        "configurable": {
            "thread_id": f"{request.user_id}-stream"
        }
    }

    async def event_generator():
        for event in compiled_app.astream(
            {"question": request.query, "user_id": request.user_id},
            config=config,
            stream_mode="custom"  # ✅ 只取自定义Event
        ):
            if event:
                yield f"data: {json.dumps({'content': event})}\n\n"
        yield f"data: {json.dumps({'done': True})}\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

# ✅ 断点恢复API（审核通过后继续）
@app_api.post("/agent/resume")
async def agent_resume(thread_id: str, decision: str = "approve"):
    """✅ 从断点恢复——人工审核后继续执行"""
    config = {"configurable": {"thread_id": thread_id}}

    # 更新人工决策
    compiled_app.update_state(
        config,
        {"human_decision": decision},
        as_node="review"
    )

    # 从断点继续
    result = await compiled_app.ainvoke(None, config=config)
    return {"answer": result.get("answer")}

# ✅ API对照表（Java: Spring Boot RESTful）
LangGraph API Spring Boot对照 说明
/agent/query POST /api/chat 同步问答
/agent/stream GET /api/chat/stream (SSE) 流式问答
/agent/resume POST /api/chat/resume 断点恢复
thread_id sessionId 用户状态隔离
ainvoke async Service调用 异步执行

坑3：Docker里SQLite单写锁——10个并发用户排队3秒

翻车现场

# ❌ Dockerfile里用SQLite作为Checkpointer
# SQLite是单写锁——同时只能有1个写入操作
# 10个用户同时写入 → 9个等待 → 平均等待3秒

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

# ❌ checkpoints.db在容器内——容器重启数据全丢
# ❌ SQLite文件锁在Docker overlay文件系统上更慢
# ❌ 没有health check——容器挂了不知道

修复：PostgreSQL + Volume持久化 + Health Check

# ✅ 生产级Dockerfile——PostgreSQL + Health Check
FROM python:3.11-slim

WORKDIR /app

# ✅ 先安装依赖（利用Docker缓存层）
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# ✅ 后复制代码（代码改动不影响依赖缓存）
COPY . .

# ✅ Health Check（Java: Spring Boot Actuator /health）
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD curl -f http://localhost:8000/docs || exit 1

EXPOSE 8000

# ✅ 生产启动（2 worker + timeout）
CMD ["uvicorn", "main:app_api", \
     "--host", "0.0.0.0", \
     "--port", "8000", \
     "--workers", "2", \
     "--timeout-keep-alive", "30"]

# ✅ docker-compose.yml——PostgreSQL + LangGraph + 持久化
version: "3.8"
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: langgraph
      POSTGRES_USER: langgraph
      POSTGRES_PASSWORD: ${PG_PASSWORD}
    volumes:
      - pg_data:/var/lib/postgresql/data  # ✅ 数据持久化
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U langgraph"]
      interval: 10s
      timeout: 5s
      retries: 5

  langgraph-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      PG_CONN: "postgresql://langgraph:${PG_PASSWORD}@postgres:5432/langgraph"
      OPENAI_API_KEY: ${OPENAI_API_KEY}
      LANGSMITH_API_KEY: ${LANGSMITH_API_KEY}
    depends_on:
      postgres:
        condition: service_healthy  # ✅ 等PG就绪再启动
    volumes:
      - user_memory:/app/user_memory  # ✅ Chroma向量库持久化

volumes:
  pg_data:       # PostgreSQL数据持久化
  user_memory:   # 用户长期记忆持久化

SQLite vs PostgreSQL Docker对比：

维度 SQLite Docker PostgreSQL Docker Java对照
并发写入 ❌ 单写锁排队3秒 ✅ 多连接并行 单线程DAO vs 连接池
数据持久化 ❌ 容器重启丢 ✅ Volume持久 内存缓存 vs Redis持久
Health Check ❌ 无 ✅ pg_isready 无 vs Actuator
连接数 ❌ 1 ✅ 5-20 pool HikariCP配置
月成本 0元 ~200元 0 vs 小型RDS

推荐：10个以上用户必须用PostgreSQL，5个以下可以用SQLite开发。

坑4：LangSmith采样率没配——API费翻了5倍

翻车现场

# ❌ LangSmith默认全量Trace——每个请求都记录
os.environ["LANGSMITH_API_KEY"] = "lsv2_pt_xxx"
os.environ["LANGSMITH_PROJECT"] = "my-rag-app"

# 100个用户每天10次对话 = 1000次Trace
# 每次Trace记录：LLM调用、工具调用、状态变化、checkpoint
# LangSmith免费额度5000条/月 → 3天就用完了
# 超额后按$5/1000条收费 → 一个月$50 → 约350元

# 更惨的：开发调试时每次都记录，测试跑了100次 = 100条Trace
# 代码里print调试也记录 → Token费翻了5倍

# 就像Java里SkyWalking全量采样——100%请求都记录链路
# 生产环境应该采样10%，不是100%

修复：分环境采样率 + 关键请求全量

import os

# ✅ 分环境配置采样率（Java: SkyWalking采样率配置）
ENVIRONMENT = os.getenv("ENV", "development")

if ENVIRONMENT == "development":
    # 开发环境：全量记录（方便调试）
    os.environ["LANGSMITH_TRACING"] = "true"   # ✅ 开发全量
    SAMPLE_RATE = 1.0  # 100%

elif ENVIRONMENT == "staging":
    # 预发布：50%采样
    os.environ["LANGSMITH_TRACING"] = "true"
    SAMPLE_RATE = 0.5  # 50%

elif ENVIRONMENT == "production":
    # ✅ 生产环境：10%采样（Java: SkyWalking 10%采样）
    os.environ["LANGSMITH_TRACING"] = "true"
    SAMPLE_RATE = 0.1  # 10%

# ✅ 关键请求强制全量（Java: 异常请求全量Trace）
async def smart_invoke(input_data, config):
    """智能采样：普通请求按比例，关键请求全量"""
    question = input_data.get("question", "").lower()

    # ✅ 涉及敏感操作的请求 → 强制全量记录
    is_sensitive = any(w in question for w in ["删除", "退款", "转账", "取消"])

    # ✅ 异常请求 → 强制全量
    is_retry = input_data.get("retry_count", 0) > 0

    # ✅ 随机采样（普通请求）
    should_trace = is_sensitive or is_retry or (random.random() < SAMPLE_RATE)

    if not should_trace:
        # ✅ 不记录Trace——省Token费
        os.environ["LANGSMITH_TRACING"] = "false"

    result = await compiled_app.ainvoke(input_data, config)

    # 恢复默认
    os.environ["LANGSMITH_TRACING"] = "true"
    return result

LangSmith采样率配置对比：

环境 采样率 月Trace条数 月成本 Java对照 适用场景
开发 100% ~3000 0元（免费内） Debug全量 调试
预发布 50% ~15000 ~$75 Staging采样 验证
生产 10% ~3000 0元 SkyWalking10% ✅ 推荐
异常请求 100% ~200 0元 错误全量 排查

成本对比：

  • ❌ 全量100%：1000用户×10次×100% = 10万条 → $500/月（3500元）
  • ✅ 10%采样：10万条×10% = 1万条 → $50/月（350元）
  • ✅ 10%采样+敏感全量：1万+200 = 1.02万条 → ~$50/月

省钱10倍，关键请求不丢！

完整实战：生产级LangGraph部署配置

把4个坑的修复全部串起来：

# ==================== config.py（生产配置） ====================
import os
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver

# ✅ 坑1修复：环境区分
ENVIRONMENT = os.getenv("ENV", "development")

# ✅ 坑4修复：LangSmith采样率
if ENVIRONMENT == "development":
    os.environ["LANGSMITH_TRACING"] = "true"
    SAMPLE_RATE = 1.0
elif ENVIRONMENT == "staging":
    os.environ["LANGSMITH_TRACING"] = "true"
    SAMPLE_RATE = 0.5
else:  # production
    os.environ["LANGSMITH_TRACING"] = "true"
    SAMPLE_RATE = 0.1  # ✅ 生产10%采样

# ✅ 坑3修复：PostgreSQL连接池（不用SQLite）
pool = ConnectionPool(
    os.getenv("PG_CONN", "postgresql://langgraph:pass@localhost:5432/lg_db"),
    min_size=5,
    max_size=20,
    open=True
)
checkpointer = PostgresSaver(pool)

# ==================== state.py（✅ 坑1修复：Pydantic） ====================
from pydantic import BaseModel, Field
from typing import Annotated
from langgraph.graph.message import add_messages

class RAGState(BaseModel):
    """✅ Pydantic强类型——JSON序列化跨后端一致"""
    messages: Annotated[list, add_messages] = Field(default_factory=list)
    question: str = ""
    answer: str = ""
    review_passed: bool = False    # ✅ bool不会变成str
    retry_count: int = 0           # ✅ int不会变成str
    user_id: str = ""
    human_decision: str = ""

# ==================== main.py（✅ 坑2修复：thread_id） ====================
from fastapi import FastAPI
from pydantic import BaseModel
import uuid
import random

app_api = FastAPI(title="LangGraph RAG API", version="1.0")
compiled_app = graph.compile(
    checkpointer=checkpointer,  # ✅ PG持久化
    interrupt_before=["review"]  # ✅ 审核前中断
)

class QueryRequest(BaseModel):
    query: str
    user_id: str = "anonymous"

@app_api.post("/agent/query")
async def agent_query(request: QueryRequest):
    """✅ 每个请求独立thread_id"""
    thread_id = f"{request.user_id}-{uuid.uuid4()}"
    config = {"configurable": {"thread_id": thread_id}}

    # ✅ 坑4修复：智能采样
    is_sensitive = any(w in request.query.lower() for w in ["删除", "退款", "转账"])
    should_trace = is_sensitive or (random.random() < SAMPLE_RATE)

    if not should_trace:
        os.environ["LANGSMITH_TRACING"] = "false"

    result = await compiled_app.ainvoke(
        {"question": request.query, "user_id": request.user_id},
        config=config
    )

    os.environ["LANGSMITH_TRACING"] = "true"
    return {
        "answer": result.get("answer", ""),
        "thread_id": thread_id  # ✅ 返回给前端保存
    }

@app_api.post("/agent/stream")
async def agent_stream(request: QueryRequest):
    """✅ 流式SSE输出"""
    config = {"configurable": {"thread_id": f"{request.user_id}-stream-{uuid.uuid4()}"}}

    async def event_generator():
        for event in compiled_app.astream(
            {"question": request.query, "user_id": request.user_id},
            config=config,
            stream_mode="custom"
        ):
            if event:
                yield f"data: {json.dumps({'content': event})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

@app_api.post("/agent/resume")
async def agent_resume(thread_id: str, decision: str = "approve"):
    """✅ 断点恢复"""
    config = {"configurable": {"thread_id": thread_id}}
    compiled_app.update_state(config, {"human_decision": decision}, as_node="review")
    result = await compiled_app.ainvoke(None, config=config)
    return {"answer": result.get("answer", "")}

@app_api.get("/health")
async def health():
    """✅ Health Check（Docker/监控用）"""
    return {"status": "healthy", "env": ENVIRONMENT}

LangGraph Server vs 自建FastAPI选择

原版文章只提了FastAPI部署，没提LangGraph自己的Server：

维度 LangGraph Server (langgraph dev) 自建FastAPI Java对照
启动速度 1条命令 需要写FastAPI代码 Spring Boot Starter vs 手写
流式API ✅ 原生支持 需要手写SSE 内置WebSocket vs 手写
Studio调试 ✅ 原生集成 ❌ 需手动配置 Actuator vs 手写监控
自定义API ❌ 只支持标准接口 ✅ 任意定制 标准REST vs 自定义
扩展性 ❌ 受限于SDK ✅ 完全自由 脚手架 vs 自建
适合场景 开发调试+简单上线 生产级定制 开发 vs 生产

推荐路径：

  • 开发阶段 → langgraph dev（一键启动+Studio调试）
  • 上线阶段 → 自建FastAPI（自定义API+采样率+Health Check）

4坑速查表

# 坑 症状 修复 Java对照
1 Checkpointer换后端类型不一致 bool变成str"True"→条件边永远错 Pydantic BaseModel + 数据迁移 HashMap→Redis类型转换
2 FastAPI没配thread_id 10用户共享State→数据错乱 userId-sessionId作为thread_id sessionId隔离
3 Docker里SQLite单写锁 10并发排队3秒 PostgreSQL+连接池+Volume+HealthCheck SQLite→MySQL+HikariCP
4 LangSmith全量采样 月3500元API费 分环境采样(开发100%/生产10%)+敏感全量 SkyWalking采样率

5篇从入门→API→能力→多Agent→部署，覆盖LangGraph从写代码到上线的完整路线。

有问题欢迎评论区讨论，特别是PostgreSQL连接池和LangSmith采样率，这两个直接影响性能和成本 👇

# python # 人工智能
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

AtomGit 2026年6月三方库下载量排行榜重磅发布！双榜格局稳固，生态深耕持续发力

avatar AtomGit开源社区
cover

AtomGit 6 月：AtomCode CodingPlan 上线 GLM-5.2！

avatar AtomGit开源社区
cover

行业 Agent 黑客松圆满收官：打通行业落地“最后一公里”

avatar AtomGit开源社区