记忆机制与对话管理

为什么需要记忆

LLM 的无状态问题

大语言模型本身是无状态的——每次调用都是独立的，模型不会记住上一次的对话内容。这意味着如果不做处理，模型完全不知道之前聊了什么：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini")

# 第一次对话
response1 = llm.invoke([HumanMessage(content="我叫小明，我今年25岁")])
print(response1.content)
# 输出：你好小明！很高兴认识你...

# 第二次对话——模型完全不记得之前的信息
response2 = llm.invoke([HumanMessage(content="我叫什么名字？")])
print(response2.content)
# 输出：抱歉，我不知道你的名字...（因为没有传入历史对话）

要实现多轮对话，必须手动把历史消息拼接到每次请求中：

LangChain 的记忆模块就是为了解决这个问题，它提供了多种策略来管理和传递对话历史。

对话记忆原理

核心思路

所有记忆机制的底层逻辑都是相同的：在每次调用 LLM 前，将存储的历史消息注入到提示词中。

记忆策略对比

不同场景需要不同的记忆策略，核心区别在于如何处理历史消息：

策略	原理	优点	缺点	适用场景
完整保留	保存所有消息	信息不丢失	Token 消耗大	短对话
滑动窗口	只保留最近 K 轮	控制 Token 量	丢失早期信息	日常聊天
摘要压缩	LLM 总结旧消息	节省 Token	摘要可能失真	长对话
混合策略	保留近期 + 摘要远期	兼顾细节和全局	实现复杂	生产环境

ConversationBufferMemory

基本用法

最简单的记忆方式，保存完整的对话历史：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini")

# 手动管理消息历史（底层原理）
chat_history = []

def chat(user_input: str) -> str:
    """简单的对话函数"""
    # 添加用户消息到历史
    chat_history.append(HumanMessage(content=user_input))

    # 构建带历史的提示词
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个友好的AI助手，请记住用户告诉你的信息。"),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ])

    # 组装链
    chain = prompt | llm | StrOutputParser()

    # 调用
    response = chain.invoke({
        "history": chat_history,
        "input": user_input,
    })

    # 保存 AI 回复到历史
    chat_history.append(AIMessage(content=response))

    return response

# 测试多轮对话
print(chat("你好，我叫小明，我是一名Python开发者"))
# 输出：你好小明！很高兴认识一位Python开发者...

print(chat("我最近在学习 LangChain"))
# 输出：那太好了！LangChain 是一个非常实用的框架...

print(chat("我刚才说我叫什么名字？"))
# 输出：你叫小明。（能记住之前的信息）

使用 RunnableWithMessageHistory

LangChain 推荐使用 RunnableWithMessageHistory 来管理对话历史，它是基于 Runnable 接口的现代方案：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# 模型与提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的AI助手。请用中文回答。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

chain = prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()

# 会话存储（内存版）
store = {}

def get_session_history(session_id: str) -> ChatMessageHistory:
    """根据 session_id 获取或创建会话历史"""
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# 包装为带历史的链
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

# 使用 —— 注意 config 中传入 session_id
response1 = chain_with_history.invoke(
    {"input": "你好，我叫小红，我喜欢画画"},
    config={"configurable": {"session_id": "user-001"}},
)
print(response1)

response2 = chain_with_history.invoke(
    {"input": "我喜欢什么？"},
    config={"configurable": {"session_id": "user-001"}},
)
print(response2)
# 输出：你喜欢画画。（能记住之前的对话）

滑动窗口记忆

保留最近 K 轮对话

完整保留所有历史会导致 Token 消耗线性增长，滑动窗口只保留最近的 K 轮对话：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages import HumanMessage, AIMessage

llm = ChatOpenAI(model="gpt-4o-mini")

class SlidingWindowMemory:
    """滑动窗口记忆管理器"""

    def __init__(self, max_rounds: int = 5):
        """
        Args:
            max_rounds: 保留的最大对话轮数（一轮 = 一问一答）
        """
        self.max_rounds = max_rounds
        self.history: list = []

    def add_user_message(self, content: str):
        """添加用户消息"""
        self.history.append(HumanMessage(content=content))
        self._trim()

    def add_ai_message(self, content: str):
        """添加 AI 消息"""
        self.history.append(AIMessage(content=content))
        self._trim()

    def _trim(self):
        """裁剪历史，只保留最近 max_rounds 轮"""
        max_messages = self.max_rounds * 2  # 每轮2条消息
        if len(self.history) > max_messages:
            self.history = self.history[-max_messages:]

    def get_messages(self) -> list:
        """获取当前历史消息"""
        return self.history

    def clear(self):
        """清空历史"""
        self.history = []


