Agent 记忆管理:从手动拼接到 PostgreSQL 持久化
Agent 记忆管理:从手动拼接到 PostgreSQL 持久化
大模型没有记忆。你跟它说了名字,下一轮它就忘了。要让对话连贯,得自己管理消息历史。
最原始的方式:手动拼消息
```python from langchain.agents import create_agent
agent = create_agent(model="openai:gpt-4o-mini") history = []
response = agent.invoke({"messages": [{"role": "user", "content": "来一首宋词"}]}) history.extend(response["messages"])
history.append({"role": "user", "content": "再来"}) response = agent.invoke({"messages": history}) ```
能用,但要自己维护消息列表,多会话管理起来很麻烦。
InMemorySaver:内存级持久化
LangGraph 提供了 Checkpointer 机制,用 `thread_id` 区分不同会话,框架自动管理消息历史:
```python from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver() agent = create_agent(model="openai:gpt-4o-mini", checkpointer=checkpointer)
config = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": [{"role": "user", "content": "来一首宋词"}]}, config=config) agent.invoke({"messages": [{"role": "user", "content": "再来"}]}, config=config) ```
不用手动拼消息了。换个 thread_id 就是新会话,互不干扰。
PostgresSaver:数据库级持久化
InMemorySaver 进程一关数据就没了。生产环境用 PostgresSaver:
```python from langgraph.checkpoint.postgres import PostgresSaver
DB_URI = "postgresql://user:pass@localhost:5432/mydb"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer: agent = create_agent(model="openai:gpt-4o-mini", checkpointer=checkpointer) config = {"configurable": {"thread_id": "1"}} agent.invoke({"messages": [{"role": "user", "content": "来一首宋词"}]}, config=config) ```
第一次用需要调 `checkpointer.setup()` 建表。之后重启程序,用同一个 thread_id 还能接着聊。
查看历史记录
```python with PostgresSaver.from_conn_string(DB_URI) as checkpointer: config = {"configurable": {"thread_id": "1"}} checkpoints = checkpointer.list(config=config) for cp in checkpoints: for msg in cp[1]["channel_values"]["messages"]: print(f"[{type(msg).name}] {msg.content}") break ```
小结
记忆管理三个层次:手动拼接理解原理用,InMemorySaver 开发调试用,PostgresSaver 生产环境用。Checkpointer 机制不光管记忆,还是后面 Human-in-the-Loop 和时间旅行功能的基础。