一个聊天机器人刚上线时，消息列表只有三五条，一切正常。

用户聊了半小时后，响应越来越慢，token 越来越多。为了控制成本，开发者粗暴删掉前一半历史。模型速度恢复了，却忘记用户姓名，还把一次工具调用的结果和另一轮问题混在一起。

这不是简单的「记住最近几轮」问题，而是上下文治理问题。

大模型通常不会替应用保存会话历史。应用需要决定消息归谁所有、保存在哪里、每次发送哪些内容、何时裁剪、何时摘要，以及哪些敏感信息根本不该留下。

本文从一个最小多轮对话开始，逐步增加会话隔离、token 预算、工具消息完整性和摘要策略。读完后，你应该能区分对话历史、短期记忆和长期记忆，而不是把它们统称为一个 messages 列表。

模型无状态，应用拥有状态

一次模型调用只看到这次传入的上下文。

模型负责生成，应用负责状态。这条边界非常重要。

如果两个用户共用同一个全局列表，就可能发生串话。如果用户刷新页面后历史全丢，说明状态只存在进程内存。如果服务扩容到多台机器，内存列表还会因为请求落到不同实例而失效。

运行前准备

下面的示例使用 Python 3.10+、OpenAI 集成和 python-dotenv。先安装依赖。

python -m pip install -U langchain langchain-openai python-dotenv

创建 .env，并将模型 ID 改为当前账号可以调用的值。

OPENAI_API_KEY=replace_with_your_openai_key
OPENAI_MODEL=gpt-5.4-mini

一个最小多轮对话

下面的程序可以直接运行。会话历史通过参数显式传入，因而不会意外借用进程级全局变量。它仍然只是内存中的演示，程序重启后历史会消失。

import os

from dotenv import load_dotenv
from langchain.messages import HumanMessage, SystemMessage
from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI


load_dotenv()

if not os.getenv("OPENAI_API_KEY"):
    raise RuntimeError("请设置 OPENAI_API_KEY")

model = ChatOpenAI(
    model=os.getenv("OPENAI_MODEL", "gpt-5.4-mini"),
    temperature=0,
)


def new_history() -> list[BaseMessage]:
    """为一个会话创建独立的初始消息历史。"""
    return [SystemMessage(content="你是简洁的学习助理。")]


def chat_once(history: list[BaseMessage], user_text: str) -> str:
    """完成一轮对话，并把双方的原始消息写回指定会话。"""
    history.append(HumanMessage(content=user_text))
    response = model.invoke(history)
    history.append(response)
    return str(response.content)


xiaoming_history = new_history()
print(chat_once(xiaoming_history, "我叫小明，正在学习 Python。"))
print(chat_once(xiaoming_history, "我叫什么，正在学什么?"))

这段代码有两个关键动作，调用前追加用户消息，调用后追加模型返回的原始 AIMessage。漏掉其中任何一个，下一轮上下文都会不完整。保留原始回复，而不是只保存它的文本，才能在后续接入工具调用、用量统计和供应商元数据时不丢协议字段。

xiaoming_history 只属于一个会话。真正的服务还要把这份状态写进会话存储，并在读取时校验身份与并发版本。

会话隔离比记忆算法更重要

每个会话至少需要稳定的 session_id 或 thread_id，并在服务端校验它属于当前用户或租户。

不要直接相信客户端传来的 thread ID。攻击者若能猜到其他会话标识，可能读取或污染别人的上下文。会话存储必须同时按用户、租户和 thread ID 做授权。

开发实验可以使用内存 Checkpointer。生产环境应使用数据库支持的 Checkpointer 或自己的持久化层，保证多实例、重启和并发更新下的一致性。

为什么不能无限追加历史

消息历史越长，模型输入 token 越多。成本和延迟会上升，最终还会超过上下文窗口。

更长也不一定更聪明。大量无关历史会稀释当前问题，旧指令可能与新要求冲突，模型还可能关注到早已失效的信息。

治理目标不是保留最多内容，而是在预算内保留完成当前任务所需的信息。

方案一, 只裁剪纯文本历史

按最近 N 轮裁剪简单，但不同轮次的长度可能差几十倍。更稳妥的是按 token 预算裁剪。下面这段只适用于没有工具调用的普通聊天历史。

from langchain.messages import AIMessage, ToolMessage, trim_messages


def contains_tool_protocol(history: list[BaseMessage]) -> bool:
    """判断历史中是否存在不能按单条消息随意切开的工具协议。"""
    return any(
        isinstance(message, ToolMessage)
        or (isinstance(message, AIMessage) and message.tool_calls)
        for message in history
    )


