第 25 期 | 长程记忆：在 Graph 中保存长期偏好

更新于 2026/4/16

各位未来的 AI 架构师们，大家好！我是你们的老朋友，AI 技术导师。欢迎来到《LangGraph 多智能体专家课》第 25 期。

我们已经构建了一个相当强大的「AI 万能内容创作机构」。Planner 运筹帷幄，Researcher 博览群书，Writer 妙笔生花，Editor 更是我们内容的最后一道防线。但你有没有遇到这样的情况：同一个客户，每次都要重新交代他的品牌口吻、偏好术语、甚至是他个人对“幽默感”的独特定义？这就像你每次去理发店，都得从头开始解释你喜欢什么发型一样，效率低下，用户体验极差。

今天，我们要解决的就是这个痛点——给我们的 AI 编辑 Agent 装上“长程记忆”，让它能记住用户的历史偏好，真正做到“懂你”。这将是你的 AI 机构从“能干活”到“会来事儿”的关键一步。

🎯 本期学习目标

在本期课程中，你将不仅仅是学习技术，更要理解如何用技术解决实际业务问题。具体来说，你将收获：

  1. 理解长程记忆的核心价值： 为什么在多智能体系统中，短期记忆（Graph State）不足以支撑复杂的用户交互，长程记忆如何提升用户体验和系统效率。
  2. 掌握 LangGraph 中长程记忆的构建范式： 学习如何巧妙结合 VectorStore（向量数据库）来存储非结构化的用户偏好，并利用 LangGraph 的 Checkpointers 来管理会话状态和记忆指针。
  3. 实战 AI 编辑 Agent 的个性化记忆： 为我们的 Editor Agent 打造一个能够“记住”用户特定口吻、风格和历史反馈的工作流，让它成为一个真正贴心的“私人编辑”。
  4. 识别并规避长程记忆的常见陷阱： 深入探讨记忆膨胀、记忆冲突、检索精度等高阶问题，并提供实用的解决方案。

准备好了吗？让我们一起给 AI 装上“记忆芯片”！

📖 原理解析

在之前的课程中，我们主要依赖 LangGraph 的 Graph State 来维护智能体之间的短期对话和任务上下文。这就像人类的“工作记忆”，处理当前正在进行的任务。然而，一旦一个会话结束，或者需要跨越多个会话来记住用户的个性化需求，短期记忆就捉襟见肘了。

想象一下我们的 AI 编辑 Agent。一个客户可能喜欢幽默风趣的口吻，另一个则偏爱严谨专业的风格。如果每次编辑任务开始时，Editor Agent 都要重新“学习”这些偏好，那它就永远无法成为一个真正高效且个性化的服务者。这就是为什么我们需要引入“长程记忆”。

长程记忆，顾名思义，是能够持久化存储，并在需要时被检索和利用的信息。在 LangGraph 的语境下，我们通常通过两种机制的组合来实现它：

  1. VectorStore（向量数据库）： 这是一个存储非结构化数据（如文本、图片、音频等）并能进行语义检索的数据库。我们将用户的个性化偏好、品牌风格指南、历史反馈等信息，以文本的形式嵌入（Embed）成向量，然后存储在 VectorStore 中。当需要时，我们可以根据当前任务或用户ID进行语义相似性搜索，检索出最相关的偏好信息，将其作为上下文注入到 LLM 的提示词中。这解决了“记忆什么”和“如何检索”的问题。

  2. Checkpointers（检查点）： LangGraph 内置的 Checkpointer 机制允许我们将 Graph 的当前状态持久化到数据库（如 SQLite）中。它主要用于恢复中断的会话，或者在多次迭代中保持状态的连贯性。在本期中，我们将利用 Checkpointer 来保存每个用户的会话 ID（thread_id），以及可能指向 VectorStore 中特定记忆条目的索引或摘要。这解决了“记忆在哪里”和“如何关联”的问题。

