🗜️ AI Agent 核心架构：会话摘要与压缩（Memory Compression）全解析与面试通关指南

在开发 AI Agent 时，如果你做的是一个“虚拟伴侣”或者“长期私人助理”，你会遇到一个极其头疼的问题：用户可能会跟你聊上几个月甚至几年。

如果用“全量记忆”，大模型的 Token 费用会让你倾家荡产；如果用“滑动窗口记忆”（直接丢弃老对话），AI 就会变成“渣男/渣女”，上一周发生的事情它全忘了。

为了打破这个僵局，工业界引入了记忆系统中最优雅的折中方案：会话摘要与压缩（Conversation Summarization & Compression）。

在高级 AI 研发面试中，面试官非常喜欢考察候选人对长文本上下文的管理能力。这篇博客将用大白话带你搞懂记忆压缩的核心策略，并附带面试级别的核心代码实现！

---

💡 一、 为什么需要会话摘要与压缩？（大白话秒懂）

通俗概念：
和 AI 聊天就像背着石头爬山。每一句话都是一块石头（Token），聊得越久，背包越重。

• 全量记忆：石头全背着，最后把你压垮（Token 超出限制，API 报错）。
• 滑动窗口：边走边扔石头（直接丢弃老对话），最后到了山顶，你忘了自己怎么上来的（硬遗忘）。
• 摘要与压缩：每当你背了 10 块石头觉得累了，你就停下来，用炼金术把这 10 块石头炼成一颗小钻石（总结提炼核心信息）。你扔掉石头，只带着钻石继续走。这样既减轻了重量，又保留了最核心的记忆。

核心收益：

1. 控制 Token 成本：将几万字的聊天记录压缩成几百字的摘要，大幅降低每一次 API 调用的花费。
2. 避免“硬遗忘”：即使是几个月前聊过的重要设定（比如“用户有一个女儿叫小芳”），也能通过摘要一直保留在当前对话中。

---

⚙️ 二、 主流的两种压缩机制（面试必背）

1. 滚动摘要机制 (Rolling Summary)

• 原理：维护一个 current_summary（当前摘要）字符串。当新增的聊天记录达到一定长度（比如 10 轮对话）时，触发一次后台 LLM 调用，让它把 【旧摘要】+【新聊的 10 轮对话】 重新总结成一段 【新摘要】。
• 优点：实现简单，上下文始终保持极其精简的状态。
• 缺点：每次压缩都可能丢失一些细节。压缩次数多了，就像传话游戏一样，可能会产生幻觉或信息模糊。

2. 实体与画像抽取 (Entity/Profile Extraction)

• 原理：不总结大段的流水账对话，而是让模型像填表一样，专门提取对话中的“关键实体”。比如提取出：{"user_name": "张三", "job": "程序员", "hobbies": ["打游戏", "看科幻小说"]}。
• 优点：信息极其结构化，不仅能作为当前记忆，还能直接落库到关系型数据库（MySQL）中长期保存。
• 缺点：只能记住“设定”，记不住具体的“聊天语境”或“情绪”。

🌟 工业界标准做法：
真实的业务线通常是 “短期原话滑动窗口” + “中长期滚动摘要” + “永久用户画像抽取” 三管齐下。

---

🎯 三、 高频面试 Q&A 实战演练

Q1：做会话摘要时，什么时候触发压缩最合适？

标准答案：
通常有两种触发机制：

1. 按阈值触发（Threshold-based）：当短期对话轮数（如 10 轮）或 Token 数量（如达到 2000 Tokens）达到上限时，触发一次压缩。适合常规问答。
2. 异步定时触发（Asynchronous/Cron-based）：不阻塞用户的正常聊天，在服务器闲时（比如用户离线 5 分钟后），由后台队列（Celery/Kafka）发起压缩任务。适合对 C 端延迟（Latency）要求极高的闲聊产品。

Q2：滚动摘要（Rolling Summary）会带来“信息磨损”，聊得越久，早期的记忆就越模糊，如何解决？

标准答案：
摘要必然会带来信息的有损压缩。为了缓解这个问题，我们在工程上通常结合 RAG（检索增强生成）：
在丢弃原始聊天记录（将其压缩）之前，先将被丢弃的原话做 Embedding 向量化，存入本地的长期向量数据库（Vector DB）。
这样，摘要作为“主线剧情”一直带在 Prompt 里；而那些被磨损的“支线细节”，当用户重新提及时，系统还能通过 RAG 瞬间从长期记忆库里捞回来。

Q3：调用大模型做摘要也是要花钱和时间的，这难道不是增加了系统的负担吗？

标准答案：
从单次看确实增加了负担，但从长期算是一笔极其划算的经济账。
假设你带了 10K Token 的上下文，每次用户跟你聊一句，你都要为这 10K Token 付费。聊 10 句就是 100K 的计费。
如果花 1K 的 Token 成本把它压缩成 500 Token 的摘要。接下来的几十轮对话，你都只需要为这 500 Token 付费。长线来看，不仅极大地节省了成本，还降低了主模型大段阅读带来的延迟（TTFT）。此外，后台摘要完全可以采用更便宜、更快速的小模型（如 GPT-4o-mini 或 Qwen-7B）来完成。

