文章目录

前言：为什么我会学这个问题

我们都会遇见这个问题 Agent 第一轮对话很聪明，但第二轮就"失忆"了——用户问"那刚才说的那个功能怎么开通"，Agent 完全不知道在说什么。

这让我意识到一个问题：LLM 本身是没有记忆的，每次调用都是独立的一次性请求。对话历史需要我们主动管理。

前文关联：本文接续《Planning Agent 和 Action Agent 有什么区别？》，在前文理解不同 Agent 架构的基础上，继续探讨所有 Agent 都会遇到的核心问题——如何让 Agent 在多轮对话中记住之前的内容。

一、问题背景

为什么 LLM 没有记忆

LLM 的运作机制是"输入-输出"。每次调用时，模型只会看到这一次输入的内容，不会自动保留之前的对话记录。

调用 1：
User: 你好               → LLM: 你好，有什么可以帮你？

调用 2：
User: 你是谁             → LLM: 我是一个 AI 助手
    
调用 3（模型"失忆"了）：
User: 刚才我说什么了？    → LLM: 抱歉，我不知道你刚才说了什么

记忆管理的挑战

  1. Token 限制：LLM 有上下文窗口限制（4K、16K、128K），对话太长装不下
  2. 成本问题：历史越长，Token 消耗越多，成本越高
  3. 相关性：不是所有历史都有用，需要筛选
  4. 隐私：对话历史可能包含敏感信息，不能全部保留

二、核心概念

对话历史的结构

在这里插入图片描述

记忆管理的三种策略

1. 完整历史（Full History）

  • 把所有对话都放进上下文
  • 简单但不可扩展
  • 适合短对话

2. 摘要历史（Summarized History）

  • 定期把对话压缩成摘要
  • 省 Token，但会丢失细节
  • 适合中等长度对话

3. 选择性记忆（Selective Memory）

  • 只保留关键信息
  • 通过向量检索相关历史
  • 适合长对话

三、错误理解 / 常见误区

误区 1：给 LLM 越多历史越好

错。过多无关历史会稀释关键信息，还浪费 Token。LLM 也可能受到早期信息的"锚定效应"影响。

误区 2：摘要会丢失所有重要信息

不一定。好的摘要策略会保留"关键实体"、“用户意图”、"已完成的任务"等核心信息。

误区 3：向量检索适合所有场景

向量检索适合"需要从大量历史中找相关信息"的场景。对于短对话或连续对话，简单的滑动窗口可能更高效。

误区 4：不需要管理历史

错。无限增长的历史会导致：1) Token 超限 2) 成本爆炸 3) 模型困惑。记忆管理是必须考虑的工程问题。

四、正确实现思路

策略选择决策树

对话长度 < 1/4 上下文窗口？
├── 是：直接用完整历史
└── 否：需要压缩
    对话有明确主线？
    ├── 是：用摘要策略
    └── 否：用向量检索策略

分层记忆架构

在这里插入图片描述

五、代码示例

基础版：固定窗口历史

class SimpleHistoryAgent:
    """简单的固定窗口历史管理"""
    
    def __init__(self, llm, max_messages: int = 10):
        self.llm = llm
        self.max_messages = max_messages  # 保留最近 N 条消息
        self.history = []
    
    def chat(self, user_input: str) -> str:
        """处理对话"""
        # 1. 添加用户消息
        self.history.append({"role": "user", "content": user_input})
        
        # 2. 保留最近 N 条消息
        if len(self.history) > self.max_messages:
            self.history = self.history[-self.max_messages:]
        
        # 3. 构建消息列表（加上系统消息）
        messages = [{"role": "system", "content": "你是一个有用的助手"}]
        messages.extend(self.history)
        
        # 4. 调用 LLM
        response = self.llm(messages)
        
        # 5. 添加助手回复到历史
        self.history.append({"role": "assistant", "content": response})
        
        return response

进阶版：自动摘要历史

import json
from typing import List, Dict


