目录编辑1. 消息Messages1.1 LLM 消息结构1.2 LangChain 消息1.2.1 BaseMessage 抽象消息类1.2.2 对话模式1.3 缓存历史消息1.3.1 多轮对话1.3.2 内存缓存1.4 管理历史消息1.4.1 前置概念1.4.1.1 上下文窗口1.4.1.2 Token1.4.2 消息裁剪1.4.2.1 基于输入 Token 数的修剪1.4.2.2 基于消息数的修剪1.4.3 消息过滤按类型进行筛选按类型 ID 进行筛选1.4.4 消息合并1. 消息Messages消息是聊天模型中的通信单位用于表示聊天模型的输入和输出以及可能与对话关联的任何其他上下文或元数据。1.1 LLM 消息结构每条消息都有一个角色和内容以及因 LLM 的不同而不同的附加元数据。消息角色Role角色描述system系统角色用于告诉聊天模型如何行为并提供额外的上下文。并非所有聊天模型提供商都支持。user用户角色表示用户与模型交互的输入通常以文本或其他交互式输入的形式。assistant助理角色表示来自模型的响应其中可以包括文本或调用工具的请求。tool工具角色用于在检索外部数据或将工具调用的结果传递回模型的消息。与支持工具调用的聊天模型一起使用。用来区分对话中不同类型的消息并帮助聊天模型了解如何响应给定的消息序列。消息内容 (Content)表示多模态数据例如图像、音频、视频的消息文本或字典列表的内容。内容的具体格式可能因底层不同的 LLM 而异。目前大多数模型都支持文本作为主要内容类型对多模态数据的支持仍然有限。消息其他元数据 (Additional metadata)元数据描述ID消息标识符。Name名称允许区分具有相同角色的不同实体。并非所有型号都支持此功能Metadata有关消息的其他信息例如时间戳、令牌使用情况等。Tool Calls模型发出的一个或多个工具的调用请求下面展示一个 OpenAI 的格式消息列表[ { role: user, content: Hello, how are you?, }, { role: assistant, content: Im doing well, thank you for asking., }, { role: user, content: Can you tell me a joke?, } ]LangChain 接受下面的格式作为聊天模型的输入chat_model.invoke([ { role: user, content: Hello, how are you?, }, { role: assistant, content: Im doing well, thank you for asking., }, { role: user, content: Can you tell me a joke?, } ])1.2 LangChain 消息LangChain 提供了一种统一的消息格式可以跨聊天模型使用允许用户使用不同的聊天模型而无需担心每个模型提供商使用的消息格式的具体细节。例如openai_model init_chat_model(gpt-4o-mini, model_provideropenai) anthropic_model init_chat_model(claude-3-5-sonnet-latest, model_provideranthropic) deepseek_model init_chat_model(deepseek-chat, model_providerdeepseek) google_genai_model init_chat_model(gemini-2.5-flash, model_providergoogle_genai) model init_chat_model(...)这些模型提供商不同但对于其输入和输出统一使用 LangChain 的消息格式。LangChain 消息格式主要分为五种分别是消息类型对应角色描述SystemMessage对应 system 系统角色用于启动 AI 模型的行为并提供额外的上下文例如指示模型采用特定角色或设定对话的基调例如你是一个后端开发的专家。HumanMessage对应 user 用户角色人类消息表示用户与模型交互的输入。大多数聊天模型都希望用户输入采用文本形式。AIMessage对应 assistant 助理角色这是来自模型的响应其中可以包括文本或调用工具的请求。它还可能包括其他媒体类型如图像、音频或视频 —— 尽管这目前仍然不常见。AIMessageChunk对应 assistant 助理角色用于流式响应通常在生成聊天模型时流式传输响应因此用户可以实时看到响应而不是等待生成整个响应后再显示。ToolMessage对应 tool 工具角色这表示一条角色为 tool 的消息其中包含调用工具的结果。这几个消息类型我们已经全部见过它们都是 LangChainBaseMessage的子类全部是作为 LangChain 聊天模型的输入和输出。1.2.1 BaseMessage 抽象消息类class langchain_core.messages.base.BaseMessage是作为 LangChain 聊天模型的输入和输出参数如下content消息的字符串内容。additional_kwargs与消息关联的其他有效负载数据。对于来自 AI 的消息可能包括模型提供程序编码的工具调用。response_metadata响应元数据。例如响应标头、logprobs、令牌计数、模型名称。type消息类型。必须是消息类型唯一的字符串。此字段的目的是在对消息进行序列化时方便地识别消息类型。name消息名称为消息提供一个人类可读的名称。该字段的使用是可选的是否使用它取决于模型实现。id消息的可选唯一标识符。理想情况下这应该由创建消息的提供者 / 模型提供。modelChatDeepSeek(modeldeepseek-chat) print(model.invoke(你好啊))content你好呀很高兴见到你 我是DeepSeek你的AI助手。有什么我可以帮你的吗无论是聊天、解答问题、写作、翻译还是其他需求尽管告诉我我会尽力帮你搞定。你今天过得怎么样 additional_kwargs{refusal: None} response_metadata{token_usage: {completion_tokens: 50, prompt_tokens: 6, total_tokens: 56, completion_tokens_details: None, prompt_tokens_details: {audio_tokens: None, cache_write_tokens: None, cached_tokens: 0}, prompt_cache_hit_tokens: 0, prompt_cache_miss_tokens: 6}, model_provider: deepseek, model_name: deepseek-v4-flash, system_fingerprint: a26a7955944dc5c60445bff77fac9c8e, id: 526f8f0f-aaad-44d0-b390-b2160d1654f2, finish_reason: stop, logprobs: None} idlc_run--01a041fb-82f7-7e82-a124-45478c326e03-0 tool_calls[] invalid_tool_calls[] usage_metadata{input_tokens: 6, output_tokens: 50, total_tokens: 56, input_token_details: {cache_read: 0}, output_token_details: {}}内置方法pretty_print() → None打印消息的漂亮表示。pretty_repr(html: bool False) → str获得消息的漂亮表示。请求是否将消息格式化为 HTML。如果为 True则消息将使用 HTML 标记进行格式化。默认值为 False。响应这是消息的漂亮表示。text() → str获取消息的文本内容。aimessmodel.invoke(你好啊) aimess.pretty_print()print(aimess.text)1.2.2 对话模式大多数对话都以设置对话上下文的系统消息开始。接下来是包含用户输入的用户消息然后是包含模型响应的助手消息。两种对话流转示意图1.3 缓存历史消息1.3.1 多轮对话在与大型语言模型交互的过程中我们常常体验到与智能助手进行连贯多轮对话的便利性。但目前我们的系统还不支持此功能代码如下modelChatDeepSeek(modeldeepseek-chat) print(model.invoke(你好啊,我的名字叫王小明).content) print(model.invoke(你知道我的名字叫什么吗).content)这里我们可以发现虽然我们在第一段对话的时候就告诉了AI我的名字但是在第二次对话的过程中AI并不知道我们的名字。在这样的对话方式里AI并不具有存储历史消息或者说记忆的功能。解决这个问题的方式就是在每次对话的过程中应该将历史对话的上下文合并为一个消息列表然后传递给AI代码如下message[ SystemMessage(你是一个人工智能助手), HumanMessage(你好啊,我的名字叫王小明), AIMessage(你好呀王小明很高兴认识你。), HumanMessage(你知道我的名字叫什么吗) ] modelChatDeepSeek(modeldeepseek-chat) print(model.invoke(message).content)可以看到当我们将历史上下文传递给AI的时候就能发现它能准确地说出我们的名字了。换句话说只要将历史消息重新发送给聊天模型那么就可以实现多轮对话的功能。1.3.2 内存缓存那么对于历史消息的管理就显得尤为重要。在 LangChain 老版本中可以使用RunnableWithMessageHistory消息历史类来包装另一个 Runnable 并为其管理聊天消息历史记录。它将跟踪模型的输入和输出并将其存储在某个数据存储中。未来的交互将加载这些消息并将其作为输入的一部分传递给链。from langchain_core.chat_history import InMemoryChatMessageHistory, BaseChatMessageHistory from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langchain_core.runnables import RunnableWithMessageHistory, configurable from langchain_deepseek import ChatDeepSeek modelChatDeepSeek(modeldeepseek-chat) store{} def Get_History_ssid(ssid:str)-BaseChatMessageHistory: if ssid not in store: store[ssid]InMemoryChatMessageHistory() return store[ssid] with_History_modelRunnableWithMessageHistory(model,Get_History_ssid) config{configurable:{session_id:1}} print(with_History_model.invoke([HumanMessage(你好我的名字叫王小明)], configconfig).content) print(with_History_model.invoke([HumanMessage(你知道我的名字吗)], configconfig).content)class langchain_core.runnables.history.RunnableWithMessageHistory类初始化参数说明runnable被包装 Runnable 实例这里就是我们定义的聊天模型Get_History_ssid返回类型为BaseChatMessageHistory的函数传入后作为回调函数。此函数接受一个session_id字符串类型并返回相应的聊天消息历史记录实例。class langchain_core.runnables.history.RunnableWithMessageHistory类方法说明.invoke()方法此方法与其他 Runnable 实例的.invoke()方法相同。只不过注意其 config 配置需要配置成config{configurable: {session_id: }}让RunnableWithMessageHistory可以读取到会话 id。记忆功能已经实现但是需要注意的是从 LangChain 的 v0.3 版本开始官方建议 LangChain 用户不要使用RunnableWithMessageHistory而是利用 LangGraph 持久性 来完成见 LangGraph 章节。原因是它们的功能有限不太适合现实世界的对话式 AI 应用程序。这些内存抽象缺乏对多用户、多对话场景的内置支持而这对于实际的对话式人工智能系统至关重要。这些实现中的大多数已在 LangChain 0.3.x 中被正式弃用取而代之的是 LangGraph 持久性。LangGraph 持久性 非常灵活可以支持比RunnableWithMessageHistory接口更广泛的用例。我们会在 LangGraph 篇章中学习它因此RunnableWithMessageHistory这部分我们讲解的并不深入。在之前对于生产环境我们还需要使用聊天消息历史记录的持久化实现例如RedisChatMessageHistory()而不是InMemoryChatMessageHistory()但现在也已不推荐新应用使用它们了。1.4 管理历史消息1.4.1 前置概念1.4.1.1 上下文窗口管理历史消息无非就是理解如何 “管理”“管理” 无非也就是一些 “CRUD”。那么在了解如何管理消息之前需要先了解下多轮对话的核心概念上下文窗口。上下文窗口可以理解为模型的 “短期工作记忆区”即 LLM 在一次处理请求时所能查看和处理的最大 Token 数量它包含了用户的输入大模型的输出有时还包括系统指令SystemMessage和对话历史。不同大模型支持的上下文窗口大小不同例如OpenAI 下 GPT‑5 模型上下文窗口为 400000最大 Token 数量GPT‑4.1 模型上下文窗口为 1047576最大 Token 数量其他模型上下文窗口可参考对应模型官网说明如 OpenAI 下模型可以参考这里。1.4.1.2 Token在自然语言处理NLP中Token 是文本的基本单位。它不是完全等同于一个单词或一个汉字而是一个更细粒度的划分。为什么用 Token计算机无法直接理解文字它需要将文本转换为数字向量。Tokenization令牌化就是这个转换过程的第一步将句子分解成模型可以理解和处理的碎片。对于英文1 个 Token ~ 4 个字符或 0.75 个单词1000 个 Tokens 约等于 750 个英文单词。一个 Token 可以是一个单词如apple、一个词根如un在unlikely中或者一个标点符号如.。例如ChatGPT is great!可能会被分成[Chat, G, PT, is, great, !]这 6 个 Token。对于中文1 个汉字 1.5‑2 个 Tokens1000 个 Tokens 大约相当于 500‑700 个汉字。常见的词和字可能是一个 Token生僻字或复杂词可能会被拆分成多个。举个例子上下文窗口就像一个固定大小的工作台再把 Token 比作一个积木零件把大模型比作一个工匠。工匠需要拼出模型必须把所需的零件输入的 Token放在工作台上一边拼装生成回复一边把拼好的部分输出的 Token也放在工作台上。整个过程输入 输出中工作台上的所有积木Tokens总数都不能超过工作台的最大容量上下文窗口大小。如果最初的零件太多占满了工作台工匠就没有空间进行拼装了。这时你就需要减少零件精简输入。如果最初的零件太多占满了工作台工匠就没有空间进行拼装了。这时你就需要减少零件精简输入1.4.2 消息裁剪有了上下文窗口和 Token 的认知再来看多轮对话的实现原理其实就是输入 系统消息 对话历史 最新用户问题对于模型来说并不真正 “记忆”而是每次都将完整的上下文重新输入。由于所有模型的上下文窗口大小都是有限的这意味着作为输入的 Token 也是有限的。如果有累积了很长的消息历史记录则需要管理传递给模型的消息的长度。trim_messages可用于将聊天历史记录的大小减小为指定的令牌计数或指定的消息计数。1.4.2.1 基于输入 Token 数的修剪下面演示一个通过trim_messages裁剪消息的示例基于输入 Token 数的修剪。先来看看不做任何输入限制的聊天from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langchain_deepseek import ChatDeepSeek modelChatDeepSeek(modeldeepseek-chat) message[ SystemMessage(contentyoure a good assistant), HumanMessage(contenthi! Im bob), AIMessage(contenthi!), HumanMessage(contentI like vanilla ice cream), AIMessage(contentnice), HumanMessage(contentwhats 2 2), AIMessage(content4), HumanMessage(contentthanks), AIMessage(contentno problem!), HumanMessage(contenthaving fun?), AIMessage(contentyes!), HumanMessage(contentWhats my name?), ] print(model.invoke(message))运行结果contentYour name is Bob! additional_kwargs{refusal: None} response_metadata{token_usage: {completion_tokens: 7, prompt_tokens: 64, total_tokens: 71, completion_tokens_details: None, prompt_tokens_details: {audio_tokens: None, cache_write_tokens: None, cached_tokens: 0}, prompt_cache_hit_tokens: 0, prompt_cache_miss_tokens: 64}, model_provider: deepseek, model_name: deepseek-v4-flash, system_fingerprint: a26a7955944dc5c60445bff77fac9c8e, id: f5f92cf5-00c1-4b8c-a36c-76efcd7963f8, finish_reason: stop, logprobs: None} idlc_run--01a04292-0afb-74b1-8b13-6ad989118566-0 tool_calls[] invalid_tool_calls[] usage_metadata{input_tokens: 64, output_tokens: 7, total_tokens: 71, input_token_details: {cache_read: 0}, output_token_details: {}}在运行结果的usage_metadata字段我们可以看到这些属性usage_metadata{input_tokens: 64, output_tokens: 7, total_tokens: 71, input_token_details: {cache_read: 0}, output_token_details: {}}从打印结果来看LLM 还认识我们且共输入了 64 tokens。接下来让我们对消息进行裁剪我们只希望将来输入时最多输入 50 tokens超出的需要按照一定的 “规则” 进行裁剪代码如下from typing import List import tiktoken from langchain_deepseek import ChatDeepSeek from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, trim_messages modelChatDeepSeek(modeldeepseek-chat) message[ SystemMessage(contentyoure a good assistant), HumanMessage(contenthi! Im bob), AIMessage(contenthi!), HumanMessage(contentI like vanilla ice cream), AIMessage(contentnice), HumanMessage(contentwhats 2 2), AIMessage(content4), HumanMessage(contentthanks), AIMessage(contentno problem!), HumanMessage(contenthaving fun?), AIMessage(contentyes!), HumanMessage(contentWhats my name?), ] # 自定义token计数器tiktoken估算不需要导入不存在的函数 def tiktoken_counter(messages: List) - int: enc tiktoken.get_encoding(cl100k_base) num_tokens 3 # 每条消息固定前缀开销 for msg in messages: num_tokens 3 num_tokens len(enc.encode(msg.content)) return num_tokens # 使用 trim_messages 减少发送给模型的消息数量 trimmer trim_messages( max_tokens50, # 修剪消息的最大令牌数根据你想要的谈话长度来调整 strategylast, # 修剪策略: # last(默认): 保留最后的消息。 # first: 保留最早的消息。 token_countertiktoken_counter, # 传入一个函数或一个语言模型(因为语言模型有消息令牌计数方法) include_systemTrue, # 如果想始终保留初始系统消息可以指定 allow_partialFalse, # 是否允许拆分消息的内容 start_onhuman, # 如果需要确保我们的第一条消息(不包括系统消息)始终是特定类型可以指定 start_on ) chain trimmer | model print(chain.invoke(message))运行结果contentI dont actually know your name—you havent told me! \n\nIf youd like to tell me, Id be happy to use it. What should I call you? additional_kwargs{refusal: None} response_metadata{token_usage: {completion_tokens: 39, prompt_tokens: 31, total_tokens: 70, completion_tokens_details: None, prompt_tokens_details: {audio_tokens: None, cache_write_tokens: None, cached_tokens: 0}, prompt_cache_hit_tokens: 0, prompt_cache_miss_tokens: 31}, model_provider: deepseek, model_name: deepseek-v4-flash, system_fingerprint: a26a7955944dc5c60445bff77fac9c8e, id: eca83f67-5368-46c8-be82-606313c5b2c4, finish_reason: stop, logprobs: None} idlc_run--01a042a8-2bbe-7571-86c6-35a5c2c7d165-0 tool_calls[] invalid_tool_calls[] usage_metadata{input_tokens: 31, output_tokens: 39, total_tokens: 70, input_token_details: {cache_read: 0}, output_token_details: {}}此时我们可以发现我们的历史消息被裁剪了AI并不知道我们的名字是谁。被修剪了哪些消息呢来看下print(trimmer.invoke(message))输出结果[SystemMessage(contentyoure a good assistant, additional_kwargs{}, response_metadata{}), HumanMessage(contentthanks, additional_kwargs{}, response_metadata{}), AIMessage(contentno problem!, additional_kwargs{}, response_metadata{}, tool_calls[], invalid_tool_calls[]), HumanMessage(contenthaving fun?, additional_kwargs{}, response_metadata{}), AIMessage(contentyes!, additional_kwargs{}, response_metadata{}, tool_calls[], invalid_tool_calls[]), HumanMessage(contentWhats my name?, additional_kwargs{}, response_metadata{})]从结果来看确实是按照我们给定的裁剪 “规则” 来完成的。修剪聊天记录后生成的聊天记录输入应该有效需遵循对话模式原则聊天记录以HumanMessage或SystemMessage开头后跟HumanMessage。这可以通过设置start_onhuman来实现。聊天记录以HumanMessage或ToolMessage结尾。这可以通过设置ends_on(human, tool)来实现。ToolMessage只能出现在涉及工具调用的AIMessage之后。如果原始聊天历史记录中存在SystemMessage则新聊天历史记录应包括SystemMessage因为SystemMessage包含对聊天模型的特殊说明。SystemMessage总是历史记录中的第一条消息如果存在。这可以通过设置include_systemTrue。1.4.2.2 基于消息数的修剪除了基于 token 的修剪还可以通过设置token_counterlen根据消息数修剪聊天记录。在这种情况下max_tokens将控制最大消息数。示例如下from langchain_deepseek import ChatDeepSeek from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, trim_messages modelChatDeepSeek(modeldeepseek-chat) message[ SystemMessage(contentyoure a good assistant), HumanMessage(contenthi! Im bob), AIMessage(contenthi!), HumanMessage(contentI like vanilla ice cream), AIMessage(contentnice), HumanMessage(contentwhats 2 2), AIMessage(content4), HumanMessage(contentthanks), AIMessage(contentno problem!), HumanMessage(contenthaving fun?), AIMessage(contentyes!), HumanMessage(contentWhats my name?), ] # 使用 trim_messages 减少发送给模型的消息数量 trimmer trim_messages( max_tokens11, # 修剪消息的最大令牌数根据你想要的谈话长度来调整 strategylast, # 修剪策略: # last(默认): 保留最后的消息。 # first: 保留最早的消息。 token_counterlen, # 传入一个函数或一个语言模型(因为语言模型有消息令牌计数方法) include_systemTrue, # 如果想始终保留初始系统消息可以指定也就是第一个SyStemMessage allow_partialFalse, # 是否允许拆分消息的内容 start_onhuman, # 如果需要确保我们的第一条消息(不包括系统消息)始终是特定类型可以指定 start_on ) chain trimmer | model print(trimmer.invoke(message))运行结果[SystemMessage(contentyoure a good assistant, additional_kwargs{}, response_metadata{}), HumanMessage(contentI like vanilla ice cream, additional_kwargs{}, response_metadata{}), AIMessage(contentnice, additional_kwargs{}, response_metadata{}, tool_calls[], invalid_tool_calls[]), HumanMessage(contentwhats 2 2, additional_kwargs{}, response_metadata{}), AIMessage(content4, additional_kwargs{}, response_metadata{}, tool_calls[], invalid_tool_calls[]), HumanMessage(contentthanks, additional_kwargs{}, response_metadata{}), AIMessage(contentno problem!, additional_kwargs{}, response_metadata{}, tool_calls[], invalid_tool_calls[]), HumanMessage(contenthaving fun?, additional_kwargs{}, response_metadata{}), AIMessage(contentyes!, additional_kwargs{}, response_metadata{}, tool_calls[], invalid_tool_calls[]), HumanMessage(contentWhats my name?, additional_kwargs{}, response_metadata{})]可以看到代码中我们指定了最大消息数为11个消息。并且第一个消息必须是HumanMessage所以一共裁剪掉了除系统消息之外最早的两个消息如图1.4.3 消息过滤按类型进行筛选在更复杂的场景下我们可能会使用消息列表来跟踪状态例如我们可能只想将这个完整消息列表的子集传递模型调用而不是所有的历史记录。filter_messages方法则可以轻松地按类型、ID 或名称过滤 message。如以下代码的效果就是获取message消息列表中的所有HumanMessage:message[ SystemMessage(content你是一个人工智能助手,id1), HumanMessage(content示例输入,id2), AIMessage(content示例输出,id3), HumanMessage(content示例输入,id4), AIMessage(content示例输出,id5) ] #获取mesage这个消息列表中所有的HumanMessage并打印 print(filter_messages(message, include_typeshuman)) #等价于 # print(filter_messages( include_typeshuman).invoke(message))运行结果[HumanMessage(content示例输入, additional_kwargs{}, response_metadata{}, id2), HumanMessage(content示例输入, additional_kwargs{}, response_metadata{}, id4)]按类型 ID 进行筛选获取mesage这个消息列表中所有除了id3的HumanMessage与AIMessage并打印message[ SystemMessage(content你是一个人工智能助手,id1), HumanMessage(content示例输入,id2), AIMessage(content示例输出,id3), HumanMessage(content示例输入,id4), AIMessage(content示例输出,id5) ] #获取mesage这个消息列表中所有除了id3的HumanMessage与AIMessage并打印 print(filter_messages(message, include_types[HumanMessage,AIMessage],exclude_ids3)) #等价于 #print(filter_messages( include_types[HumanMessage,AIMessage],exclude_ids3).invoke(message))运行结果[HumanMessage(content示例输入, additional_kwargs{}, response_metadata{}, id2), HumanMessage(content示例输入, additional_kwargs{}, response_metadata{}, id4), AIMessage(content示例输出, additional_kwargs{}, response_metadata{}, id5, tool_calls[], invalid_tool_calls[])]1.4.4 消息合并若我们的消息列表存在连续某种类型相同的消息但实际上某些模型不支持传递相同类型的连续消息。因此对于这种情况我们可以使用merge_message_runs方法轻松合并相同类型的连续消息。# 定义大模型 model ChatDeepSeek(modeldeepseek-chat) # 历史消息记录 messages [ SystemMessage(你是一个聊天助手。), SystemMessage(你总是以笑话回应。), HumanMessage(为什么要使用 LangChain?), HumanMessage(为什么要使用 LangGraph?), AIMessage(因为当你试图让你的代码更有条理时LangGraph 会让你感到“节点”是个好主意), AIMessage(不过别担心它不会“分散”你的注意力), HumanMessage(选择LangChain还是LangGraph?), ] mergemerge_message_runs(messages) #调用大模型方法一 print(model.invoke(messages).content) # #调用大模型方法二 # mergemerge_message_runs() # # chainmerge | model # # print(chain.invoke(messages).content)运行结果选 LangChain 还是 LangGraph这就像问“我该用锤子还是螺丝刀”——如果我想把代码钉进墙里就选 LangChain如果我想画一幅流程图选 LangGraph 准没错不过说真的如果你总是选错我们可能需要聊聊“链”接心理医生的问题了 LangChain 消息组件：从基础消息到历史消息处理
目录编辑1. 消息Messages1.1 LLM 消息结构1.2 LangChain 消息1.2.1 BaseMessage 抽象消息类1.2.2 对话模式1.3 缓存历史消息1.3.1 多轮对话1.3.2 内存缓存1.4 管理历史消息1.4.1 前置概念1.4.1.1 上下文窗口1.4.1.2 Token1.4.2 消息裁剪1.4.2.1 基于输入 Token 数的修剪1.4.2.2 基于消息数的修剪1.4.3 消息过滤按类型进行筛选按类型 ID 进行筛选1.4.4 消息合并1. 消息Messages消息是聊天模型中的通信单位用于表示聊天模型的输入和输出以及可能与对话关联的任何其他上下文或元数据。1.1 LLM 消息结构每条消息都有一个角色和内容以及因 LLM 的不同而不同的附加元数据。消息角色Role角色描述system系统角色用于告诉聊天模型如何行为并提供额外的上下文。并非所有聊天模型提供商都支持。user用户角色表示用户与模型交互的输入通常以文本或其他交互式输入的形式。assistant助理角色表示来自模型的响应其中可以包括文本或调用工具的请求。tool工具角色用于在检索外部数据或将工具调用的结果传递回模型的消息。与支持工具调用的聊天模型一起使用。用来区分对话中不同类型的消息并帮助聊天模型了解如何响应给定的消息序列。消息内容 (Content)表示多模态数据例如图像、音频、视频的消息文本或字典列表的内容。内容的具体格式可能因底层不同的 LLM 而异。目前大多数模型都支持文本作为主要内容类型对多模态数据的支持仍然有限。消息其他元数据 (Additional metadata)元数据描述ID消息标识符。Name名称允许区分具有相同角色的不同实体。并非所有型号都支持此功能Metadata有关消息的其他信息例如时间戳、令牌使用情况等。Tool Calls模型发出的一个或多个工具的调用请求下面展示一个 OpenAI 的格式消息列表[ { role: user, content: Hello, how are you?, }, { role: assistant, content: Im doing well, thank you for asking., }, { role: user, content: Can you tell me a joke?, } ]LangChain 接受下面的格式作为聊天模型的输入chat_model.invoke([ { role: user, content: Hello, how are you?, }, { role: assistant, content: Im doing well, thank you for asking., }, { role: user, content: Can you tell me a joke?, } ])1.2 LangChain 消息LangChain 提供了一种统一的消息格式可以跨聊天模型使用允许用户使用不同的聊天模型而无需担心每个模型提供商使用的消息格式的具体细节。例如openai_model init_chat_model(gpt-4o-mini, model_provideropenai) anthropic_model init_chat_model(claude-3-5-sonnet-latest, model_provideranthropic) deepseek_model init_chat_model(deepseek-chat, model_providerdeepseek) google_genai_model init_chat_model(gemini-2.5-flash, model_providergoogle_genai) model init_chat_model(...)这些模型提供商不同但对于其输入和输出统一使用 LangChain 的消息格式。LangChain 消息格式主要分为五种分别是消息类型对应角色描述SystemMessage对应 system 系统角色用于启动 AI 模型的行为并提供额外的上下文例如指示模型采用特定角色或设定对话的基调例如你是一个后端开发的专家。HumanMessage对应 user 用户角色人类消息表示用户与模型交互的输入。大多数聊天模型都希望用户输入采用文本形式。AIMessage对应 assistant 助理角色这是来自模型的响应其中可以包括文本或调用工具的请求。它还可能包括其他媒体类型如图像、音频或视频 —— 尽管这目前仍然不常见。AIMessageChunk对应 assistant 助理角色用于流式响应通常在生成聊天模型时流式传输响应因此用户可以实时看到响应而不是等待生成整个响应后再显示。ToolMessage对应 tool 工具角色这表示一条角色为 tool 的消息其中包含调用工具的结果。这几个消息类型我们已经全部见过它们都是 LangChainBaseMessage的子类全部是作为 LangChain 聊天模型的输入和输出。1.2.1 BaseMessage 抽象消息类class langchain_core.messages.base.BaseMessage是作为 LangChain 聊天模型的输入和输出参数如下content消息的字符串内容。additional_kwargs与消息关联的其他有效负载数据。对于来自 AI 的消息可能包括模型提供程序编码的工具调用。response_metadata响应元数据。例如响应标头、logprobs、令牌计数、模型名称。type消息类型。必须是消息类型唯一的字符串。此字段的目的是在对消息进行序列化时方便地识别消息类型。name消息名称为消息提供一个人类可读的名称。该字段的使用是可选的是否使用它取决于模型实现。id消息的可选唯一标识符。理想情况下这应该由创建消息的提供者 / 模型提供。modelChatDeepSeek(modeldeepseek-chat) print(model.invoke(你好啊))content你好呀很高兴见到你 我是DeepSeek你的AI助手。有什么我可以帮你的吗无论是聊天、解答问题、写作、翻译还是其他需求尽管告诉我我会尽力帮你搞定。你今天过得怎么样 additional_kwargs{refusal: None} response_metadata{token_usage: {completion_tokens: 50, prompt_tokens: 6, total_tokens: 56, completion_tokens_details: None, prompt_tokens_details: {audio_tokens: None, cache_write_tokens: None, cached_tokens: 0}, prompt_cache_hit_tokens: 0, prompt_cache_miss_tokens: 6}, model_provider: deepseek, model_name: deepseek-v4-flash, system_fingerprint: a26a7955944dc5c60445bff77fac9c8e, id: 526f8f0f-aaad-44d0-b390-b2160d1654f2, finish_reason: stop, logprobs: None} idlc_run--01a041fb-82f7-7e82-a124-45478c326e03-0 tool_calls[] invalid_tool_calls[] usage_metadata{input_tokens: 6, output_tokens: 50, total_tokens: 56, input_token_details: {cache_read: 0}, output_token_details: {}}内置方法pretty_print() → None打印消息的漂亮表示。pretty_repr(html: bool False) → str获得消息的漂亮表示。请求是否将消息格式化为 HTML。如果为 True则消息将使用 HTML 标记进行格式化。默认值为 False。响应这是消息的漂亮表示。text() → str获取消息的文本内容。aimessmodel.invoke(你好啊) aimess.pretty_print()print(aimess.text)1.2.2 对话模式大多数对话都以设置对话上下文的系统消息开始。接下来是包含用户输入的用户消息然后是包含模型响应的助手消息。两种对话流转示意图1.3 缓存历史消息1.3.1 多轮对话在与大型语言模型交互的过程中我们常常体验到与智能助手进行连贯多轮对话的便利性。但目前我们的系统还不支持此功能代码如下modelChatDeepSeek(modeldeepseek-chat) print(model.invoke(你好啊,我的名字叫王小明).content) print(model.invoke(你知道我的名字叫什么吗).content)这里我们可以发现虽然我们在第一段对话的时候就告诉了AI我的名字但是在第二次对话的过程中AI并不知道我们的名字。在这样的对话方式里AI并不具有存储历史消息或者说记忆的功能。解决这个问题的方式就是在每次对话的过程中应该将历史对话的上下文合并为一个消息列表然后传递给AI代码如下message[ SystemMessage(你是一个人工智能助手), HumanMessage(你好啊,我的名字叫王小明), AIMessage(你好呀王小明很高兴认识你。), HumanMessage(你知道我的名字叫什么吗) ] modelChatDeepSeek(modeldeepseek-chat) print(model.invoke(message).content)可以看到当我们将历史上下文传递给AI的时候就能发现它能准确地说出我们的名字了。换句话说只要将历史消息重新发送给聊天模型那么就可以实现多轮对话的功能。1.3.2 内存缓存那么对于历史消息的管理就显得尤为重要。在 LangChain 老版本中可以使用RunnableWithMessageHistory消息历史类来包装另一个 Runnable 并为其管理聊天消息历史记录。它将跟踪模型的输入和输出并将其存储在某个数据存储中。未来的交互将加载这些消息并将其作为输入的一部分传递给链。from langchain_core.chat_history import InMemoryChatMessageHistory, BaseChatMessageHistory from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langchain_core.runnables import RunnableWithMessageHistory, configurable from langchain_deepseek import ChatDeepSeek modelChatDeepSeek(modeldeepseek-chat) store{} def Get_History_ssid(ssid:str)-BaseChatMessageHistory: if ssid not in store: store[ssid]InMemoryChatMessageHistory() return store[ssid] with_History_modelRunnableWithMessageHistory(model,Get_History_ssid) config{configurable:{session_id:1}} print(with_History_model.invoke([HumanMessage(你好我的名字叫王小明)], configconfig).content) print(with_History_model.invoke([HumanMessage(你知道我的名字吗)], configconfig).content)class langchain_core.runnables.history.RunnableWithMessageHistory类初始化参数说明runnable被包装 Runnable 实例这里就是我们定义的聊天模型Get_History_ssid返回类型为BaseChatMessageHistory的函数传入后作为回调函数。此函数接受一个session_id字符串类型并返回相应的聊天消息历史记录实例。class langchain_core.runnables.history.RunnableWithMessageHistory类方法说明.invoke()方法此方法与其他 Runnable 实例的.invoke()方法相同。只不过注意其 config 配置需要配置成config{configurable: {session_id: }}让RunnableWithMessageHistory可以读取到会话 id。记忆功能已经实现但是需要注意的是从 LangChain 的 v0.3 版本开始官方建议 LangChain 用户不要使用RunnableWithMessageHistory而是利用 LangGraph 持久性 来完成见 LangGraph 章节。原因是它们的功能有限不太适合现实世界的对话式 AI 应用程序。这些内存抽象缺乏对多用户、多对话场景的内置支持而这对于实际的对话式人工智能系统至关重要。这些实现中的大多数已在 LangChain 0.3.x 中被正式弃用取而代之的是 LangGraph 持久性。LangGraph 持久性 非常灵活可以支持比RunnableWithMessageHistory接口更广泛的用例。我们会在 LangGraph 篇章中学习它因此RunnableWithMessageHistory这部分我们讲解的并不深入。在之前对于生产环境我们还需要使用聊天消息历史记录的持久化实现例如RedisChatMessageHistory()而不是InMemoryChatMessageHistory()但现在也已不推荐新应用使用它们了。1.4 管理历史消息1.4.1 前置概念1.4.1.1 上下文窗口管理历史消息无非就是理解如何 “管理”“管理” 无非也就是一些 “CRUD”。那么在了解如何管理消息之前需要先了解下多轮对话的核心概念上下文窗口。上下文窗口可以理解为模型的 “短期工作记忆区”即 LLM 在一次处理请求时所能查看和处理的最大 Token 数量它包含了用户的输入大模型的输出有时还包括系统指令SystemMessage和对话历史。不同大模型支持的上下文窗口大小不同例如OpenAI 下 GPT‑5 模型上下文窗口为 400000最大 Token 数量GPT‑4.1 模型上下文窗口为 1047576最大 Token 数量其他模型上下文窗口可参考对应模型官网说明如 OpenAI 下模型可以参考这里。1.4.1.2 Token在自然语言处理NLP中Token 是文本的基本单位。它不是完全等同于一个单词或一个汉字而是一个更细粒度的划分。为什么用 Token计算机无法直接理解文字它需要将文本转换为数字向量。Tokenization令牌化就是这个转换过程的第一步将句子分解成模型可以理解和处理的碎片。对于英文1 个 Token ~ 4 个字符或 0.75 个单词1000 个 Tokens 约等于 750 个英文单词。一个 Token 可以是一个单词如apple、一个词根如un在unlikely中或者一个标点符号如.。例如ChatGPT is great!可能会被分成[Chat, G, PT, is, great, !]这 6 个 Token。对于中文1 个汉字 1.5‑2 个 Tokens1000 个 Tokens 大约相当于 500‑700 个汉字。常见的词和字可能是一个 Token生僻字或复杂词可能会被拆分成多个。举个例子上下文窗口就像一个固定大小的工作台再把 Token 比作一个积木零件把大模型比作一个工匠。工匠需要拼出模型必须把所需的零件输入的 Token放在工作台上一边拼装生成回复一边把拼好的部分输出的 Token也放在工作台上。整个过程输入 输出中工作台上的所有积木Tokens总数都不能超过工作台的最大容量上下文窗口大小。如果最初的零件太多占满了工作台工匠就没有空间进行拼装了。这时你就需要减少零件精简输入。如果最初的零件太多占满了工作台工匠就没有空间进行拼装了。这时你就需要减少零件精简输入1.4.2 消息裁剪有了上下文窗口和 Token 的认知再来看多轮对话的实现原理其实就是输入 系统消息 对话历史 最新用户问题对于模型来说并不真正 “记忆”而是每次都将完整的上下文重新输入。由于所有模型的上下文窗口大小都是有限的这意味着作为输入的 Token 也是有限的。如果有累积了很长的消息历史记录则需要管理传递给模型的消息的长度。trim_messages可用于将聊天历史记录的大小减小为指定的令牌计数或指定的消息计数。1.4.2.1 基于输入 Token 数的修剪下面演示一个通过trim_messages裁剪消息的示例基于输入 Token 数的修剪。先来看看不做任何输入限制的聊天from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langchain_deepseek import ChatDeepSeek modelChatDeepSeek(modeldeepseek-chat) message[ SystemMessage(contentyoure a good assistant), HumanMessage(contenthi! Im bob), AIMessage(contenthi!), HumanMessage(contentI like vanilla ice cream), AIMessage(contentnice), HumanMessage(contentwhats 2 2), AIMessage(content4), HumanMessage(contentthanks), AIMessage(contentno problem!), HumanMessage(contenthaving fun?), AIMessage(contentyes!), HumanMessage(contentWhats my name?), ] print(model.invoke(message))运行结果contentYour name is Bob! additional_kwargs{refusal: None} response_metadata{token_usage: {completion_tokens: 7, prompt_tokens: 64, total_tokens: 71, completion_tokens_details: None, prompt_tokens_details: {audio_tokens: None, cache_write_tokens: None, cached_tokens: 0}, prompt_cache_hit_tokens: 0, prompt_cache_miss_tokens: 64}, model_provider: deepseek, model_name: deepseek-v4-flash, system_fingerprint: a26a7955944dc5c60445bff77fac9c8e, id: f5f92cf5-00c1-4b8c-a36c-76efcd7963f8, finish_reason: stop, logprobs: None} idlc_run--01a04292-0afb-74b1-8b13-6ad989118566-0 tool_calls[] invalid_tool_calls[] usage_metadata{input_tokens: 64, output_tokens: 7, total_tokens: 71, input_token_details: {cache_read: 0}, output_token_details: {}}在运行结果的usage_metadata字段我们可以看到这些属性usage_metadata{input_tokens: 64, output_tokens: 7, total_tokens: 71, input_token_details: {cache_read: 0}, output_token_details: {}}从打印结果来看LLM 还认识我们且共输入了 64 tokens。接下来让我们对消息进行裁剪我们只希望将来输入时最多输入 50 tokens超出的需要按照一定的 “规则” 进行裁剪代码如下from typing import List import tiktoken from langchain_deepseek import ChatDeepSeek from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, trim_messages modelChatDeepSeek(modeldeepseek-chat) message[ SystemMessage(contentyoure a good assistant), HumanMessage(contenthi! Im bob), AIMessage(contenthi!), HumanMessage(contentI like vanilla ice cream), AIMessage(contentnice), HumanMessage(contentwhats 2 2), AIMessage(content4), HumanMessage(contentthanks), AIMessage(contentno problem!), HumanMessage(contenthaving fun?), AIMessage(contentyes!), HumanMessage(contentWhats my name?), ] # 自定义token计数器tiktoken估算不需要导入不存在的函数 def tiktoken_counter(messages: List) - int: enc tiktoken.get_encoding(cl100k_base) num_tokens 3 # 每条消息固定前缀开销 for msg in messages: num_tokens 3 num_tokens len(enc.encode(msg.content)) return num_tokens # 使用 trim_messages 减少发送给模型的消息数量 trimmer trim_messages( max_tokens50, # 修剪消息的最大令牌数根据你想要的谈话长度来调整 strategylast, # 修剪策略: # last(默认): 保留最后的消息。 # first: 保留最早的消息。 token_countertiktoken_counter, # 传入一个函数或一个语言模型(因为语言模型有消息令牌计数方法) include_systemTrue, # 如果想始终保留初始系统消息可以指定 allow_partialFalse, # 是否允许拆分消息的内容 start_onhuman, # 如果需要确保我们的第一条消息(不包括系统消息)始终是特定类型可以指定 start_on ) chain trimmer | model print(chain.invoke(message))运行结果contentI dont actually know your name—you havent told me! \n\nIf youd like to tell me, Id be happy to use it. What should I call you? additional_kwargs{refusal: None} response_metadata{token_usage: {completion_tokens: 39, prompt_tokens: 31, total_tokens: 70, completion_tokens_details: None, prompt_tokens_details: {audio_tokens: None, cache_write_tokens: None, cached_tokens: 0}, prompt_cache_hit_tokens: 0, prompt_cache_miss_tokens: 31}, model_provider: deepseek, model_name: deepseek-v4-flash, system_fingerprint: a26a7955944dc5c60445bff77fac9c8e, id: eca83f67-5368-46c8-be82-606313c5b2c4, finish_reason: stop, logprobs: None} idlc_run--01a042a8-2bbe-7571-86c6-35a5c2c7d165-0 tool_calls[] invalid_tool_calls[] usage_metadata{input_tokens: 31, output_tokens: 39, total_tokens: 70, input_token_details: {cache_read: 0}, output_token_details: {}}此时我们可以发现我们的历史消息被裁剪了AI并不知道我们的名字是谁。被修剪了哪些消息呢来看下print(trimmer.invoke(message))输出结果[SystemMessage(contentyoure a good assistant, additional_kwargs{}, response_metadata{}), HumanMessage(contentthanks, additional_kwargs{}, response_metadata{}), AIMessage(contentno problem!, additional_kwargs{}, response_metadata{}, tool_calls[], invalid_tool_calls[]), HumanMessage(contenthaving fun?, additional_kwargs{}, response_metadata{}), AIMessage(contentyes!, additional_kwargs{}, response_metadata{}, tool_calls[], invalid_tool_calls[]), HumanMessage(contentWhats my name?, additional_kwargs{}, response_metadata{})]从结果来看确实是按照我们给定的裁剪 “规则” 来完成的。修剪聊天记录后生成的聊天记录输入应该有效需遵循对话模式原则聊天记录以HumanMessage或SystemMessage开头后跟HumanMessage。这可以通过设置start_onhuman来实现。聊天记录以HumanMessage或ToolMessage结尾。这可以通过设置ends_on(human, tool)来实现。ToolMessage只能出现在涉及工具调用的AIMessage之后。如果原始聊天历史记录中存在SystemMessage则新聊天历史记录应包括SystemMessage因为SystemMessage包含对聊天模型的特殊说明。SystemMessage总是历史记录中的第一条消息如果存在。这可以通过设置include_systemTrue。1.4.2.2 基于消息数的修剪除了基于 token 的修剪还可以通过设置token_counterlen根据消息数修剪聊天记录。在这种情况下max_tokens将控制最大消息数。示例如下from langchain_deepseek import ChatDeepSeek from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, trim_messages modelChatDeepSeek(modeldeepseek-chat) message[ SystemMessage(contentyoure a good assistant), HumanMessage(contenthi! Im bob), AIMessage(contenthi!), HumanMessage(contentI like vanilla ice cream), AIMessage(contentnice), HumanMessage(contentwhats 2 2), AIMessage(content4), HumanMessage(contentthanks), AIMessage(contentno problem!), HumanMessage(contenthaving fun?), AIMessage(contentyes!), HumanMessage(contentWhats my name?), ] # 使用 trim_messages 减少发送给模型的消息数量 trimmer trim_messages( max_tokens11, # 修剪消息的最大令牌数根据你想要的谈话长度来调整 strategylast, # 修剪策略: # last(默认): 保留最后的消息。 # first: 保留最早的消息。 token_counterlen, # 传入一个函数或一个语言模型(因为语言模型有消息令牌计数方法) include_systemTrue, # 如果想始终保留初始系统消息可以指定也就是第一个SyStemMessage allow_partialFalse, # 是否允许拆分消息的内容 start_onhuman, # 如果需要确保我们的第一条消息(不包括系统消息)始终是特定类型可以指定 start_on ) chain trimmer | model print(trimmer.invoke(message))运行结果[SystemMessage(contentyoure a good assistant, additional_kwargs{}, response_metadata{}), HumanMessage(contentI like vanilla ice cream, additional_kwargs{}, response_metadata{}), AIMessage(contentnice, additional_kwargs{}, response_metadata{}, tool_calls[], invalid_tool_calls[]), HumanMessage(contentwhats 2 2, additional_kwargs{}, response_metadata{}), AIMessage(content4, additional_kwargs{}, response_metadata{}, tool_calls[], invalid_tool_calls[]), HumanMessage(contentthanks, additional_kwargs{}, response_metadata{}), AIMessage(contentno problem!, additional_kwargs{}, response_metadata{}, tool_calls[], invalid_tool_calls[]), HumanMessage(contenthaving fun?, additional_kwargs{}, response_metadata{}), AIMessage(contentyes!, additional_kwargs{}, response_metadata{}, tool_calls[], invalid_tool_calls[]), HumanMessage(contentWhats my name?, additional_kwargs{}, response_metadata{})]可以看到代码中我们指定了最大消息数为11个消息。并且第一个消息必须是HumanMessage所以一共裁剪掉了除系统消息之外最早的两个消息如图1.4.3 消息过滤按类型进行筛选在更复杂的场景下我们可能会使用消息列表来跟踪状态例如我们可能只想将这个完整消息列表的子集传递模型调用而不是所有的历史记录。filter_messages方法则可以轻松地按类型、ID 或名称过滤 message。如以下代码的效果就是获取message消息列表中的所有HumanMessage:message[ SystemMessage(content你是一个人工智能助手,id1), HumanMessage(content示例输入,id2), AIMessage(content示例输出,id3), HumanMessage(content示例输入,id4), AIMessage(content示例输出,id5) ] #获取mesage这个消息列表中所有的HumanMessage并打印 print(filter_messages(message, include_typeshuman)) #等价于 # print(filter_messages( include_typeshuman).invoke(message))运行结果[HumanMessage(content示例输入, additional_kwargs{}, response_metadata{}, id2), HumanMessage(content示例输入, additional_kwargs{}, response_metadata{}, id4)]按类型 ID 进行筛选获取mesage这个消息列表中所有除了id3的HumanMessage与AIMessage并打印message[ SystemMessage(content你是一个人工智能助手,id1), HumanMessage(content示例输入,id2), AIMessage(content示例输出,id3), HumanMessage(content示例输入,id4), AIMessage(content示例输出,id5) ] #获取mesage这个消息列表中所有除了id3的HumanMessage与AIMessage并打印 print(filter_messages(message, include_types[HumanMessage,AIMessage],exclude_ids3)) #等价于 #print(filter_messages( include_types[HumanMessage,AIMessage],exclude_ids3).invoke(message))运行结果[HumanMessage(content示例输入, additional_kwargs{}, response_metadata{}, id2), HumanMessage(content示例输入, additional_kwargs{}, response_metadata{}, id4), AIMessage(content示例输出, additional_kwargs{}, response_metadata{}, id5, tool_calls[], invalid_tool_calls[])]1.4.4 消息合并若我们的消息列表存在连续某种类型相同的消息但实际上某些模型不支持传递相同类型的连续消息。因此对于这种情况我们可以使用merge_message_runs方法轻松合并相同类型的连续消息。# 定义大模型 model ChatDeepSeek(modeldeepseek-chat) # 历史消息记录 messages [ SystemMessage(你是一个聊天助手。), SystemMessage(你总是以笑话回应。), HumanMessage(为什么要使用 LangChain?), HumanMessage(为什么要使用 LangGraph?), AIMessage(因为当你试图让你的代码更有条理时LangGraph 会让你感到“节点”是个好主意), AIMessage(不过别担心它不会“分散”你的注意力), HumanMessage(选择LangChain还是LangGraph?), ] mergemerge_message_runs(messages) #调用大模型方法一 print(model.invoke(messages).content) # #调用大模型方法二 # mergemerge_message_runs() # # chainmerge | model # # print(chain.invoke(messages).content)运行结果选 LangChain 还是 LangGraph这就像问“我该用锤子还是螺丝刀”——如果我想把代码钉进墙里就选 LangChain如果我想画一幅流程图选 LangGraph 准没错不过说真的如果你总是选错我们可能需要聊聊“链”接心理医生的问题了