# 使用
memory = SlidingWindowMemory(max_rounds=3)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的AI助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])
chain = prompt | llm | StrOutputParser()

# 模拟多轮对话
conversations = [
    "我住在上海",
    "我在一家互联网公司工作",
    "我养了一只猫叫小花",
    "我喜欢周末去公园跑步",
    "我的猫是什么颜色的？",  # 可能不记得了（超出窗口）
    "我住在哪里？",          # 可能不记得了
]

for user_input in conversations:
    memory.add_user_message(user_input)
    response = chain.invoke({
        "history": memory.get_messages(),
        "input": user_input,
    })
    memory.add_ai_message(response)
    print(f"用户: {user_input}")
    print(f"AI: {response}")
    print(f"当前历史条数: {len(memory.get_messages())}")
    print("---")

摘要记忆

自动摘要旧对话

当对话很长时，可以让 LLM 自动将旧对话压缩成摘要，只保留最近的原始消息：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

llm = ChatOpenAI(model="gpt-4o-mini")

class SummaryMemory:
    """摘要记忆管理器"""

    def __init__(self, max_recent_messages: int = 6):
        """
        Args:
            max_recent_messages: 保留的近期消息条数（不摘要）
        """
        self.summary = ""           # 当前摘要
        self.recent_messages = []   # 近期原始消息
        self.max_recent = max_recent_messages

    def _summarize(self, old_messages: list) -> str:
        """使用 LLM 生成摘要"""
        # 将旧消息转为文本
        conversation_text = "\n".join(
            f"{'用户' if isinstance(m, HumanMessage) else 'AI'}: {m.content}"
            for m in old_messages
        )

        summarize_prompt = ChatPromptTemplate.from_template(
            "请将以下对话历史压缩成一段简洁的摘要，保留关键信息（姓名、偏好、重要事实）：\n\n"
            "{conversation}\n\n"
            "摘要："
        )

        chain = summarize_prompt | llm | StrOutputParser()
        result = chain.invoke({"conversation": conversation_text})
        return result

    def add_messages(self, human_msg: str, ai_msg: str):
        """添加一轮对话"""
        self.recent_messages.append(HumanMessage(content=human_msg))
        self.recent_messages.append(AIMessage(content=ai_msg))

        # 超出窗口时，摘要旧消息
        if len(self.recent_messages) > self.max_recent:
            # 取出需要摘要的消息
            to_summarize = self.recent_messages[:-self.max_recent]
            self.recent_messages = self.recent_messages[-self.max_recent:]

            # 生成新摘要（合并旧摘要 + 新旧消息）
            new_summary = self._summarize(to_summarize)
            if self.summary:
                self.summary = f"{self.summary}\n后续发展：{new_summary}"
            else:
                self.summary = new_summary

    def get_context_messages(self) -> list:
        """获取用于发送给 LLM 的完整上下文"""
        messages = []
        if self.summary:
            messages.append(
                SystemMessage(content=f"之前对话的摘要：{self.summary}")
            )
        messages.extend(self.recent_messages)
        return messages


# 使用示例
memory = SummaryMemory(max_recent_messages=4)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的AI助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])
chain = prompt | llm | StrOutputParser()

conversations = [
    "我叫小李，今年28岁，住在北京",
    "我在一家AI公司做算法工程师",
    "我最近在研究大语言模型",
    "我有一只金毛犬叫旺财",
    "周末我喜欢去爬山",
    "我之前说我叫什么？养了什么宠物？",
]

for user_input in conversations:
    response = chain.invoke({
        "history": memory.get_context_messages(),
        "input": user_input,
    })
    memory.add_messages(user_input, response)
    print(f"用户: {user_input}")
    print(f"AI: {response[:80]}...")
    if memory.summary:
        print(f"当前摘要: {memory.summary[:80]}...")
    print("---")

自定义记忆存储

基于 Redis 的持久化存储

生产环境中，对话历史需要持久化存储，不能只在内存中：

import json
from typing import Optional
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage, messages_from_dict

