Agent 记忆系统设计：从短期对话到长期知识

与上午的「可观测性架构」形成互补：如果说可观测性解决的是”如何看见 Agent”，那么记忆系统解决的就是”如何让 Agent 记住”。

一、问题的本质：为什么 Agent 需要记忆？

想象一个场景：用户昨天告诉客服 Agent “我喜欢用支付宝付款”，今天再次对话时 Agent 却问”您想用微信还是银行卡？”——这种失忆体验会让用户抓狂。

传统 LLM 的局限：

• 无状态：每次请求都是独立的，不保留历史上下文
• 上下文窗口有限：即使保留对话历史，长对话也会溢出
• 无法跨会话学习：无法从过去互动中积累知识

Agent 记忆的三大价值：

1. 效率：避免重复询问已知信息
2. 个性化：基于用户历史提供定制服务
3. 能力进化：从经验中学习，持续改进

二、记忆系统的双轨架构

LangGraph 将记忆系统划分为两个互补的层级：

┌─────────────────────────────────────────────────────────────┐
│                    Agent 记忆系统架构                         │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌──────────────────┐         ┌──────────────────────────┐  │
│  │  短期记忆          │         │      长期记忆            │  │
│  │  (Short-term)     │         │    (Long-term)           │  │
│  ├──────────────────┤         ├──────────────────────────┤  │
│  │ • Thread-scoped   │         │ • Cross-thread           │  │
│  │ • 对话历史         │         │ • 用户画像               │  │
│  │ • 临时状态         │         │ • 经验积累               │  │
│  │ • 文件/检索结果    │         │ • 领域知识               │  │
│  ├──────────────────┤         ├──────────────────────────┤  │
│  │ Checkpointer      │         │ Memory Store             │  │
│  │ (自动保存状态)     │         │ (显式读写记忆)           │  │
│  └──────────────────┘         └──────────────────────────┘  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

2.1 短期记忆：对话的上下文

核心机制： 通过 checkpointer 自动保存和恢复图状态

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph

# 创建 checkpointer
checkpointer = MemorySaver()

# 编译图时绑定 checkpointer
graph = workflow.compile(checkpointer=checkpointer)

# 每次调用时指定 thread_id，自动管理对话历史
config = {"configurable": {"thread_id": "user-123-session-1"}}
result = graph.invoke({"messages": [user_input]}, config)

短期记忆存储的内容：

• 完整的消息历史（HumanMessage + AIMessage）
• 中间计算结果（检索到的文档、生成的草稿等）
• 用户上传的文件引用
• 当前对话的临时变量

关键特性：

# 1. 自动状态恢复
# 第二次调用时，自动加载之前的 state
result2 = graph.invoke({"messages": [new_input]}, config)  # 包含历史上下文

# 2. 随时中断和恢复
# 用户可以随时离开，回来后从断点继续
state = graph.get_state(config)  # 获取当前状态快照

# 3. 时间旅行调试
# 可以回到任意历史节点重新执行
graph.update_state(config, state_snapshot)  # 回溯到特定状态

2.2 长期记忆：跨会话的知识沉淀

核心机制： 通过 store 显式读写跨线程的记忆

from langgraph.store.memory import InMemoryStore

# 创建 store（生产环境使用 Redis/Postgres 后端）
store = InMemoryStore(
    index={
        "embed": embed_fn,  # 嵌入函数
        "dims": 1536        # 向量维度
    }
)

# 保存记忆：namespace + key + value
store.put(
    namespace=("user_profile", "user_123"),  # 命名空间
    key="preferences",                        # 记忆键
    value={
        "payment_method": "alipay",
        "language": "zh-CN",
        "timezone": "Asia/Shanghai"
    }
)

# 检索记忆
memories = store.search(
    namespace=("user_profile", "user_123"),
    query="payment preferences",  # 语义搜索
    limit=5
)

三、长期记忆的三重分类

借鉴人类记忆类型，AI Agent 的长期记忆可分为：

记忆类型	存储内容	人类类比	Agent 示例
语义记忆	事实、概念	学校里学的知识	用户偏好、产品规格
情景记忆	经验、事件	个人经历	成功案例、失败教训
程序记忆	规则、技能	骑自行车的能力	系统提示词、工作流

3.1 语义记忆：用户画像与事实

两种管理策略：

策略 A：统一画像（Profile）

# 单个 JSON 文档，持续更新
user_profile = {
    "user_id": "user_123",
    "preferences": {
        "communication_style": "concise",
        "technical_level": "expert",
        "interests": ["AI", "blockchain", "trading"]
    },
    "constraints": {
        "max_response_length": 500,
        "preferred_language": "zh-CN"
    }
}