核心思想是： Checkpointer 负责记住“这是谁的会话，以及他上次进展到哪了”，而 VectorStore 则负责存储“这个用户的所有长期偏好和历史经验”。当一个编辑任务到来时，Graph 首先通过 Checkpointer 识别用户，然后利用这个用户ID去 VectorStore 中检索其专属的长期偏好，最后将这些偏好作为额外上下文，指导 Editor Agent 进行个性化编辑。如果用户对编辑结果不满意并提供了反馈，这些反馈又可以被结构化或非结构化地添加到 VectorStore 中，从而“教会”AI 编辑 Agent 下次做得更好。

AI 编辑 Agent 的长程记忆工作流：

  1. 任务接收： AI Content Agency 接收到新的内容编辑请求，并附带 user_id 和 content_to_edit。
  2. 记忆检索（Retrieve Preferences）：
    • Graph 首先通过 Checkpointer 识别当前 thread_id 对应的 user_id。
    • 使用 user_id 作为查询条件，从 VectorStore 中检索该用户的所有历史偏好、风格指南和之前的反馈。
    • 将检索到的偏好信息整合成一个清晰的指令集。
  3. 个性化编辑（Editor Agent）：
    • 将原始内容和检索到的个性化偏好注入到 Editor Agent 的 LLM 提示词中。
    • LLM 根据这些指令进行内容润色和优化。
  4. 用户反馈（Process Feedback）：
    • 如果用户对编辑结果提供反馈（例如：“太正式了，要更活泼一点！”）。
    • Graph 接收反馈，并将其处理（可以是由另一个 LLM 总结，或者直接存储）。
    • 将处理后的反馈作为新的记忆条目，添加到 VectorStore 中，并与 user_id 关联。
  5. 记忆更新： Checkpointer 自动保存 Graph 的最新状态，包括任何可能更新的记忆指针或摘要。

通过这个循环，我们的 AI 编辑 Agent 将不再是“一次性”的工具，而是能够随着每次交互而不断学习和适应的“私人助理”。

Mermaid 图解核心架构

让我们通过一个 Mermaid 图来直观地理解这个流程：

graph TD
    A[用户提交编辑请求] --> B{识别用户/会话ID}
    B -- thread_id --> C(LangGraph Checkpointer)
    C -- user_id --> D[检索用户长期偏好]
    D -- 查询 --> E(VectorStore: 偏好/反馈)
    E -- 检索结果 --> D
    D -- 注入上下文 --> F[Editor Agent (LLM)]
    F --> G[生成编辑内容]
    G --> H{用户审核/反馈?}
    H -- 是 --> I[处理用户反馈]
    I -- 更新记忆 --> E
    I --> G'[[编辑完成/再次编辑]]'
    H -- 否 --> J[完成编辑，交付]

    subgraph LangGraph 工作流
        B --- F
        F --- I
    end

    style C fill:#f9f,stroke:#333,stroke-width:2px
    style E fill:#ccf,stroke:#333,stroke-width:2px
    style D fill:#afa,stroke:#333,stroke-width:2px
    style I fill:#afa,stroke:#333,stroke-width:2px

图例解析：

  • 用户提交编辑请求 (A): 工作的起点，包含内容和用户标识。
  • 识别用户/会话ID (B): LangGraph 内部机制，通过 configurable: thread_id 识别当前会话。
  • LangGraph Checkpointer (C): 持久化会话状态，保存 user_id 或其关联信息，确保会话连贯性。
  • 检索用户长期偏好 (D): 这是一个 LangGraph 节点，负责根据 user_id 从 VectorStore 中获取相关记忆。
  • VectorStore: 偏好/反馈 (E): 我们的长程记忆库，存储所有用户的个性化偏好、风格指南和历史反馈的向量表示。
  • Editor Agent (LLM) (F): 我们的核心编辑智能体，接收原始内容和检索到的个性化偏好进行编辑。
  • 生成编辑内容 (G): Editor Agent 的输出。
  • 用户审核/反馈? (H): 模拟用户对编辑结果的判断。
  • 处理用户反馈 (I): 另一个 LangGraph 节点，负责接收用户反馈，并将其转化为可存储的记忆，更新 VectorStore。
  • 完成编辑，交付 (J): 工作流的终点。