plain_history = xiaoming_history
if contains_tool_protocol(plain_history):
    raise ValueError("含工具调用的历史请使用下一节的完整轮次裁剪策略")


trimmed_history = trim_messages(
    plain_history,
    max_tokens=2_000,
    token_counter=model,
    strategy="last",
    include_system=True,
    start_on="human",
    allow_partial=False,
)

next_messages = [
    *trimmed_history,
    HumanMessage(content="请用一句话概括我正在学习什么。"),
]
response = model.invoke(next_messages)
print(response.content)

include_system=True 尝试保留系统消息，strategy="last" 优先保留最近内容，start_on="human" 帮助得到合理的对话起点。allow_partial=False 不截断一条普通文本消息。

token 统计可能依赖模型的计数能力，也可能只是近似值。应给实际输出预留空间，不能把整个上下文窗口都分给历史。这个示例刻意在发现 AIMessage.tool_calls 或 ToolMessage 时立即拒绝，因为工具协议不能被当成普通聊天文本处理。

裁剪时不能破坏工具调用协议

历史并不是任意消息数组。带工具调用的 AIMessage 与对应 ToolMessage 是一个协议单元。

如果裁剪后只剩 ToolMessage，模型不知道它对应哪个请求。如果保留工具调用却删掉结果，多数供应商也会拒绝消息序列。

trim_messages 可以参与消息裁剪，但它不会替业务定义一轮工具链的语义边界。对含工具调用的历史，先确认每个调用都有对应结果，再以完整用户轮次为最小裁剪单位。不要简单执行 history[-10:] 后就假定序列合法。

下面的辅助函数适用于常见的 SystemMessage* + (HumanMessage ...)* 会话形状。它有意做得严格，遇到未完成的工具调用、孤立工具结果或不以用户消息开始的历史就报错，而不是把无效上下文交给模型。

from collections.abc import Callable

from langchain.messages import (
    AIMessage,
    HumanMessage,
    SystemMessage,
    ToolMessage,
)
from langchain_core.messages import BaseMessage


TokenCounter = Callable[[list[BaseMessage]], int]


def validate_tool_call_pairs(history: list[BaseMessage]) -> None:
    """确认每个工具调用都有紧随其后的、ID 唯一匹配的工具结果。"""
    index = 0
    while index < len(history):
        message = history[index]
        if isinstance(message, ToolMessage):
            raise ValueError("发现孤立的 ToolMessage")

        if not isinstance(message, AIMessage) or not message.tool_calls:
            index += 1
            continue

        expected_ids = {call.get("id") for call in message.tool_calls}
        if None in expected_ids:
            raise ValueError("工具调用缺少 ID，不能安全裁剪")

        index += 1
        returned_ids: set[str] = set()
        while index < len(history) and isinstance(history[index], ToolMessage):
            tool_result = history[index]
            if tool_result.tool_call_id not in expected_ids:
                raise ValueError("ToolMessage 与前一个 AIMessage 的调用 ID 不匹配")
            if tool_result.tool_call_id in returned_ids:
                raise ValueError("同一个 tool_call_id 不能出现两次结果")
            returned_ids.add(tool_result.tool_call_id)
            index += 1

        if returned_ids != expected_ids:
            raise ValueError("AIMessage.tool_calls 没有得到完整的 ToolMessage 结果")


def split_complete_turns(
    history: list[BaseMessage],
) -> tuple[list[BaseMessage], list[list[BaseMessage]]]:
    """拆分开头系统消息和以 HumanMessage 开始的完整用户轮次。"""
    system_messages: list[BaseMessage] = []
    index = 0
    while index < len(history) and isinstance(history[index], SystemMessage):
        system_messages.append(history[index])
        index += 1

    turns: list[list[BaseMessage]] = []
    current_turn: list[BaseMessage] = []
    for message in history[index:]:
        if isinstance(message, HumanMessage):
            if current_turn:
                turns.append(current_turn)
            current_turn = [message]
        elif not current_turn:
            raise ValueError("除开开头系统消息后，每轮历史必须从 HumanMessage 开始")
        else:
            current_turn.append(message)

    if current_turn:
        turns.append(current_turn)
    return system_messages, turns


def flatten(turns: list[list[BaseMessage]]) -> list[BaseMessage]:
    """按原有顺序展开多轮消息。"""
    return [message for turn in turns for message in turn]