# 更新策略：传入旧画像，生成新画像
def update_profile(old_profile, new_interaction):
    prompt = f"""
    基于旧画像：{old_profile}
    和新交互：{new_interaction}
    生成更新后的用户画像，保持相同 JSON 结构
    """
    return llm.invoke(prompt)

优点：结构清晰，易于提供给模型作为上下文
缺点：大画像更新时容易丢失信息

策略 B：文档集合（Collection）

# 多条独立记忆，语义检索
memories = [
    {"fact": "User prefers Alipay", "timestamp": "2026-01-15"},
    {"fact": "User is a Python expert", "timestamp": "2026-01-20"},
    {"fact": "User dislikes verbose explanations", "timestamp": "2026-02-01"}
]

# 检索时按语义相似度排序
relevant = store.search(
    namespace=("facts", "user_123"),
    query="payment method",  # 会匹配到第一条
    limit=3
)

优点：不易丢失信息，检索灵活
缺点：需要更多 tokens 提供给模型

3.2 情景记忆：从经验中学习

核心价值：让 Agent 记住”怎么做”而不是”做什么”

# 保存成功案例
store.put(
    namespace=("episodes", "coding_tasks"),
    key="task_001",
    value={
        "task": "优化 Python 数据处理脚本",
        "approach": "使用 pandas vectorization 替代循环",
        "result": "性能提升 100x",
        "code_snippet": "df['col'] = df['col'].apply(...)"
    }
)

# 检索相似任务的经验
similar_episodes = store.search(
    namespace=("episodes", "coding_tasks"),
    query="Python performance optimization",
    limit=3
)

# 用于 few-shot prompting
few_shot_examples = format_for_prompt(similar_episodes)

Few-shot 学习的记忆化：

def agent_with_episodic_memory(user_input, store):
    # 1. 检索相关经验
    episodes = store.search(
        namespace=("episodes", "user_123"),
        query=user_input,
        limit=5
    )
    
    # 2. 构建包含经验的提示词
    system_prompt = f"""
    你是编程助手。参考以下成功案例：
    
    {format_episodes(episodes)}
    
    用户问题：{user_input}
    """
    
    # 3. 调用模型
    return llm.invoke(system_prompt)

3.3 程序记忆：自我进化的指令

最激进的记忆形式：Agent 修改自己的系统提示词

# 初始系统提示词
initial_instructions = """
你是客服助手。请用礼貌、简洁的语言回答用户问题。
"""

# 保存到 store
store.put(
    namespace=("instructions", "customer_service_bot"),
    key="system_prompt",
    value={"instructions": initial_instructions}
)

def call_model(state, store):
    """使用存储的指令调用模型"""
    # 获取最新指令
    result = store.get(
        ("instructions", "customer_service_bot"),
        key="system_prompt"
    )
    instructions = result.value["instructions"]
    
    # 构建提示词
    prompt = f"{instructions}\n\n用户：{state['messages'][-1].content}"
    return llm.invoke(prompt)

def update_instructions(state, store):
    """基于反馈更新系统提示词"""
    # 获取当前指令
    current = store.get(
        ("instructions", "customer_service_bot"),
        key="system_prompt"
    )
    
    # 分析对话，提取改进建议
    feedback_prompt = f"""
    当前指令：{current.value['instructions']}
    
    最近对话：{state['messages']}
    
    用户反馈：{extract_feedback(state)}
    
    请生成优化后的系统提示词。
    """
    
    new_instructions = llm.invoke(feedback_prompt)
    
    # 保存新指令
    store.put(
        namespace=("instructions", "customer_service_bot"),
        key="system_prompt",
        value={"instructions": new_instructions}
    )

四、记忆写入的两种策略

4.1 热路径写入（Hot Path）

特点：在 Agent 响应用户前同步写入记忆

def agent_with_hotpath_memory(state, store):
    # 1. 检索现有记忆
    memories = store.search(namespace=("user", user_id), query=state["input"])
    
    # 2. 调用模型生成响应
    response = llm.invoke(build_prompt(state, memories))
    
    # 3. 同步写入新记忆（在返回前）
    store.put(
        namespace=("user", user_id),
        key=f"interaction_{timestamp}",
        value={
            "input": state["input"],
            "response": response.content,
            "extracted_facts": extract_facts(state["input"])
        }
    )
    
    return response

优点：

• 新记忆立即可用
• 用户可见记忆创建过程（透明度高）

缺点：

