告别“你是谁？”：手把手教你给大模型装上记忆大脑

你有没有遇到过这样的对话：

你问：“法国的首都是哪里？”
AI答：“巴黎。”
你再问：“那人口是多少？”
AI反问：“哪里的？”

这就是无记忆聊天机器人的典型症状。每一次消息都被当作全新对话，它不知道“那里”指的是巴黎。

真正的对话是这样工作的：上下文会延续，引用会累积，聊天机器人必须知道之前发生过什么。这篇文章就带你从零构建一个能记住对话历史的机器人——它会记得你两轮前说了啥，当前在聊什么主题，以及之前做出的决定。

---

你将从本文学到什么

• 为什么大语言模型（LLM）天生无状态，以及如何“伪造”记忆
• 对话历史模式：它到底是怎么工作的
• 上下文窗口的限制以及为什么它重要
• 滑动窗口记忆：只保留最后 N 条消息
• 摘要记忆：压缩旧对话
• 实体记忆：记住关于用户的具体事实
• 用 LangChain 构建完整的多轮对话机器人
• 跨会话持久化记忆

---

为什么 LLM 是无状态的

每次你调用 LLM 的 API，它都是从零开始。它对之前的调用完全没有记忆。它唯一能知道的上下文就是你放在当前 prompt 里的内容。

让聊天机器人工作的诀窍是：把整个对话历史都塞进每次的 prompt 里。

第一轮：
  USER: “法国的首都是哪里？”
  → 发送给 LLM: "User: 法国的首都是哪里？"
  → LLM 回复: "巴黎"

第二轮：
  USER: “那人口是多少？”
  → 发送给 LLM:
      "User: 法国的首都是哪里？
       Assistant: 巴黎。
       User: 那人口是多少？"
  → LLM 看到完整上下文，知道“那”= 巴黎

第三轮：
  → 把第一轮、第二轮、以及第三轮的所有内容一起发过去

每条消息都会追加到一个不断增长的列表中。这个列表会被放进后续的每一次 prompt 里。LLM 可以回看，因为这些东西就在当前的上下文里。

很简单。但它有一个硬限制：上下文窗口。

---

上下文窗口问题

每个 LLM 都有一个它能同时处理的最大 token 数。GPT-3.5-turbo：16k tokens。GPT-4：128k tokens。LLaMA-7B：4k tokens。

一段很长的对话会填满这个窗口。当对话超出限制时，你不能把一切都塞进去。你需要一个策略。

# 估计 token 数量（粗略：1 token ≈ 4 个英文字符）
def estimate_tokens(text: str) -> int:
    return len(text) // 4

def estimate_conversation_tokens(messages: list) -> int:
    total = 0
    for msg in messages:
        total += estimate_tokens(msg['content'])
        total += 4   # 每条消息的开销（角色、格式等）
    return total

# 演示对话如何快速填满窗口
messages = []
example_turns = [
    ("user", "给我讲讲机器学习。"),
    ("assistant", "机器学习是人工智能的一个领域，它让计算机能够从数据中学习，而无需显式编程。它包括监督学习（模型用标注样本训练）、无监督学习（不依赖标签发现模式）和强化学习（代理通过试错学习）。"),
    ("user", "那深度学习具体是什么？"),
    ("assistant", "深度学习是机器学习的一个子集，它使用多层神经网络。这些网络学习数据的分层表示，对图像、音频和文本任务尤其强大。2017 年提出的 Transformer 架构已成为现代深度学习系统的基础。"),
    ("user", "能举一些实际应用的例子吗？"),
    ("assistant", "当然！实际应用包括医学诊断中的图像分类、翻译和聊天机器人的自然语言处理、Netflix 和 Spotify 的推荐系统、银行欺诈检测，以及自动驾驶。深度学习通过大规模模式识别驱动了大多数这些应用。"),
]

print(f"{'轮次':<6} {'新增 tokens':<14} {'总 tokens':<14} {'占 4k 限制的百分比'}")
print("-" * 50)
for role, content in example_turns:
    messages.append({'role': role, 'content': content})
    total = estimate_conversation_tokens(messages)
    new   = estimate_tokens(content)
    print(f"{len(messages):<6} {new:<14} {total:<14} {total/4000:.1%}")

输出：

