生产最佳实践

把一个 demo 跑通不难，把它放到生产环境稳定服务却要趟过无数坑。本篇把社区踩过的坑总结成一份「生产清单」，覆盖状态设计、持久化、防死循环、超时熔断、并发隔离、错误降级、成本控制、安全、版本管理、性能优化十大主题。读完对照自查，能避掉 80% 的线上事故。

一、状态设计

state 是图的「血液」，设计好坏直接影响可维护性和序列化稳定性。

原则 1：可序列化优先 state 会被 checkpointer 落库、被 Studio 传输，必须 JSON 可序列化。只放 str/int/float/bool/list/dict，不放数据库连接、文件句柄、锁等。

原则 2：最小化 只存「下一步节点需要」的字段，不要把所有中间产物都塞 state。临时数据用节点局部变量。

原则 3：敏感信息不入 state API Key、用户密码、token 永远不进 state——一旦进 state 就会被 checkpointer 落库、被 LangSmith 上报。

python

# 反例
class BadState(TypedDict):
    api_key: str          # ❌ 不要存
    db_conn: Any          # ❌ 不可序列化
    full_history: list    # ❌ 冗余，messages 已有

# 正例
class GoodState(TypedDict):
    messages: Annotated[list, add_messages]  # 对话
    user_id: str                              # 业务标识
    retrieved_docs: list[dict]                # 检索片段（结构化）

二、持久化选型

Checkpointer	场景	特点
`MemorySaver`	本地开发、单测	进程内 dict，重启即丢
`SqliteSaver`	单机小流量、PoC	本地文件，无并发
`AsyncPostgresSaver`	生产	跨进程、高并发、可备份

python

# 生产标配
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from psycopg_pool import AsyncConnectionPool

pool = AsyncConnectionPool(conninfo=PG_DSN, min_size=5, max_size=20, open=False)
await pool.open()
saver = AsyncPostgresSaver(pool)
await saver.setup()  # 建表
graph = builder.compile(checkpointer=saver)

别用 SqliteSaver 上生产

SQLite 写锁是库级，多请求并发写会 database is locked。仅适合单机调试。

三、recursion_limit 与防死循环

LangGraph 默认 recursion_limit=25（图执行的最大「超级步」数）。条件边如果成环（A→B→A），不限制会无限跑。

python

# 显式设上限，按业务定
result = await graph.ainvoke(
    inputs,
    config={
        "configurable": {"thread_id": tid},
        "recursion_limit": 50,  # 防失控
    },
)

报 RecursionError: Recursion limit of 25 reached 时，先别急着调大，先排查是不是真的死循环。ReAct 智能体工具连续调用 5+ 次仍没结论，多半是 prompt 不清晰或工具描述有歧义，让模型在「调用工具—不满意—再调用」里打转。

防死循环的工程手段：

在 state 里加 iteration 计数，节点里判断超阈值就 goto END
工具调用前做去重：同一参数连续调 3 次直接拦截
用 recursion_limit 兜底

python

def chatbot(state: State):
    it = state.get("iteration", 0) + 1
    if it > 8:
        return {"messages": [HumanMessage("已达最大轮次，停止")]}  # 实际用 AIMessage
    # 正常逻辑...
    return {"messages": [resp], "iteration": it}

四、超时与熔断

LLM 调用慢且不可控，必须加超时和熔断。

python

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o-mini",
    timeout=30,        # 单次请求 30s 超时
    max_retries=2,     # 重试 2 次（指数退避）
)

# 图层超时：用 asyncio.wait_for 包一层
import asyncio
try:
    result = await asyncio.wait_for(
        graph.ainvoke(inputs, config=config),
        timeout=120,
    )
except asyncio.TimeoutError:
    logger.error("图执行超时", extra={"thread_id": tid})
    # 降级：返回兜底回复
    return {"reply": "服务繁忙，请稍后重试"}