• 增加响应延迟
• Agent 需要多任务处理（回复 + 记忆）

ChatGPT 的实现方式：

# OpenAI 的 save_memories 工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "save_memories",
            "description": "Save important facts about the user",
            "parameters": {...}
        }
    }
]

# 模型自主决定是否保存记忆
response = openai.chat.completions.create(
    model="gpt-4",
    messages=messages,
    tools=tools  # 模型可选择调用 save_memories
)

4.2 后台写入（Background）

特点：异步处理记忆写入，不阻塞主流程

from langgraph.func import task

@task
async def background_memory_writer(thread_id, store):
    """后台任务：分析对话并写入记忆"""
    # 获取完整对话历史
    history = await get_conversation_history(thread_id)
    
    # 提取需要记忆的事实
    facts = await extract_facts_with_llm(history)
    
    # 批量写入
    for fact in facts:
        store.put(
            namespace=("user", user_id),
            key=f"fact_{fact['id']}",
            value=fact
        )

def agent_with_background_memory(state, store):
    # 1. 检索记忆（可能不包含最新对话的）
    memories = store.search(namespace=("user", user_id), query=state["input"])
    
    # 2. 快速响应（不等待记忆写入）
    response = llm.invoke(build_prompt(state, memories))
    
    # 3. 触发后台任务（不阻塞）
    background_memory_writer.submit(state["thread_id"], store)
    
    return response

触发策略：

# 策略 1：定时触发（每 5 分钟）
@cron.schedule(interval="5m")
def periodic_memory_consolidation():
    recent_conversations = get_recent_chats(minutes=5)
    for conv in recent_conversations:
        extract_and_save_memories(conv)

# 策略 2：对话结束后触发
def on_conversation_end(thread_id):
    background_memory_writer.submit(thread_id)

# 策略 3：手动触发（用户点击"记住这个"）
def user_triggered_memory_save(content, store):
    store.put(namespace=("user", user_id), key=f"saved_{timestamp}", value=content)

对比：

维度	热路径	后台写入
延迟	高（同步写入）	低（异步）
即时性	新记忆立即可用	可能延迟几秒到几分钟
复杂度	简单	需要任务队列
一致性	强	最终一致
适用场景	关键事实必须记住	大批量记忆处理

五、生产级记忆系统架构

5.1 完整架构图

┌─────────────────────────────────────────────────────────────────────────────┐
│                              Agent Memory System                             │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  ┌──────────────────────────────────────────────────────────────────────┐  │
│  │                         Short-term Memory Layer                       │  │
│  │  ┌─────────────────┐    ┌─────────────────┐    ┌──────────────────┐  │  │
│  │  │   MemorySaver   │    │  Redis Checkpt  │    │ Postgres Checkpt │  │  │
│  │  │   (in-memory)   │    │   (distributed) │    │   (persistent)   │  │  │
│  │  └─────────────────┘    └─────────────────┘    └──────────────────┘  │  │
│  │                           thread_id + checkpoint_id                   │  │
│  └──────────────────────────────────────────────────────────────────────┘  │
│                                    │                                        │
│                                    ▼                                        │
│  ┌──────────────────────────────────────────────────────────────────────┐  │
│  │                          Long-term Memory Layer                       │  │
│  │  ┌────────────────────────────────────────────────────────────────┐  │  │
│  │  │                      Memory Store Interface                      │  │  │
│  │  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────┐  │  │  │
│  │  │  │  InMemory    │  │    Redis     │  │  Postgres + pgvector │  │  │  │
│  │  │  │   Store      │  │    Store     │  │      Store           │  │  │  │
│  │  │  └──────────────┘  └──────────────┘  └──────────────────────┘  │  │  │
│  │  └────────────────────────────────────────────────────────────────┘  │  │
│  │                         namespace + key + vector                     │  │
│  └──────────────────────────────────────────────────────────────────────┘  │
│                                    │                                        │
│                                    ▼                                        │
│  ┌──────────────────────────────────────────────────────────────────────┐  │
│  │                         Memory Management Logic                       │  │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  ┌──────────┐  │  │
│  │  │   Semantic   │  │   Episodic   │  │  Procedural  │  │  Write   │  │  │
│  │  │   Memory     │  │   Memory     │  │   Memory     │  │ Strategy │  │  │
│  │  └──────────────┘  └──────────────┘  └──────────────┘  └──────────┘  │  │
│  └──────────────────────────────────────────────────────────────────────┘  │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

5.2 完整实现代码

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore
from langchain_openai import OpenAIEmbeddings
import psycopg

