你是否曾向AI助手询问公司内部政策，得到的却是“我不知道”？或者在重启对话后，发现AI完全忘记了之前的交流历史？这些尴尬正源于AI缺乏专属知识库和持续记忆的能力。
今天，我们就来解决这两个痛点。我们将基于 LangGraph 框架，一步步为AI智能体注入两大核心能力：检索增强生成（RAG） 和 长期记忆。学完本文，你将能构建一个真正“博闻强识”的AI助手，它可以：
- 从指定文档中查找答案，突破预训练知识的限制。
- 记住跨会话的对话历史，实现真正连贯的个性化交流。
一、核心概念：RAG与长期记忆
在动手编码前，我们先快速理解这两个概念。
RAG（检索增强生成）
想象你的AI助手知识渊博，但它的知识库在训练完成后就定格了。RAG就像为它配备了一个智能书架。当用户提问时，助手会先从这个书架上快速查找相关的资料（检索），然后再结合自身知识组织语言回答（生成）。
传统方式的局限性：
# 传统方式：只依赖预训练知识
user_question = "我们公司最新的报销政策是什么？"
# AI: "抱歉，我不知道你们公司的具体政策..."
RAG方式的优势：
# RAG方式：结合文档知识
user_question = "我们公司最新的报销政策是什么？"
# 1. 从公司文档库中检索相关政策文档
# 2. 基于检索到的文档生成答案
# AI: "根据2024年3月发布的《费用报销管理办法》第5条..."
长期记忆
如果说短期记忆让AI能在一次对话中记住上下文，那么长期记忆就是让它能记住不同会话之间的对话历史。这相当于为AI建造了一个“记忆宫殿”，使它能回忆起昨天、上周甚至更早的交流内容，让每次对话都建立在之前的基础上。
二、为LangGraph智能体添加RAG能力
我们以让AI学习 React Native ExecuTorch 技术文档为例，构建一个能回答相关问题的智能助手。
步骤1：加载和分割文档
首先，需要获取文档并将其切分成适合处理的“知识片段”。
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 1. 加载文档 - 这里以React Native ExecuTorch文档为例
loader = WebBaseLoader("https://docs.swmansion.com/react-native-executorch/")
docs = loader.load()
print(f"加载了 {len(docs)} 个文档，总字符数：{len(docs[0].page_content)}")
# 2. 分割文档 - 就像把长文章切成便于查找的小卡片
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 每个块约1000字符
chunk_overlap=200 # 块之间重叠200字符，避免信息割裂
)
all_splits = text_splitter.split_documents(docs)
print(f"分割成 {len(all_splits)} 个文本块")
关键参数解释：
chunk_size=1000：每个文本块大小适中，避免信息过于碎片化或冗长。chunk_overlap=200：块间重叠确保关键信息不会因为恰好位于块边界而被割裂。
步骤2：创建向量数据库（智能书架）
现在我们需要一个能快速进行语义查找的“书架”。
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_huggingface import HuggingFaceEmbeddings
# 1. 选择嵌入模型 - 这相当于文档的“指纹生成器”
# all-MiniLM-L6-v2是一个轻量级但效果不错的模型
embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/all-MiniLM-L6-v2"
)
# 2. 创建向量存储 - 我们的“智能书架”
vector_store = InMemoryVectorStore(embeddings)
# 3. 将文本块添加到书架
_ = vector_store.add_documents(documents=all_splits)
print("向量数据库准备就绪！")
什么是嵌入？
简单说，嵌入就是将文本转换成数学向量。语义相似的文本，其向量表示在数学空间中也更接近，这样我们就能通过计算向量相似度来找到相关的文档。
步骤3：改造提问函数，集成RAG
接下来，让我们的智能体学会“先查资料，再回答问题”。
from langchain_core.messages import HumanMessage, AIMessage
def ask_llm_with_rag(state):
"""增强版提问函数：先检索，后生成"""
user_query = input("请输入您的问题: ")
# 1. 检索相关文档 - 从书架上找到最相关的卡片
retrieved_docs = vector_store.similarity_search(user_query, k=3)
print(f"检索到 {len(retrieved_docs)} 个相关文档片段")
# 2. 构建上下文
context = "\n\n---\n\n".join([doc.page_content for doc in retrieved_docs])
# 3. 构建增强提示
user_message = HumanMessage(
f"""请基于以下上下文回答问题。如果上下文不包含相关信息，请诚实地说不知道。
上下文：
{context}
用户问题：
{user_query}
请提供准确、有用的回答："""
)
# 4. 调用模型生成回答
answer_message = model.invoke(
state["messages"] + [user_message]
)
# 5. 打印并保存结果
print(f"\n🤖 AI回答: {answer_message.content}\n")
return {
"messages": [user_message, answer_message],
}
# 小练习：尝试添加一个中间步骤，让AI先根据查询和上下文生成优化的提示，再回答
步骤4：完整RAG智能体代码示例
让我们把各部分组合起来，看看完整的代码。
# 完整代码：带RAG的LangGraph智能体
from typing import TypedDict, List
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
import os
# 设置OpenAI API密钥（请替换为你的密钥）
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
# 定义状态结构
class State(TypedDict):
messages: List[BaseMessage]
iteration: int
# 初始化模型
model = ChatOpenAI(model="gpt-3.5-turbo")
# 构建图
graph_builder = StateGraph(State)
# 添加节点
graph_builder.add_node("ask", ask_llm_with_rag)
# 设置入口点
graph_builder.set_entry_point("ask")
graph_builder.add_edge("ask", END)
# 编译图
graph = graph_builder.compile()
# 测试RAG功能
initial_state = {"messages": [], "iteration": 0}
result = graph.invoke(initial_state)
print("=" * 50)
print("测试问题1: React Native ExecuTorch是什么？")
print("=" * 50)
实际运行效果示例：
请输入您的问题: React Native ExecuTorch是什么？
检索到 3 个相关文档片段
🤖 AI回答: React Native ExecuTorch是一个为React Native生态系统量身定制的设备端AI和大语言模型工具包。它基于Meta的ExecuTorch AI框架，允许开发者在移动设备上本地运行AI模型和LLM。主要特点包括：
1. 设备端模型执行：AI模型直接在设备上运行，保护用户隐私，无需外部API调用
2. 成本效益：减少对云基础设施的依赖，降低服务器成本和延迟
3. 隐私优先：数据完全保留在设备上
4. 开发者友好：提供声明式API，无需深厚的AI专业知识
5. 属于PyTorch Edge生态系统
看，AI已经能基于我们提供的特定文档进行准确回答了！
三、为智能体添加长期记忆
解决了知识库问题，我们再来攻克第二个痛点：如何让AI记住跨会话的对话？
步骤1：设置记忆存储系统
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
# 1. 创建检查点保存器 - 保存单个对话状态
checkpointer = InMemorySaver()
# 2. 创建存储 - 跨线程/会话存储状态
store = InMemoryStore()
# 3. 编译时集成记忆系统
workflow = graph.compile(
checkpointer=checkpointer, # 记住每次对话的状态
store=store # 跨会话记忆
)
步骤2：使用会话ID管理连续性对话
# 为每个对话会话设置唯一ID
config = {
"recursion_limit": 100,
"configurable": {
"thread_id": "user_123_session_1" # 实际中可以使用用户ID+时间戳
}
}
# 第一次对话
print("=== 第一次对话 ===")
workflow.invoke(
{"messages": [], "iteration": 0},
config=config,
)
# 获取当前对话状态
current_state = workflow.get_state(config)
print(f"当前对话轮次: {current_state.values['iteration']}")
# 第二次对话（延续上次）
print("\n=== 第二次对话（延续上次）===")
workflow.invoke(
current_state,
config=config, # 相同的thread_id，AI会记得上次对话
)
步骤3：实现可持久化的长期记忆
对于更复杂的应用，我们可能需要将记忆保存到本地文件或数据库中。
import json
from datetime import datetime
class LongTermMemoryManager:
"""长期记忆管理器"""
def __init__(self, storage_path="memory_storage.json"):
self.storage_path = storage_path
self.memories = self.load_memories()
def load_memories(self):
"""加载历史记忆"""
try:
with open(self.storage_path, 'r', encoding='utf-8') as f:
return json.load(f)
except FileNotFoundError:
return {}
def save_memory(self, user_id, conversation_summary, key_points):
"""保存重要对话记忆"""
if user_id not in self.memories:
self.memories[user_id] = []
memory_entry = {
"timestamp": datetime.now().isoformat(),
"summary": conversation_summary,
"key_points": key_points
}
self.memories[user_id].append(memory_entry)
# 保持最近50条记忆
if len(self.memories[user_id]) > 50:
self.memories[user_id] = self.memories[user_id][-50:]
self.save_to_disk()
def save_to_disk(self):
"""保存到文件"""
with open(self.storage_path, 'w', encoding='utf-8') as f:
json.dump(self.memories, f, ensure_ascii=False, indent=2)
def get_user_memories(self, user_id, limit=5):
"""获取用户最近记忆"""
return self.memories.get(user_id, [])[-limit:]
# 使用示例：在对话结束时调用AI总结并保存
def end_conversation_and_save(user_id, messages):
"""结束对话并保存重要信息"""
# 让AI总结对话要点
summary_prompt = f"""请总结以下对话的要点：
对话记录：
{messages}
请提取：
1. 讨论的核心话题
2. 重要的决策或结论
3. 需要后续跟进的事项
总结："""
# 调用AI生成总结
summary = model.invoke(summary_prompt)
# 保存到长期记忆
memory_manager.save_memory(
user_id=user_id,
conversation_summary=summary.content,
key_points=["技术讨论", "React Native ExecuTorch"] # 实际中可以让AI自动提取关键词
)
return summary.content
四、完整项目：融合RAG与记忆的智能文档助手
现在，让我们将两大能力结合，创建一个功能完整的智能体。
class SmartDocumentAssistant:
"""智能文档助手：RAG + 长期记忆"""
def __init__(self, document_url):
# 初始化组件
self.vector_store = self.setup_rag(document_url)
self.memory_manager = LongTermMemoryManager()
self.model = ChatOpenAI(model="gpt-3.5-turbo")
# 构建对话图
self.workflow = self.build_workflow()
def setup_rag(self, document_url):
"""设置RAG系统"""
# 加载和分割文档
loader = WebBaseLoader(document_url)
docs = loader.load()
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, chunk_overlap=200
)
all_splits = text_splitter.split_documents(docs)
# 创建向量存储
embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/all-MiniLM-L6-v2"
)
vector_store = InMemoryVectorStore(embeddings)
vector_store.add_documents(documents=all_splits)
return vector_store
def build_workflow(self):
"""构建工作流"""
graph_builder = StateGraph(State)
# 添加增强的提问节点
graph_builder.add_node("smart_ask", self.smart_ask_with_memory)
graph_builder.set_entry_point("smart_ask")
graph_builder.add_edge("smart_ask", END)
# 编译带记忆的工作流
return graph_builder.compile(
checkpointer=InMemorySaver(),
store=InMemoryStore()
)
def smart_ask_with_memory(self, state: State) -> State:
"""智能提问：RAG + 记忆"""
user_id = "current_user" # 实际中从用户登录获取
# 1. 获取用户历史记忆
past_memories = self.memory_manager.get_user_memories(user_id)
memory_context = ""
if past_memories:
memory_context = "\n\n之前的对话要点：\n"
for mem in past_memories:
memory_context += f"- {mem['summary'][:100]}...\n"
# 2. 获取用户问题
user_query = input("\n💬 你的问题: ")
# 3. RAG检索
retrieved_docs = self.vector_store.similarity_search(user_query, k=3)
rag_context = "\n\n".join([doc.page_content for doc in retrieved_docs])
# 4. 构建综合提示
prompt = f"""{memory_context}
相关文档内容：
{rag_context}
用户当前问题：
{user_query}
请基于以上信息回答问题。如果文档中没有相关信息，请说明。"""
# 5. 调用AI
response = self.model.invoke(prompt)
print(f"\n🤖 助手: {response.content}")
# 6. 检查是否需要保存到长期记忆（示例逻辑）
if "重要" in user_query or "记住" in user_query:
print("(已将此对话标记为重要，会长期记住)")
# 实际中可以让AI判断重要性并调用 memory_manager.save_memory
return {"messages": [{"role": "user", "content": user_query},
{"role": "assistant", "content": response.content}]}
def chat(self, user_id="default_user"):
"""启动对话"""
config = {
"recursion_limit": 50,
"configurable": {"thread_id": user_id}
}
print("=" * 60)
print("智能文档助手已启动！")
print("我可以：1. 回答文档相关问题 2. 记住我们的重要对话")
print("输入 '退出' 结束对话")
print("=" * 60)
# 开始对话
self.workflow.invoke(
{"messages": [], "iteration": 0},
config=config
)
# 使用示例
if __name__ == "__main__":
# 创建助手（基于React Native ExecuTorch文档）
assistant = SmartDocumentAssistant(
"https://docs.swmansion.com/react-native-executorch/"
)
# 开始对话
assistant.chat("user_001")
五、避坑指南与最佳实践
在实现过程中，你可能会遇到一些常见问题，以下是一些解决方案：
RAG常见问题
-
检索不准确
- 问题：返回的文档片段与问题无关。
- 解决：调整
chunk_size（如500或1500），尝试不同的嵌入模型，或为文档块添加元数据（如标题、章节）进行过滤。
-
上下文过长
- 问题：检索到的内容太多，导致提示超出模型Token限制。
- 解决：采用“映射-归纳”策略，先让AI对每个检索片段做摘要，再基于摘要回答；或使用更智能的检索器，只返回最核心的片段。
长期记忆最佳实践
-
记忆总结策略
- 不要存储所有原始消息，这会导致存储膨胀和检索效率低下。定期（例如每5轮对话或对话结束时）让AI自动总结对话要点并保存。
- 可以按话题对记忆进行分类存储，便于按主题回顾。
-
隐私与用户体验
- 明确告知用户哪些信息会被长期记住。
- 务必提供让用户查看、管理（如删除特定记忆）和清除所有记忆的选项。
写在最后
通过本文的实践，我们成功地为基于 LangGraph 的 AI 智能体装备了两项关键能力。RAG 赋予了它查阅特定知识库的“外脑”，而长期记忆则让它拥有了持续成长的“经验库”。这不再是简单的聊天轮次，而是向着真正个性化、有深度的 AI 伙伴迈进了一步。
关键收获：
- RAG是增强而非替代：它在AI的通用能力之上，叠加了精准的领域知识查询功能。
- 记忆需要主动管理：智能的记忆总结与归档，远比无差别存储所有对话记录更重要。
- LangGraph提供了优雅的范式：其状态图和检查点机制，让复杂的多轮对话和状态管理变得清晰可控。
动手挑战：
如果你已掌握基础，可以尝试：
- 扩展RAG源，让智能体能同时查询多个不同来源的文档（如本地PDF、Confluence页面、数据库）。
- 实现一个记忆重要性自动评分系统，让AI能智能判断哪些对话值得存入长期记忆。
- 将记忆存储从JSON文件迁移到如 SQLite 或 PostgreSQL 等数据库中，实现更稳定、可查询的持久化。
构建更智能的 Agent 是一个持续探索的过程，希望本文的 Python 实战代码能为你提供一个坚实的起点。欢迎在 云栈社区 分享你在实现过程中遇到的独特挑战或巧妙的解决方案。
发表于 16 小时前
|
查看: 3|
回复: 0