这个架构清晰地展示了 Checkpointer 和 VectorStore 如何协同工作，为我们的智能体提供强大的长程记忆能力。

💻 实战代码演练 (Agency 项目中的具体应用)

好了，理论说得再多，不如撸起袖子干一场。现在，我们将把上述原理转化为代码，为我们的 AI Content Agency 里的 Editor Agent 注入长程记忆。我们将使用 Python 和一些流行的 LangChain/LangGraph 组件。

1. 环境准备与依赖安装

首先，确保你的环境已安装必要的库：

pip install -U langchain-openai langgraph langchain_community chromadb tiktoken

2. 代码实现

import os
import sqlite3
from typing import TypedDict, Annotated, List, Literal
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter # 用于处理大文本
from langgraph.graph import StateGraph, END
from langgraph.checkpoint import MemorySaver, SqliteSaver # 用于持久化状态

# 设置你的 OpenAI API 密钥
# 建议通过环境变量设置，例如：export OPENAI_API_KEY="sk-..."
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY" # 请替换为你的真实密钥或通过环境变量设置

# 确保API密钥已设置
if not os.getenv("OPENAI_API_KEY"):
    raise ValueError("请设置环境变量 OPENAI_API_KEY")

print("--- 环境准备完成，API 密钥已加载 ---")

### 1. 定义 Graph 状态 ###
# TypedDict 用于定义 LangGraph 的状态模式，确保类型安全和清晰
class AgentState(TypedDict):
    """
    定义我们的多智能体工作流中的共享状态。
    这个状态会在所有节点之间传递。
    """
    content: str # 待编辑的原始内容 (Original content to be edited)
    user_id: str # 用户的唯一标识符，用于检索个性化偏好 (Unique user ID for retrieving preferences)
    preferences: str # 从 VectorStore 检索到的用户偏好 (User preferences retrieved from VectorStore)
    edited_content: str # AI 编辑后的内容 (Content after AI editing)
    feedback: str # 用户对编辑内容的反馈 (User feedback on the edited content)
    # 可以添加一个历史消息列表，用于短期对话上下文，但本期侧重长程记忆
    # chat_history: List[BaseMessage]

print("--- AgentState 定义完成 ---")

### 2. 初始化 VectorStore (长程记忆库) ###
# 我们使用 ChromaDB 作为向量数据库，用于存储用户偏好和历史反馈。
# 它支持本地持久化，方便演示。

# 初始化 OpenAI 嵌入模型，用于将文本转换为向量
embeddings = OpenAIEmbeddings()

# 初始化 ChromaDB。我们将数据持久化到本地文件系统，以便跨会话保持记忆。
# persist_directory 是存储数据库文件的路径
VECTOR_DB_PATH = "./chroma_db_editor_memory"
vectorstore = Chroma(embedding_function=embeddings, persist_directory=VECTOR_DB_PATH)

# 模拟一些初始的用户偏好文档
# 真实场景中，这些数据可能来自客户的品牌指南、风格手册或初期沟通
user_preferences_data = [
    {"user_id": "client_A", "preference": "口吻：活泼、幽默、简洁。避免冗长和过于正式的表达。", "doc_id": "pref_A_1"},
    {"user_id": "client_A", "preference": "关键词：创新、前沿技术、未来趋势。多用比喻和类比。", "doc_id": "pref_A_2"},
    {"user_id": "client_B", "preference": "口吻：专业、严谨、细节丰富。强调数据支持和逻辑推理。", "doc_id": "pref_B_1"},
    {"user_id": "client_B", "preference": "关键词：市场分析、投资回报、风险管理、合规性。避免主观判断。", "doc_id": "pref_B_2"},
    {"user_id": "client_C", "preference": "口吻：平易近人、通俗易懂。避免行业术语和复杂句式。目标读者是普通大众。", "doc_id": "pref_C_1"},
    {"user_id": "client_C", "preference": "关键词：健康生活、美食、旅行、亲子教育。内容要积极向上。", "doc_id": "pref_C_2"},
]