class SummarizingHistoryAgent:
    """带自动摘要的记忆管理"""
    
    def __init__(self, llm, max_messages: int = 20, summary_threshold: int = 15):
        self.llm = llm
        self.max_messages = max_messages
        self.summary_threshold = summary_threshold
        self.history = []
        self.summary = ""  # 对话摘要
    
    def chat(self, user_input: str) -> str:
        """处理对话"""
        # 添加用户消息
        self.history.append({"role": "user", "content": user_input})
        
        # 检查是否需要摘要
        if len(self.history) >= self.summary_threshold:
            self._summarize_history()
        
        # 构建消息
        messages = self._build_messages()
        response = self.llm(messages)
        
        # 添加助手回复
        self.history.append({"role": "assistant", "content": response})
        
        return response
    
    def _summarize_history(self):
        """生成对话摘要"""
        # 用 LLM 生成摘要
        prompt = f"""请总结以下对话的要点：

{chr(10).join([f"{m['role']}: {m['content']}" for m in self.history])}

要求：
- 保留关键实体（人名、地点、产品名等）
- 保留用户的主要意图和需求
- 保留已确定的信息和结论
- 用简洁的语言总结
"""
        self.summary = self.llm([{"role": "user", "content": prompt}])
        
        # 清空历史，只保留摘要
        self.history = [{"role": "system", "content": f"对话摘要：{self.summary}"}]
        print(f"📝 生成摘要：{self.summary[:50]}...")
    
    def _build_messages(self) -> List[Dict]:
        """构建发送给 LLM 的消息列表"""
        messages = [{"role": "system", "content": "你是一个有用的助手。"}]
        
        # 如果有摘要，添加摘要作为上下文
        if self.summary:
            messages.append({
                "role": "system", 
                "content": f"之前的对话摘要：{self.summary}"
            })
        
        messages.extend(self.history)
        return messages

高级版：向量检索记忆

import hashlib
from typing import List, Dict, Tuple, Optional


class VectorMemoryAgent:
    """基于向量检索的长期记忆管理"""
    
    def __init__(self, llm, vector_store, max_context_messages: int = 10):
        self.llm = llm
        self.vector_store = vector_store  # 向量数据库
        self.max_context_messages = max_context_messages
        self.conversation_history = []  # 当前对话
        self.current_session_id = self._new_session_id()
    
    def chat(self, user_input: str) -> str:
        """处理对话"""
        # 1. 添加用户消息
        self.conversation_history.append({
            "role": "user",
            "content": user_input,
            "session_id": self.current_session_id
        })
        
        # 2. 保存用户消息到向量存储
        self._add_to_memory(user_input, "user")
        
        # 3. 检索相关记忆
        relevant_memories = self._retrieve_memories(user_input)
        
        # 4. 构建上下文
        messages = self._build_context(relevant_memories)
        
        # 5. 调用 LLM
        response = self.llm(messages)
        
        # 6. 保存助手回复
        self.conversation_history.append({
            "role": "assistant",
            "content": response
        })
        self._add_to_memory(response, "assistant")
        
        return response
    
    def _add_to_memory(self, content: str, role: str):
        """添加内容到向量记忆"""
        self.vector_store.add(
            text=content,
            metadata={
                "role": role,
                "session_id": self.current_session_id
            }
        )
    
    def _retrieve_memories(self, query: str, top_k: int = 3) -> List[Dict]:
        """检索相关记忆"""
        results = self.vector_store.search(query, top_k=top_k)
        return results
    
    def _build_context(self, memories: List[Dict]) -> List[Dict]:
        """构建上下文"""
        messages = [{"role": "system", "content": "你是一个有用的助手。"}]
        
        # 添加相关记忆
        if memories:
            memory_context = "以下是相关的历史记忆：\n"
            for m in memories:
                memory_context += f"- [{m['metadata'].get('role', 'unknown')}]: {m['text']}\n"
            messages.append({"role": "system", "content": memory_context})
        
        # 添加当前对话历史（限制数量）
        messages.extend(self.conversation_history[-self.max_context_messages:])
        
        return messages
    
    def _new_session_id(self) -> str:
        """生成新的会话 ID"""
        import time
        return hashlib.md5(str(time.time()).encode()).hexdigest()[:8]
    
    def start_new_session(self):
        """开始新会话"""
        self.conversation_history = []
        self.current_session_id = self._new_session_id()


