LangChain消息

消息

介绍

在 LangChain 中，消息（Messages） 是与大语言模型（LLM）交互时传递上下文的基本单位。

• 它们用于表示模型的输入和输出，既包含实际内容（如文本、图像等），也包含元数据（如角色、工具调用信息、token 使用情况等）。
• LangChain 提供了一套统一的消息类型系统，使得无论使用哪个模型提供商（如 OpenAI、Anthropic 等），开发者都能以一致的方式构建和处理对话。

消息组成

角色（Role）： 表示消息的发送者类型，决定了模型如何解释该消息

• SystemMessage：系统指令，用于设定模型行为、角色或规则。
• HumanMessage：用户输入。
• AIMessage：模型生成的回复。
• ToolMessage：工具调用的执行结果（用于函数调用场景）。

内容（Content）： 消息的实际“载荷”

• 纯文本（string）
• 多模态内容（如图像、音频、PDF 等），通过结构化列表表示
• 标准化的 Content Blocks（LangChain v1 引入的新机制，支持跨模型标准化）

元数据（Metadata）（可选）

• id：消息唯一标识（用于追踪）
• name：用户或工具名称（部分模型支持）
• usage_metadata：token 使用统计（输入/输出 token 数等）
• response_metadata：模型响应的额外信息（如 finish reason）
• tool_calls：模型请求调用的工具列表（在 AIMessage 中）

4种消息类型

SystemMessage：用于初始化模型的行为准则。

from langchain.messages import SystemMessage
system_msg = SystemMessage("You are a helpful coding assistant.")

HumanMessage：代表用户输入，支持多模态内容

• text、reasoning（模型推理步骤）
• image、audio、video、file（多模态）
• tool_call、server_tool_result（工具交互）
• non_standard（用于模型特有功能）

from langchain.messages import HumanMessage
# 纯文本
msg = HumanMessage("What is AI?")

# 多模态（图像 + 文本）
human_message = HumanMessage(content_blocks=[
    {"type": "text", "text": "Hello, how are you?"},
    {"type": "image", "url": "https://example.com/image.jpg"},
])

AIMessage：模型的输出，可能包含文本、工具调用、推理过程等

response = model.invoke([HumanMessage("What's the weather in Paris?")])
print(response.tool_calls)  # 如果绑定了工具，这里会显示调用请求
print(response.usage_metadata)  # 如 {'input_tokens': 10, 'output_tokens': 25}

ToolMessage：用于将工具执行结果返回给模型

tool_message = ToolMessage(
    content="Sunny, 72°F",
    tool_call_id="call_123",  # 必须与 AIMessage 中的 tool_call.id 匹配
    name="get_weather"
)

消息列表

通过提供消息对象列表，将消息列表传递给模型。可用于管理多轮对话

from langchain.messages import SystemMessage, HumanMessage, AIMessage

messages = [
    SystemMessage("You are a poetry expert"),
    HumanMessage("Write a haiku about spring"),
    AIMessage("Cherry blossoms bloom...")
]
response = model.invoke(messages)

或

messages = [
    {"role": "system", "content": "You are a poetry expert"},
    {"role": "user", "content": "Write a haiku about spring"},
    {"role": "assistant", "content": "Cherry blossoms bloom..."}
]
response = model.invoke(messages)

参考资料

• LangChain消息
• LangChain AIMessageChunk类

AIMessage实战示例

非流式请求

from langchain.messages import HumanMessage, AIMessage
from langchain.chat_models import init_chat_model

MODEL = "qwen-flash"
API_KEY = "sk-"
BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"

# 创建聊天模型
model = init_chat_model(
    model=MODEL, model_provider="openai", api_key=API_KEY, base_url=BASE_URL
)
response = model.invoke([HumanMessage("北京大学是211吗？回答是与否")])

print(type(response))
print(response)
response.pretty_print()  # 美式打印消息

响应类型

<class 'langchain_core.messages.ai.AIMessage'>

AIMessage的组成结构