熔断：连续 N 次失败后短时间直接拒绝请求，防止雪崩。可用 pybreaker 库或网关层（nginx/istio）实现。

五、并发与线程隔离

每个会话唯一 thread_id：同一用户多个会话各自独立
thread_id 服务端生成：从鉴权 token 解析用户身份后拼接，禁止前端直传
同 thread 串行写：同一 thread 高并发写入会导致 checkpointer 状态错乱，业务层加分布式锁

python

import redis.asyncio as redis

r = redis.from_url(REDIS_URL)
lock_key = f"thread:{tid}"
# 简化：用 setnx 抢锁
if not await r.set(lock_key, "1", nx=True, ex=30):
    raise HTTPException(429, "会话忙，请稍后")
try:
    result = await graph.ainvoke(inputs, config={"configurable": {"thread_id": tid}})
finally:
    await r.delete(lock_key)

六、错误处理与降级

智能体不可能 100% 成功，关键是失败得优雅。

python

def safe_chatbot(state: State):
    try:
        resp = llm.invoke(state["messages"])
        return {"messages": [resp]}
    except Exception as e:
        logger.exception("LLM 失败，降级")
        # 降级 1：换便宜模型重试
        try:
            resp = cheap_llm.invoke(state["messages"])
            return {"messages": [resp]}
        except Exception:
            # 降级 2：返回兜底
            from langchain_core.messages import AIMessage
            return {"messages": [AIMessage(content="服务暂时不可用，请稍后再试。")]}

工具调用失败也要捕获，否则一个工具崩了整个图崩：

python

@tool
def search_db(query: str) -> str:
    """查数据库"""
    try:
        return db.query(query)
    except Exception as e:
        return f"查询失败：{e}（请换个问法）"  # 返回字符串，不抛异常

七、成本控制

LLM 是按 token 计费，成本失控是生产头号杀手。

1. 模型分级

路由/分类、简单判断 → 小模型（gpt-4o-mini / 国产小模型）
复杂推理、工具调用 → 大模型（gpt-4o / Claude）
不要所有节点都用最贵模型

2. 缓存 重复 prompt 用 langchain_core.globals.InMemoryCache 或 Redis 缓存 LLM 响应：

python

from langchain_core.globals import set_llm_cache
from langchain_community.cache import RedisCache
import redis
set_llm_cache(RedisCache(redis_=redis.from_url(REDIS_URL)))

3. token 预算 在 state 里累加 token，超预算强制结束：

python

def chatbot(state: State):
    used = state.get("tokens_used", 0)
    if used > 10000:
        return {"messages": [AIMessage(content="本会话已达用量上限")]}
    resp = llm.invoke(state["messages"])
    return {"messages": [resp], "tokens_used": used + resp.usage_metadata["total_tokens"]}

4. 历史裁剪 长对话历史会爆 token，用 trim_messages 只保留最近 N 轮：

python

from langchain_core.messages import trim_messages
trimmer = trim_messages(max_tokens=2000, strategy="last")
state["messages"] = trimmer(state["messages"])

八、安全

维度	措施
API Key	用 `.env` + Vault/SSM，不入代码不入 state
工具白名单	按用户角色决定能用哪些工具，运行时动态 bind
输入校验	Pydantic 校验入参，防 prompt 注入（加系统 prompt 防护）
沙箱执行	执行用户代码的工具用 Docker/沙箱，限制文件/网络
速率限制	按用户/IP 限流，防滥用
审计日志	关键操作（工具调用、数据修改）落审计表

python

# 工具白名单示例
def get_tools_for_user(role: str):
    base = [search_tool]
    if role == "admin":
        return base + [delete_tool, write_tool]
    return base

llm = ChatOpenAI().bind_tools(get_tools_for_user(current_user.role))

九、版本管理与平滑升级