轮次    新增 tokens    总 tokens      占 4k 限制的百分比
--------------------------------------------------
1       12            16            0.4%
2       73            93            2.3%
3       13            110           2.8%
4       65            179           4.5%
5       15            198           5.0%
6       72            274           6.9%

一个关于复杂话题的长时间对话很容易达到 2000-3000 tokens。再加上 RAG 上下文和系统提示，很快就会触及限制。

---

策略 1：滑动窗口记忆

只保留最近 N 条消息。简单而有效。

from collections import deque
from typing import List, Optional

class SlidingWindowChatbot:
    """滑动窗口聊天机器人：只保留最近 N 条消息"""
    def __init__(self, model_pipeline, window_size: int = 10,
                 system_prompt: str = "你是一个乐于助人的助手。"):
        self.model         = model_pipeline
        self.window_size   = window_size  # 最多保留的消息条数
        self.system_prompt = system_prompt
        self.history       = deque(maxlen=window_size)  # 自动丢弃超出部分

    def chat(self, user_message: str) -> str:
        # 将用户消息添加到历史
        self.history.append({'role': 'user', 'content': user_message})

        # 构建带历史的 prompt
        messages = [
            {'role': 'system', 'content': self.system_prompt}
        ] + list(self.history)

        # 调用模型（演示中使用简单的文本格式）
        prompt = self._format_prompt(messages)
        response = self.model(prompt)

        # 将助手回复添加到历史
        self.history.append({'role': 'assistant', 'content': response})

        return response

    def _format_prompt(self, messages: List[dict]) -> str:
        formatted = ""
        for msg in messages:
            if msg['role'] == 'system':
                formatted += f"System: {msg['content']}\n\n"
            elif msg['role'] == 'user':
                formatted += f"Human: {msg['content']}\n"
            else:
                formatted += f"Assistant: {msg['content']}\n"
        formatted += "Assistant:"
        return formatted

    def get_history(self) -> list:
        return list(self.history)

    def clear(self):
        self.history.clear()
        print("对话历史已清空。")

# 模拟对话（用模拟模型演示）
def mock_model(prompt: str) -> str:
    """模拟的模型响应，实际使用请替换为真实 LLM 调用"""
    if "capital of france" in prompt.lower():
        return "法国的首都是巴黎。"
    elif "population" in prompt.lower() and "巴黎" in prompt:
        return "巴黎市常住人口约 210 万，大巴黎地区约 1200 万。"
    elif "famous landmark" in prompt.lower():
        return "巴黎以埃菲尔铁塔、卢浮宫、巴黎圣母院和凯旋门闻名。"
    elif "eiffel tower" in prompt.lower():
        return "埃菲尔铁塔建于 1887 年至 1889 年，由工程师古斯塔夫·埃菲尔设计，高 330 米。"
    else:
        return "我明白了。能告诉我更多吗？"

bot = SlidingWindowChatbot(mock_model, window_size=6)

# 模拟多轮对话
turns = [
    "法国的首都是哪里？",
    "那里的人口是多少？",
    "那个城市有哪些著名景点？",
    "跟我说说埃菲尔铁塔。",
    "它是什么时候建的？",
]

for user_input in turns:
    print(f"\n用户: {user_input}")
    response = bot.chat(user_input)
    print(f"机器人: {response}")

print(f"\n历史中有 {len(bot.get_history())} 条消息（最多 {bot.window_size}）")

输出：

用户: 法国的首都是哪里？
机器人: 法国的首都是巴黎。

用户: 那里的人口是多少？
机器人: 巴黎市常住人口约 210 万，大巴黎地区约 1200 万。

用户: 那个城市有哪些著名景点？
机器人: 巴黎以埃菲尔铁塔、卢浮宫、巴黎圣母院和凯旋门闻名。

用户: 跟我说说埃菲尔铁塔。
机器人: 埃菲尔铁塔建于 1887 年至 1889 年，由工程师古斯塔夫·埃菲尔设计，高 330 米。

用户: 它是什么时候建的？
机器人: 我明白了。能告诉我更多吗？

历史中有 6 条消息（最多 6）

机器人通过上下文理解了“那里”（巴黎）和“那个城市”（巴黎）。滑动窗口保留了最后 6 条消息。

---

策略 2：摘要记忆

