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

第 12 章生产部署与性能优化12.1 本章目标学完本章你将能够了解 LangGraph Platform 和 LangGraph Server 的部署方式掌握生产环境的 Checkpointer 和 Store 配置学会错误处理、重试和并发控制了解监控、日志和成本优化策略12.2 核心概念生产部署架构监控数据库LangGraph Server客户端Web/App 客户端API Server:2024langgraph_sdkPostgreSQLCheckpointer StoreLangSmith追踪 监控开发 vs 生产对比维度开发环境生产环境CheckpointerMemorySaverPostgresSaverStoreInMemoryStorePostgresStoreLLM 调用直连 API加缓存 重试错误处理打印日志结构化日志 告警部署方式langgraph devlanggraph deploy或 Docker监控无LangSmith 集成12.3 实战实战 1langgraph.json 项目配置{graphs:{agent:./agent.py:graph},store:{index:{embed:openai:text-embeddings-3-small,dims:1536,fields:[$]}},env:{OPENAI_API_KEY:your-api-key}}实战 2Postgres 持久化配置importosfromlanggraph.checkpoint.postgresimportPostgresSaverfromlanggraph.store.postgresimportPostgresStoreimportpsycopg2# 连接数据库connpsycopg2.connect(os.environ[DATABASE_URL])# 配置 CheckpointercheckpointerPostgresSaver(conn)checkpointer.setup()# 创建表和索引# 配置 StorestorePostgresStore(conn)store.setup()# 编译图graphbuilder.compile(checkpointercheckpointer,storestore,)实战 3SDK 远程调用fromlanggraph_sdkimportget_client# 连接到 LangGraph Serverclientget_client(urlhttp://localhost:2024)# 流式调用asyncforchunkinclient.runs.stream(None,# thread_idNone 无状态agent,# assistant_id对应 langgraph.json 中的 graphs keyinput{messages:[{role:human,content:你好帮我查询订单}]},):print(chunk.event,chunk.data)实战 4错误重试策略fromlanggraph.typesimportRetryPolicy# 在 add_node 时配置重试builder.add_node(api_call,api_node,retry_policyRetryPolicy(max_attempts3,# 最多重试 3 次initial_interval1.0,# 首次重试等待 1 秒backoff_factor2,# 指数退避因子1s → 2s → 4sretry_onException,# 重试的错误类型),)12.4 API 速查API完整签名入参说明返回值说明langgraph devCLI 命令无开发服务器启动本地开发服务器langgraph deployCLI 命令无部署部署到 LangSmith Cloudget_client(url)get_client(url: str)url: 服务地址ClientSDK 客户端RetryPolicy(max_attempts, ...)RetryPolicy(max_attempts, initial_interval, backoff_factor)max_attempts: 最大重试次数;initial_interval: 初始等待秒数;backoff_factor: 退避因子RetryPolicy节点重试策略PostgresSaver(conn).setup().setup()无无初始化数据库表结构12.5 错误与避坑指南坑 1生产环境用 MemorySaver# ❌ 生产环境错误写法checkpointerMemorySaver()# 服务重启后所有对话状态丢失# ✅ 生产环境正确写法fromlanggraph.checkpoint.postgresimportPostgresSaver checkpointerPostgresSaver(conn)checkpointer.setup()坑 2未设置数据库连接池# ❌ 错误单连接高并发时瓶颈connpsycopg2.connect(DATABASE_URL)checkpointerPostgresSaver(conn)# ✅ 正确使用连接池frompsycopg2.poolimportThreadedConnectionPool poolThreadedConnectionPool(minconn5,maxconn20,dsnDATABASE_URL)checkpointerPostgresSaver(pool)坑 3忽略工具调用超时# ❌ 错误工具调用可能无限等待tooldefexternal_api(query:str)-str:responserequests.get(...)# 没有超时设置returnresponse.text# ✅ 正确设置超时 重试tooldefexternal_api(query:str)-str:try:responserequests.get(...,timeout10)returnresponse.textexceptrequests.Timeout:return服务暂时不可用请稍后重试12.6 最佳实践总结开发用langgraph dev生产用langgraph deploy或 Docker环境统一使用 PostgresSaver PostgresStore 持久化高可用、支持并发关键节点设置 RetryPolicy自动重试 指数退避集成 LangSmith 监控追踪每次 LLM 调用和图执行工具调用设置超时防止外部服务拖垮整个 Agent附录 A完整 API 速查手册A.1 图构建 APIAPI导入路径签名说明StateGraphlanggraph.graphStateGraph(state_schema)创建状态图构建器add_nodeStateGraph 方法.add_node(name, action)注册节点add_edgeStateGraph 方法.add_edge(start, end)添加普通边add_conditional_edgesStateGraph 方法.add_conditional_edges(source, path, path_map)添加条件边add_sequenceStateGraph 方法.add_sequence(nodes)批量添加顺序节点compileStateGraph 方法.compile(checkpointer, store, ...)编译为可执行图STARTlanggraph.graph常量__start__图入口ENDlanggraph.graph常量__end__图出口A.2 State 与 Reducer APIAPI导入路径说明TypedDicttyping定义 State 的推荐方式Annotatedtyping_extensions声明字段的 Reduceradd_messageslanggraph.graph消息专用 Reduceroperator.addoperator数值/列表累加 ReducerMessagesStatelanggraph.graph预构建的 messages StateA.3 消息 APIAPI导入路径说明HumanMessagelangchain_core.messages用户消息AIMessagelangchain_core.messagesAI 回复ToolMessagelangchain_core.messages工具执行结果SystemMessagelangchain_core.messages系统提示词A.4 工具 APIAPI导入路径说明toollangchain_core.tools将函数标记为工具ToolNodelanggraph.prebuilt自动处理工具调用tools_conditionlanggraph.prebuilt判断是否有 tool_callscreate_react_agentlanggraph.prebuilt预构建 AgentA.5 持久化 APIAPI导入路径说明MemorySaverlanggraph.checkpoint.memory内存 CheckpointerSqliteSaverlanggraph.checkpoint.sqliteSQLite CheckpointerPostgresSaverlanggraph.checkpoint.postgresPostgres CheckpointerInMemoryStorelanggraph.store.memory内存 StorePostgresStorelanggraph.store.postgresPostgres StoreA.6 流程控制 APIAPI导入路径说明interruptlanggraph.types暂停执行Commandlanggraph.types动态控制流程Sendlanggraph.types并行分发RetryPolicylanggraph.types重试策略Runtimelanggraph.runtime运行时上下文A.7 流式输出 APIAPI导入路径说明.stream()CompiledGraph 方法同步流式.astream()CompiledGraph 方法异步流式.stream_events()CompiledGraph 方法v3 事件流式get_stream_writerlanggraph.config自定义流式事件