# ============ 状态定义 ============
class AgentState(TypedDict):
    messages: Annotated[List, "对话消息历史"]
    user_id: str
    thread_id: str
    extracted_facts: List[str]
    retrieved_memories: List[dict]

# ============ 存储初始化 ============
def init_stores(connection_string: str):
    """初始化短期和长期记忆存储"""
    
    # 短期记忆：Postgres Checkpointer
    conn = psycopg.connect(connection_string)
    checkpointer = PostgresSaver(conn)
    checkpointer.setup()  # 创建必要的表
    
    # 长期记忆：Postgres Store（支持向量搜索）
    embeddings = OpenAIEmbeddings()
    store = PostgresStore(
        connection_string,
        index_config={
            "dims": 1536,
            "embed": embeddings.embed_query,
            "distance": "cosine"
        }
    )
    
    return checkpointer, store

# ============ 记忆检索节点 ============
def retrieve_memories(state: AgentState, store: PostgresStore):
    """检索相关长期记忆"""
    user_id = state["user_id"]
    last_message = state["messages"][-1].content
    
    # 语义搜索：用户画像
    profile_results = store.search(
        namespace=("profile", user_id),
        query=last_message,
        limit=3
    )
    
    # 语义搜索：历史经验
    episode_results = store.search(
        namespace=("episodes", user_id),
        query=last_message,
        limit=3
    )
    
    # 获取系统指令（程序记忆）
    instructions = store.get(
        namespace=("instructions", "default"),
        key="system_prompt"
    )
    
    state["retrieved_memories"] = {
        "profile": profile_results,
        "episodes": episode_results,
        "instructions": instructions.value if instructions else None
    }
    
    return state

# ============ 模型调用节点 ============
def call_model(state: AgentState):
    """调用 LLM 生成响应"""
    memories = state["retrieved_memories"]
    
    # 构建增强提示词
    system_prompt = build_enhanced_prompt(memories)
    
    messages = [{"role": "system", "content": system_prompt}] + [
        {"role": m.type, "content": m.content} for m in state["messages"]
    ]
    
    response = llm.invoke(messages)
    state["messages"].append(response)
    
    return state

# ============ 记忆提取节点 ============
def extract_facts(state: AgentState):
    """从对话中提取事实"""
    conversation = format_conversation(state["messages"][-4:])  # 最近4轮
    
    extraction_prompt = f"""
    从以下对话中提取值得长期保存的事实（用户偏好、重要信息等）。
    以 JSON 列表格式返回，每条包含 "fact" 和 "category" 字段。
    如果没有值得保存的事实，返回空列表 []。
    
    对话：
    {conversation}
    """
    
    result = llm.invoke(extraction_prompt)
    state["extracted_facts"] = parse_json(result.content)
    
    return state

# ============ 记忆写入节点 ============
def save_memories(state: AgentState, store: PostgresStore):
    """保存提取的事实到长期记忆"""
    user_id = state["user_id"]
    
    for fact in state["extracted_facts"]:
        store.put(
            namespace=("profile", user_id),
            key=f"fact_{generate_id()}",
            value={
                "fact": fact["fact"],
                "category": fact["category"],
                "timestamp": datetime.now().isoformat()
            }
        )
    
    return state

# ============ 构建工作流 ============
def build_agent(checkpointer, store):
    """构建带记忆系统的 Agent"""
    
    workflow = StateGraph(AgentState)
    
    # 添加节点
    workflow.add_node("retrieve", lambda s: retrieve_memories(s, store))
    workflow.add_node("agent", call_model)
    workflow.add_node("extract", extract_facts)
    workflow.add_node("save", lambda s: save_memories(s, store))
    
    # 定义边
    workflow.set_entry_point("retrieve")
    workflow.add_edge("retrieve", "agent")
    workflow.add_edge("agent", "extract")
    workflow.add_edge("extract", "save")
    workflow.add_edge("save", END)
    
    # 编译（绑定 checkpointer 实现短期记忆）
    return workflow.compile(checkpointer=checkpointer)

# ============ 使用示例 ============
connection_string = "postgresql://user:pass@localhost/agent_db"
checkpointer, store = init_stores(connection_string)
agent = build_agent(checkpointer, store)

# 第一次对话
config = {"configurable": {"thread_id": "session_001", "user_id": "user_123"}}
result1 = agent.invoke({
    "messages": [{"role": "user", "content": "我喜欢用支付宝付款"}],
    "user_id": "user_123",
    "thread_id": "session_001"
}, config)