当历史变长时，将旧消息总结成摘要，只保留近期完整消息。

class SummaryMemoryChatbot:
    """摘要记忆聊天机器人：对旧对话进行自动总结，只保留近期完整消息"""
    def __init__(self, model_pipeline, summarizer_pipeline,
                 max_recent: int = 6, summary_threshold: int = 10,
                 system_prompt: str = "你是一个乐于助人的助手。"):
        self.model       = model_pipeline
        self.summarizer  = summarizer_pipeline
        self.max_recent  = max_recent  # 保留的最近完整消息数
        self.threshold   = summary_threshold  # 触发摘要的消息总数阈值
        self.system      = system_prompt
        self.history     = []
        self.summary     = ""  # 较早轮次的压缩记忆

    def _maybe_summarize(self):
        """如果历史达到阈值，触发摘要"""
        if len(self.history) < self.threshold:
            return

        # 摘要历史中最早的一半
        n_to_summarize = len(self.history) // 2
        old_messages   = self.history[:n_to_summarize]
        self.history   = self.history[n_to_summarize:]

        # 将旧消息格式化为文本
        old_text = "\n".join([
            f"{m['role'].title()}: {m['content']}"
            for m in old_messages
        ])

        # 调用摘要函数（生产环境中应调用 LLM）
        new_summary_input = f"{self.summary}\n\n{old_text}" if self.summary else old_text
        self.summary = self._summarize(new_summary_input)

        print(f"[记忆] 已将 {n_to_summarize} 条消息压缩为摘要")

    def _summarize(self, text: str) -> str:
        """模拟的摘要生成，实际应使用 LLM 或专门的摘要模型"""
        return "[早期对话摘要：用户询问了法国、巴黎、其人口（约 210 万）以及巴黎的地标，包括埃菲尔铁塔。]"

    def _format_prompt(self) -> str:
        parts = [f"System: {self.system}\n"]

        if self.summary:
            parts.append(f"[早期对话摘要]: {self.summary}\n")

        # 只保留最近 max_recent 条消息作为完整内容
        for msg in self.history[-self.max_recent:]:
            role = "Human" if msg['role'] == 'user' else "Assistant"
            parts.append(f"{role}: {msg['content']}")

        parts.append("Assistant:")
        return "\n".join(parts)

    def chat(self, user_message: str) -> str:
        self.history.append({'role': 'user', 'content': user_message})
        self._maybe_summarize()

        prompt   = self._format_prompt()
        response = self.model(prompt)
        self.history.append({'role': 'assistant', 'content': response})

        return response

    def memory_status(self):
        print(f"摘要: {'存在' if self.summary else '无'}")
        print(f"完整保留的最近消息数: {min(len(self.history), self.max_recent)}")
        print(f"总历史消息数: {len(self.history)}")

summary_bot = SummaryMemoryChatbot(mock_model, None, max_recent=6, summary_threshold=8)

for user_input in turns * 2:  # 重复对话两次以触发摘要
    response = summary_bot.chat(user_input)

summary_bot.memory_status()

---

策略 3：实体记忆

提取并存储关于用户或对话实体的具体事实。

import re
from typing import Dict

