人工智能大模型AI AgentAgent 框架RAG【免费下载链接】langchainThe agent engineering platform.项目地址https://gitcode.com/GitHub_Trending/la/langchain点击查看免费下载LangChain 的 Message 抽象层为大型语言模型LLM的对话输入输出提供了一套统一、与具体模型厂商无关的表示方式。其核心是BaseMessage——一个可序列化的消息容器既能容纳纯文本字符串也能容纳结构化的**内容块Content Block**列表TypedDict 对象可表示文本、图像、音频、视频、工具调用、推理过程等。本文将结合langchain-core源码系统讲解消息层次结构、统一多模态内容块标准、流式 chunk 聚合机制以及 block translator 如何把标准内容翻译成各家厂商的专有格式帮助你写出可移植、可流式、可扩展的多模态 Agent 应用。一、BaseMessage 层次结构与核心字段源码位置base.pyBaseMessage是所有消息类型的抽象基类继承自Serializable参见 serializable.py这意味着所有消息对象都可以被序列化、反序列化并在 LangChain 生态中安全传递。其核心字段如下字段类型说明contentstr \| list[str \| dict[Any, Any]]消息内容要么是纯文本字符串要么是字符串与字典内容块 dict混合的列表。字符串在内部被视为文本块typestr消息类型的唯一标识schema 必填字段用于反序列化时路由到正确的类如human、ai、system、tool、chat、function以及各自的 chunk 变体additional_kwargsdict[Any, Any]保留给尚未映射到标准字段的厂商专有数据例如 Ollama、DeepSeek 返回的reasoning_contentresponse_metadatadict[Any, Any]响应元数据响应头、logprobs、token 计数、模型名、厂商名model_provider、输出版本output_version等namestr \| None可选人类可读的消息名称大多数模型不会使用idstr \| None可选唯一标识符通常由模型/厂商分配coerce_numbers_to_strTrue允许数字 ID 自动转字符串值得注意的细节BaseMessage.__init__同时接受content和content_blocks两个入口——如果传了content_blocks它会直接赋值给content见 base.py。model_config设置了extraallow允许子类携带额外的自定义字段。1.1 核心消息类型HumanMessagehuman.py表示用户输入用于构造提示词、问题与用户对话轮次type为human流式场景下有HumanMessageChunk变体。AIMessageai.py表示模型输出包含三个专用字段tool_callsToolCalldict 列表结构化的工具调用请求invalid_tool_calls解析失败的ToolCall如 JSON 参数格式错误usage_metadata标准化的 token 计数input_tokens、output_tokens、total_tokens以及可选的input_token_details/output_token_details分类明细见 ai.py。流式场景下的AIMessageChunk变体持有tool_call_chunks部分工具调用块而不是完整的tool_calls。另外AIMessage内置了_backwards_compat_tool_calls模型校验器能够把旧版additional_kwargs[tool_calls]自动解析到标准字段见 ai.py保证向后兼容。SystemMessagesystem.py用于设定模型行为通常是对话中的第一条消息同样支持SystemMessageChunk。ToolMessagetool.py表示一次工具调用的执行结果必填字段tool_call_id把该结果与发起它的AIMessage.tool_calls[].id关联起来支持并行多次工具调用时的配对content工具输出字符串或内容块列表statussuccess或errorartifact可选完整工具输出不发送给模型例如只把摘要放进content、把原始数据放进artifact的场景见 tool.py。此外还有ChatMessage通用消息带role字段与FunctionMessage已废弃的函数调用格式仅保留兼容。ToolOutputMixin见 tool.py保证自定义工具返回非ToolOutputMixin实例时会被自动转成字符串并包装为ToolMessage。RemoveMessagemodifier.py一种特殊消息用于从对话历史中删除其他消息。它只接收id参数要删除的消息 ID不接受任何content传入content会抛ValueError。在 Agent 工作流中常用来裁剪前几轮对话、控制上下文长度。二、Message Chunk 与流式聚合源码位置base.py流式输出时模型会增量地发出AIMessageChunk对象。这些 chunk 被设计为可合并mergeable通过运算符合并时它们会累加内容、按 index 合并工具调用块、并汇总 token 用量。合并实现位于BaseMessageChunk.__add__内部调用merge_contentbase.py合并内容、merge_dicts合并additional_kwargs与response_metadata。AIMessageChunk还支持与 chunk 列表直接相加。AIMessageChunk关键字段ai.pytool_call_chunks部分工具调用对象name与args可为Noneargs是 JSON 字符串片段。相同index的 chunk 在聚合时被合并chunk_position最后一个 chunk 上标记为last作为触发完成动作的信号例如把累积的tool_call_chunks解析为完整的tool_calls。chunk 聚合的 5 步行为对应 ai.py 的init_tool_calls校验器内容合并字符串内容直接拼接列表内容按顺序合并内容块merge_content会把相邻字符串拼到一起工具调用块合并按index字段合并tool_call_chunks同一 index 的两个块拼接name与args字符串Token 用量汇总跨 chunk 累加usage_metadatainput/output token 及分类明细chunk_position 处理当chunk_positionlast末块时累积的tool_call_chunks通过parse_partial_json()解析为完整ToolCalldict见 utils/json.py解析失败则进入invalid_tool_calls并记录error服务端工具调用补全当chunk_positionlast且output_versionv1时server_tool_call_chunk块会被解析为带完整args对象的server_tool_call块见 ai.py 之后的init_server_tool_calls。合并示例chunk1 AIMessageChunk(tool_call_chunks[ToolCallChunk(namesearch, args{q:, index0)]) chunk2 AIMessageChunk(tool_call_chunks[ToolCallChunk(nameNone, argshello}, index0)]) final chunk1 chunk2 # final.tool_call_chunks[0].args {q: hello} # 当末块带 chunk_positionlast 时解析为 # {type: tool_call, name: search, args: {q: hello}, ...}三、内容块统一的多模态表示源码位置content.py内容块是TypedDict 对象代表不同类型的消息内容。不同厂商使用互不兼容的 API schema例如 OpenAI 的image_urlvs. Anthropic 的documentsource 块LangChain 把所有内容归一化到统一格式应用因此可以可移植地处理多模态消息只有到真正调用时才由适配器block translator转换为厂商专有格式。一条发给/来自模型的消息本质上就是一组内容块的有序列表天然支持文本、图像等内容的自然交错。3.1 标准块类型TextContentBlock文本{ type: text, text: str, id: str (可选自动生成), annotations: list[Annotation] (可选引文/元数据), index: int | str (可选用于流式), extras: dict (可选厂商专有字段), }模型的纯文本输出。annotations里的Citation支持指向源文档的引用其start_index/end_index是相对于模型响应文本而非源文档的偏移见 content.py。ReasoningContentBlock推理{ type: reasoning, reasoning: str (可选), id: str (可选), index: int | str (可选), extras: dict (可选), }o1、o3 等模型的链式思考chain-of-thought或中间推理过程。通常从think标签或additional_kwargs中的厂商专有字段提取。ToolCall工具调用{ type: tool_call, id: str | None, name: str, args: dict, index: int | str (可选), extras: dict (可选), }模型发起工具调用的请求。id在同一消息内必须唯一以便与ToolMessage响应配对。ToolCallChunk流式工具调用片段{ type: tool_call_chunk, id: str | None, name: str | None, args: str | None, index: int | str (可选), extras: dict (可选), }流式输出时发出的部分工具调用。字符串args累积 JSON 片段相同index的 chunk 到达后即被合并。InvalidToolCall解析失败的工具调用{ type: invalid_tool_call, id: str | None, name: str | None, args: str | None, error: str | None, index: int | str (可选), extras: dict (可选), }解析失败的工具调用error字段捕获异常信息。3.2 多模态数据块ImageContentBlock图像{ type: image, url: str (可选), base64: str (可选), file_id: str (可选), mime_type: str (可选base64 时必填), id: str (可选), index: int | str (可选), extras: dict (可选), }图像数据可通过 URL、base64 编码或云端文件引用如 OpenAI Files API 的file_id提供。AudioContentBlock、VideoContentBlock结构与ImageContentBlock类似type分别为audio与video。FileContentBlock通用文件{ type: file, url: str (可选), base64: str (可选), file_id: str (可选), mime_type: str (可选), id: str (可选), index: int | str (可选), extras: dict (可选), }PDF、Word 等未被图像/音频/纯文本类型覆盖的通用文件数据。PlainTextContentBlock纯文本文档{ type: text-plain, text: str (可选), base64: str (可选), url: str (可选), file_id: str (可选), mime_type: Literal[text/plain], title: str (可选), context: str (可选), id: str (可选), index: int | str (可选), extras: dict (可选), }带可选标题title与上下文context的纯文本文档帮助模型理解。3.3 服务端工具调用ServerToolCall、ServerToolCallChunk、ServerToolResult支持服务端执行工具如代码执行、网络搜索。模型直接发出这些块请求执行无需本地 handler 代码。3.4 NonStandardContentBlock{ type: non_standard, value: dict, id: str (可选), index: int | str (可选), }保存无法映射到标准块类型的厂商专有内容。block translator 会在content_blocks属性求值时尝试解析这些非标准块。3.5 内容块类型总览与 DataContentBlock 联合类型标准块类型可归纳为五组文本输出TextContentBlocktext、ReasoningContentBlockreasoning链式思考工具调用ToolCalltool_call、ToolCallChunktool_call_chunk流式、InvalidToolCallinvalid_tool_call解析失败服务端工具ServerToolCallserver_tool_call、ServerToolCallChunkserver_tool_call_chunk、ServerToolResultserver_tool_result多模态数据ImageContentBlockimage、AudioContentBlockaudio、VideoContentBlockvideo、FileContentBlockfile、PlainTextContentBlocktext-plain厂商专有NonStandardContentBlocknon_standard。所有多模态数据块构成联合类型DataContentBlock定义见 content.pyDataContentBlock ( ImageContentBlock | VideoContentBlock | AudioContentBlock | PlainTextContentBlock | FileContentBlock )所有块类型都支持可选的extras: dict[str, Any]字段用于承载厂商专有元数据而不破坏标准结构。3.6 访问内容块content_blocks属性源码位置base.pycontent_blocks属性把消息内容归一化为类型化内容块 dict 列表property def content_blocks(self) - list[types.ContentBlock]:其规范化流程6 步content是字符串时包装为{type: text, text: content}解析列表元素字符串变为文本块type为已知块类型的 dict 原样保留其余包装为{type: non_standard, value: ...}通过一系列厂商专有解析器v0 块、Chat Completions 格式、Anthropic 格式、Google GenAI 格式、Bedrock 格式尝试拆解非标准块对AIMessage检查response_metadata[model_provider]若该厂商已注册 translator 则使用之如 OpenAI、Anthropic无 translator 时回退到尽力而为best-effort解析对AIMessage把tool_calls中尚未出现在 content 里的工具调用追加为工具调用块若additional_kwargs[reasoning_content]存在则提取推理块并插入到块列表开头见 base.py 的_extract_reasoning_from_additional_kwargs。四、Block Translators适配各家厂商格式源码位置block_translators/init.py 及 block_translators 目录下的各厂商模块。Block translator 负责在 LangChain 标准块与厂商专有格式之间互转。每个厂商模块注册 translator 函数当AIMessage.content_blocks被访问且response_metadata[model_provider]匹配时调用。4.1 注册系统register_translator与get_translatorinit.pydef register_translator( provider: str, translate_content: Callable[[AIMessage], list[ContentBlock]], translate_content_chunk: Callable[[AIMessageChunk], list[ContentBlock]], ) - None:translator 存储在PROVIDER_TRANSLATORS字典中key 为厂商名value 为{translate_content: ..., translate_content_chunk: ...}模块加载时通过_register_translators()自动初始化init.py目前已内置注册Bedrock、Bedrock Converse、Anthropic、Google GenAI、Google VertexAI、Groq、OpenAI 七家。外部集成可以在运行时调用register_translator()追加自己的厂商 translator。4.2 各厂商 translator 要点OpenAIopenai.py处理 Chat Completions 格式——把 OpenAI 的image_url块转为标准ImageContentBlock解析函数调用的tool_calls为ToolCall块支持 Responses API 的input_audio、input_file、input_image类型并提供公开工具函数convert_to_openai_image_block()与convert_to_openai_data_block()供模型与集成方复用。Anthropicanthropic.py处理 Anthropic 格式——把带source字段类型可为base64、url、file或text的document块转为标准 file/plaintext 块把各种 source 类型的image块转为ImageContentBlock用extras填充如cache_control等厂商专有字段。Google GenAI 与 Google VertexAI把 Google 格式的块转为标准类型。BedrockClassic与 Bedrock Converse把 AWS Bedrock 格式的块转为标准类型。Groq处理 Groq API 响应格式。LangChain v0向后兼容langchain_v0.py解析旧的source_type风格块如{type: image, source_type: url, url: ...}为 v1 块确保用 v0 块格式构造消息的旧代码依然可用。4.3content_blocks中的翻译流程访问AIMessage.content_blocks时ai.py检查response_metadata[output_version]是否为v1——若为 v1且content是列表直接短路返回内容已被归一化为标准 v1 块避免重复解析若response_metadata[model_provider]已设置尝试厂商专有翻译translator 抛NotImplementedError则跳过回退到BaseMessage.content_blocks的尽力而为解析对AIMessage额外追加 content 中缺失的 tool calls并从 kwargs 提取推理块。五、消息操作工具集源码位置utils.pyget_buffer_string(messages, formatprefix)utils.py把消息序列转为单个字符串用于日志、提示词拼接或调试formatprefix默认带角色前缀的格式如Human: ...\nAI: ...。多模态内容块会被跳过只保留文本与text块formatxmlXML 格式输出结构为message typerolecontent/message。支持安全渲染复杂的多模态内容图像、音频、视频、推理、工具调用并正确转义字符base64 编码数据会被跳过。当消息内容可能包含类似角色前缀、造成歧义时使用此格式更稳妥。convert_to_messages与convert_to_openai_messagesutils.py 附近把各种输入格式dict、字符串、MessageLikeRepresentation联合类型强制转换为类型化消息对象。filter_messages(messages, include_types..., exclude_types...)utils.py 附近按类型、名称或 ID 过滤消息序列。trim_messages(messages, max_tokens..., strategy...)utils.py 附近把消息序列裁剪到 token 预算内支持多种策略keep start、keep end、keep first/last 等。merge_message_runs(messages)utils.py 附近去重并合并连续的同类型消息例如连续多条AIMessage。message_chunk_to_message(chunk: BaseMessageChunk) - BaseMessageutils.py 附近把消息 chunk或 chunk 列表转换为完整消息。AnyMessage联合类型utils.pyAnyMessage Annotated[ Annotated[AIMessage, Tag(tagai)] | Annotated[HumanMessage, Tag(taghuman)] | Annotated[ChatMessage, Tag(tagchat)] | Annotated[SystemMessage, Tag(tagsystem)] | Annotated[FunctionMessage, Tag(tagfunction)] | Annotated[ToolMessage, Tag(tagtool)] | Annotated[AIMessageChunk, Tag(tagAIMessageChunk)] | Annotated[HumanMessageChunk, Tag(tagHumanMessageChunk)] | Annotated[ChatMessageChunk, Tag(tagChatMessageChunk)] | Annotated[SystemMessageChunk, Tag(tagSystemMessageChunk)] | Annotated[FunctionMessageChunk, Tag(tagFunctionMessageChunk)] | Annotated[ToolMessageChunk, Tag(tagToolMessageChunk)], Field(discriminatorDiscriminator(_get_type)), ]这是用于 Pydantic 反序列化的带标签联合类型涵盖全部消息类型及 chunk 变体_get_type()判别函数提取每条消息的type字段在反序列化时路由到正确的类。六、内容表示字符串 vs. 块列表消息的content接受两种形式字符串内容简单、向后兼容内部被当作单个文本块处理AIMessage(contentHello, world!)块列表内容AIMessage( content[ {type: text, text: What is this?}, {type: image, url: https://example.com/img.png}, ] )或使用类型化的content_blockskwarg推荐自动生成 ID、无需手写type字段AIMessage( content_blocks[ create_text_block(What is this?), create_image_block(urlhttps://example.com/img.png), ] )工厂函数create_text_block、create_image_block、create_citation等定义见 content.py的好处包括未提供时自动生成 IDUUID4 前缀lc_、创建时严格校验必填参数。6.1text属性源码位置base.pytext属性返回一个TextAccessorstr的子类提取所有文本类型内容块的拼接文本msg AIMessage(content[ {type: text, text: Hello}, {type: image, url: ...}, {type: text, text: World}, ]) print(msg.text) # Hello World为保持向后兼容TextAccessor同时支持属性与方法两种访问方式见 base.py现代写法v1.0msg.text属性访问旧版写法pre-1.0msg.text()方法调用已标记废弃并发出警告计划在 2.0.0 移除。图像、音频、视频、工具调用、推理等非文本块在提取文本时会被自动跳过。七、与 Chat 模型的集成Chat 模型使用消息抽象归一化消息输入与输出输入侧用户提供消息字符串、dict 或MessageLikeRepresentation。模型调用_normalize_messages()转换为BaseMessage对象并可选地为目标厂商展开多模态内容输出侧模型返回AIMessage其中content模型的文本响应多模态时为块列表response_metadata填充model_provider、output_version、token 计数等tool_calls从厂商格式解析为结构化ToolCalldictusage_metadata标准化的 token 计数。模型调用与流式生命周期的完整说明见 chat-models.md。八、厂商专有扩展extras字段内容块的extras字段允许承载厂商元数据而不破坏标准结构{ type: text, text: Response text, extras: { thought_signature: EpoWCpc..., # Google cache_control: {type: ephemeral}, # Anthropic }, }这一设计在保持类型安全的同时支持新兴厂商能力。content.py 的模块文档还预告随着 PEP 728 的广泛采用计划给内容块增加extra_itemsAny参数届时厂商专有字段可以直接放在块顶层见 content.py。九、消息版本化与向后兼容LangChain 1.0 引入了 v1 内容块格式取代了 v0 的source_type风格系统对两者透明处理9.1 v0 块识别与转换v0 块如{type: image, source_type: url, url: ...}通过是否存在source_type字段来识别。在content_blocks归一化过程中v0 块先被包装为{type: non_standard, value: ...}_convert_v0_multimodal_input_to_v1()解析器把它拆解为 v1 格式langchain_v0.py随后作为标准 v1 块继续处理。9.2 厂商专有块的拆解原始厂商块如 OpenAI 的{type: image_url, image_url: {url: ...}}在初次解析时被包装为非标准块在content_blocks属性访问时注册在PROVIDER_TRANSLATORS中的厂商 translator 再将其拆解为标准类型。9.3 输出版本追踪response_metadata[output_version]字段标识内容归一化状态v1内容已被归一化为 v1 块标准 dict 列表。当output_version v1且content是列表非字符串时content_blocks直接返回内容、不再解析——这种短路优化避免了模型响应已经符合 v1 格式时的冗余解析ai.pyNone或缺省内容需要通过厂商 translator 与回退解析进行归一化。对AIMessageChunkv1 短路尤其关键即便output_versionv1如果content是字符串例如纯文本流式内容仍会落到厂商 translator以便从tool_call_chunks构建ContentBlockdict——否则字符串内容会被直接返回工具调用会被静默丢弃ai.py。十、示例工作流10.1 发送多模态消息from langchain_core.messages import HumanMessage, create_text_block, create_image_block message HumanMessage( content_blocks[ create_text_block(Describe this chart.), create_image_block(urlhttps://example.com/chart.png, mime_typeimage/png), ] ) # 访问文本 print(message.text) # Describe this chart. # 获取归一化后的块 for block in message.content_blocks: print(block[type]) # text, image10.2 处理模型返回的工具调用ai_msg model.invoke([...]) # ai_msg.tool_calls [ # {type: tool_call, id: call_1, name: search, args: {query: ...}}, # ] for tool_call in ai_msg.tool_calls: result invoke_tool(tool_call[name], tool_call[args]) tool_response ToolMessage( contentstr(result), tool_call_idtool_call[id], )注意ToolMessage.tool_call_id必须与ai_msg.tool_calls[].id一一对应模型才能正确匹配多次并行工具调用的结果。10.3 流式与 chunk 聚合chunks [] for chunk in model.stream(input_msg): chunks.append(chunk) print(fReceived: {chunk.content}) # 聚合所有 chunk final chunks[0] for chunk in chunks[1:]: final final chunk # final.tool_calls 现在完整了由 tool_call_chunks 解析而来10.4 使用 Block Translators当模型设置了response_metadata[model_provider]时block translator 会被透明调用# OpenAI 模型 ai_msg openai_model.invoke(msg) # response_metadata 包含 model_provideropenai blocks ai_msg.content_blocks # 如果内容来自 OpenAI APItranslator 会把 image_url → ImageContentBlock自定义厂商集成可以注册自己的 translatorfrom langchain_core.messages.block_translators import register_translator def my_translate_content(msg: AIMessage) - list[ContentBlock]: # 自定义逻辑 pass def my_translate_content_chunk(chunk: AIMessageChunk) - list[ContentBlock]: # 自定义逻辑 pass register_translator(my_provider, my_translate_content, my_translate_content_chunk)注册后凡是response_metadata[model_provider] my_provider的消息访问content_blocks时都会走你的翻译逻辑从而把自定义厂商的原始响应归一化为标准内容块。总结LangChain 的消息体系由三个层次支撑BaseMessage类层次提供统一的对话消息载体与流式 chunk 聚合能力**标准内容块Content Block**提供与厂商无关的多模态表示文本、推理、工具调用、图像、音频、视频、文件等Block Translator在调用期把标准块翻译成 OpenAI、Anthropic、Google、Bedrock、Groq 等各家专有格式同时通过 v0 兼容层与output_version追踪保证新旧代码无缝迁移。理解这三层结构是编写可移植多模态应用、实现流式工具调用、以及为自定义厂商接入 LangChain 的基础。赞分享人工智能大模型AI AgentAgent 框架RAG【免费下载链接】langchainThe agent engineering platform.项目地址https://gitcode.com/GitHub_Trending/la/langchain点击查看免费下载相关推荐AI SDK 消息分层架构解析从 UI 消息到 Provider 私有请求的四层消息体系AI SDK 消息分层架构解析从 UI 消息到 Provider 私有请求的四层消息体系 AI SDKThe AI Toolkit for TypeScri人工智能AI 应用AI Agent工具调用MCP ClientsElectron LanguageModelMessage 结构详解本地 AIPrompt API消息模型与多模态内容设计Electron LanguageModelMessage 结构详解本地 AIPrompt API消息模型与多模态内容设计 在 Electron 的实验性桌面应用跨平台前端Wagtail分类法多层次的内容组织结构Wagtail分类法多层次的内容组织结构 痛点内容管理的分类困境 你是否曾经面临这样的困境网站内容越来越多却缺乏有效的组织结构产品分类混乱用户难以找CMS后端上一篇Linux B站客户端 bilibili-linux 快速上手下一篇Windows HEIC缩略图如何免费显示windows-heic-thumbnails 快速安装指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考