微信扫码
添加专属顾问
我要投稿
让AI Agent告别"金鱼记忆"!深度解析如何通过LangGraph+PostgreSQL实现长期记忆,打造真正智能的AI助手。核心内容:1. AI Agent"失忆症"的痛点分析与长期记忆的重要性2. LangGraph核心机制:状态管理、图结构与检查点技术3. PostgreSQL数据库实现持久化存储的详细技术方案
在当今 AI 飞速发展的时代,LLM (大型语言模型) 驱动的 AI Agent 正在成为自动化复杂任务、提升生产力的核心力量。从智能客服、招聘助理到自动化代码生成,Agent 的应用场景日益广泛。然而,许多开发者在构建 Agent 时,都会遇到一个核心痛点:Agent 的“金鱼记忆”问题。
标准的 LLM 调用是无状态的。每次交互,它都像第一次见到你一样,无法记住上文、无法积累经验、更无法在多轮对话或跨会话中保持上下文连贯性。这极大地限制了 Agent 的应用深度和用户体验。想象一下,一个招聘 Agent 每次都让你重复职位要求;一个客服 Agent 无法记住你上次的投诉详情,这是多么糟糕的体验!
为了解决这一问题,赋予 AI Agent **长期记忆(Long-Term Memory)**能力成为了构建真正智能、高效 Agent 的关键。它让 Agent 不仅能理解当前指令,更能记住过往的交互、学习到的知识,甚至从历史经验中进行“反思”和“成长”。
本文将深入探讨如何在 LangGraph (LangChain 家族中用于编排复杂 Agent 行为的框架) 中,结合PostgreSQL (一个强大、稳定、可扩展的关系型数据库) 来实现 Agent 的长期记忆。我们将提供详细的技术原理、工程实践、可直接使用的代码示例,并讨论其带来的变革、面临的挑战与最佳实践。
在深入实现之前,我们需要先理解 LangGraph 的几个核心概念:
TypedDict
,存储着 Agent 在特定时刻的所有相关信息。例如,对于一个对话 Agent,它的状态可能包含 messages
(消息列表)、current_task
(当前正在执行的任务)、tool_outputs
(工具执行结果) 等。在 Agent 的每次运行中,这个状态会被不断读取、修改和传递。通过 Checkpointer 机制,LangGraph 解耦了 Agent 的运行时状态与持久化存储,使得开发者可以灵活选择不同的后端(如 SQLite、PostgreSQL、Redis 等)来实现长期记忆。
在众多持久化存储方案中,PostgreSQL 作为关系型数据库的佼佼者,具备以下显著优势,使其成为 Agent 长期记忆的理想选择:
pgvector
等扩展,甚至可以直接在数据库中进行向量相似度搜索,实现更高级的知识管理。PostgresSaver
,简化了与 PostgreSQL 的集成过程。接下来,我们将通过一个实际的 ReAct Agent 示例,详细演示如何将其与 PostgreSQL 进行集成,赋予它强大的长期记忆能力。
首先,确保你的 Python 环境就绪,并安装必要的库:
# 核心依赖:LangGraph, LangChain核心库
pip install langgraph langchain-core
# PostgreSQL 驱动和 SQLAlchemy ORM
# psycopg2-binary 是一个预编译版本,安装更简便
pip install psycopg2-binary sqlalchemy
# 如果需要 OpenAI 的 LLM
pip install langchain-openai
# (可选但强烈推荐) 用于调试和可观测性的 LangSmith
# pip install langsmith
确保你有一个运行中的 PostgreSQL 实例,并创建用于 Agent 记忆的数据库和用户。
方法一:使用 Docker 快速启动 (适合本地开发和测试)
# 1. 拉取 PostgreSQL 镜像
docker pull postgres:latest
# 2. 运行 PostgreSQL 容器
# -p 5432:5432: 将容器的5432端口映射到宿主机的5432端口
# -e POSTGRES_PASSWORD=mysecretpassword: 设置PostgreSQL的root用户密码
# --name some-postgres: 给容器命名
# -d: 后台运行
docker run --name some-postgres -e POSTGRES_PASSWORD=mysecretpassword -p 5432:5432 -d postgres
方法二:通过 psql
创建数据库和用户
连接到你的 PostgreSQL 服务器(例如,使用 psql -U postgres -h localhost
并输入密码),然后执行以下 SQL 命令:
-- 创建一个独立的数据库用于Agent的记忆
CREATEDATABASE langgraph_agent_memory;
-- 创建一个专门的用户,增强安全性
CREATEUSER langgraph_user WITHPASSWORD'your_secure_password';
-- 授予用户对数据库的所有权限
GRANTALLPRIVILEGESONDATABASE langgraph_agent_memory TO langgraph_user;
PostgreSQL 连接字符串 (URI)
根据你的设置,连接字符串通常遵循以下格式:postgresql+psycopg2://<user>:<password>@<host>:<port>/<database_name>
例如:postgresql+psycopg2://langgraph_user:your_secure_password@localhost:5432/langgraph_agent_memory
在后续代码中,我们会将这个字符串赋值给 POSTGRES_CONNECTION_STRING
变量。
我们将构建一个经典的 ReAct Agent,它能够接收用户输入,决定是调用工具还是直接回答,并执行相应操作。
3.1 Agent 状态定义
Agent 的状态是其内部信息的载体,messages
是最核心的部分,用于存储对话历史。
from typing import List, Dict, TypedDict, Union, Annotated
import operator
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
"""
AgentState 定义了 Agent 在 LangGraph 图中传递的状态。
'messages':核心,用于存储对话历史。
使用 Annotated[List[BaseMessage], add_messages],
LangGraph 会自动将新消息添加到列表中。
'current_plan':可选,Agent 对当前任务的规划。
'tool_output': 可选,上次工具的输出结果。
"""
messages: Annotated[List[BaseMessage], add_messages]
# 你可以根据需要添加更多状态变量,例如:
# current_job_query: Optional[str] = None # 招聘Agent可能需要记住当前职位查询
# relevant_docs: List[str] = [] # RAG Agent可能需要存储检索到的相关文档
3.2 工具定义
Agent 通过调用工具来与外部世界交互,执行特定任务。这里我们定义一个简单的模拟网页搜索工具。
from langchain_core.tools import tool
@tool
def search_web(query: str) -> str:
"""
模拟一个网页搜索工具。
在实际应用中,这里会调用真正的搜索引擎API(如Google Search API, Serper API等)。
"""
print(f"\n--- 🌍 Agent 正在执行工具:search_web,查询内容:'{query}' ---")
if"current weather"in query.lower():
return"模拟结果:当前天气晴朗,气温25摄氏度。"
elif"capital of France"in query.lower():
return"模拟结果:法国的首都是巴黎。"
else:
returnf"模拟结果:关于 '{query}' 的信息 (这是一个模拟的搜索结果)。"
tools = [search_web]
3.3 LLM 配置
我们将使用 OpenAI 的模型,并绑定我们定义的工具。bind_tools
让 LLM 知道何时以及如何调用这些工具。
import os
from langchain_openai import ChatOpenAI
# 确保你的 OpenAI API Key 已经设置在环境变量中
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"
# (可选) 配置 LangSmith 提高可观测性,便于调试
# os.environ["LANGCHAIN_API_KEY"] = "YOUR_LANGSMITH_API_KEY"
# os.environ["LANGCHAIN_TRACING_V2"] = "true"
# os.environ["LANGCHAIN_PROJECT"] = "LangGraph_Postgres_Memory_ReAct_Agent"
# 初始化 LLM 并绑定工具
# streaming=True 对于实时响应很有用
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, streaming=True)
llm_with_tools = llm.bind_tools(tools)
3.4 Agent 节点与决策逻辑
我们将 Agent 的核心逻辑拆分为两个节点 (call_llm
和 call_tool
) 和一个条件函数 (should_continue
)。
from langgraph.graph import StateGraph, END
# Agent 节点:调用 LLM
def call_llm(state: AgentState) -> AgentState:
"""
LLM 节点:将当前对话历史发送给 LLM,获取 LLM 的响应。
响应可能是直接回答,也可能是调用工具的指令。
"""
messages = state['messages']
print(f"\n--- 🧠 Agent 思考中:调用LLM处理消息:{messages[-1].content[:50]}... ---")
response = llm_with_tools.invoke(messages)
# LLM 的响应会自动添加到 state['messages'] 中 (因为使用了 add_messages 注解)
return {"messages": [response]}
# Agent 节点:执行工具
def call_tool(state: AgentState) -> AgentState:
"""
工具节点:执行 LLM 决定的工具调用。
"""
messages = state['messages']
last_message = messages[-1] # 获取 LLM 的最新响应,检查是否有工具调用
tool_outputs = []
print(f"\n--- 🛠️ Agent 执行工具调用 ---")
for tool_call in last_message.tool_calls:
print(f" - 工具名称: {tool_call.name}, 参数: {tool_call.args}")
try:
# 调用实际的工具函数
tool_output = globals()[tool_call.name].invoke(tool_call.args)
tool_outputs.append(ToolMessage(content=str(tool_output), tool_call_id=tool_call.id))
except Exception as e:
tool_outputs.append(ToolMessage(content=f"工具执行失败: {e}", tool_call_id=tool_call.id))
# 工具输出也会添加到 state['messages'] 中
return {"messages": tool_outputs}
# 条件函数:决定下一步流程
def should_continue(state: AgentState) -> str:
"""
条件函数:根据 LLM 的最新响应决定 Agent 的下一步。
如果 LLM 请求调用工具,则返回 "continue" (走向工具节点);
否则,返回 "end" (终止 Agent 运行,LLM 的响应即为最终答案)。
"""
messages = state['messages']
last_message = messages[-1]
# 检查 LLM 是否包含工具调用指令
if last_message.tool_calls:
print(f"--- 决策:LLM 请求调用工具,继续执行工具。---")
return"continue"
else:
print(f"--- 决策:LLM 直接给出答案,Agent 运行结束。---")
return"end"
# 构建 LangGraph 图
graph_builder = StateGraph(AgentState)
# 添加节点
graph_builder.add_node("llm", call_llm)
graph_builder.add_node("tool", call_tool)
# 定义图的边
# 1. 工具执行后,总会回到 LLM 节点,让 LLM 根据工具结果进行下一步判断或回答。
graph_builder.add_edge("tool", "llm")
# 2. 从 LLM 节点出发,根据条件决定是继续调用工具还是结束。
graph_builder.add_conditional_edges(
"llm", # 源节点:LLM 的响应
should_continue, # 条件函数:根据 LLM 的响应判断
{
"continue": "tool", # 如果 should_continue 返回 "continue",则走向 "tool" 节点
"end": END # 如果 should_continue 返回 "end",则结束图的运行
}
)
# 设置图的入口点
graph_builder.set_entry_point("llm") # 每次运行都从 LLM 思考开始
现在,我们将 PostgresSaver
引入到我们的 LangGraph 应用中。
# --- 整合 PostgreSQL Checkpointer ---
from langgraph.checkpoint.postgres import PostgresSaver
from sqlalchemy import create_engine # PostgresSaver 内部使用 SQLAlchemy
# 替换为你的 PostgreSQL 连接字符串
POSTGRES_CONNECTION_STRING = "postgresql+psycopg2://langgraph_user:your_secure_password@localhost:5432/langgraph_agent_memory"
# 初始化 PostgresSaver
try:
# `sync_connection` 用于同步操作(例如 get_state, get_graph_state)
# `async_connection` 用于异步操作(例如 aget_state, aget_graph_state)
# 建议同时提供以支持同步和异步场景
memory = PostgresSaver(
sync_connection=POSTGRES_CONNECTION_STRING,
async_connection=POSTGRES_CONNECTION_STRING
)
print("✅ PostgresSaver 已成功初始化,数据库连接就绪。")
except Exception as e:
print(f"❌ 初始化 PostgresSaver 失败:{e}")
print("请检查 PostgreSQL 服务是否运行,以及连接字符串 (POSTGRES_CONNECTION_STRING) 是否正确。")
# 在生产环境中,这里可能需要更复杂的错误处理或重试机制
exit()
# **编译图并绑定 Checkpointer**
# 这一步将 Agent 的状态持久化能力注入到其运行图中。
app = graph_builder.compile(checkpointer=memory)
print("\n--- LangGraph Agent 已编译,并拥有 PostgreSQL 长期记忆能力!---")
print("现在,我们可以启动/恢复与 Agent 的多轮对话了。")
现在,我们可以像使用普通 LangGraph 应用一样运行我们的 Agent,但每次调用 invoke
时,通过 config
参数传入一个唯一的 thread_id
。这个 thread_id
就是 LangGraph 在 PostgreSQL 中存储和检索特定对话历史的键。
print("\n=== 场景一:启动/恢复会话 'user_session_abc' ===")
# 使用一个唯一的 thread_id 来标识一个会话
# 首次运行会创建新会话并保存;后续运行会加载该会话的最新状态
session_id_1 = "user_session_abc"
config_1 = {"configurable": {"thread_id": session_id_1}}
# 第一次交互:提问一个需要工具的问题
user_input_1 = "请问现在巴黎的天气怎么样?"
print(f"\n用户 ({session_id_1}): {user_input_1}")
# invoke 返回的是最终的状态,包括所有消息
final_state_1 = app.invoke({"messages": [HumanMessage(content=user_input_1)]}, config=config_1)
print(f"Agent ({session_id_1}): {final_state_1['messages'][-1].content}")
# 检查数据库,你会发现 'langgraph_checkpoints' 表中多了一条记录
print("\n--- (模拟Agent进程重启或时间流逝) ---")
# 第二次交互:在同一个会话中继续提问,Agent应该记得上下文
user_input_2 = "那法国的首都在哪里?"# 故意不提“法国”
print(f"\n用户 ({session_id_1}): {user_input_2}")
# LangGraph 会自动从数据库加载 user_session_abc 的最新状态
final_state_2 = app.invoke({"messages": [HumanMessage(content=user_input_2)]}, config=config_1)
print(f"Agent ({session_id_1}): {final_state_2['messages'][-1].content}")
# 此时 Agent 应该能利用历史对话中的“法国”信息来回答。
print("\n=== 场景二:启动一个全新的会话 'user_session_xyz' ===")
session_id_2 = "user_session_xyz"
config_2 = {"configurable": {"thread_id": session_id_2}}
user_input_3 = "你好,我是新用户,你叫什么名字?"
print(f"\n用户 ({session_id_2}): {user_input_3}")
final_state_3 = app.invoke({"messages": [HumanMessage(content=user_input_3)]}, config=config_2)
print(f"Agent ({session_id_2}): {final_state_3['messages'][-1].content}")
# 此时 Agent 应该从头开始对话,因为它是一个新的 session_id。
print("\n=== 场景三:手动获取和检查会话状态 ===")
# 你可以随时通过 get_state 方法获取特定会话的最新状态
retrieved_state = app.get_state(config=config_1)
print(f"\n从数据库中获取 '{session_id_1}' 的最新状态:")
for i, msg in enumerate(retrieved_state.values['messages']):
print(f" 消息 {i+1} ({msg.type}): {msg.content}")
# 注意:`get_state` 返回的是一个 `GraphState` 对象,它的 `values` 属性是一个字典,包含了 AgentState 的内容。
# `messages` 是一个 List[BaseMessage]
运行结果的观察点:
search_web
工具调用: 当你问“巴黎天气”时,你会看到 Agent 打印出调用 search_web
工具的信息。langgraph_checkpoints
表中存储了每个 thread_id
的会话快照。你可以看到 thread_id
、checkpoint_id
(每次状态更新都会生成新的检查点ID)、metadata
和 state
(序列化后的 Agent 状态)。仅仅保存对话历史只是长期记忆的起点。要让 Agent 具备更强大的“智慧”,还需要结合以下高级技术:
pgvector
)
对于招聘 Agent 而言,仅仅记住聊天历史是不够的,它还需要拥有海量的招聘知识:职位描述范本、行业薪资标准、公司文化、面试题库等。
pgvector
扩展,就能原生支持向量存储和相似度搜索,非常适合构建 RAG(Retrieval Augmented Generation,检索增强生成)系统。pgvector
中搜索最相关的知识片段。# 伪代码示例:结合 pgvector 的 RAG
# pip install pgvector langchain-community openai
# (假设你已在PostgreSQL中安装并配置pgvector扩展)
# from langchain_community.vectorstores import PGVector
# from langchain_openai import OpenAIEmbeddings
# embeddings = OpenAIEmbeddings()
# vectorstore = PGVector(
# connection_string=POSTGRES_CONNECTION_STRING,
# embedding_function=embeddings,
# collection_name="recruitment_knowledge" # 存储招聘知识的向量集合
# )
# @tool
# def retrieve_knowledge(query: str) -> str:
# """
# 从招聘知识库中检索与查询相关的文档。
# """
# docs = vectorstore.similarity_search(query, k=3)
# return "\n".join([doc.page_content for doc in docs])
# # 在 Agent 图中添加一个新的节点或在 call_llm 中加入检索逻辑
# # ... (LLM可能会决定何时调用 retrieve_knowledge 工具)
长时间的对话会导致 messages
列表越来越长,增加 LLM 的输入 Token 数量,进而提高成本并可能超出上下文窗口。
thread_id
关联),作为该会话的“长期记忆摘要”。在每次新的对话开始时,将这个摘要作为系统指令(System Prompt)或额外的上下文注入到 LLM 的输入中,从而提供历史背景,而无需传入全部原始消息。# 伪代码示例:摘要Agent节点
# def summarize_conversation_node(state: AgentState) -> AgentState:
# # 假设 messages 列表很长
# long_history = state['messages'][:-N] # 获取需要摘要的历史消息
# # 调用一个专门的LLM进行摘要
# summary_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.1)
# summary_prompt = f"请总结以下对话内容的关键信息和结论:\n{''.join([m.content for m in long_history])}"
# summary_text = summary_llm.invoke(summary_prompt).content
# # 将摘要存储到PostgreSQL(例如一个独立的表 `session_summaries`)
# # db_client.save_summary(state['configurable']['thread_id'], summary_text)
# # 移除旧消息,只保留少量最新消息和摘要
# state['messages'] = [AIMessage(content=f"历史对话摘要:{summary_text}")] + state['messages'][-N:]
# return state
# # 在LangGraph中,可以通过条件边在特定时机触发这个摘要节点
更高级的长期记忆意味着 Agent 能够“学习”并适应用户、任务或环境。
AI Agent 拥有长期记忆能力后,将对传统行业带来以下颠覆性变革:
招聘行业:从“广撒网”到“精匹配”
客户服务:从“被动响应”到“主动关怀”
软件开发:从“重复劳动”到“智能协作”
尽管潜力巨大,但实现和管理 AI Agent 的长期记忆也面临一些挑战:
最佳实践建议:
通过 LangGraph 的强大编排能力与 PostgreSQL 的稳定持久化特性,我们成功地为 AI Agent 赋予了“永不遗忘”的长期记忆超能力。这不仅解决了 Agent 的核心痛点,更将它从一个简单的“对话机器”提升为能够持续学习、积累经验、提供个性化服务的“智能伙伴”。
未来,随着 Agent 技术的不断演进和数据处理能力的提升,具备强大长期记忆的 AI Agent 将在各行各业扮演越来越重要的角色,成为我们数字化转型浪潮中不可或缺的驱动力。
你的下一个 AI Agent 项目,打算如何设计它的“记忆力”呢?你认为最值得用 AI Agent 改造的传统行业或业务流程是什么?欢迎在评论区分享你的真知灼见!
53AI,企业落地大模型首选服务商
产品:场景落地咨询+大模型应用平台+行业解决方案
承诺:免费场景POC验证,效果验证后签署服务协议。零风险落地应用大模型,已交付160+中大型企业
2025-07-04
Karpathy:我不是要造新词,是「上下文工程」对 Agent 来说太重要了
2025-07-04
AI Agent的核心:Context Engineering(上下文工程)
2025-07-04
AI Agent与AI Workflow:“对决”与“共生”,未来属于“混血儿”!
2025-07-04
破局AI内卷:揭秘驱动10倍效能的AI工作流三大核心技术支柱
2025-07-04
深度揭秘:下一代AI生产力,颠覆你的工作与认知?99%的人还没看懂!
2025-07-04
AI Agent时代的AI Workflow,重构未来工作流设计准则!
2025-07-04
MCP对AI Agent意味什么?深度解剖MCP的本质与未来影响力
2025-07-04
喂给AI的第一口饭:文本预处理
2025-05-29
2025-04-11
2025-04-12
2025-04-06
2025-04-29
2025-04-12
2025-04-29
2025-05-07
2025-05-23
2025-05-07