class EntityMemoryChatbot:
    """实体记忆聊天机器人：自动提取并记住用户的姓名、地点、兴趣等实体"""
    def __init__(self, model_pipeline,
                 system_prompt: str = "你是一个乐于助人的助手。"):
        self.model   = model_pipeline
        self.system  = system_prompt
        self.history = []
        self.entities: Dict[str, str] = {}  # 实体存储器，如 { 'name': '张三', 'topic': '机器学习' }

    def _extract_entities(self, message: str):
        """从用户消息中提取实体（简化版，生产环境应使用 NER 模型或 LLM）"""
        patterns = {
            'name':     r"(?:我叫|我是|我的名字是)\s+([\u4e00-\u9fa5]+)",  # 匹配中文姓名
            'location': r"(?:我住在|我在|我来自)\s+([\u4e00-\u9fa5]+(?:\s*[\u4e00-\u9fa5]+)?)",
            'job':      r"(?:我是|我是一名|我从事)\s+([\u4e00-\u9fa5]+(?:\s*[\u4e00-\u9fa5]+)?)",
            'topic':    r"(?:我想学|我在学|我需要帮助的方面是)\s+([\u4e00-\u9fa5\s]+)"
        }

        for entity_type, pattern in patterns.items():
            match = re.search(pattern, message)
            if match:
                self.entities[entity_type] = match.group(1).strip()

    def _build_entity_context(self) -> str:
        """构建包含已记住实体的上下文文本"""
        if not self.entities:
            return ""
        lines = ["关于用户的已知信息："]
        for entity, value in self.entities.items():
            lines.append(f"  - {entity}: {value}")
        return "\n".join(lines)

    def _format_prompt(self) -> str:
        parts = [f"System: {self.system}"]

        entity_ctx = self._build_entity_context()
        if entity_ctx:
            parts.append(entity_ctx)

        for msg in self.history[-8:]:
            role = "Human" if msg['role'] == 'user' else "Assistant"
            parts.append(f"{role}: {msg['content']}")

        parts.append("Assistant:")
        return "\n".join(parts)

    def chat(self, user_message: str) -> str:
        self._extract_entities(user_message)
        self.history.append({'role': 'user', 'content': user_message})

        prompt   = self._format_prompt()
        response = self.model(prompt)
        self.history.append({'role': 'assistant', 'content': response})

        return response

# 测试实体记忆
def entity_mock_model(prompt: str) -> str:
    if "name" in prompt.lower() and "Alex" in prompt:
        return "很高兴认识你，Alex！"
    elif "Alex" in prompt and "推荐" in prompt:
        return "根据你对机器学习的兴趣，Alex，我建议从 Python 和 scikit-learn 开始。"
    elif "课程" in prompt:
        return "对于机器学习，Andrew Ng 的 Coursera 课程对初学者非常棒。"
    else:
        return "请告诉我你想学什么。"

entity_bot = EntityMemoryChatbot(entity_mock_model)

conversations = [
    "你好，我叫 Alex。",
    "我想学机器学习。",
    "你能推荐些什么吗？",
    "有什么课程吗？",
]

for user_input in conversations:
    print(f"\n用户: {user_input}")
    response = entity_bot.chat(user_input)
    print(f"机器人: {response}")

print(f"\n提取到的实体: {entity_bot.entities}")

输出：

用户: 你好，我叫 Alex。
机器人: 很高兴认识你，Alex！

用户: 我想学机器学习。
机器人: 请告诉我你想学什么。

用户: 你能推荐些什么吗？
机器人: 根据你对机器学习的兴趣，Alex，我建议从 Python 和 scikit-learn 开始。

用户: 有什么课程吗？
机器人: 对于机器学习，Andrew Ng 的 Coursera 课程对初学者非常棒。

提取到的实体: {'name': 'Alex', 'topic': '机器学习'}

机器人在所有轮次中记住了用户的名字和话题。

---

使用 OpenAI API 构建完整聊天机器人

import openai
import json
from datetime import datetime

class ProductionChatbot:
    """生产级聊天机器人，基于 OpenAI API，支持历史管理和持久化"""
    def __init__(
        self,
        system_prompt: str = "你是一个乐于助人的 AI 助手。",
        model: str = "gpt-3.5-turbo",
        max_history: int = 20,
        max_tokens: int = 500,
        temperature: float = 0.7
    ):
        self.client      = openai.OpenAI()
        self.model       = model
        self.max_history = max_history    # 最多保留多少条对话历史
        self.max_tokens  = max_tokens
        self.temperature = temperature
        self.history     = []
        self.system      = system_prompt
        self.created_at  = datetime.now()

    def chat(self, user_message: str) -> str:
        self.history.append({'role': 'user', 'content': user_message})

        # 若历史过长则裁剪
        if len(self.history) > self.max_history:
            self.history = self.history[-self.max_history:]

        # 构建发送给 API 的消息列表
        messages = [
            {'role': 'system', 'content': self.system}
        ] + self.history

        # 调用 OpenAI API
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            max_tokens=self.max_tokens,
            temperature=self.temperature,
        )

        assistant_message = response.choices[0].message.content
        self.history.append({'role': 'assistant', 'content': assistant_message})

        return assistant_message

    def save_conversation(self, filepath: str):
        """将对话历史保存到文件"""
        data = {
            'created_at': self.created_at.isoformat(),
            'saved_at':   datetime.now().isoformat(),
            'model':      self.model,
            'system':     self.system,
            'messages':   self.history
        }
        with open(filepath, 'w') as f:
            json.dump(data, f, indent=2)
        print(f"已将 {len(self.history)} 条消息保存到 {filepath}")

    def load_conversation(self, filepath: str):
        """从文件加载对话历史"""
        with open(filepath, 'r') as f:
            data = json.load(f)
        self.history = data['messages']
        self.system  = data.get('system', self.system)
        print(f"已从 {filepath} 加载 {len(self.history)} 条消息")

    def reset(self):
        """重置对话"""
        self.history = []
        print("对话已重置。")

    def get_stats(self) -> dict:
        """获取对话统计信息"""
        n_user      = sum(1 for m in self.history if m['role'] == 'user')
        n_assistant = sum(1 for m in self.history if m['role'] == 'assistant')
        total_chars = sum(len(m['content']) for m in self.history)

        return {
            '轮次':            n_user,
            '总消息数':        len(self.history),
            '估计 token 数':   total_chars // 4,
            '历史深度':        len(self.history)
        }