def trim_complete_turns(
    history: list[BaseMessage],
    max_tokens: int,
    token_counter: TokenCounter,
) -> list[BaseMessage]:
    """在 token 预算内保留最近完整轮次，不拆分工具调用和结果。"""
    validate_tool_call_pairs(history)
    system_messages, turns = split_complete_turns(history)
    if token_counter(system_messages) > max_tokens:
        raise ValueError("系统消息自身超过 token 预算，请缩短系统提示词或提高预算")

    kept_turns: list[list[BaseMessage]] = []

    for turn in reversed(turns):
        candidate = [*system_messages, *turn, *flatten(kept_turns)]
        if token_counter(candidate) > max_tokens:
            if not kept_turns:
                raise ValueError("最近一轮自身超过 token 预算，请缩短输入或提高预算")
            break
        kept_turns.insert(0, turn)

    return [*system_messages, *flatten(kept_turns)]

tool_history = [
    SystemMessage(content="你是天气助手。"),
    HumanMessage(content="北京天气如何?"),
    AIMessage(
        content="",
        tool_calls=[
            {
                "name": "get_weather",
                "args": {"city": "北京"},
                "id": "call_weather_001",
                "type": "tool_call",
            }
        ],
    ),
    ToolMessage(
        content="北京晴，最高温度 28 摄氏度。",
        tool_call_id="call_weather_001",
        name="get_weather",
    ),
    AIMessage(content="北京晴，最高温度 28 摄氏度。"),
    HumanMessage(content="请用一句话重述刚才的天气。"),
]

trimmed_tool_history = trim_complete_turns(
    tool_history,
    max_tokens=2_000,
    token_counter=model.get_num_tokens_from_messages,
)
answer = model.invoke(trimmed_tool_history)
print(answer.content)

降低 max_tokens 时，这个策略会保留或丢弃整个用户轮次，绝不会只留下半段工具协议。协议正确不等于语义一定充足，若早期工具结果仍是当前问题的必要依据，应把结论写入摘要或权威的结构化状态，而不是强行保留全部原始消息。

方案二, 摘要早期历史

直接删除消息会丢信息。对于长对话，可以把较早内容压缩成摘要，再保留最近原始消息。

摘要应保留稳定事实、用户目标、已经做出的决定、未完成事项和必要约束。闲聊、重复表达和已经失效的中间细节可以压缩。

摘要不是无损压缩。生成摘要的模型可能遗漏或改写事实，因此高风险信息不要只存在自然语言摘要中。订单 ID、金额、审批状态和权限信息更适合结构化存储。

LangChain Agent 可以使用 SummarizationMiddleware 在达到 token 条件时自动摘要。普通聊天程序也可以在自己的会话层实现同样策略。

方案三, 把稳定事实从消息中分离

用户姓名、偏好、账户状态和任务进度不应该永远靠翻历史寻找。

短期历史负责当前对话连贯性。长期存储负责跨会话稳定信息。业务数据库负责订单、权限和交易等权威状态。

这三类数据的生命周期、修改权限和可信度不同，不应全部塞进 messages。

历史应该保存多久

没有统一答案，需要同时考虑产品体验、法律合规、成本和用户选择。

数据	可能策略	主要风险
当前会话消息	会话期或短期保存	上下文成本、敏感信息
会话摘要	随会话保存	摘要失真、事实遗漏
用户偏好	用户可查看和删除	过度画像、错误偏好
业务状态	按业务系统规则保存	权限与一致性
Trace	按观测策略采样和保留	隐私与第三方传输

给用户提供清除会话、关闭个性化或删除长期偏好的能力，比偷偷保存所有历史更可靠。

一套实用的上下文组装顺序

每次调用前，可以按以下顺序构造上下文。

上下文组装应该是一个可测试的模块。给定同一会话状态，它应能输出可预测的消息序列，并保证不超过预算。

多轮聊天的检查清单

• 每个会话都有稳定且经过授权的 ID。
• 用户消息和 AI 回复都被追加，失败调用有明确处理。
• 历史按 token 而不是只按轮数控制。
• system 消息和工具调用协议在裁剪后仍然有效。
• 关键业务事实存入结构化系统，不只存在摘要。
• 会话、长期偏好和业务状态分开管理。
• 敏感数据有保留期限、删除入口和观测脱敏策略。

真正成熟的聊天系统，不是记住一切，而是知道什么该留、什么该忘、什么必须交给权威系统保存。

对话历史只是上下文。治理它，才接近记忆。

延伸阅读

• LangChain Short-term memory
• LangChain Messages
• trim_messages API
• LangGraph Persistence