核心内容摘要
深度测评10个降AIGC网站 千笔·专业降AI率智能体解决论文查重难题
LangGraph 生产环境持久化学习
总结核心概念
三种持久化方式对比方式说明适用场景持久化InMemorySaver内存存储测试环境❌ 重启丢失PostgresSaverPostgreSQL 存储生产环境推荐✅ 持久化RedisSaverRedis 存储高并发场景✅ 持久化
State状态- 工作流中共享的数据结构classcustomstate(AgentState):user_name:str# 用户姓名conversation_count:int0# 对话次数
Checkpointer检查点- 短期记忆作用保存对话历史和工作流状态记录每次对话的状态支持多轮对话提供 Time Travel 功能状态回溯
Store存储- 长期记忆作用保存用户画像、知识库等跨会话数据用户信息姓名、偏好等知识库元数据跨会话的业务数据
PostgreSQL Checkpointer 使用
1 数据库准备-- 创建数据库如果还没有CREATEDATABASEdengmingfang OWNER dengmingfang;-- 赋予权限GRANTALLPRIVILEGESONDATABASEdengmingfangTOdengmingfang;
2 代码实现fromlanggraph.checkpoint.postgresimportPostgresSaver# 数据库连接字符串DB_URLpostgresql://dengmingfang:123456localhost:5432/dengmingfang# 使用上下文管理器withPostgresSaver.from_conn_string(DB_URL)ascheckpointer:# 第一次使用必须调用 setup() 初始化数据库表checkpointer.setup()# 创建 graph绑定 checkpointergraphcreate_react_agent(modelllm,tools[username,greet_user],state_schemacustomstate,checkpointercheckpointer)# 配置参数config{configurable:{thread_id:user_123,# 会话 IDusername:张三}}# 多轮对话状态自动持久化result1graph.invoke({messages:[{role:user,content:第一次对话}]},config)result2graph.invoke({messages:[{role:user,content:第二次对话}]},config# 相同 thread_id状态会累加)
3 查看状态# 查看当前状态state_snapshotgraph.get_state(config)print(fnext:{state_snapshot.next})print(fvalues:{state_snapshot.values})# 查看历史记录forsnapshotingraph.get_state_history(config):print(fcheckpoint_id:{snapshot.config[configurable][checkpoint_id]})
Time Travel时间回溯
1 获取历史状态# 获取所有历史记录historylist(graph.get_state_history(config))# 倒数第二条记录snapshothistory[-2]
2 回溯并修改状态# 从历史记录恢复new_configgraph.update_state(snapshot.config,{messages:[这是人工修改的消息]})# 从修改后的状态继续resultgraph.invoke({messages:[{role:user,content:继续执行}]},new_config)
3 应用场景调试回溯到某个节点重新执行错误回滚撤销错误的工具调用流程优化对比不同执行路径的结果
Store长期记忆
1 保存用户信息fromlanggraph.store.postgresimportPostgresStorewithPostgresStore.from_conn_string(DB_URL)asstore:# 初始化表store.setup()# 保存用户画像store.put(namespace(profile,),# 命名空间keyuser_123,value{name:邓,age:26,preferences:[技术,阅读]})
2 读取用户信息# 读取用户数据user_profilestore.get((profile,),user_
ifuser_profile:print(user_profile.value)# 输出{name: 邓茗芳, age: 27, preferences: [技术, 阅读]}
3 更新用户信息# 更新会覆盖原值store.put((profile,),user_123,{**user_profile.value,last_login:
,login_count:user_profile.value.get(login_count,
1})
完整生产环境架构
1 架构图┌─────────────────────────────────────────────────────────┐ │ LangGraph Application │ ├─────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────┐ ┌──────────────┐ │ │ │ Graph │────────▶│ Checkpointer │ │ │ │ (State) │ │ (PostgreSQL) │ │ │ └─────────────┘ └──────────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌─────────────┐ ┌──────────────┐ │ │ │ Store │ │ Vector Store │ │ │ │ (PostgreSQL)│ │ (FAISS) │ │ │ └─────────────┘ └──────────────┘ │ │ │ │ │ │ └───────────────────────┘ │ │ │ │ │ ▼ │ │ PostgreSQL 数据库 │ └─────────────────────────────────────────────────────────┘
2 完整代码示例fromlanggraph.checkpoint.postgresimportPostgresSaverfromlanggraph.store.postgresimportPostgresStorefromlangchain_community.vectorstoresimportFAISSfromlanggraph.prebuilt.chat_agent_executorimportcreate_react_agent DB_URLpostgresql://dengmingfang:123456localhost:5432/dengmingfangwithPostgresSaver.from_conn_string(DB_URL)ascheckpointer,\ PostgresStore.from_conn_string(DB_URL)asstore:# 初始化checkpointer.setup()store.setup()# 加载向量库vector_storeFAISS.load_local(/path/to/faiss_index,embeddingsembeddings,allow_dangerous_deserializationTrue)# 创建 graphgraphcreate_react_agent(modelllm,tools[username,greet_user,rag_query],state_schemacustomstate,checkpointercheckpointer,storestore)# 执行resultgraph.invoke({messages:[{role:user,content:查询商品}]},{configurable:{thread_id:user_123}})
数据库表结构
1 Checkpointer 自动创建的表-- 检查点表CREATETABLEcheckpoints(thread_idTEXT,checkpoint_nsTEXT,checkpoint_idTEXT,parent_checkpoint_idTEXT,typeTEXT,checkpointJSONB,metadata JSONB,PRIMARYKEY(thread_id,checkpoint_ns,checkpoint_id));-- 写入表CREATETABLEcheckpoint_writes(thread_idTEXT,checkpoint_nsTEXT,checkpoint_idTEXT,task_idTEXT,idxINTEGER,channelTEXT,typeTEXT,valueJSONB,PRIMARYKEY(thread_id,checkpoint_ns,checkpoint_id,task_id,idx));
2 Store 自动创建的表-- 存储表CREATETABLEstore(namespaceTEXT,keyTEXT,valueJSONB,created_atTIMESTAMP,updated_atTIMESTAMP,PRIMARYKEY(namespace,key));
最佳实践
1 配置优化# 使用连接池importpsycopg
pool DB_URLpostgresql://dengmingfang:123456localhost:5432/dengmingfang?pool_size10max_overflow20# 设置超时checkpointerPostgresSaver.from_conn_string(DB_URL,checkpointer_options{connection_timeout:30,command_timeout:60})
2 错误处理frompsycopg2importOperationalErrortry:checkpointerPostgresSaver.from_conn_string(DB_URL)checkpointer.setup()exceptOperationalErrorase:print(f数据库连接失败:{e})# 使用降级方案如 InMemorySavercheckpointerInMemorySaver()
3 数据备份# 备份数据库pg_dump -U dengmingfang dengmingfangbackup.sql# 恢复数据库psql -U dengmingfang dengmingfangbackup.sql
测试环境 vs 生产环境
1 测试环境InMemorySaverfromlanggraph.checkpoint.memoryimportInMemorySaver# 简单、快速但重启后丢失checkpointerInMemorySaver()graphcreate_react_agent(modelllm,toolstools,checkpointercheckpointer)
2 生产环境PostgresSaverfromlanggraph.checkpoint.postgresimportPostgresSaver# 持久化、可靠、支持回溯checkpointerPostgresSaver.from_conn_string(DB_URL)checkpointer.setup()graphcreate_react_agent(modelllm,toolstools,checkpointercheckpointer)
八、
常见问题
1 数据库连接失败# 错误psycopg
OperationalError# 解决检查数据库服务、用户名密码、网络连接# 验证连接importpsycopg2 connpsycopg
connect(DB_URL)print(连接成功)
2 表已存在错误# 第二次运行时 setup() 会报错# 解决捕获异常或先检查try:checkpointer.setup()exceptExceptionase:print(f表可能已存在:{e})
3 状态丢失# 确保 thread_id 一致config{configurable:{thread_id:same_id}}# 每次都用同一个 configresult1graph.invoke({messages:[...]},config)result2graph.invoke({messages:[...]},config)# 状态会累加
性能优化
1 批量操作# 批量存储用户数据foruser_id,user_datainuser_list:store.put((profile,),user_id,user_data)
2 定期清理-- 清理 30 天前的历史记录DELETEFROMcheckpointsWHEREcreated_atNOW()-INTERVAL30 days;
3 索引优化-- 为常用查询创建索引CREATEINDEXidx_checkpoints_thread_idONcheckpoints(thread_id);CREATEINDEXidx_store_namespaceONstore(namespace);
十、