# 使用示例
# bot = ProductionChatbot(
#     system_prompt="你是一个专注于实际例子的 ML 导师。",
#     model="gpt-3.5-turbo",
#     max_history=20
# )
# response = bot.chat("解释一下过拟合。")
# print(response)
# bot.save_conversation('session_001.json')
print("ProductionChatbot 已准备就绪（需要设置 OPENAI_API_KEY）")

---

LangChain 记忆：更轻松的方式

from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
from langchain.chains import ConversationChain
from langchain_community.llms import HuggingFacePipeline
from transformers import pipeline as hf_pipeline

# 创建 LLM
gen_pipe = hf_pipeline('text-generation', model='gpt2', max_new_tokens=100)
llm = HuggingFacePipeline(pipeline=gen_pipe)

# 缓冲区记忆：保留所有消息
buffer_memory = ConversationBufferMemory()

# 摘要记忆：自动在过长时进行摘要（需要 LLM 实例）
# summary_memory = ConversationSummaryMemory(llm=llm)

# 构建对话链
conversation = ConversationChain(
    llm=llm,
    memory=buffer_memory,
    verbose=False
)

# 开始聊天
result = conversation.predict(input="你好，我叫 Alex。")
print(f"机器人: {result[:100]}...")

result = conversation.predict(input="我叫什么名字？")
print(f"机器人: {result[:100]}...")

# 查看记忆内容
print(f"\n记忆缓冲区:\n{buffer_memory.buffer}")

---

跨会话持久化记忆

import json
import os

class PersistentChatbot:
    """持久化聊天机器人：将对话历史保存到磁盘，支持跨会话恢复"""
    def __init__(self, model_pipeline, session_id: str,
                 storage_dir: str = './chat_sessions',
                 max_history: int = 50):
        self.model       = model_pipeline
        self.session_id  = session_id
        self.storage_dir = storage_dir
        self.max_history = max_history
        self.history     = []
        self.metadata    = {}

        os.makedirs(storage_dir, exist_ok=True)
        self._load_session()  # 启动时自动加载历史

    def _session_path(self) -> str:
        return os.path.join(self.storage_dir, f"{self.session_id}.json")

    def _load_session(self):
        """从文件加载会话"""
        path = self._session_path()
        if os.path.exists(path):
            with open(path, 'r') as f:
                data = json.load(f)
            self.history  = data.get('history', [])
            self.metadata = data.get('metadata', {})
            print(f"已加载会话 '{self.session_id}'，包含 {len(self.history)} 条消息")
        else:
            print(f"新会话 '{self.session_id}' 已创建")

    def _save_session(self):
        """将会话保存到文件"""
        data = {
            'session_id':   self.session_id,
            'last_updated': datetime.now().isoformat(),
            'history':      self.history,
            'metadata':     self.metadata
        }
        with open(self._session_path(), 'w') as f:
            json.dump(data, f, indent=2)

    def chat(self, user_message: str) -> str:
        self.history.append({'role': 'user', 'content': user_message})

        if len(self.history) > self.max_history:
            self.history = self.history[-self.max_history:]

        response = self.model(self._format_prompt())
        self.history.append({'role': 'assistant', 'content': response})
        self._save_session()  # 每次聊天后自动保存

        return response

    def _format_prompt(self) -> str:
        parts = []
        for msg in self.history[-10:]:
            role = "Human" if msg['role'] == 'user' else "Assistant"
            parts.append(f"{role}: {msg['content']}")
        parts.append("Assistant:")
        return "\n".join(parts)

    def list_sessions(self) -> list:
        """列出存储目录中的所有会话 ID"""
        sessions = []
        for f in os.listdir(self.storage_dir):
            if f.endswith('.json'):
                sessions.append(f.replace('.json', ''))
        return sessions