# 简单的向量存储实现（用于演示）
class SimpleVectorStore:
    """简化的向量存储（实际生产用 Milvus/Pinecone 等）"""
    
    def __init__(self):
        self.documents = []
    
    def add(self, text: str, metadata: dict = None):
        """添加文档"""
        # 简化：用文本长度做伪向量
        self.documents.append({
            "text": text,
            "metadata": metadata or {},
            "vector": self._simple_vector(text)
        })
    
    def search(self, query: str, top_k: int = 3) -> List[Dict]:
        """搜索相关文档（简化实现）"""
        query_vec = self._simple_vector(query)
        
        # 计算相似度（简化：向量点积）
        results = []
        for doc in self.documents:
            similarity = sum(q * d for q, d in zip(query_vec, doc["vector"]))
            results.append((similarity, doc))
        
        # 返回 top_k
        results.sort(key=lambda x: x[0], reverse=True)
        return [r[1] for r in results[:top_k]]
    
    def _simple_vector(self, text: str) -> List[float]:
        """简化向量化：用字符编码和"""
        return [ord(c) / 255.0 for c in text[:50]]

生产级完整实现

from datetime import datetime
from typing import Optional


class ProductionMemoryAgent:
    """生产级记忆 Agent"""
    
    def __init__(
        self,
        llm,
        vector_store=None,
        max_session_messages: int = 50,
        max_context_tokens: int = 4000,
        summarize_threshold: int = 20
    ):
        self.llm = llm
        self.vector_store = vector_store
        
        # 配置
        self.max_session_messages = max_session_messages
        self.max_context_tokens = max_context_tokens
        self.summarize_threshold = summarize_threshold
        
        # 记忆层次
        self.short_term = []      # 当前对话的原始消息
        self.summary = ""          # 压缩后的摘要
        self.long_term = []        # 持久化的关键记忆
        
        # 元信息
        self.session_id = self._new_session_id()
        self.created_at = datetime.now()
    
    def chat(self, user_input: str, user_id: str = None) -> str:
        """完整的对话处理"""
        
        # 1. 记录用户消息
        user_msg = {
            "role": "user",
            "content": user_input,
            "timestamp": datetime.now().isoformat(),
            "user_id": user_id
        }
        self.short_term.append(user_msg)
        
        # 2. 决定使用哪个记忆层次
        context = self._build_context()
        
        # 3. 调用 LLM
        response = self.llm(context)
        
        # 4. 记录助手回复
        assistant_msg = {
            "role": "assistant",
            "content": response,
            "timestamp": datetime.now().isoformat()
        }
        self.short_term.append(assistant_msg)
        
        # 5. 更新记忆
        self._update_memory()
        
        # 6. 如果配置了向量存储，持久化关键信息
        if self.vector_store and self._is_key_information(user_input):
            self._persist_to_long_term(user_input, response)
        
        return response
    
    def _build_context(self) -> list:
        """构建上下文"""
        messages = [
            {"role": "system", "content": "你是一个专业的客服助手。"}
        ]
        
        # 添加长期记忆（摘要）
        if self.summary:
            messages.append({
                "role": "system",
                "content": f"对话摘要：{self.summary}"
            })
        
        # 添加长期记忆（关键事实）
        if self.long_term:
            facts = "用户关键信息：\n"
            for fact in self.long_term[-5:]:  # 最近 5 条
                facts += f"- {fact}\n"
            messages.append({"role": "system", "content": facts})
        
        # 添加短期记忆（最近的消息）
        messages.extend(self.short_term[-self.max_session_messages:])
        
        return messages
    
    def _update_memory(self):
        """更新记忆"""
        # 如果短期记忆太长，生成摘要
        if len(self.short_term) >= self.summarize_threshold:
            self._generate_summary()
    
    def _generate_summary(self):
        """生成对话摘要"""
        messages = [
            {"role": "system", "content": "你是一个对话摘要专家。"},
            {"role": "user", "content": f"请总结以下对话的要点：\n\n{self._format_history(self.short_term)}"}
        ]
        
        self.summary = self.llm(messages)
        
        # 保留摘要和最近几轮对话
        self.short_term = [{"role": "system", "content": f"摘要：{self.summary}"}] + self.short_term[-6:]
        print(f"📝 记忆压缩：生成新摘要")
    
    def _is_key_information(self, text: str) -> bool:
        """判断是否为关键信息"""
        keywords = ["地址", "电话", "订单号", "账号", "密码", "银行卡", "邮箱"]
        return any(kw in text for kw in keywords)
    
    def _persist_to_long_term(self, user_input: str, response: str):
        """持久化关键信息到长期记忆"""
        # 简化：直接存储文本
        # 实际应该用 LLM 提取实体和关系
        if "地址" in user_input:
            self.long_term.append(f"用户地址：{user_input}")
        if "订单" in user_input or "订单号" in user_input:
            self.long_term.append(f"订单相关信息：{user_input}")
    
    def _format_history(self, history: list) -> str:
        """格式化历史消息"""
        return "\n".join([
            f"{msg['role']}: {msg['content']}"
            for msg in history[-20:]  # 最近 20 条
        ])
    
    def _new_session_id(self) -> str:
        """生成会话 ID"""
        import hashlib
        import time
        return hashlib.md5(str(time.time()).encode()).hexdigest()[:8]
    
    def reset(self):
        """重置记忆（新会话）"""
        self.short_term = []
        self.summary = ""
        self.session_id = self._new_session_id()
        self.created_at = datetime.now()

