【LangChain-AI】核心组件--消息
·
1. LangChain 消息
LangChain 提供了一种统一的消息格式，可以跨聊天模型使用，允许用户使用不同的聊天模型，而无需担心每个模型提供商使用的消息格式的具体细节。
LangChain 消息格式主要分为五种，分别是：
1.1 BaseMessage 抽象消息类
LangChain中的消息都继承了class langchain_core.messages.base.BaseMessage这个抽象类。
参数：
- content ：消息的字符串内容。
- additional_kwargs ：与消息关联的其他有效负载数据。对于来自 AI 的消息，可能包括模
型提供程序编码的工具调用。 - response_metadata ：响应元数据。例如：响应标头、logprobs、令牌计数、模型名称。
- type ：消息的类型。必须是消息类型唯一的字符串。此字段的目的是在对消息进行反序列化时
方便地识别消息类型。 - name ：消息名称，为消息提供一个人类可读的名称。该字段的使用是可选的，是否使用它取决
于模型实现。 - id ：消息的可选唯一标识符。理想情况下，这应该由创建消息的提供者/模型提供。
2. 缓存历史消息
在 LangChain 老版本中，可以使用RunnableWithMessageHistory 消息历史类来包装另一个 Runnable 并为其管理聊天消息历史记录。它将跟踪模型的输入和输出，并将其存储在某个数据存储中。未来的交互将加载这些消息，并将其作为输入的一部分传递给链。
从 LangChain 的 v0.3 版本开始，官方建议 LangChain 用户不要使用RunnableWithMessageHistory ，而是利用 LangGraph 持久性来完成。
原因是它们的功能有限，不太适合现实世界的对话式 AI 应用程序。这些内存抽象缺乏对多用户、多对话场景的内置支持，而这对于实际的对话式人工智能系统至关重要。这些实现中的大多数已在LangChain 0.3.x 中被正式弃用，取而代之的是 LangGraph 持久性。LangGraph 持久性非常灵活，可以支持比 RunnableWithMessageHistory 接口更广泛的用例。
3. 管理历史消息
3.1 上下文窗口
上下文窗口可以理解为模型的**“短期工作记忆区”**，即 LLM 在一次处理请求时，所能查看和处理的最大 Token 数量，它包含了（用户的输入、大模型的输出、有时还包括系统指令和对话历史）。
3.2 消息裁剪
由于所有模型的上下文窗口大小都是有限的，这意味着作为输入的 Token 也是有限的。如果有累积了很长的消息历史记录，则需要管理传递给模型的消息的长度。trim_messages 可用于将聊天历史记录的大小减小为指定的令牌计数或指定的消息计数。
3.2.1 基于 token 数的修剪
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, trim_messages
model = init_chat_model(model="deepseek-chat", model_provider="deepseek")
# 历史消息记录
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?"),
]
print(model.invoke(messages))
从输出结果中我们可以看到，现在的总token数是64，下面我们来对其进行裁剪，将token数限制在55。
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, trim_messages
from langchain_core.messages.utils import count_tokens_approximately
model = init_chat_model(model="deepseek-chat", model_provider="deepseek")
# 历史消息记录
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?"),
]
# trim
# 使用 trim_messages 减少发送给模型的消息数量
trimmer = trim_messages(
max_tokens=65, # 修剪消息的最大令牌数，根据你想要的谈话长度来调整
strategy="last", # 修剪策略：
# “last”（默认）：保留最后的消息。
# “first”：保留最早的消息。
token_counter=model, # 传入一个函数或一个语言模型（因为语言模型有消息令牌计数方法）
include_system=True, # 如果想始终保留初始系统消息，可以指定 include_system=True
allow_partial=False, # 是否允许拆分消息的内容
start_on="human", # 如果需要确保我们的第一条消息（不包括系统消息）始终是特定类型，可以指定 start_on
)
chain = trimmer | model
print(chain.invoke(messages))
注意：上面在定义链的时候，不能写成chain = model | trimmer ，应该是chain = trimmer | model应该是先将消息进行裁剪，再发送给大模型。
3.2.2 基于消息数的修剪
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, trim_messages
from langchain_core.messages.utils import count_tokens_approximately
model = init_chat_model(model="deepseek-chat", model_provider="deepseek")
# 历史消息记录
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?"),
]
# trim
# 使用 trim_messages 减少发送给模型的消息数量
trimmer = trim_messages(
max_tokens=11, # 修剪消息的最大令牌消息数，根据你想要的谈话长度来调整
strategy="last", # 修剪策略：
# “last”（默认）：保留最后的消息。
# “first”：保留最早的消息。
token_counter=len, # 传入一个函数或一个语言模型（因为语言模型有消息令牌计数方法）
include_system=True, # 如果想始终保留初始系统消息，可以指定 include_system=True
allow_partial=False, # 是否允许拆分消息的内容
start_on="human", # 如果需要确保我们的第一条消息（不包括系统消息）始终是特定类型，可以指定 start_on
)
chain = trimmer | model
print(chain.invoke(messages))
输出结果：
content="I don't actually know your name — I can't see or remember who I'm talking to from one chat to the next. If you told me a name earlier, I don't have that memory. But you can tell me now, and for this conversation, I can call you that!" additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 59, 'prompt_tokens': 53, 'total_tokens': 112, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 53}, 'model_provider': 'deepseek', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': 'e1500561-4c5b-4a19-b460-c48c7be4cbec', 'finish_reason': 'stop', 'logprobs': None} id='lc_run--019e9bca-4227-7791-bb5c-3025f8f3ebac-0' tool_calls=[] invalid_tool_calls=[] usage_metadata={'input_tokens': 53, 'output_tokens': 59, 'total_tokens': 112, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}
3.3 消息过滤
filter_messages 方法则可以轻松地按类型、ID 或名称过滤 message。
按照类型筛选
# 历史消息记录
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"),
]
# 按照类型筛选（两种调用方式）
print(filter_messages(include_types="human").invoke(messages))
print(filter_messages(messages, include_types="human"))
输出结果：
[HumanMessage(content='示例输入', additional_kwargs={}, response_metadata={}, id='2'), HumanMessage(content='真实输入', additional_kwargs={}, response_metadata={}, id='4')]
[HumanMessage(content='示例输入', additional_kwargs={}, response_metadata={}, id='2'), HumanMessage(content='真实输入', additional_kwargs={}, response_metadata={}, id='4')]
按照id筛选
# 历史消息记录
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"),
]
# 按照id筛选
print(filter_messages(messages, exclude_ids=["3"]))
输出结果：
[SystemMessage(content='你是一个聊天助手', additional_kwargs={}, response_metadata={}, id='1'), HumanMessage(content='示例输入', additional_kwargs={}, response_metadata={}, id='2'), HumanMessage(content='真实输入', additional_kwargs={}, response_metadata={}, id='4'), AIMessage(content='真实输出', additional_kwargs={}, response_metadata={}, id='5', tool_calls=[], invalid_tool_calls=[])]
可以看到id为3的消息已经被排除掉了。
按照id和类型筛选
# 历史消息记录
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"),
]
# 按照id+类型筛选
print(filter_messages(messages, exclude_ids=["3"], include_types=[HumanMessage, AIMessage]))
输出结果：
[HumanMessage(content='示例输入', additional_kwargs={}, response_metadata={}, id='2'), HumanMessage(content='真实输入', additional_kwargs={}, response_metadata={}, id='4'), AIMessage(content='真实输出', additional_kwargs={}, response_metadata={}, id='5', tool_calls=[], invalid_tool_calls=[])]
可以看到，系统消息和id为3的消息已经被筛选掉了。
3.4 消息合并
若我们的消息列表存在连续某种类型相同的消息，但实际上某些模型不支持传递相同类型的连续消息。因此对于这种情况，我们可以使用 merge_message_runs 方法轻松合并相同类型的连续消息。
# 历史消息记录
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, filter_messages, merge_message_runs
# messages = [
# SystemMessage("你是一个聊天助手", id="1"),
# HumanMessage("示例输入", id="2"),
# AIMessage("示例输出", id="3"),
# HumanMessage("真实输入", id="4"),
# AIMessage("真实输出", id="5"),
# ]
# 按照类型筛选
# print(filter_messages(include_types="human").invoke(messages))
# print(filter_messages(messages, include_types="human"))
# 按照id筛选
# print(filter_messages(messages, exclude_ids=["3"]))
# 按照id+类型筛选
# print(filter_messages(messages, exclude_ids=["3"], include_types=[HumanMessage, AIMessage]))
model = init_chat_model(model="deepseek-chat", model_provider="deepseek")
# 历史消息记录
messages = [
SystemMessage("你是一个聊天助手。"),
SystemMessage("你总是以笑话回应。"),
HumanMessage("为什么要使用 LangChain?"),
HumanMessage("为什么要使用 LangGraph?"),
AIMessage("因为当你试图让你的代码更有条理时，LangGraph 会让你感到“节点”是个好主意！"),
AIMessage("不过别担心，它不会“分散”你的注意力！"),
HumanMessage("选择LangChain还是LangGraph?"),
]
# 方式1
merger_message = merge_message_runs(messages)
model.invoke(merger_message).pretty_print()
# 方式2
merger = merge_message_runs()
chain = merger | model
chain.invoke(messages).pretty_print()
输出结果：
================================== Ai Message ==================================
这就像问要吃面条还是吃刀削面 - LangChain 是已经切好的面条，直接就能下锅；LangGraph 是把面团放在厨师手里，想要什么形状都可以！
简单说：想要快速搞定就选LangChain，想要秀操作就选LangGraph！
================================== Ai Message ==================================
这个问题就像问“该用筷子还是叉子吃饭？”
如果只是简单对话，用LangChain就像用勺子喝汤；
但如果你想写一个会自我循环、判断、分支的AI，LangGraph就是那个能让你的代码跳探戈的舞伴！
不过记住：用LangGraph时别太兴奋，否则你的AI可能会陷入"思绪的迷宫"，需要找出口！🤪
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
2
0- 0
已为社区贡献7条内容
所有评论(0)