记忆与对话历史

LLM 为什么是无状态的？

LLM 本身是完全无状态的——每次 API 调用都是独立的，模型不会"记住"上一次你问了什么。这是其底层架构决定的：模型只处理当前请求的输入 token，不存储任何用户会话状态。

要实现多轮对话，应用层需要自己维护对话历史，并在每次请求时将历史消息拼入上下文发送给模型。LangChain 的记忆（Memory）系统就是为此而生的。

InMemoryChatMessageHistory：内存存储

最简单的历史存储，适合原型开发和测试（重启服务后历史丢失）：

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage

# 直接使用 ChatMessageHistory
history = InMemoryChatMessageHistory()
history.add_user_message("我叫小明")
history.add_ai_message("你好，小明！")

print(history.messages)
# [HumanMessage(content='我叫小明'), AIMessage(content='你好，小明！')]

RunnableWithMessageHistory：LCEL 对话链

这是构建多轮对话应用的标准模式：

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 内存存储（生产中替换为 Redis 等）
store: dict[str, InMemoryChatMessageHistory] = {}

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

# 带 MessagesPlaceholder 的提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位友好的 AI 助手，记住用户的偏好并个性化回复。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

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

# 包裹成有记忆的链
chain_with_history = RunnableWithMessageHistory(
    base_chain,
    get_session_history,
    input_messages_key="input",     # 哪个字段是用户输入
    history_messages_key="history", # 历史注入到哪个占位符
)

# 第一轮对话
response1 = chain_with_history.invoke(
    {"input": "我叫小明，我喜欢 Python 编程"},
    config={"configurable": {"session_id": "user_001"}},
)
print(response1)

# 第二轮对话（模型记得上文）
response2 = chain_with_history.invoke(
    {"input": "你还记得我的名字和爱好吗？"},
    config={"configurable": {"session_id": "user_001"}},
)
print(response2)  # 模型会回复小明和 Python 的内容

RedisChatMessageHistory：生产级持久化

生产环境中，对话历史需要持久化存储，以在服务重启后保留，并支持多实例水平扩展。Redis 是最常用的选择：

from langchain_community.chat_message_histories import RedisChatMessageHistory

def get_redis_history(session_id: str) -> RedisChatMessageHistory:
    return RedisChatMessageHistory(
        session_id=session_id,
        url="redis://localhost:6379",
        ttl=86400,  # 24小时过期，防止历史无限积累
        key_prefix="chat:",  # Redis key 前缀
    )

chain_with_redis = RunnableWithMessageHistory(
    base_chain,
    get_redis_history,
    input_messages_key="input",
    history_messages_key="history",
)

# 用法与内存版本完全相同
response = chain_with_redis.invoke(
    {"input": "你好"},
    config={"configurable": {"session_id": "session_abc"}},
)

其他持久化后端

历史窗口截断：避免 Context 溢出

随着对话轮次增加，历史消息会越来越多，可能超出模型的 context window 限制（GPT-4o 为 128K token，Claude 3.5 为 200K token）。需要实施截断策略：

from langchain_core.messages import trim_messages
from langchain_core.runnables import RunnablePassthrough
from operator import itemgetter

# trim_messages：按 token 数截断
trimmer = trim_messages(
    max_tokens=2000,         # 最多保留 2000 token 的历史
    strategy="last",          # 保留最近的消息
    token_counter=model,        # 用模型计算 token 数
    include_system=True,      # 保留 SystemMessage
    allow_partial=False,      # 不保留不完整的消息
    start_on="human",         # 截断后从 HumanMessage 开始
)

# 将 trimmer 插入链中
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是 AI 助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

chain = (
    RunnablePassthrough.assign(
        history=itemgetter("history") | trimmer  # 历史先截断再注入
    )
    | prompt
    | model
    | StrOutputParser()
)

手动管理对话历史的模式

对于更复杂的场景（如 Agent 决策、工具调用），有时需要手动控制历史消息的增减：

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# 维护一个本地消息列表
messages = [
    SystemMessage(content="你是一个代码审查助手。")
]

def chat(user_input: str) -> str:
    messages.append(HumanMessage(content=user_input))
    response = model.invoke(messages)
    messages.append(response)  # AIMessage 直接追加
    return response.content

# 多轮对话
print(chat("帮我审查这段代码：for i in range(10): print(i)"))
print(chat("有没有更 Pythonic 的写法？"))  # 记得上下文
print(chat("如果需要同时打印下标和值呢？"))

摘要记忆：ConversationSummaryMemory

当对话非常长时，窗口截断会丢失早期重要信息。摘要记忆用 LLM 将旧历史压缩为摘要，兼顾历史信息的保留与长度控制：

from langchain.memory import ConversationSummaryBufferMemory

# 混合策略：最近消息原文 + 更早的对话摘要
memory = ConversationSummaryBufferMemory(
    llm=model,
    max_token_limit=1000,       # 超过 1000 token 就开始摘要
    return_messages=True,
)

memory.save_context(
    {"input": "我正在学 LangChain"},
    {"output": "很好！LangChain 是构建 LLM 应用的强大框架。"}
)

# 当历史太长时，自动调用 LLM 生成摘要
history = memory.load_memory_variables({})
print(history["history"])

多用户会话隔离

在 Web 应用中，需要为每个用户维护独立的对话历史。session_id 就是实现隔离的关键：

import uuid
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    message: str
    session_id: str = None

@app.post("/chat")
async def chat(req: ChatRequest):
    session_id = req.session_id or str(uuid.uuid4())
    response = await chain_with_history.ainvoke(
        {"input": req.message},
        config={"configurable": {"session_id": session_id}},
    )
    return {"response": response, "session_id": session_id}

本章小结

• LLM 本身无状态，对话历史需由应用层维护并每次传入
• RunnableWithMessageHistory + ChatMessageHistory 是 LangChain 0.3.x 的标准对话模式
• InMemoryChatMessageHistory 适合开发测试，RedisChatMessageHistory 适合生产
• trim_messages() 按 token 数截断历史，防止 context 溢出
• 使用唯一 session_id 隔离不同用户的对话历史
• 摘要记忆（ConversationSummaryBufferMemory）适合超长对话场景