LangChain 核心组件解析 (消息)
目录
2.2 使用 RunnableWithMessageHistory 自动管理
在大语言模型（LLM）的应用开发中，如何高效、规范地与模型交互是构建可靠应用的关键。LangChain 作为 LLM 应用开发的主流框架，通过一系列精心设计的核心组件，将复杂的交互流程抽象化、标准化。本文将深入解析其中最基础也最重要的部分 ——消息（Messages）与历史消息管理，带你掌握构建连贯、智能对话系统的核心能力。
一、消息（Messages）：LLM 交互的通用语言
消息是聊天模型的通信单位，它不仅承载了用户的输入和模型的输出，还包含了引导对话的上下文信息。理解消息的结构和类型，是使用 LangChain 的第一步。
1.1 LLM 原生消息结构
在与 LLM 交互时，每条消息都包含一个角色（Role）和内容（Content），以及因模型而异的附加元数据。
| 角色 (Role) | 描述 |
|---|---|
| system (系统角色) | 用于告诉聊天模型如何行为并提供额外的上下文，例如设定角色或对话基调。 |
| user (用户角色) | 表示用户与模型交互的输入，通常是文本或其他交互式输入。 |
| assistant (助理角色) | 表示来自模型的响应，其中可以包含文本或调用工具的请求。 |
| tool (工具角色) | 用于在检索外部数据或将工具调用的结果传递回模型的消息。 |
以 OpenAI API 为例，一个典型的消息列表格式如下：
[
{"role": "user", "content": "Hello, how are you?"},
{"role": "assistant", "content": "I'm doing well, thank you for asking."},
{"role": "user", "content": "Can you tell me a joke?"}
]
1.2 LangChain 统一消息格式
为了解决不同 LLM 提供商消息格式不统一的问题，LangChain 提供了一套抽象的消息类型，让开发者可以无缝切换模型，而无需关心底层细节。
| LangChain 消息类型 | 对应角色 | 描述 |
|---|---|---|
| SystemMessage | system | 用于启动 AI 模型的行为并提供额外的上下文。 |
| HumanMessage | user | 表示用户与模型交互的输入。 |
| AIMessage | assistant | 来自模型的响应，可包含文本或工具调用请求。 |
| AIMessageChunk | assistant | 用于流式响应，在生成聊天模型时逐块传输响应。 |
| ToolMessage | tool | 表示工具角色的消息，包含调用工具的结果。 |
这些消息类型都是 BaseMessage 抽象类的子类，它们共同构成了 LangChain 聊天模型的输入和输出。
1.3 BaseMessage 抽象类
langchain_core.messages.base.BaseMessage 是所有消息的基类，它定义了消息的核心属性和方法：
-
核心参数：
content: 消息的字符串内容。additional_kwargs: 与消息关联的其他有效负载数据，如模型提供的工具调用。response_metadata: 响应元数据，如响应标头、令牌计数、模型名称等。type: 消息的类型，用于反序列化时识别消息类。name: 可选的消息名称，提供人类可读的标识。id: 可选的唯一标识符。
-
内置方法：
pretty_print(): 打印消息的漂亮表示。pretty_repr(): 获得消息的漂亮表示。text(): 获取消息的文本内容。
二、缓存历史消息：实现多轮对话的关键
LLM 本身是无状态的，每次调用都是独立的。要实现连贯的多轮对话，就必须将历史消息作为上下文，在每次新请求时重新发送给模型。
2.1 手动管理对话历史
最直接的方法是手动维护一个消息列表，每次交互后将新的 HumanMessage 和 AIMessage 添加进去。
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
model = ChatOpenAI(model="gpt-4o-mini")
# 手动构建包含历史的消息列表
messages = [
HumanMessage(content="Hi! I'm Bob"),
AIMessage(content="Hello Bob! How can I assist you today?"),
HumanMessage(content="What's my name?"),
]
# 将完整历史发送给模型
model.invoke(messages).pretty_print()
输出结果：
==================================================
Ai Message
==================================================
Your name is Bob! How can I help you today, Bob?
这种方法虽然有效，但在复杂应用中会变得繁琐且容易出错。
2.2 使用 RunnableWithMessageHistory 自动管理
为了简化历史管理，LangChain 提供了 RunnableWithMessageHistory 包装器。它可以自动跟踪模型的输入和输出，并将其存储在指定的存储介质中，未来的交互会自动加载这些历史。
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
model = ChatOpenAI(model="gpt-4o-mini")
# 定义一个存储，用于区分不同会话
store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:
if session_id not in store:
store[session_id] = InMemoryChatMessageHistory()
return store[session_id]
# 包装模型，使其具备记忆能力
with_message_history = RunnableWithMessageHistory(model, get_session_history)
# 配置会话ID
config = {"configurable": {"session_id": "1"}}
# 第一次对话
with_message_history.invoke([HumanMessage(content="Hi! I'm Bob")], config=config).pretty_print()
# 第二次对话，模型会自动记住之前的信息
with_message_history.invoke([HumanMessage(content="What's my name?")], config=config).pretty_print()
输出结果：
==================================================
Ai Message
==================================================
Hi Bob! How can I assist you today?
==================================================
Ai Message
==================================================
Your name is Bob! How can I help you today, Bob?
⚠️ 重要提示：从 LangChain v0.3 开始，官方建议使用 LangGraph 持久性 来替代
RunnableWithMessageHistory，因为它更灵活，能更好地支持多用户、多对话场景。
三、管理历史消息：在 Token 限制下高效对话
模型的上下文窗口（Context Window）是有限的，它决定了模型一次能处理的最大 Token 数量。如果历史消息过长，就会超出 Token 限制，导致请求失败。因此，管理历史消息的核心在于在保持对话连贯性的同时，有效控制 Token 数量。
3.1 理解上下文窗口与 Token
- 上下文窗口：可以理解为模型的 “短期工作记忆区”，是 LLM 在一次处理请求时，所能查看和处理的最大 Token 数量。它包含了用户输入、模型输出、系统指令和对话历史。
- Token：是文本的基本单位。对于英文，1 个 Token ≈ 4 个字符或 0.75 个单词；对于中文，1 个汉字 ≈ 1.5-2 个 Token。
不同模型的上下文窗口大小不同，例如：
- OpenAI GPT-5: 400,000 Token
- GPT-4 Turbo: 128,000 Token
四、消息裁剪：在 Token 限制下的智能对话管理
在上一部分中，我们了解了消息的基本结构和历史管理。但在实际应用中，模型的上下文窗口（Context Window）是有限的。当对话历史不断累积，Token 数量超过模型上限时，就会导致请求失败。因此，** 消息裁剪（Message Trimming）** 成为了在 Token 限制下维持对话连贯性的关键技术。
4.1 基于 Token 数的智能裁剪
LangChain 提供了强大的 trim_messages 函数，它允许我们根据指定的 Token 数量，智能地裁剪历史消息列表，确保每次发送给模型的上下文都在安全范围内。
核心参数解析
max_tokens: 裁剪后消息列表的最大 Token 数。strategy: 裁剪策略，决定保留哪些消息。"last"(默认): 保留列表中最后的消息，即最新的对话。"first": 保留列表中最早的消息。
token_counter: 用于计算 Token 数的函数。可以传入模型本身（如model），或简单的len函数（基于消息条数裁剪）。include_system: 是否始终保留SystemMessage，默认为True。allow_partial: 是否允许拆分消息，默认为False。start_on: 指定裁剪后列表的起始消息类型，确保对话结构完整。
实战示例
假设我们有一个较长的对话历史，模型的输入 Token 数为 88，而我们希望将其限制在 65 Token 以内：
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, trim_messages
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?"),
]
# 创建裁剪器
trimmer = trim_messages(
max_tokens=65,
strategy="last",
token_counter=model,
include_system=True,
allow_partial=False,
start_on="human",
)
# 执行裁剪并调用模型
chain = trimmer | model
print(chain.invoke(messages))
输出结果：
content='I don't know your name. Would you like to share it with me?'
...
usage_metadata={
'input_tokens': 60, # Token 数已被成功裁剪
'output_tokens': 16,
'total_tokens': 76
}
从结果可以看出，模型不再认识 “Bob”，因为早期的对话历史（包括自我介绍）已被裁剪，只保留了最近的交互。这证明了裁剪策略 "last" 正在生效。
4.2 基于消息条数的简易裁剪
除了基于 Token 数，我们还可以使用 token_counter=len 来根据消息条数进行裁剪，这在快速原型开发中非常方便：
trimmer = trim_messages(
max_tokens=1, # 这里的“tokens”指消息条数
strategy="last",
token_counter=len,
include_system=True,
)
print(trimmer.invoke(messages))
结果：
[
SystemMessage(content="You're a good assistant", ...),
HumanMessage(content="what's my name?", ...)
]
4.3 对话模式的完整性
无论采用何种裁剪策略，都必须遵循对话模式的基本原则，以确保模型能正确理解上下文：
- 对话开头：聊天记录必须以
HumanMessage或SystemMessage开头，后跟HumanMessage。可通过start_on="human"强制。 - 对话结尾：聊天记录必须以
HumanMessage或ToolMessage结尾。可通过ends_on="human"强制。 - ToolMessage：只能出现在调用工具的
AIMessage之后。
五、消息过滤与合并：复杂场景下的消息流优化
在更复杂的应用场景中，我们不仅需要裁剪消息，还需要对消息流进行更精细的控制。LangChain 提供了 filter_messages 和 merge_message_runs 两个强大的工具。
5.1 消息过滤（Filtering）
filter_messages 允许我们根据消息的类型（type）、ID 或名称（name）来筛选消息，只将我们关心的部分传递给模型。
按类型筛选
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, filter_messages
messages = [
SystemMessage(content="你是一个聊天助手", id="1"),
HumanMessage(content="示例输入", additional_kwargs={"name": "example_user"}, id="2"),
AIMessage(content="示例输出", id="3"),
HumanMessage(content="真实输入", id="4"),
AIMessage(content="真实输出", id="5"),
]
# 只保留 HumanMessage
filtered = filter_messages(messages, include_types="human")
for msg in filtered:
print(f"{msg.type}: {msg.content} (id={msg.id})")
输出：
human: 示例输入 (id=2)
human: 真实输入 (id=4)
按类型和 ID 组合筛选
# 保留 HumanMessage 和 AIMessage，但排除 id="3"
filtered = filter_messages(messages, include_types=[HumanMessage, AIMessage], exclude_ids=["3"])
for msg in filtered:
print(f"{msg.type}: {msg.content} (id={msg.id})")
输出：
human: 示例输入 (id=2)
human: 真实输入 (id=4)
ai: 真实输出 (id=5)
5.2 合并连续消息（Merging）
某些模型不支持连续的相同类型消息（例如，连续两条 HumanMessage）。merge_message_runs 可以轻松地将连续的同类型消息合并为一条，避免模型报错。
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, merge_message_runs
messages = [
SystemMessage(content="你是一个聊天助手。"),
SystemMessage(content="你总是以笑话回应。"),
HumanMessage(content="为什么要使用 LangChain?"),
HumanMessage(content="为什么要使用 LangGraph?"),
AIMessage(content="因为当你试图让你的代码更有条理时，LangGraph 会让你感到\"节点\"是个好主意！"),
AIMessage(content="不过别担心，它不会\"分散\"你的注意力！"),
HumanMessage(content="选择LangChain还是LangGraph?"),
]
merged = merge_message_runs(messages)
for x in merged:
print(f"\n{x.type}:")
print(x.content)
合并结果：
system:
你是一个聊天助手。
你总是以笑话回应。
human:
为什么要使用 LangChain?
为什么要使用 LangGraph?
ai:
因为当你试图让你的代码更有条理时，LangGraph 会让你感到"节点"是个好主意！
不过别担心，它不会"分散"你的注意力！
human:
选择LangChain还是LangGraph?
合并后的消息列表更加整洁，符合大多数模型的输入要求。我们可以将 merge_message_runs 作为预处理步骤，直接集成到我们的处理链中：
merger = merge_message_runs()
chain = merger | model
chain.invoke(messages).pretty_print()
六、总结与实践路径
通过对消息裁剪、过滤和合并的学习，我们已经掌握了在复杂场景下管理对话历史的完整工具箱：
- 消息裁剪 (
trim_messages)：核心是控制 Token 数量，确保在模型上下文窗口内。通过strategy和start_on等参数，我们可以灵活地保留关键信息，同时丢弃冗余内容。 - 消息过滤 (
filter_messages)：用于精确控制哪些信息进入模型，保护敏感数据或简化模型输入。 - 消息合并 (
merge_message_runs)：用于处理模型兼容性问题，确保消息流格式正确。
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
20
0- 0
已为社区贡献17条内容
所有评论(0)