class RedisChatMessageHistory:
    """基于 Redis 的对话历史存储"""

    def __init__(self, redis_client, session_id: str, ttl: int = 86400):
        """
        Args:
            redis_client: Redis 客户端
            session_id: 会话 ID（如 user_id 或 session_uuid）
            ttl: 过期时间（秒），默认24小时
        """
        self.redis = redis_client
        self.key = f"chat_history:{session_id}"
        self.ttl = ttl

    def add_message(self, message: BaseMessage) -> None:
        """添加消息"""
        msg_dict = {
            "type": message.type,
            "content": message.content,
        }
        self.redis.rpush(self.key, json.dumps(msg_dict, ensure_ascii=False))
        self.redis.expire(self.key, self.ttl)

    def add_user_message(self, content: str) -> None:
        self.add_message(HumanMessage(content=content))

    def add_ai_message(self, content: str) -> None:
        self.add_message(AIMessage(content=content))

    def get_messages(self) -> list[BaseMessage]:
        """获取所有消息"""
        raw_messages = self.redis.lrange(self.key, 0, -1)
        dicts = [json.loads(m) for m in raw_messages]
        return messages_from_dict(dicts)

    def clear(self) -> None:
        """清空历史"""
        self.redis.delete(self.key)


# 使用示例
import redis

redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

def get_history(session_id: str) -> RedisChatMessageHistory:
    return RedisChatMessageHistory(redis_client, session_id, ttl=3600)

# 测试
history = get_history("user-001")
history.add_user_message("你好")
history.add_ai_message("你好！有什么可以帮你的？")

# 重新获取，数据依然在
history2 = get_history("user-001")
print(len(history2.get_messages()))  # 2

基于文件的存储

适合开发和测试环境：

import json
from pathlib import Path
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage

