[LangChain语言模型组件的设计与实现]消息——Agent与模型交互的媒介

JaydenAI

1102人浏览 · 2026-02-28 11:59:30

JaydenAI · 2026-02-28 11:59:30 发布

Agent开发语境下的模型大体分两种，即传统的文本补齐Completions模型和基于多方交谈的Chat模型，在LangChain中，它们都继承自如下这个BaseLanguageModel抽象类。它继承自RunnableSerializable，所以能成为LCEL链上的一环。

1. 作为语言模型输入/出的消息

class BaseLanguageModel(
    RunnableSerializable[LanguageModelInput, LanguageModelOutputVar], ABC
)

BaseLanguageModel的输入和输出被定义成泛型参数LanguageModelInput和LanguageModelOutputVar。LanguageModelInput是针对多个类型的联合，联合类型包括PromptValue和MessageLikeRepresentation序列，表示提示词的抽象类PromptValue利用抽象方法to_messages赋予了“消息生成”的能力，MessageLikeRepresentation也是一个“消息兼容”联合类型，也可实现针对消息的转换。输出类型LanguageModelOutputVar被约束AIMessage类型或字符串。由此可见，对于LangChain语言模型来说，其输入和输出都可视为“消息”，所以我们先来认识一下LangChain定义的众多消息类型。

LanguageModelOutputVar = TypeVar("LanguageModelOutputVar", AIMessage, str)
LanguageModelInput = PromptValue | str | Sequence[MessageLikeRepresentation]

class PromptValue(Serializable, ABC):
    @abstractmethod
    def to_string(self) -> str:

    @abstractmethod
    def to_messages(self) -> list[BaseMessage]

MessageLikeRepresentation = (
    BaseMessage | list[str] | tuple[str, str] | str | dict[str, Any]
)

如下这个UML类图囊括了语言模型的输入和输出涉及到的绝大部分类型，其中包括描述消息的基类BaseMessage以及针对不同角色的消息子类（SystemMessage、HumanMessage、AIMessage和ToolMessage等），还包括表示提示词的PromptValue及其子类（StringPromptValue、ChatPromptValue和PromptValueConcrete等）。消息的主体内容通过ContentBlock表示，不同形式的内容对应不同的类型，ContentBlock是这些类型的联合。框起来的部分就是我们接下来着重介绍的部分。

Alternative Text

2. BaseMessage

LangChain的消息类型直接或者间接地继承自如下这个BaseMessage基类，这是一个派生自Serializable的可序列化的类型。BaseMessageChunk派生于BaseMessage，表示“流式输出”场景下的消息。Chat模型的消息继承自ChatMessage和ChatMessageChunk，具体的消息类型（SystemMessage/SystemMessageChunk、HumanMessage/HumanMessageChunk、AIMessage/AIMessageChunk和ToolMessage/ToolMessageChunk等）会关联一个具体的角色。

对于基类的BaseMessage来说，表示消息内容的content字段可以是单纯的字符串或字典（Key为字符串）的列表，其他于消息相关信息存储在additional_kwargs字段对应的字典中，比如语言模型返回的AIMessage可以利用它来保存涉及的工具调用。response_metadata字段用于存储响应的元数据，比如响应的Header、语言模型的名称，涉及的Token消费数据等。

class BaseMessage(Serializable):
    content: str | list[str | dict]
    additional_kwargs: dict = Field(default_factory=dict)
    response_metadata: dict = Field(default_factory=dict)
    type: str
    name: str | None = None
    id: str | None = Field(default=None, coerce_numbers_to_str=True)

    @property
    def content_blocks(self) -> list[types.ContentBlock]

    @property
    def text(self) -> TextAccessor

    def pretty_repr(
        self,
        html: bool = False,
    ) -> str
    def pretty_print(self) -> None