图结构变更：节点/边变了，旧 thread 的 state 可能不兼容，建议新版本用新 checkpointer 表或加 version 前缀
模型升级：换模型先灰度 10% 流量，对比效果再全量
API 版本：FastAPI 路由加 /v1/、/v2/ 前缀，给客户端迁移时间
配置即代码：图定义、prompt 模板都进 git，CI/CD 部署

十、性能优化

异步全链路：用 ainvoke/astream 而非 invoke，避免阻塞 worker
批处理：多用户请求合并，用 asyncio.gather 并发调 LLM
预编译图：应用启动时编译一次，全局复用，不要每请求 compile
连接池复用：DB、Redis、HTTP 客户端都用连接池
流式优先：长响应用 SSE 流式，提升首字时间

python

# 批处理示例
async def batch_invoke(messages_list: list[str]):
    tasks = [graph.ainvoke({"messages": [HumanMessage(m)]},
                           config={"configurable": {"thread_id": str(i)}})
             for i, m in enumerate(messages_list)]
    return await asyncio.gather(*tasks, return_exceptions=True)

十一、生产 Checklist

类别	检查项	✅
状态	state 可序列化、无敏感信息、最小化	☐
持久化	用 PostgresSaver，连接池配好	☐
防死循环	设 recursion_limit，有兜底退出	☐
超时	LLM timeout + 图整体 timeout	☐
熔断	连续失败能熔断降级	☐
并发	thread_id 服务端生成，同 thread 加锁	☐
错误	节点 try/except，工具不抛异常	☐
成本	模型分级、缓存、token 预算、历史裁剪	☐
安全	Key 管理、工具白名单、输入校验、限流	☐
监控	LangSmith 追踪 + 结构化日志 + Prometheus	☐
健康	/health /ready 端点	☐
部署	Dockerfile 多阶段、多 worker、PgBouncer	☐
版本	API 版本前缀、灰度发布	☐

十二、与 LangGraph4j 生产经验互通点

Java 同学的生产经验大多可迁移：

状态序列化 ≈ Java 的 Serializable/Record，原则一致：只放数据
Checkpointer ≈ 自实现 JDBC 持久化，PostgresSaver 已替你做好
超时熔断 ≈ Resilience4j/Hystrix，Python 用 asyncio.wait_for + pybreaker
限流 ≈ Sentinel/Guava RateLimiter，Python 用 slowapi 或网关层
可观测 ≈ SkyWalking/Pinpoint，LangSmith + OTel 是等价物
沙箱 ≈ Java 的 SecurityManager（已废弃）/ Docker 隔离，Python 同样用 Docker

需要重新理解的：Python 的 GIL 让「多 worker」倾向多进程而非多线程；async 协程比 Java 线程更轻，但要注意别在 async 里调同步阻塞函数（会卡住事件循环）。

十三、小结

生产化是把「能跑」变成「稳定跑」的工程过程，核心就三件事：

限制失控：recursion_limit、timeout、熔断、token 预算
隔离故障：thread_id 隔离会话、try/except 隔离错误、降级保可用
看得见：追踪、日志、指标、健康检查

对照上面的 Checklist 逐项落实，你的 LangGraph 应用就具备了上线条件。下一模块进入附录：FAQ、API 速查、与 LangGraph4j 对比、学习路线。

生产最佳实践 ​

一、状态设计 ​

二、持久化选型 ​

三、recursion_limit 与防死循环 ​

四、超时与熔断 ​

五、并发与线程隔离 ​

六、错误处理与降级 ​

七、成本控制 ​

八、安全 ​

九、版本管理与平滑升级 ​

十、性能优化 ​

十一、生产 Checklist ​

十二、与 LangGraph4j 生产经验互通点 ​

十三、小结 ​

生产最佳实践

一、状态设计

二、持久化选型

三、recursion_limit 与防死循环

四、超时与熔断

五、并发与线程隔离

六、错误处理与降级

七、成本控制

八、安全

九、版本管理与平滑升级

十、性能优化

十一、生产 Checklist

十二、与 LangGraph4j 生产经验互通点

十三、小结