【langgraph 从入门到精通graphApi 篇】生产部署与性能优化

第 12 章：生产部署与性能优化

12.1 本章目标

学完本章你将能够：

1. 了解 LangGraph Platform 和 LangGraph Server 的部署方式
2. 掌握生产环境的 Checkpointer 和 Store 配置
3. 学会错误处理、重试和并发控制
4. 了解监控、日志和成本优化策略

12.2 核心概念

生产部署架构

开发 vs 生产

12.3 实战

实战 1：langgraph.json 项目配置

{"graphs":{"agent":"./agent.py:graph"},"store":{"index":{"embed":"openai:text-embeddings-3-small","dims":1536,"fields":["$"]}},"env":{"OPENAI_API_KEY":"your-api-key"}}

实战 2：Postgres 持久化配置

importosfromlanggraph.checkpoint.postgresimportPostgresSaverfromlanggraph.store.postgresimportPostgresStoreimportpsycopg2# 连接数据库conn=psycopg2.connect(os.environ["DATABASE_URL"])# 配置 Checkpointercheckpointer=PostgresSaver(conn)checkpointer.setup()# 创建表和索引# 配置 Storestore=PostgresStore(conn)store.setup()# 编译图graph=builder.compile(checkpointer=checkpointer,store=store,)

实战 3：SDK 远程调用

fromlanggraph_sdkimportget_client# 连接到 LangGraph Serverclient=get_client(url="http://localhost:2024")# 流式调用asyncforchunkinclient.runs.stream(None,# thread_id（None = 无状态）"agent",# assistant_id（对应 langgraph.json 中的 graphs key）input={"messages":[{"role":"human","content":"你好，帮我查询订单"}]},):print(chunk.event,chunk.data)

实战 4：错误重试策略

fromlanggraph.typesimportRetryPolicy# 在 add_node 时配置重试builder.add_node("api_call",api_node,retry_policy=RetryPolicy(max_attempts=3,# 最多重试 3 次initial_interval=1.0,# 首次重试等待 1 秒backoff_factor=2,# 指数退避因子（1s → 2s → 4s）retry_on=Exception,# 重试的错误类型),)

12.4 API 速查

12.5 错误与避坑指南

坑 1：生产环境用 MemorySaver

# ❌ 生产环境错误写法checkpointer=MemorySaver()# 服务重启后所有对话状态丢失！# ✅ 生产环境正确写法fromlanggraph.checkpoint.postgresimportPostgresSaver checkpointer=PostgresSaver(conn)checkpointer.setup()

坑 2：未设置数据库连接池

# ❌ 错误：单连接，高并发时瓶颈conn=psycopg2.connect(DATABASE_URL)checkpointer=PostgresSaver(conn)# ✅ 正确：使用连接池frompsycopg2.poolimportThreadedConnectionPool pool=ThreadedConnectionPool(minconn=5,maxconn=20,dsn=DATABASE_URL)checkpointer=PostgresSaver(pool)

坑 3：忽略工具调用超时

# ❌ 错误：工具调用可能无限等待@tooldefexternal_api(query:str)->str:response=requests.get(...)# 没有超时设置returnresponse.text# ✅ 正确：设置超时 + 重试@tooldefexternal_api(query:str)->str:try:response=requests.get(...,timeout=10)returnresponse.textexceptrequests.Timeout:return"服务暂时不可用，请稍后重试"

12.6 最佳实践总结

1. 开发用langgraph dev，生产用langgraph deploy或 Docker：环境统一
2. 使用 PostgresSaver + PostgresStore 持久化：高可用、支持并发
3. 关键节点设置 RetryPolicy：自动重试 + 指数退避
4. 集成 LangSmith 监控：追踪每次 LLM 调用和图执行
5. 工具调用设置超时：防止外部服务拖垮整个 Agent

附录 A：完整 API 速查手册

A.1 图构建 API

A.2 State 与 Reducer API

A.3 消息 API

A.4 工具 API

A.5 持久化 API

A.6 流程控制 API

A.7 流式输出 API