为了明确消息承载的内容形式，以便于其内容能够被正常反序列化，我们必须利用其type字段指定一个标准的类型。除此之外，我们还可以利用id和name字典赋予消息一个唯一标识和可读性的名称，这两个字段都是可以缺省的。BaseMesage的text属性提供的TextAccessor是一个返回类型为字符串的可执行对象，我们可以利用它得到消息内容的文本表示，不过TextAccessor类型已经被标注为deprecated。它的两个pretty_print方法可以以指定的形式（HTML或者简单文本）输出响应的内容来描述当前的消息。

消息承载的内容多种多样，可以是简单的文本，还可以是多媒体图片、音频和视频，还可以是一个任意的二进制文件，不同类型的内容具有不同的处理方式，所以LangChain利用ConentBlock这个类型实现了“内容的标准化”。类似于HTTP的媒体类型（Media Type，也成为MIME类型），每个ConentBlock对象都关联一个标准的类型名称（很多采用的就是MIME类型）。BaseMessage的content_blocks属性实现了原始形态的内容到ConentBlock列表的转换。

语言模型可能需要经历耗时的处理流程后才能生成完整的内容，但它可以利用“流式传输”实时返回当前生成的“消息碎片”。此消息碎片通过如下这个BaseMessageChunk类表示，虽然我们以“碎片”称呼它，其他一个BaseMessageChunk也可以视为一个消息，因为它继承了BaseMessage基类。

class BaseMessageChunk(BaseMessage):
    def __add__(self, other: Any) -> BaseMessageChunk

由于分块传输的“碎片”是完整消息的一部分，所以它们应该可以“拼接”成完整的消息，这个“拼接”的能力以重写的__add__方法被赋予，所以我们可以采用如下的形式使用“+”操作符对其实施拼接。

result = AIMessageChunk(content="Hello", ...) + AIMessageChunk(content=" World", ...)
# AIMessageChunk(content="Hello World", ...)

3. ChatMessage/ChatMessageChunk

传统的基于“文本补齐”的消息交互方式已经全面转向了“多角色参与”的聊天模式，后者涉及的消息类型以ChatMessage/ChatMessageChunk为基类。ChatMessage利用role字段命名发出该消息的“角色”，它的默认值为“chat”，派生于它的ChatMessageChunk的类型为“ChatMessageChunk”。

class ChatMessage(BaseMessage):
    role: str
    type: Literal["chat"] = "chat"

class ChatMessageChunk(ChatMessage, BaseMessageChunk):
    type: Literal["ChatMessageChunk"] = "ChatMessageChunk"

4. SystemMessage/SystemMessageChunk

在 LangChain 以及底层的 Chat API 架构中，系统消息是用于定义模型“人格”与“运行规则”的核心组件。它告诉 AI “你是谁”（例如资深 Python 开发者、苏格拉底式的导师、或是一只可爱的猫娘）。规定模型“不能做什么”（例如严禁提及竞争对手、不准输出代码、只能用 JSON 格式回答）。它通常位于消息列表的最顶端，作为整个对话的“宪法”，其权重通常高于普通的消息。它们具有专属的类型“system”和“SystemMessageChunk”。

class SystemMessage(BaseMessage):
    type: Literal["system"] = "system"

class SystemMessageChunk(SystemMessage, BaseMessageChunk):
    type: Literal["SystemMessageChunk"] = "SystemMessageChunk"

5. HumanMessage/HumenMessageChunk

在 LangChain 的消息架构中，HumanMessage/HumanMessageChunk代表了对话的“需求侧”，即真实用户发送给模型的消息。它是用户意图的直接表达，包含了模型需要完成的具体任务或提出的疑问，它们对应的专属类型分别为“human”和“HumanMessageChunk”。

class HumanMessage(BaseMessage):
    type: Literal["human"] = "human"
class HumanMessageChunk(HumanMessage, BaseMessageChunk):
    type: Literal["HumanMessageChunk"] = "HumanMessageChunk"

6. AIMessage/AIMessageChunk

