必学干货！用LangGraph为AI助手添加知识库+长期记忆，小白也能学会，建议收藏！

模型优化师

314人浏览 · 2026-03-24 10:09:02

模型优化师 · 2026-03-24 10:09:02 发布

你的AI助手总是“一问三不知”？今天，我们要给它装上知识库和长期记忆，让它真正成为你的得力伙伴！

你是否遇到过这样的尴尬场景：

向AI询问公司内部文档的内容，它却回答“我不知道”？
与AI进行多轮对话后重启程序，之前的对话历史全部丢失？
每次都要向AI重复背景信息，感觉像是在跟金鱼聊天？

如果你也有这些烦恼，那么今天这篇文章就是为你准备的！我们将基于LangGraph，为智能体添加两大核心能力：**检索增强生成（RAG）**和 长期记忆。学完本文，你的AI助手将能够：

从指定文档中查找答案，不再局限于预训练知识
记住跨会话的对话历史，实现真正的连续对话

一、什么是RAG与长期记忆？

在深入代码之前，让我们快速理解这两个核心概念：

RAG（检索增强生成）

想象一下，你有一个超级聪明的助手，但它的知识只停留在某个时间点。RAG就像给了它一个智能书架：当用户提问时，助手先快速浏览书架上的相关书籍（检索），然后结合自己的知识回答问题（生成）。

传统AI的局限性：

# 传统方式：只依赖预训练知识user_question = "我们公司最新的报销政策是什么？"# AI: "抱歉，我不知道你们公司的具体政策..."

RAG的优势：

# RAG方式：结合文档知识user_question = "我们公司最新的报销政策是什么？"# 1. 从公司文档库中检索相关政策文档# 2. 基于检索到的文档生成答案# AI: "根据2024年3月发布的《费用报销管理办法》第5条..."

长期记忆

如果说短期记忆是对话中记住上下文，那么长期记忆就是记住不同会话间的对话历史。这就像给AI装了一个“记忆宫殿”，让它能记住昨天、上周甚至上个月的对话内容。

二、为LangGraph智能体添加RAG能力

让我们一步步构建一个能理解特定文档的智能助手。我们将以React Native ExecuTorch文档为例，让AI学会回答相关技术问题。

步骤1：加载和分割文档

首先，我们需要获取文档内容并将其分割成适合处理的“知识块”：

from langchain_community.document_loaders import WebBaseLoaderfrom 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：每个文本块约1000字符，既不会太碎片化，也不会太冗长
chunk_overlap=200：相邻块重叠200字符，确保相关信息不会因为恰好跨块而被割裂

步骤2：创建向量数据库

现在我们需要一个能快速查找相关文本的“书架”：

from langchain_core.vectorstores import InMemoryVectorStorefrom 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("向量数据库准备就绪！")

什么是嵌入（Embedding）？简单来说，嵌入就是将文本转换成数学向量的过程。相似的文本会有相似的向量表示，这样我们就能通过数学计算找到语义上相关的文档。

步骤3：改造提问函数，集成RAG

现在让我们的智能体学会先查资料再回答：

from langchain_core.messages import HumanMessage, AIMessagedef 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, Listfrom langchain_core.messages import BaseMessagefrom langgraph.graph import StateGraph, ENDfrom langchain_openai import ChatOpenAIimport 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记住跨会话的对话？

步骤1：设置记忆存储系统

from langgraph.checkpoint.memory import InMemorySaverfrom langgraph.store.memory import InMemoryStore# 1. 创建检查点保存器 - 保存单个对话状态checkpointer = InMemorySaver()# 2. 创建存储 - 跨线程/会话存储状态store = InMemoryStore()# 3. 编译时集成记忆系统workflow = graph.compile(    checkpointer=checkpointer,  # 记住每次对话的状态    store=store                 # 跨会话记忆)

步骤2：使用会话ID管理对话

# 为每个对话会话设置唯一IDconfig = {    "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 jsonfrom datetime import datetimeclass 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 notin 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:]# 使用示例memory_manager = LongTermMemoryManager()# 对话结束时保存记忆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判断重要性                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")