---

💻 四、 面试加分代码：手写“滚动摘要记忆管理器”

在面试白板环节，如果让你手写记忆压缩逻辑，面试官最想看的是你何时触发压缩、如何拼接 Prompt、以及如何替换旧状态。

以下是一段保姆级注释的滚动摘要实现代码：

import json

# ==========================================
# 1. 模拟环境：大模型 API
# ==========================================
class MockLLM:
    """模拟大模型接口，专门处理摘要任务"""
    def summarize(self, text_to_summarize: str) -> str:
        # 在真实环境中，这里应该是一个调用 OpenAI/大模型的 Prompt 链
        # 比如："请将以下【旧摘要】和【新对话】合并提炼，保留用户的核心信息、偏好和未完成的任务..."
        print("⚙️ [后台服务] 正在调用小模型执行长文本压缩压缩...")
        return f"【核心摘要】用户张三是一名程序员，之前探讨了 Python 的基础语法，当前正在研究如何写并发代码。"

# ==========================================
# 2. 核心大管家：会话压缩记忆类
# ==========================================
class SummarizationMemory:
    """
    带滚动摘要机制的短期记忆管理器
    """
    def __init__(self, token_limit: int = 100):
        # 1. 保存极少量的高清近期对话（滑动窗口）
        self.recent_messages = []
        # 2. 保存中长期的模糊记忆（滚动摘要）
        self.current_summary = ""
        # 3. 触发压缩的阈值（为了方便演示，这里把长度设得很小）
        self.token_limit = token_limit
        
        self.llm = MockLLM()

    def _estimate_length(self) -> int:
        """简单估算当前近期对话的长度"""
        return sum(len(msg["content"]) for msg in self.recent_messages)

    def add_message(self, role: str, content: str):
        """添加新消息，并检查是否需要触发压缩"""
        self.recent_messages.append({"role": role, "content": content})
        print(f"📥 [存入原话] {role}: {content[:15]}...")

        # 检查：如果近期对话太长了，立刻触发压缩机制！
        if self._estimate_length() > self.token_limit:
            self._trigger_compression()

    def _trigger_compression(self):
        """核心逻辑：滚动摘要的炼金术"""
        print("\n⚠️ 警告：近期记忆太长，触发【会话压缩】机制！")
        
        # 1. 将当前的旧摘要和需要被压缩的近期对话拼接起来
        chat_text = "\n".join([f"{msg['role']}: {msg['content']}" for msg in self.recent_messages])
        text_to_compress = f"【旧的摘要】\n{self.current_summary}\n\n【需要补充的新对话】\n{chat_text}"
        
        # 2. 调用模型生成新摘要
        new_summary = self.llm.summarize(text_to_compress)
        
        # 3. 状态更新（极其重要）
        self.current_summary = new_summary # 用新摘要替换旧摘要
        self.recent_messages = []          # ⚠️ 清空被压缩过的近期原话，释放出宝贵的上下文空间！
        print(f"✅ [压缩完成] 历史对话已清空，当前记忆已更新为：\n{self.current_summary}\n")

    def get_prompt_context(self) -> str:
        """每次向大模型发请求前，获取当前的最优上下文"""
        context = ""
        if self.current_summary:
            context += f"System 背景信息 (历史摘要):\n{self.current_summary}\n\n"
            
        context += "近期对话原话:\n"
        for msg in self.recent_messages:
            context += f"{msg['role']}: {msg['content']}\n"
            
        return context

# ==========================================
# 3. 测试运行与面试讲解
# ==========================================
if __name__ == "__main__":
    memory = SummarizationMemory(token_limit=50) # 设积极小的阈值以触发压缩
    
    # 回合 1：闲聊，字数不多，正常存入
    memory.add_message("user", "你好，我是张三，是个程序员。")
    memory.add_message("assistant", "你好张三，请问在研究什么技术？")
    
    print("\n--- 第一轮聊完后的内存状态 ---")
    print(memory.get_prompt_context())
    print("-" * 40)
    
    # 回合 2：突然发了一大堆话，导致总长度超过 50，触发大压缩！
    long_text = "我之前一直在用 Python 写基础脚本，但是现在业务量上来了，我必须要学习一下 Python 的异步编程和多线程并发控制，你能教教我吗？"
    memory.add_message("user", long_text)
    
    print("--- 触发压缩后的内存状态 ---")
    print(memory.get_prompt_context())

# 💡 面试讲解要点：
# 向面试官解释：“在真正的工程落地中，`_trigger_compression` 通常是被推送到 MQ（消息队列）中异步执行的。
# 通过维护 `current_summary` 这个字符串变量，我们把大模型看做是一个‘状态机’的更新函数：
# 新状态 = Update(旧状态, 新事件)。
# 采用这种滚动压缩的策略，即使用户和我们的 Agent 聊了 3 年，
# Prompt 的长度也永远会稳定在一个安全的范围内，实现了成本和体验的完美平衡。”