消息 Messages
LangChain 中使用消息(Messages)作为模型输入和输出的核心数据结构。不同类型消息用于表示不同的对话角色和内容类型。理解消息系统是构建 Agent 的基础。
消息类型
LangChain 提供了四种核心消息类型:
HumanMessage(用户消息)
表示用户或终端用户的输入:
python
from langchain.messages import HumanMessage
# 基本文本消息
message = HumanMessage(content="你好,今天天气怎么样?")
# 多内容消息(支持文本和图像)
message = HumanMessage(content=[
{"type": "text", "text": "这张图片里有什么?"},
{"type": "image_url", "image_url": {"url": "https://example.com/photo.jpg"}},
])
print(message.content) # 消息内容
print(message.role) # "user" - 角色标识
print(message.id) # 自动生成的唯一 IDSystemMessage(系统消息)
设置模型的系统行为、角色和约束:
python
from langchain.messages import SystemMessage
# 基础系统提示
system_message = SystemMessage(content="你是一个专业的编程助手。代码尽量简洁。")
# 带详细指令的系统提示
system_message = SystemMessage(content="""你是一个客服助手。
规则:
1. 始终使用中文回复
2. 保持礼貌和专业
3. 如无法回答,转接人工客服
4. 不要编造信息""")
# 在使用 Agent 时,系统消息通过 system_prompt 参数设置AIMessage(AI 消息)
表示模型的回复输出。在 Agent 运行时,AIMessage 还可能包含工具调用(tool_calls):
python
from langchain.messages import AIMessage
# 简单回复
response = AIMessage(content="今天的天气是晴天,气温 25°C。")
# 包含工具调用的回复(由 Agent 自动生成)
response = AIMessage(
content="我来查询一下天气。",
tool_calls=[
{
"id": "call_123",
"name": "get_weather",
"args": {"city": "北京", "date": "2025-05-27"}
}
]
)
print(response.tool_calls) # [ToolCall(...)]
print(response.role) # "assistant"
print(response.usage_metadata) # 如果启用,包含 token 用量信息ToolMessage(工具消息)
表示工具执行后的返回结果:
python
from langchain.messages import ToolMessage
# 工具执行结果
tool_message = ToolMessage(
content="当前北京天气:晴朗,25°C,湿度 40%",
tool_call_id="call_123", # 与 AIMessage 中的 tool_call id 对应
name="get_weather",
)
print(tool_message.role) # "tool"
print(tool_message.name) # "get_weather"消息结构
每条消息都包含以下核心属性:
python
from langchain.messages import HumanMessage
message = HumanMessage(
content="消息内容",
id="optional-custom-id", # 可选,默认自动生成 UUID
additional_kwargs={}, # 模型特定的附加参数
response_metadata={}, # 响应元数据(由模型填充)
)
# 所有类型共享的属性
print(message.content) # 主要内容
print(message.role) # 角色标识(user/assistant/tool/system)
print(message.id) # 唯一标识
print(message.type) # 消息类型字符串消息序列
对话历史是一组有序的消息列表:
python
from langchain.messages import SystemMessage, HumanMessage, AIMessage
# 构建对话历史
messages = [
SystemMessage(content="你是一个友好的助手。"),
HumanMessage(content="你好!"),
AIMessage(content="你好!有什么可以帮助你的吗?"),
HumanMessage(content="Python 中如何读取文件?"),
]
# 消息列表是可序列化的
serialized = [m.to_dict() for m in messages]
print(serialized)消息操作工具
消息合并
python
from langchain.messages import merge_message_runs
from langchain.messages import HumanMessage, AIMessage
# 合并连续的同类型消息
messages = [
HumanMessage(content="第一步说明"),
HumanMessage(content="第二步说明"), # 连续的用户消息将被合并
AIMessage(content="好的。"),
]
merged = merge_message_runs(messages)
# 结果:HumanMessage(content="第一步说明\n第二步说明"), AIMessage(content="好的。")消息筛选
python
from langchain.messages import filter_messages
# 筛选特定类型的消息
messages = [...] # 混合类型消息列表
# 保留用户和 AI 消息,过滤系统消息
filtered = filter_messages(
messages,
include_types=["human", "ai"],
)
# 按角色筛选
from operator import attrgetter
user_messages = [m for m in messages if m.role == "user"]消息修剪
python
from langchain.messages import trim_messages
# 限制消息数量和总 token 数
trimmed = trim_messages(
messages,
max_tokens=4096,
strategy="last", # 保留最新的消息
token_counter=len, # 可使用实际 tokenizer
include_system=True, # 始终保留系统消息
)在 Agent 中使用消息
python
from langchain import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langchain.messages import SystemMessage, HumanMessage
@tool
def get_weather(city: str) -> str:
"""获取指定城市的天气。"""
return f"{city}的天气:晴天,25°C"
agent = create_agent(
model=init_chat_model("openai/gpt-4o"),
tools=[get_weather],
system_prompt="你是一个天气助手。",
)
# Agent 自动处理消息类型转换
result = agent.invoke({
"messages": [HumanMessage(content="北京今天天气怎么样?")]
})
# 或者使用字典格式
result = agent.invoke({
"messages": [{"role": "user", "content": "北京今天天气怎么样?"}]
})消息格式转换
python
from langchain.messages import message_to_dict, messages_from_dict
# 消息的序列化与反序列化
messages = [HumanMessage(content="你好")]
dict_list = [message_to_dict(m) for m in messages]
# 从字典恢复
restored = messages_from_dict(dict_list)
# 转为 OpenAI 格式(用于 API 调用)
openai_messages = [m.to_openai_format() for m in messages]使用消息构建对话链
python
from langchain.messages import SystemMessage, HumanMessage, AIMessage
from langchain.chat_models import init_chat_model
model = init_chat_model("openai/gpt-4o")
# 构建带历史的多轮对话
conversation = [
SystemMessage(content="你是一个翻译助手,将英文翻译为中文。"),
HumanMessage(content="Hello, world!"),
AIMessage(content=""你好,世界!""),
HumanMessage(content="How are you?"),
]
response = model.invoke(conversation)
print(response.content) # "你好吗?"