【LangChain】模型为什么记不住上次对话？从消息类型封装到上下文裁剪与过滤合并全解析

feng_qi0325

6087人浏览 · 2026-08-17 22:26:34

feng_qi0325 · 2026-08-17 22:26:34 发布

11. 聊天模型-消息管理

LangChain 消息系统是对原生大模型消息格式的统一抽象封装。理解消息的底层机制，才能高效实现多轮对话、上下文窗口管理以及消息列表的增删改查。

1. 消息系统概述

1.1 原生大模型消息结构

在了解 LangChain 的消息之前，先回顾原生大语言模型的消息格式。以 OpenAI 为例，每条消息包含两个核心字段：

role —— 身份标识（system / user / assistant / tool）
content —— 消息正文

# 原生 OpenAI 调用示例
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "你是一个非常有帮助的助手"},
        {"role": "user", "content": "请介绍一下 LangChain"},
    ]
)
print(response.choices[0].message.role)      # assistant
print(response.choices[0].message.content)   # 模型回复内容

常见角色说明：

角色	含义	说明
`system`	系统角色	定义对话行为与场景上下文
`user`	用户角色	用户的输入内容
`assistant`	助理角色	模型生成的响应
`tool`	工具角色	工具调用的返回结果

不同模型对角色名称的命名可能不同。例如 OpenAI 的 system 角色实际使用 developer 标识，但语义一致。

消息内容可以是多模态的（文本、图像、音频），具体取决于底层模型的能力。当前我们以文本消息为主，但 LangChain 的消息体系同样支持多模态内容的封装。

1.2 LangChain 消息类型封装

LangChain 对原生消息格式做了一层统一封装，所有模型只认一套消息类型：

LangChain 类型	对应原生角色	说明
`SystemMessage`	`system`	系统消息
`HumanMessage`	`user`	用户消息
`AIMessage`	`assistant`	模型回复
`AIMessageChunk`	—	流式传输的数据块
`ToolMessage`	`tool`	工具调用结果

跨模型的统一性：使用 LangChain 时，无论底层是 GPT、Claude、DeepSeek 还是 Gemini，都使用同一套消息类型。LangChain 在调用时自动将其转换为目标模型的原生格式。

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

messages = [
    SystemMessage(content="你是一个有用的助手。"),
    HumanMessage(content="什么是 LangChain？"),
]

1.3 BaseMessage 抽象类

所有消息类型均继承自 BaseMessage，它定义了消息的基础结构：

核心属性：

属性	类型	说明
`content`	`str`	消息的文本正文
`additional_kwargs`	`dict`	工具调用等附加负载数据
`response_metadata`	`dict`	响应元数据（响应头 token、模型名称等）
`type`	`str`	消息类型标识
`name`	`str`	消息可选名称
`id`	`str`	消息唯一标识

内置方法：

# pretty_print —— 美化打印消息
ai_msg = model.invoke("介绍一下 LangChain")
ai_msg.pretty_print()
# 输出格式：
# ========== AIMessage ==========
# LangChain 是一个用于构建 LLM 应用的框架...

# pretty_print 支持设置截断字数，避免内容过长
ai_msg.pretty_print(truncate=50)

# pretty_repr() —— 获取消息的美化文本表示（可选 HTML 格式），返回字符串而非直接打印
text_repr = ai_msg.pretty_repr()
text_repr_html = ai_msg.pretty_repr(html=True)

# content() —— 获取文本内容，等价于 .content
text = ai_msg.content()

# to_string() —— 字符串化输出

2. 多轮对话实现

2.1 原生模型无记忆

直接调用原生大模型时，每次请求都是独立的，模型不会记住之前的对话内容：

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

# 第一轮：自我介绍
model.invoke("我是小明，你好！").pretty_print()
# → "你好，小明！很高兴认识你！今天有什么想聊的呢？"

# 第二轮：询问身份
model.invoke("你知道我是谁吗？").pretty_print()
# → "抱歉，我不知道你是谁"  ← 没有记忆！

2.2 手动消息列表实现记忆

多轮对话的本质是将历史消息一并发送给模型。手动保存 AI 的回复并打包到消息列表中：

from langchain_core.messages import HumanMessage, AIMessage

# 手动构建多轮对话消息列表
messages = [
    HumanMessage(content="我是小明，你好！"),
    AIMessage(content="你好，小明！很高兴认识你！今天有什么想聊的呢？"),
    HumanMessage(content="你知道我是谁吗？"),
]
model.invoke(messages).pretty_print()
# "我知道你叫小明，但是不了解你的其他信息"

核心规律：只要将聊天模型的返回结果保存下来，与新的消息一起打包发送，模型就能表现出有记忆功能的效果。

2.3 RunnableWithMessageHistory

LangChain 提供了 RunnableWithMessageHistory 类，用于自动管理历史消息的存储与注入。

from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_core.runnables import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

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

# 1. 定义内存存储
store = {}