# 检查 VectorStore 是否已填充，避免重复添加
# 这是一个简单的检查，更严谨的做法是检查 doc_id 是否存在
existing_docs = vectorstore.get()
if not existing_docs["ids"]: # 如果数据库是空的，则填充
    for data in user_preferences_data:
        vectorstore.add_texts(
            texts=[data["preference"]],
            metadatas=[{"user_id": data["user_id"], "doc_id": data["doc_id"]}]
        )
    print(f"--- VectorStore 已初始化并填充 {len(user_preferences_data)} 条初始用户偏好。---")
else:
    print(f"--- VectorStore 已存在 {len(existing_docs['ids'])} 条数据，跳过初始填充。---")


### 3. 定义 Agent (LLM) 和 Graph 节点 ###

# 初始化 ChatOpenAI 模型，作为我们的编辑大脑
llm = ChatOpenAI(model="gpt-4o", temperature=0.5) # 温度设低一点，让编辑更稳定

# --- Graph 节点 1: 检索用户偏好 ---
def retrieve_preferences(state: AgentState):
    """
    根据 user_id 从 VectorStore 中检索用户的长期偏好和历史反馈。
    """
    user_id = state["user_id"]
    if not user_id:
        print("警告：未提供 user_id，将使用通用编辑模式。")
        return {"preferences": "无特定用户偏好。请以通用、专业且清晰的风格编辑。"}

    # 构造查询，尝试检索与该用户相关的偏好
    # 可以在这里加入更复杂的查询逻辑，例如结合当前内容类型进行检索
    query = f"用户 {user_id} 的内容创作偏好、风格要求以及历史反馈。"
    
    # 执行相似性搜索，并过滤出与当前 user_id 匹配的文档
    # LangChain 的 Chroma 默认不支持直接在 similarity_search 中过滤 metadata
    # 所以我们先检索，再在代码中过滤，或者使用 Chroma 的高级查询接口（如果需要）
    # 为了简化演示，我们假设检索结果中包含了足够多的相关信息，并会通过LLM进行提炼
    
    # 实际生产中，更好的做法是使用 Chroma 的 `similarity_search_with_score` 或 `similarity_search_by_vector`
    # 并结合 `where` 参数进行过滤，但这里为了演示，我们先检索再手动过滤
    
    # 检索所有相关文档，然后手动过滤
    results = vectorstore.similarity_search_with_score(query, k=5) # 检索前5个最相关的
    
    # 过滤出真正属于当前 user_id 的偏好
    filtered_preferences = []
    for doc, score in results:
        if doc.metadata.get("user_id") == user_id:
            filtered_preferences.append(doc.page_content)
            # print(f"  - 匹配偏好 (Score: {score:.2f}): {doc.page_content[:50]}...")
    
    preferences_str = "\n".join(filtered_preferences)
    if not preferences_str:
        preferences_str = "无特定用户偏好。请以通用、专业且清晰的风格编辑。"
        print(f"--- 未检索到用户 {user_id} 的特定偏好，使用通用模式。---")
    else:
        print(f"--- 成功检索到用户 {user_id} 的偏好：\n{preferences_str}\n---")
    
    return {"preferences": preferences_str}

# --- Graph 节点 2: AI 编辑器 ---
def editor_agent_node(state: AgentState):
    """
    AI 编辑器节点，根据原始内容和检索到的用户偏好进行编辑。
    """
    content = state["content"]
    preferences = state["preferences"]
    user_id = state["user_id"]

    prompt = f"""
    你是一位经验丰富、洞察力敏锐的资深内容编辑。
    你的任务是为用户 {user_id} 润色和优化以下内容。
    
    请务必严格遵循以下用户个性化偏好和风格要求进行编辑。
    这些偏好是用户长期积累的，代表了他们的品牌声音和期望。
    
    --- 用户偏好和风格要求 ---
    {preferences}
    ---
    
    原始