# 使用示例
persistent_bot = PersistentChatbot(mock_model, session_id='user_alex_001')
persistent_bot.chat("法国的首都是哪里？")
persistent_bot.chat("那里的人口是多少？")

print(f"\n已保存的会话: {persistent_bot.list_sessions()}")
print(f"历史消息数: {len(persistent_bot.history)} 条")

---

聊天机器人质量检查清单

checklist = {
    "记忆管理": [
        "机器人能记住 5 轮之前的上下文吗？",
        "它能正确处理指代关系吗？（如“那里”、“它”、“他们”）",
        "它会重复用户已经提供的信息吗？"
    ],
    "上下文窗口": [
        "它能在超长对话中不崩溃吗？",
        "当历史过长时，有没有优雅的回退方案？",
        "摘要后的信息是否准确，没有丢失关键内容？"
    ],
    "对话质量": [
        "能保持对话始终围绕主题吗？",
        "能正确引用之前的决策吗？",
        "能优雅处理话题切换吗？"
    ],
    "持久化": [
        "能保存对话供以后使用吗？",
        "能从之前的会话中恢复继续吗？",
        "存储格式是否可读且易于调试？"
    ],
    "边缘情况": [
        "用户询问不在记忆中的内容时怎么办？",
        "用户自己矛盾怎么办？",
        "用户消息非常短或非常长时能处理吗？"
    ]
}

for category, items in checklist.items():
    print(f"\n{category}:")
    for item in items:
        print(f"  [ ] {item}")

---

快速速查表

记忆类型与适用场景

• 缓冲区（全部历史）：适合短对话。保留所有消息，每次都全部传入。
• 滑动窗口：适合中等长度对话。只保留最近 N 条消息。
• 摘要记忆：适合长对话。将旧消息总结成摘要，保留近期完整消息。
• 实体记忆：适合需要记住用户具体信息的场景。自动提取并存储命名实体。
• 持久化记忆：适合多会话聊天机器人。从磁盘或数据库保存/加载记录。

常见操作模式与代码

• 添加历史：history.append({'role': 'user', 'content': msg})
• 裁剪历史：history = history[-max_size:]
• 构建消息：[{'role': 'system', 'content': system}] + history
• 保存会话：json.dump({'history': history}, f)
• 加载会话：history = json.load(f)['history']
• LangChain 缓冲区记忆：ConversationBufferMemory()
• LangChain 摘要记忆：ConversationSummaryMemory(llm=llm)

---

练习挑战

Level 1：
用本地 GPT-2 构建一个 SlidingWindowChatbot。进行 10 轮关于你感兴趣话题的对话。结束后打印完整历史，验证机器人能正确引用前几轮的内容。

Level 2：
实现 SummaryMemoryChatbot 并用真实的摘要模型（如 T5-small）进行总结。每 8 轮对话后，将前半部分用模型总结。测试 20 轮对话，打印每次摘要触发的时间点，检查摘要是否准确。

Level 3：
构建 PersistentChatbot，将会话保存到磁盘。先开启一个对话，关闭程序，重新启动，加载会话，继续对话。验证机器人记得上一轮说了什么。添加一个 /history 命令，打印之前所有会话的简短摘要。

---

参考文献

• LangChain: Memory 文档
• OpenAI: Chat Completions API
• LangChain: ConversationBufferMemory
• Llama Index: Chat Engine

---

最终篇，第 100 篇：OpenAI API: Build With GPT-4。API 设置、聊天补全、函数调用、流式输出和成本管理。系列最后一篇将把所有内容收束在一起。

直达网址：https://python.langchain.com/docs/modules/memory/