# 2. 根据会话 ID 查询消息列表
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# 3. 包装 model，使其具备存储历史消息的能力
with_history_model = RunnableWithMessageHistory(model, get_session_history)

# 4. 配置会话 ID
config = {"configurable": {"session_id": "1"}}

# 5. 多轮对话
with_history_model.invoke(
    [HumanMessage(content="我是小明，你好！")],
    config=config,
).pretty_print()
# → "你好，小明！很高兴认识你！"

with_history_model.invoke(
    [HumanMessage(content="你知道我是谁吗？")],
    config=config,
).pretty_print()
# → "我知道你是小明"  ← 记住了！

核心机制：

参数说明：

参数	类型	说明
`model`	`Runnable`	需要包装的聊天模型
`get_session_history`	`Callable`	根据 session_id 返回 `BaseChatMessageHistory` 的函数
`session_id` (config)	`str`	通过 `config` 传入，用于区分不同会话

2.4 注意事项：v0.3 弃用说明

从 LangChain 0.3 版本开始，官方不再推荐使用 RunnableWithMessageHistory，原因如下：

功能有限，不适合复杂的 AI 应用程序
历史消息管理应作为独立组件实现，而非与模型耦合
推荐做法：在 LangGraph 中通过持久化（Persistence）来管理记忆

配套的 InMemoryChatMessageHistory 同样已随 v0.3 弃用，不推荐在生产环境中使用。

后续关于记忆管理（内存缓存、第三方数据库如 Redis）的完整方案将在 LangGraph 持久性中讲解。

3. 上下文窗口与 Token

3.1 上下文窗口概念

上下文窗口（Context Window） 指大模型一次可以处理的最大 token 数，包含输入和输出两部分。

模型不同，上下文窗口大小也各异。以 OpenAI 系列为例：GPT-4o 为 128K tokens，GPT-4.1 约 1M tokens，GPT-5 达 400K tokens（具体数值以各模型官方文档为准）。若输入占据过多空间，输出的可用 token 就会减少，可能导致回复不完整甚至完全无法输出。

3.2 Token 计算基础

Token 是自然语言处理中的基本文本单位，不同语言的 token 换算关系不同：

语言	换算关系
英文	1 token ≈ 0.75 个单词，1000 tokens ≈ 750 个单词
中文	1 汉字 ≈ 1.5~2 tokens，1000 tokens ≈ 500~700 个汉字

常见词（如 apple）可能是一个 token，生僻字或复杂词可能被拆成多个 token。

3.3 管理消息列表的必要性

多轮对话的本质是将完整上下文反复发送给模型：

Round 1: [SystemMessage, Human1] → AIMessage1
Round 2: [SystemMessage, Human1, AIMessage1, Human2] → AIMessage2
Round 3: [SystemMessage, Human1, AIMessage1, Human2, AIMessage2, Human3] → AIMessage3
...

随着轮数增加，历史消息不断累积，最终会超出上下文窗口限制。因此必须对消息列表进行管理，主要操作包括：

裁剪（Trim） —— 按 token 数或消息数移除早期消息
过滤（Filter） —— 按类型或 ID 筛选保留的消息
合并（Merge） —— 合并连续的同类型消息

4. 消息裁剪（Trimming）

trim_messages 是 LangChain 提供的消息裁剪工具，支持按 token 数和按消息条数两种裁剪方式。

4.1 按 Token 数裁剪

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, trim_messages
from langchain_openai import ChatOpenAI

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

# 模拟历史消息
messages = [
    SystemMessage(content="you're a good assistant"),
    HumanMessage(content="hi! I'm bob"),
    AIMessage(content="hi!"),
    HumanMessage(content="I like vanilla ice cream"),
    AIMessage(content="nice"),
    HumanMessage(content="whats 2 + 2"),
    AIMessage(content="4"),
    HumanMessage(content="thanks"),
    AIMessage(content="no problem!"),
    HumanMessage(content="having fun?"),
    AIMessage(content="yes!"),
    HumanMessage(content="What's my name?"),
]

# 定义裁剪器：将输入限制在 65 tokens 以内
trimmer = trim_messages(
    max_tokens=65,          # token 数量上限
    strategy="last",        # 保留策略："last"（保留最近的）或 "first"
    token_counter=model,    # 使用模型的 token 计算方法
    include_system=True,    # 始终保留第一条 system 消息
    allow_partial=False,    # 不允许拆分单条消息内容
    start_on="human",       # 裁剪后第一条消息（除 system 外）必须是 human
)

# 构建链：先裁剪 -> 再调用模型
chain = trimmer | model
result = chain.invoke(messages)

# 查看裁剪后的 token 数
print(result.response_metadata["token_usage"])
# 输入 tokens 从 88 -> 60, 符合 65 上限

4.2 按消息条数裁剪

将 token_counter 设为 len 函数，max_tokens 的含义变为消息条数上限：

trimmer = trim_messages(
    max_tokens=5,           # 保留 5 条消息
    strategy="last",
    token_counter=len,      # 使用 len 计算消息数量而非 token 数
    include_system=True,
    allow_partial=False,
    start_on="human",
)