六、运行结果 / 流程图

运行效果示意：

用户: 我想查询订单号 12345 的状态
助手: 正在为您查询...

用户: 那这个订单预计什么时候到？
助手: 根据查询结果，您的订单预计明天送达。

用户: 刚才说的订单号是多少？
助手: 您刚才查询的是订单号 12345。

记忆管理流程：

长期记忆 摘要记忆 短期记忆 Agent 用户 长期记忆 摘要记忆 短期记忆 Agent 用户 alt [短期记忆有空间] [短期记忆满了] 如果是关键信息 用户输入 添加消息 生成摘要 保留摘要 + 最近消息 提取相关长期记忆 构建完整上下文 生成回复 持久化存储

七、项目中怎么用

实际选型建议

场景 推荐方案
简单客服（10 轮以内） 固定窗口历史
复杂咨询（需要跨会话） 摘要 + 长期记忆
个人助手（需要记住用户偏好） 向量检索记忆
高价值场景（金融、医疗） 分层记忆 + 审计日志

注意事项

  1. 隐私合规：对话历史可能包含敏感信息，需要脱敏处理
  2. Token 预算：始终监控 Token 消耗，设置上限
  3. 记忆一致性：多轮对话中，同一实体应该保持一致
  4. 降级策略：向量服务不可用时，降级到简单历史管理

系列关联：本文探讨了 Agent 的对话历史管理问题，这是构建实用 Agent 的基础技术。下一篇《Multi-Agent 协作怎么设计？》会讲到，当单个 Agent 能力有限时，如何通过多 Agent 协作来完成任务，每个 Agent 可能有不同的记忆配置。

八、总结 + 下篇预告

核心要点

  1. LLM 没有记忆：每次调用是独立的，需要主动管理历史
  2. 三种策略：固定窗口（简单）、摘要压缩（中等）、向量检索（复杂）
  3. 分层记忆：短期记忆 + 摘要 + 长期记忆，各有用途
  4. 隐私优先：关键信息脱敏，敏感数据不上传

下篇预告

下一篇我们将探讨《Multi-Agent 协作怎么设计？》，理解当单个 Agent 无法完成复杂任务时，如何设计多 Agent 协作架构，包括 Agent 间的通信、任务分解和结果聚合。

延伸阅读

  • LangChain ConversationMemory
  • RAG（检索增强生成）：结合向量检索和 LLM 生成
  • MemGPT：类似操作系统的多层记忆架构
# 开发语言 # AIGC # python # pytorch
Logo
龙虾开发者社区

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用

更多推荐

cover

大模型安全实战】智能体安全的攻防二元性：同一种能力，为什么既是防线也是攻击面？（第4期）

avatar 龙虾开发者社区
cover

SparkSQL 之 Hive On Spark 原理分析

avatar 龙虾开发者社区

RL训练Agentic模型实现并行多轮上下文检索：匹配前沿模型且快10倍

基于RL的Agentic检索模型正在成为AI Agent效率革命的关键驱动力。并行化：从串行单次调用转向多轮高并行调用自编辑：模型在检索过程中主动裁剪不相关信息迭代优化：检索器与Agent在RL训练中双向迭代当Agentic检索的速度提升10倍以上、成本大幅下降，AI系统将能在数秒内完成以往需要数分钟的复杂信息收集——这是Agent从“对话助手”走向“行动执行者”的关键一步。参考文献：, ACL

avatar 龙虾开发者社区