content='你好！我是通义千问（Qwen），是阿里巴巴集团旗下的通义实验室自主研发的超大规模语言模型。我能够回答问题、创作文字，比如写故事、写公文、写邮件、写剧本、逻辑推理、编程等等，还能表达观点，玩游戏等。如果你有任何问题或需要帮助，欢迎随时告诉我！😊' 
additional_kwargs={'refusal': None} 
response_metadata={'token_usage': {'completion_tokens': 72, 'prompt_tokens': 11, 'total_tokens': 83, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'qwen-flash', 'system_fingerprint': None, 'id': 'chatcmpl-93210c1e-8eef-9bad-82c0-e5827f87c283', 'finish_reason': 'stop', 'logprobs': None} id='lc_run--019b598d-083d-70a2-9a28-c57c5bd71327-0' 
usage_metadata={'input_tokens': 11, 'output_tokens': 72, 'total_tokens': 83, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}

美式打印AIMessage

================================== Ai Message ==================================

你好！我是通义千问（Qwen），是阿里巴巴集团旗下的通义实验室自主研发的超大规模语言模型。我能够回答问题、创作文字，比如写故事、写公文、写邮件、写剧本、逻辑推理、编程等等，还能表达观点，玩游戏等。如果你有任何问题或需要帮助，欢迎随时告诉我！😊'

AIMessageChunk

介绍

在 LangChain 中，AIMessageChunk 是用于表示大语言模型（LLM）流式输出（streaming output）过程中单个片段（chunk）的消息类型。它是 BaseMessageChunk 的一个子类，专门用于代表 AI 生成内容的增量部分。

• AIMessageChunk：AI 回复的一个片段
• HumanMessageChunk、SystemMessageChunk 等：较少使用

组成

AIMessageChunk 主要属性

• content：该 chunk 包含的文本内容（字符串或列表，取决于是否支持多模态）。
• tool_calls（可选）：如果模型支持函数调用/工具调用，chunk 可能包含部分 tool_calls 信息（如 OpenAI 的 function calling 流式响应）。
• response_metadata：附加的元数据（如 token 使用、finish_reason 等，但通常在最后一个 chunk 中才有）。
• id、name 等辅助字段。

测试代码

for chunk in model.stream("你是谁？"):
    # print(type(chunk))
    print(chunk)

结果类型

<class 'langchain_core.messages.ai.AIMessageChunk'>

结果

• 最后一个包含finish_reason

content='' additional_kwargs={} response_metadata={'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c'
content='你好' additional_kwargs={} response_metadata={'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c'
content='！' additional_kwargs={} response_metadata={'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c'  
content='我是' additional_kwargs={} response_metadata={'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c'
content='Q' additional_kwargs={} response_metadata={'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c'   
content='wen，是阿里' additional_kwargs={} response_metadata={'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c'
content='云研发的超' additional_kwargs={} response_metadata={'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c'
content='大规模语言模型。' additional_kwargs={} response_metadata={'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c'
content='我可以帮助你回答' additional_kwargs={} response_metadata={'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c'
content='问题、创作文字' additional_kwargs={} response_metadata={'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c'
content='、编程、逻辑' additional_kwargs={} response_metadata={'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c'
content='推理等。有什么' additional_kwargs={} response_metadata={'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c'
content='我可以帮你的吗' additional_kwargs={} response_metadata={'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c'
content='？' additional_kwargs={} response_metadata={'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c'
content='' additional_kwargs={} response_metadata={'finish_reason': 'stop', 'model_name': 'qwen-flash', 'model_provider': 'openai'} id='lc_run--019b58ba-b3db-7b23-81e0-17b38803e23c' chunk_position='last'

拼接

AIMessageChunk.aggregate() 是一个类方法，可以将多个 chunk 合并成一个完整的 AIMessage（或保留为 AIMessageChunk，取决于实现）。

from langchain_core.messages import AIMessageChunk

chunks = []
for chunk in model.stream("你是谁？"):
    chunks.append(chunk)
    print(chunk.content, end="", flush=True)

# 合并所有 chunk
full_message = AIMessageChunk.aggregate(chunks)
print("\n\n完整消息内容:", full_message.content)

与 AIMessage 的区别

特性	AIMessage	AIMessageChunk
用途	完整 AI 回复	流式回复的片段
是否可聚合	否	是（可通过 .aggregate() 合并）
tool_calls	完整列表	可能是增量更新（需合并）
使用场景	普通调用	流式调用（stream() 或 astream()）