trimmed = trimmer.invoke(messages)
print(len(trimmed))  # 输出 ≤ 5 条消息

4.3 裁剪参数详解

参数	说明	建议
`max_tokens`	保留的上限值（token 数或消息数）	根据模型上下文窗口调整
`strategy`	`"last"` 保留最近消息 / `"first"` 保留最早消息	多数场景用 `"last"`
`token_counter`	token 计算方式：传 model 用模型算法，传 `len` 按消息条数	进阶可用自定义函数
`include_system`	是否始终保留第一条 system 消息	建议 `True`
`allow_partial`	是否允许从消息中间拆分	建议 `False`，避免语义破坏
`start_on`	裁剪后第一条消息类型（如 `"human"`）	确保对话以人类消息开始
`ends_on`	裁剪后最后一条消息类型，如 `("human", "tool")`	与 `start_on` 对称，控制尾部消息角色

5. 消息过滤（Filtering）

filter_messages 可按类型、ID 或组合条件从消息列表中筛选所需消息。

5.1 按类型过滤

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, filter_messages

messages = [
    SystemMessage("你是一个聊天助手", id="1"),
    HumanMessage("示例输入", id="2"),
    AIMessage("示例输出", id="3"),
    HumanMessage("真实输入", id="4"),
    AIMessage("真实输出", id="5"),
]

# 方式一：使用 invoke
result = filter_messages(include_types="human").invoke(messages)
# → 只保留 human 消息（id="2" 和 id="4"）

# 方式二：直接调用（等价）
result = filter_messages(messages, include_types="human")

5.2 按 ID 过滤

# 排除指定 ID 的消息
result = filter_messages(messages, exclude_ids=["3"])
# 保留了 id=1,2,4,5，排除了 id=3（AIMessage）

# 仅保留指定 ID
result = filter_messages(messages, include_ids=["1", "4"])

5.3 组合过滤

# 同时按 ID 和类型筛选：排除 id=3，且只保留 HumanMessage 和 AIMessage
result = filter_messages(
    messages,
    exclude_ids=["3"],
    include_types=[HumanMessage, AIMessage],
)
# → 保留了 id=2,4,5
#   id=1 (SystemMessage) 被类型过滤掉
#   id=3 (AIMessage) 被 ID 排除掉

filter_messages 完整参数（可在官方 API Reference 中查询）：

参数	类型	说明
`include_types`	`str` / `List[Type]`	要保留的消息类型
`exclude_types`	`str` / `List[Type]`	要排除的消息类型
`include_ids`	`List[str]`	要保留的消息 ID
`exclude_ids`	`List[str]`	要排除的消息 ID
`include_names`	`List[str]`	按 name 字段包含
`exclude_names`	`List[str]`	按 name 字段排除

6. 消息合并（Merging）

merge_message_runs 用于将连续的同类型消息合并为一条，简化消息列表。

6.1 合并连续同类型消息

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, merge_message_runs
from langchain_openai import ChatOpenAI

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

messages = [
    SystemMessage("你是一个聊天助手。"),
    SystemMessage("你总是以笑话回应。"),          # 连续 system，将被合并
    HumanMessage("为什么要使用 LangChain?"),
    HumanMessage("为什么要使用 LangGraph?"),      # 连续 human，将被合并
    AIMessage("因为当你试图让你的代码更有条理时，LangGraph 会让你感到"节点"是个好主意！"),
    AIMessage("不过别担心，它不会"分散"你的注意力！"),  # 连续 ai，将被合并
    HumanMessage("选择LangChain还是LangGraph?"),
]

# 方式一：直接调用合并
merged = merge_message_runs(messages)
# 合并后：
#   [SystemMessage("你是一个聊天助手。\n你总是以笑话回应。"),
#    HumanMessage("为什么要使用 LangChain?\n为什么要使用 LangGraph?"),
#    AIMessage("因为...好主意！\n不过别担心..."),
#    HumanMessage("选择LangChain还是LangGraph?")]

# 方式二：定义成链，合并后再调用模型
chain = merge_message_runs | model
result = chain.invoke(messages)

6.2 与模型集成

合并操作通常放在链的最前端，先精简消息列表再传入模型：

# 推荐：裁剪 + 合并 + 模型组合使用
from langchain_core.messages import trim_messages, merge_message_runs

pipeline = merge_message_runs | trim_messages(max_tokens=100, ...) | model
result = pipeline.invoke(messages)

总结

操作	工具	适用场景
消息回顾	`BaseMessage` 体系	理解消息类型、属性和内置方法
多轮对话	`RunnableWithMessageHistory`	让模型具备短期记忆（v0.3 前推荐，现推荐 LangGraph 方案）
裁剪	`trim_messages`	控制上下文窗口大小，避免超限
过滤	`filter_messages`	按类型/ID 提取或排除特定消息
合并	`merge_message_runs`	精简连续同类型消息，减少 token 消耗