class FileChatMessageHistory:
    """基于文件的对话历史存储"""

    def __init__(self, file_path: str):
        self.file_path = Path(file_path)
        self.file_path.parent.mkdir(parents=True, exist_ok=True)
        if not self.file_path.exists():
            self.file_path.write_text("[]", encoding="utf-8")

    def _read(self) -> list[dict]:
        return json.loads(self.file_path.read_text(encoding="utf-8"))

    def _write(self, messages: list[dict]):
        self.file_path.write_text(
            json.dumps(messages, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )

    def add_user_message(self, content: str):
        messages = self._read()
        messages.append({"type": "human", "content": content})
        self._write(messages)

    def add_ai_message(self, content: str):
        messages = self._read()
        messages.append({"type": "ai", "content": content})
        self._write(messages)

    def get_messages(self) -> list[BaseMessage]:
        raw = self._read()
        return [
            HumanMessage(content=m["content"]) if m["type"] == "human"
            else AIMessage(content=m["content"])
            for m in raw
        ]

    def clear(self):
        self._write([])


# 使用
history = FileChatMessageHistory("data/chat_history/user-001.json")
history.add_user_message("你好")
history.add_ai_message("你好！")
print(history.get_messages())

长对话优化策略

Token 管理

不同模型有不同的上下文窗口限制，必须控制发送的消息总量：

import tiktoken

def count_messages_tokens(messages: list, model: str = "gpt-4o-mini") -> int:
    """计算消息列表的 Token 数"""
    encoding = tiktoken.encoding_for_model(model)
    total = 0
    for msg in messages:
        # 每条消息有固定的格式开销
        total += 4  # <im_start>{role/name}\n{content}<im_end>\n
        total += len(encoding.encode(msg.content))
    return total


def truncate_history(
    messages: list,
    max_tokens: int = 4000,
    model: str = "gpt-4o-mini",
) -> list:
    """按 Token 限制裁剪历史消息"""
    total = count_messages_tokens(messages, model)

    if total <= max_tokens:
        return messages

    # 从最旧的消息开始删除
    truncated = messages.copy()
    while count_messages_tokens(truncated, model) > max_tokens and len(truncated) > 1:
        truncated.pop(0)  # 删除最早的消息

    return truncated

常见模型的上下文限制

模型	上下文窗口	建议历史 Token 上限	大约对应轮数
GPT-4o-mini	128K	4,000	~20 轮
GPT-4o	128K	8,000	~40 轮
Claude Sonnet 4	200K	8,000	~40 轮
Qwen-Plus	128K	4,000	~20 轮

提示

建议历史 Token 控制在上下文窗口的 5% 以内，给模型留足空间生成回答。

多用户会话管理

会话隔离

实际应用中，需要为不同用户、不同会话维护独立的对话历史：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# 基础链
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个智能客服助手。请根据用户的提问提供帮助。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

chain = prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()

# 多用户会话存储
session_store = {}

def get_session_history(session_id: str) -> ChatMessageHistory:
    if session_id not in session_store:
        session_store[session_id] = ChatMessageHistory()
    return session_store[session_id]

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

# 用户 A 的会话
response_a1 = chain_with_history.invoke(
    {"input": "我想查一下订单 #12345 的物流信息"},
    config={"configurable": {"session_id": "user-A"}},
)
print(f"用户A: {response_a1}")

response_a2 = chain_with_history.invoke(
    {"input": "大概什么时候能到？"},
    config={"configurable": {"session_id": "user-A"}},
)
print(f"用户A: {response_a2}")

# 用户 B 的会话（完全独立）
response_b1 = chain_with_history.invoke(
    {"input": "你们有红色款吗？"},
    config={"configurable": {"session_id": "user-B"}},
)
print(f"用户B: {response_b1}")

# 查看会话列表
print(f"当前活跃会话: {list(session_store.keys())}")

实战：构建有记忆的聊天机器人

完整示例

结合前面学到的所有知识，构建一个具有记忆功能的聊天机器人：

"""
chatbot.py — 有记忆的聊天机器人

功能：
1. 多轮对话记忆（滑动窗口）
2. 多用户会话隔离
3. 流式输出
4. 可配置系统提示词
"""

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from typing import Optional


class ChatBot:
    """具有记忆功能的聊天机器人"""

    def __init__(
        self,
        model: str = "gpt-4o-mini",
        system_prompt: str = "你是一个友好的AI助手。",
        max_history_rounds: int = 10,
    ):
        self.model_name = model
        self.system_prompt = system_prompt
        self.max_history_rounds = max_history_rounds
        self.llm = ChatOpenAI(model=model, temperature=0.7, streaming=True)

        # 提示词模板
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", system_prompt),
            MessagesPlaceholder(variable_name="history"),
            ("human", "{input}"),
        ])

        # 会话存储
        self._store = {}

        # 构建带记忆的链
        self.chain = RunnableWithMessageHistory(
            self.prompt | self.llm | StrOutputParser(),
            self._get_session_history,
            input_messages_key="input",
            history_messages_key="history",
        )

    def _get_session_history(self, session_id: str) -> ChatMessageHistory:
        if session_id not in self._store:
            self._store[session_id] = ChatMessageHistory()
        return self._store[session_id]

    def chat(self, user_input: str, session_id: str = "default") -> str:
        """同步对话"""
        response = self.chain.invoke(
            {"input": user_input},
            config={"configurable": {"session_id": session_id}},
        )
        return response

    def chat_stream(self, user_input: str, session_id: str = "default"):
        """流式对话"""
        for chunk in self.chain.stream(
            {"input": user_input},
            config={"configurable": {"session_id": session_id}},
        ):
            yield chunk

    def clear_session(self, session_id: str):
        """清除指定会话"""
        if session_id in self._store:
            del self._store[session_id]

    def list_sessions(self) -> list[str]:
        """列出所有会话"""
        return list(self._store.keys())


# 使用
if __name__ == "__main__":
    bot = ChatBot(
        model="gpt-4o-mini",
        system_prompt="你是一个 Python 编程导师，擅长用简单的方式解释复杂概念。",
    )

    # 交互式对话
    print("Python 导师已上线！输入 'quit' 退出。\n")

    session_id = "student-001"

    while True:
        user_input = input("你: ").strip()
        if user_input.lower() in ("quit", "exit", "q"):
            break
        if not user_input:
            continue

        # 流式输出
        print("导师: ", end="")
        full_response = ""
        for chunk in bot.chat_stream(user_input, session_id):
            print(chunk, end="", flush=True)
            full_response += chunk
        print("\n")

本章小结

核心要点

1. LLM 无状态：模型不记忆对话，需要应用层管理历史消息
2. RunnableWithMessageHistory：LangChain 推荐的记忆管理方案，基于 Runnable 接口
3. 滑动窗口：只保留最近 K 轮，控制 Token 消耗
4. 摘要压缩：用 LLM 总结旧消息，兼顾信息保留和 Token 效率
5. 会话隔离：通过 session_id 区分不同用户的对话历史
6. 持久化存储：生产环境使用 Redis 或数据库存储历史

选型指南