AIMessage/AIMessageChunk代表模型生成的响应。它是对话闭环的关键，承载了 AI 的回答、推理逻辑及工具调用指令。它们是模型在接收到SystemMessage和HumanMessage后产生的输出。对于OpenAI来说，它对应 OpenAI中的assistant角色。它们对应的专属类型为“ai”和“AIMessageChunk”。如果涉及针对工具的调用，描述每个工具调用的ToolCall会出现在tool_calls字段返回的列表中，另一个invalid_tool_calls字段返回于工具调用相关的错误。出现在AIMessageChunk中针对工具调用的描述类型为“AIMessageChunk”。

class AIMessage(BaseMessage):
    tool_calls: list[ToolCall] = Field(default_factory=list)
    invalid_tool_calls: list[InvalidToolCall] = Field(default_factory=list)
    usage_metadata: UsageMetadata | None = None
    type: Literal["ai"] = "ai"

    @property
    def content_blocks(self) -> list[types.ContentBlock]
    @override
    def pretty_repr(self, html: bool = False) -> str:

class AIMessageChunk(AIMessage, BaseMessageChunk):
    type: Literal["AIMessageChunk"] = "AIMessageChunk" 
    tool_call_chunks: list[ToolCallChunk] = Field(default_factory=list)
    chunk_position: Literal["last"] | None = None

在流式传输中，模型返回的是tool_call_chunks（这是碎片的、不完整的、无法直接调用的 JSON 片段）。为了性能和鲁棒性，LangChain 不会在每一个碎片到达时都尝试进行昂贵的完全解析，而是选择将这些碎片“累加”起来。当一个带有chunk_position="last"的AIMessageChunk被合并进当前的流时，它像一个“发令枪”告诉框架：“流已经结束，现在可以安全地将累积的tool_call_chunks解析成正式的、结构化的tool_calls列表了”。

ToolCall和ToolCallChunk定义如下，它们都是一个类型化字典。共同字段name、args和id分别表示调用的工具名称、传入的参数和当前工具调用的唯一标识。它们同样具有专属的类型，分别为“tool_call”和“tool_call_chunk”。ToolCallChunk具有一个表示偏移量的index字段。

class ToolCall(TypedDict):
    name: str
    args: dict[str, Any]
    id: str | None
    type: NotRequired[Literal["tool_call"]]

class ToolCallChunk(TypedDict):
    name: str | None
    args: str | None
    id: str | None
    index: int | None
    type: NotRequired[Literal["tool_call_chunk"]]

7. ToolMessage/ToolMessageChunk

ToolMessage/ToolMessageChunk属于对话的“执行层”，用于向模型反馈外部工具执行的结果。当 Agent接收到语言模型发出的带有tool_calls的AIMessage后，它会执行对应的工具，并将结果包装在ToolMessage中反馈给语言模型。它们专属的类型分别是“tool”和“ToolMessageChunk”。tool_call_id和status字段分别标识工具调用的标识和状态。

class ToolMessage(BaseMessage, ToolOutputMixin):
    tool_call_id: str
    type: Literal["tool"] = "tool"
    artifact: Any = None
    status: Literal["success", "error"] = "success"

class ToolMessageChunk(ToolMessage, BaseMessageChunk):
    type: Literal["ToolMessageChunk"] = "ToolMessageChunk"

有时候工具返回的数据非常庞大（如几千行的 DataFrame、原始图像字节流、复杂的 API 响应对象），如果全部放入content字段传给 LLM，会导致 Token 爆炸或者超出模型上下文窗口的限制，我们在这种情况下可以将它们存储在artifact字段上，该字段允许我们将原始执行结果保留在消息对象中，但不发送给模型。

8. FunctionMessage/FunctionMessageChunk

工具调用的前身是“函数调用”，函数调用的结果被封装成FunctionMessage/FunctionMessageChunk对象后被发送给语言模型，它们已经逐渐被ToolMessage/ToolMessageChunk代替。这两个类型对应的专属类型为“function”和“FunctionMessageChunk”。

class FunctionMessage(BaseMessage):
    name: str
    type: Literal["function"] = "function"

class FunctionMessageChunk(FunctionMessage, BaseMessageChunk):
    type: Literal["FunctionMessageChunk"] = "FunctionMessageChunk"