# 第二次对话（短期记忆自动恢复，长期记忆包含"支付宝偏好"）
result2 = agent.invoke({
    "messages": [{"role": "user", "content": "我要买这个"}]
}, config)  # 自动使用支付宝作为默认支付方式

六、记忆系统的挑战与对策

6.1 记忆冲突

问题：新旧记忆矛盾（用户之前说喜欢支付宝，现在说更喜欢微信）

对策：

def resolve_memory_conflicts(new_fact, existing_memories):
    """检测并解决记忆冲突"""
    
    # 1. 语义相似度检测
    similar = [m for m in existing_memories 
               if semantic_similarity(m["fact"], new_fact["fact"]) > 0.8]
    
    if similar:
        # 2. 时间戳判断（新记忆优先）
        if new_fact["timestamp"] > similar[0]["timestamp"]:
            # 标记旧记忆为过期
            store.update(
                namespace=("profile", user_id),
                key=similar[0]["key"],
                value={**similar[0]["value"], "status": "deprecated"}
            )
    
    return new_fact

6.2 记忆膨胀

问题：长期记忆过多，检索效率下降

对策：

def memory_maintenance(store, user_id):
    """定期清理和压缩记忆"""
    
    # 1. 删除过期记忆
    old_memories = store.search(
        namespace=("profile", user_id),
        filter={"timestamp": {"lt": "2025-01-01"}}
    )
    for m in old_memories:
        store.delete(namespace=("profile", user_id), key=m.key)
    
    # 2. 压缩相似记忆
    similar_groups = cluster_similar_memories(store, user_id)
    for group in similar_groups:
        compressed = summarize_memories(group)
        store.put(namespace=("profile", user_id), key=f"summary_{id}", value=compressed)
        for m in group:
            store.delete(namespace=("profile", user_id), key=m.key)

6.3 隐私与安全

问题：敏感信息泄露（密码、个人身份信息）

对策：

def sanitize_memory(fact: str) -> str:
    """敏感信息过滤"""
    
    # PII 检测
    pii_patterns = [
        (r"\b\d{18}\b", "[ID_NUMBER]"),      # 身份证号
        (r"\b1[3-9]\d{9}\b", "[PHONE]"),     # 手机号
        (r"\b\d{16,19}\b", "[CARD_NUMBER]"), # 银行卡号
    ]
    
    sanitized = fact
    for pattern, replacement in pii_patterns:
        sanitized = re.sub(pattern, replacement, sanitized)
    
    # 敏感词检测
    if contains_sensitive_keywords(sanitized):
        return None  # 拒绝保存
    
    return sanitized

七、最佳实践与 Checklist

7.1 记忆设计黄金法则

1. 分层存储：短期记忆放 checkpointer，长期记忆放 store
2. 显式优于隐式：让用户知道你在记住什么
3. 可遗忘：支持用户删除特定记忆
4. 可解释：能回答”为什么这样回复”（基于哪些记忆）
5. 渐进增强：没有记忆时也能正常工作

7.2 生产部署 Checklist

• 选择合适的存储后端（开发：内存，生产：Postgres/Redis）
• 配置向量索引支持语义搜索
• 实现记忆写入策略（热路径 vs 后台）
• 添加记忆冲突检测和解决
• 实施敏感信息过滤
• 设置记忆清理和压缩策略
• 监控记忆检索延迟和命中率
• 提供用户管理记忆的界面

7.3 与可观测性的结合

# 在 LangSmith 中追踪记忆使用
@traceable(run_type="memory")
def retrieve_memories(query, store):
    results = store.search(query)
    
    # 记录检索到的记忆（用于调试）
    return {
        "output": results,
        "metadata": {
            "query": query,
            "num_results": len(results),
            "top_score": results[0].score if results else 0
        }
    }

八、总结

Agent 记忆系统的核心在于分层和分类：

1. 短期记忆解决”对话连续性”问题——通过 checkpointer 自动管理
2. 长期记忆解决”知识积累”问题——通过 store 显式读写
3. 语义/情景/程序三种记忆类型覆盖不同应用场景
4. 热路径/后台两种写入策略权衡延迟和一致性

与上午的「可观测性架构」结合，构成完整 Agent 系统：

• 可观测性让你知道 Agent 在做什么
• 记忆系统让 Agent 记得之前做过什么

两者相辅相成，缺一不可。

---

参考资源：

• LangGraph Memory Concepts
• LangGraph Persistence Guide
• CoALA: Cognitive Architectures for Language Agents

本文是「Agent 架构系列」的第二篇，第一篇《Agent 可观测性架构：从黑箱到白盒的监控革命》已发布。