flowchart TB
    subgraph UserLayer["应用层"]
        USER[用户应用]
        AGENT[Agents<br/>智能代理]
        CHAIN[Chains<br/>链式调用]
    end

    subgraph CoreLayer["核心抽象层"]
        RUNNABLE[Runnables<br/>可执行接口]
        PROMPT[Prompts<br/>提示词模板]
        MESSAGES[Messages<br/>消息抽象<br/>⭐本模块]
        CALLBACK[Callbacks<br/>回调机制]
    end

    subgraph ModelLayer["模型层"]
        LLM[LLMs<br/>文本生成模型]
        CHAT[ChatModels<br/>对话模型]
        EMBEDDINGS[Embeddings<br/>嵌入模型]
    end

    subgraph ToolLayer["工具层"]
        TOOLS[Tools<br/>工具定义]
        RETRIEVER[Retrievers<br/>检索器]
        MEMORY[Memory<br/>记忆存储]
    end

    subgraph DataLayer["数据层"]
        VECTOR[VectorStores<br/>向量数据库]
        DOCUMENT[Document<br/>文档加载器]
        OUTPUT[OutputParsers<br/>输出解析]
    end

    USER --> AGENT
    USER --> CHAIN
    AGENT --> RUNNABLE
    CHAIN --> RUNNABLE

    RUNNABLE --> PROMPT
    RUNNABLE --> CHAT
    RUNNABLE --> TOOLS

    PROMPT --> MESSAGES
    CHAT --> MESSAGES
    AGENT --> MESSAGES

    CHAT --> CALLBACK
    MESSAGES --> OUTPUT

    TOOLS --> MESSAGES
    MEMORY --> MESSAGES
    RETRIEVER --> DOCUMENT

    style MESSAGES fill:#ff6b6b,stroke:#c92a2a,stroke-width:3px,color:#fff
    style CHAT fill:#4ecdc4,stroke:#0e9594
    style PROMPT fill:#95e1d3,stroke:#38ada9
    style TOOLS fill:#f9ca24,stroke:#f0932b

sequenceDiagram
    autonumber
    participant App as 应用层
    participant Chain as Chain/Agent
    participant Prompt as PromptTemplate
    participant Messages as Messages模块
    participant ChatModel as ChatModel
    participant Provider as 模型提供商API

    App->>Chain: 用户输入
    Chain->>Prompt: format_messages({input})

    Note over Prompt,Messages: 1️⃣ 消息生成阶段
    Prompt->>Messages: convert_to_messages(...)
    Messages->>Messages: 创建HumanMessage/SystemMessage
    Messages-->>Prompt: [BaseMessage]

    Prompt-->>Chain: messages列表
    Chain->>ChatModel: invoke(messages)

    Note over ChatModel,Messages: 2️⃣ 消息转换阶段
    ChatModel->>Messages: convert_to_messages(input)
    Messages->>Messages: 验证并标准化
    Messages-->>ChatModel: 标准化的messages

    ChatModel->>ChatModel: _convert_to_provider_format()
    ChatModel->>Provider: API调用(provider格式)

    Note over Provider,Messages: 3️⃣ 响应构建阶段
    Provider-->>ChatModel: API响应(JSON)
    ChatModel->>Messages: 构造AIMessage
    Messages->>Messages: 解析tool_calls/usage_metadata
    Messages-->>ChatModel: AIMessage实例

    ChatModel-->>Chain: AIMessage
    Chain-->>App: 最终响应

flowchart TB
    subgraph Base["基类层"]
        BASE_MSG[BaseMessage<br/>消息基类]
        BASE_CHUNK[BaseMessageChunk<br/>可合并消息块]
    end

    subgraph CoreTypes["核心消息类型"]
        HUMAN[HumanMessage<br/>用户消息]
        AI[AIMessage<br/>AI消息]
        SYSTEM[SystemMessage<br/>系统消息]
        TOOL[ToolMessage<br/>工具结果消息]
        FUNCTION[FunctionMessage<br/>函数结果(废弃)]
        CHAT[ChatMessage<br/>自定义角色]
    end

    subgraph Chunks["消息块类型"]
        HUMAN_CHUNK[HumanMessageChunk]
        AI_CHUNK[AIMessageChunk]
        SYSTEM_CHUNK[SystemMessageChunk]
        TOOL_CHUNK[ToolMessageChunk]
    end

    subgraph Content["内容类型"]
        TEXT[TextContentBlock<br/>文本]
        IMAGE[ImageContentBlock<br/>图片]
        AUDIO[AudioContentBlock<br/>音频]
        VIDEO[VideoContentBlock<br/>视频]
        FILE[FileContentBlock<br/>文件]
        REASONING[ReasoningContentBlock<br/>推理过程]
    end

    subgraph ToolCalling["工具调用"]
        TOOL_CALL[ToolCall<br/>工具调用请求]
        TOOL_CALL_CHUNK[ToolCallChunk<br/>工具调用块]
        INVALID_TOOL[InvalidToolCall<br/>无效工具调用]
        SERVER_TOOL[ServerToolCall<br/>服务端工具]
    end

    subgraph Utils["工具函数"]
        CONVERT[convert_to_messages]
        FILTER[filter_messages]
        TRIM[trim_messages]
        MERGE[merge_message_runs]
        TO_DICT[messages_to_dict]
        FROM_DICT[messages_from_dict]
    end

    BASE_MSG <|-- BASE_CHUNK
    BASE_MSG <|-- HUMAN
    BASE_MSG <|-- AI
    BASE_MSG <|-- SYSTEM
    BASE_MSG <|-- TOOL
    BASE_MSG <|-- FUNCTION
    BASE_MSG <|-- CHAT

    BASE_CHUNK <|-- HUMAN_CHUNK
    BASE_CHUNK <|-- AI_CHUNK
    BASE_CHUNK <|-- SYSTEM_CHUNK
    BASE_CHUNK <|-- TOOL_CHUNK

    AI --> TOOL_CALL
    AI --> TOOL_CALL_CHUNK
    AI --> INVALID_TOOL

    BASE_MSG --> Content
    BASE_MSG --> Utils

    style BASE_MSG fill:#e1f5ff
    style AI fill:#fff4e1
    style HUMAN fill:#e8f5e9
    style TOOL_CALL fill:#ffe8e8

classDiagram
    class BaseMessage {
        <<abstract>>
        +content: str|list
        +additional_kwargs: dict
        +response_metadata: dict
        +type: str
        +name: str|None
        +id: str|None
        +text: TextAccessor
        +pretty_repr() str
    }

    class HumanMessage {
        +type: "human"
        +example: bool
    }

    class AIMessage {
        +type: "ai"
        +tool_calls: list[ToolCall]
        +invalid_tool_calls: list[InvalidToolCall]
        +usage_metadata: UsageMetadata|None
    }

    class SystemMessage {
        +type: "system"
    }

    class ToolMessage {
        +type: "tool"
        +tool_call_id: str
        +artifact: Any
        +status: Literal["success", "error"]
    }

    class ToolCall {
        +name: str
        +args: dict
        +id: str
        +type: Literal["tool_call"]
    }

    class UsageMetadata {
        +input_tokens: int
        +output_tokens: int
        +total_tokens: int
        +input_token_details: InputTokenDetails
        +output_token_details: OutputTokenDetails
    }

    class ContentBlock {
        <<union>>
        TextContentBlock
        ImageContentBlock
        AudioContentBlock
        VideoContentBlock
        FileContentBlock
    }

    BaseMessage <|-- HumanMessage
    BaseMessage <|-- AIMessage
    BaseMessage <|-- SystemMessage
    BaseMessage <|-- ToolMessage

    AIMessage --> ToolCall : contains
    AIMessage --> UsageMetadata : contains
    BaseMessage --> ContentBlock : content

{
    "type": "image",
    "source": "url",  # 或 "base64"
    "url": "https://...",  # 或 "data": "base64..."
    "detail": "auto"  # 或 "low"/"high"
}

{
    "type": "audio",
    "source": "base64",
    "data": "base64..."
}

messages: Sequence[MessageLikeRepresentation]
# MessageLikeRepresentation = BaseMessage | list | tuple | str | dict

list[BaseMessage]  # 标准消息列表

def convert_to_messages(
    messages: Sequence[MessageLikeRepresentation],
) -> list[BaseMessage]:
    """将各种格式转换为标准消息列表

    Args:
        messages: 消息序列,支持多种格式

    Returns:
        标准BaseMessage列表
    """
    result = []
    for message in messages:
        result.append(_convert_single_message(message))
    return result

def _convert_single_message(message: MessageLikeRepresentation) -> BaseMessage:
    # 1) 已经是BaseMessage,直接返回
    if isinstance(message, BaseMessage):
        return message

    # 2) 字符串 → HumanMessage
    elif isinstance(message, str):
        return HumanMessage(content=message)

    # 3) 元组 (role, content) → 对应类型消息
    elif isinstance(message, tuple) and len(message) == 2:
        role, content = message
        return _role_to_message(role, content)

    # 4) 字典 {role: ..., content: ...} → 对应类型消息
    elif isinstance(message, dict):
        if "role" in message:
            return _role_to_message(message["role"], message["content"])
        else:
            # 尝试直接构造消息
            return _message_from_dict(message)

    else:
        msg = f"Unsupported message type: {type(message)}"
        raise ValueError(msg)

def _role_to_message(role: str, content: str) -> BaseMessage:
    """根据角色创建对应类型消息"""
    role = role.lower()
    if role in ("human", "user"):
        return HumanMessage(content=content)
    elif role in ("ai", "assistant"):
        return AIMessage(content=content)
    elif role == "system":
        return SystemMessage(content=content)
    elif role == "tool":
        # 工具消息需要tool_call_id
        return ToolMessage(content=content, tool_call_id="")
    else:
        # 自定义角色使用ChatMessage
        return ChatMessage(content=content, role=role)

# 在模型调用中自动转换
from langchain_openai import ChatOpenAI

model = ChatOpenAI()

# 各种输入格式都会自动转换
result = model.invoke("hello")  # str → [HumanMessage("hello")]
result = model.invoke([("system", "You are..."), ("user", "hi")])  # tuple → messages
result = model.invoke([{"role": "user", "content": "hi"}])  # dict → messages

# 在Prompt模板中
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are {role}"),
    ("user", "{input}"),
])

# format_messages内部调用convert_to_messages
messages = prompt.format_messages(role="assistant", input="hello")

sequenceDiagram
    autonumber
    participant User
    participant Convert as convert_to_messages
    participant Model as ChatModel

    User->>Model: invoke([<br/>  ("system", "You are..."),<br/>  "What is AI?"<br/>])
    Model->>Model: _convert_input(input)
    Model->>Convert: convert_to_messages([...])

    Convert->>Convert: 处理元组("system", ...)
    Convert->>Convert: SystemMessage("You are...")

    Convert->>Convert: 处理字符串"What is AI?"
    Convert->>Convert: HumanMessage("What is AI?")

    Convert-->>Model: [SystemMessage, HumanMessage]
    Model->>Model: 发送到API
    Model-->>User: AIMessage(response)

# messages_to_dict
messages: Sequence[BaseMessage]

# messages_from_dict
messages: list[dict]  # 包含__class__字段的字典

# messages_to_dict
list[dict]  # 可序列化的字典列表

# messages_from_dict
list[BaseMessage]  # 恢复的消息对象

def messages_to_dict(messages: Sequence[BaseMessage]) -> list[dict]:
    """将消息列表序列化为字典列表

    Args:
        messages: BaseMessage序列

    Returns:
        可JSON序列化的字典列表
    """
    return [message_to_dict(m) for m in messages]

def message_to_dict(message: BaseMessage) -> dict:
    """将单个消息序列化为字典"""
    # 使用dumpd进行序列化(处理Pydantic模型)
    data = dumpd(message)

    # 添加类型信息用于反序列化
    data["__class__"] = {
        "module": message.__class__.__module__,
        "name": message.__class__.__name__,
    }

    return data

def messages_from_dict(messages: list[dict]) -> list[BaseMessage]:
    """从字典列表反序列化消息

    Args:
        messages: 字典列表,包含__class__字段

    Returns:
        恢复的BaseMessage列表
    """
    return [_message_from_dict(m) for m in messages]

def _message_from_dict(message: dict) -> BaseMessage:
    """从单个字典反序列化消息"""
    # 1) 提取类型信息
    if "__class__" in message:
        class_info = message["__class__"]
        module = class_info["module"]
        name = class_info["name"]

        # 2) 动态导入类
        module_obj = importlib.import_module(module)
        cls = getattr(module_obj, name)

        # 3) 构造实例
        data = {k: v for k, v in message.items() if k != "__class__"}
        return cls(**data)

    else:
        # 兼容旧格式,根据type字段推断
        msg_type = message.get("type")
        if msg_type == "human":
            return HumanMessage(**message)
        elif msg_type == "ai":
            return AIMessage(**message)
        elif msg_type == "system":
            return SystemMessage(**message)
        elif msg_type == "tool":
            return ToolMessage(**message)
        else:
            return ChatMessage(**message)

import json

# 序列化消息保存到数据库
messages = [
    HumanMessage("Hello"),
    AIMessage("Hi, how can I help?"),
]

# 转换为字典
dicts = messages_to_dict(messages)

# 保存到JSON文件
with open("chat_history.json", "w") as f:
    json.dump(dicts, f)

# 从JSON恢复
with open("chat_history.json") as f:
    loaded_dicts = json.load(f)

# 反序列化
loaded_messages = messages_from_dict(loaded_dicts)

assert loaded_messages[0].content == "Hello"
assert isinstance(loaded_messages[1], AIMessage)

messages: Sequence[BaseMessage]
include_names: Sequence[str] | None  # 包含的名称
exclude_names: Sequence[str] | None  # 排除的名称
include_types: Sequence[type | str] | None  # 包含的类型
exclude_types: Sequence[type | str] | None  # 排除的类型
include_ids: Sequence[str] | None  # 包含的ID
exclude_ids: Sequence[str] | None  # 排除的ID

list[BaseMessage]  # 过滤后的消息列表

def filter_messages(
    messages: Sequence[BaseMessage],
    *,
    include_names: Sequence[str] | None = None,
    exclude_names: Sequence[str] | None = None,
    include_types: Sequence[type | str] | None = None,
    exclude_types: Sequence[type | str] | None = None,
    include_ids: Sequence[str] | None = None,
    exclude_ids: Sequence[str] | None = None,
) -> list[BaseMessage]:
    """根据条件过滤消息

    Args:
        messages: 输入消息列表
        include_names: 只保留这些名称的消息
        exclude_names: 排除这些名称的消息
        include_types: 只保留这些类型的消息
        exclude_types: 排除这些类型的消息
        include_ids: 只保留这些ID的消息
        exclude_ids: 排除这些ID的消息

    Returns:
        过滤后的消息列表
    """
    filtered = []

    for message in messages:
        # 1) 检查ID过滤
        if include_ids and message.id not in include_ids:
            continue
        if exclude_ids and message.id in exclude_ids:
            continue

        # 2) 检查名称过滤
        if include_names and message.name not in include_names:
            continue
        if exclude_names and message.name in exclude_names:
            continue

        # 3) 检查类型过滤
        if include_types:
            type_match = any(
                isinstance(message, t) if isinstance(t, type) else message.type == t
                for t in include_types
            )
            if not type_match:
                continue

        if exclude_types:
            type_match = any(
                isinstance(message, t) if isinstance(t, type) else message.type == t
                for t in exclude_types
            )
            if type_match:
                continue

        # 4) 通过所有过滤条件,保留消息
        filtered.append(message)

    return filtered

# 只保留用户和AI消息,排除系统消息
filtered = filter_messages(
    messages,
    include_types=[HumanMessage, AIMessage]
)

# 排除特定名称的消息
filtered = filter_messages(
    messages,
    exclude_names=["debug_user"]
)

# 组合条件:只保留human类型且名称为"alice"
filtered = filter_messages(
    messages,
    include_types=["human"],
    include_names=["alice"]
)

messages: Sequence[BaseMessage]
max_tokens: int | None  # 最大token数
token_counter: Callable | BaseLanguageModel | None  # token计数器
strategy: Literal["first", "last"]  # 保留策略
allow_partial: bool  # 是否允许部分消息
start_on: str | None  # 起始消息类型
end_on: str | None  # 结束消息类型
include_system: bool  # 是否始终包含系统消息

list[BaseMessage]  # 裁剪后的消息列表

def trim_messages(
    messages: Sequence[BaseMessage],
    *,
    max_tokens: int | None = None,
    token_counter: Callable | BaseLanguageModel | None = None,
    strategy: Literal["first", "last"] = "last",
    allow_partial: bool = False,
    start_on: Literal["human", "ai", "system"] | None = None,
    end_on: Literal["human", "ai", "system"] | None = None,
    include_system: bool = False,
) -> list[BaseMessage]:
    """根据token限制裁剪消息列表

    Args:
        messages: 输入消息列表
        max_tokens: 最大token数
        token_counter: token计数函数或模型
        strategy: "first"保留最早,"last"保留最新
        allow_partial: 是否允许截断消息
        start_on: 起始消息类型要求
        end_on: 结束消息类型要求
        include_system: 是否始终保留系统消息

    Returns:
        裁剪后的消息列表
    """
    # 1) 确定token计数器
    if token_counter is None:
        token_counter = _get_token_ids_default_method
    elif isinstance(token_counter, BaseLanguageModel):
        token_counter = token_counter.get_num_tokens

    # 2) 分离系统消息
    system_messages = []
    other_messages = []
    for msg in messages:
        if include_system and isinstance(msg, SystemMessage):
            system_messages.append(msg)
        else:
            other_messages.append(msg)

    # 3) 计算系统消息token数
    system_tokens = sum(token_counter(msg.content) for msg in system_messages)
    remaining_tokens = max_tokens - system_tokens if max_tokens else None

    # 4) 根据策略选择消息
    if strategy == "last":
        # 从后向前保留
        selected = _trim_from_end(other_messages, remaining_tokens, token_counter, allow_partial)
    else:
        # 从前向后保留
        selected = _trim_from_start(other_messages, remaining_tokens, token_counter, allow_partial)

    # 5) 应用start_on/end_on约束
    if start_on:
        selected = _ensure_start_on(selected, start_on)
    if end_on:
        selected = _ensure_end_on(selected, end_on)

    # 6) 合并系统消息
    return system_messages + selected

def _trim_from_end(
    messages: list[BaseMessage],
    max_tokens: int | None,
    token_counter: Callable,
    allow_partial: bool,
) -> list[BaseMessage]:
    """从后向前保留消息,不超过token限制"""
    if max_tokens is None:
        return messages

    selected = []
    total_tokens = 0

    for msg in reversed(messages):
        msg_tokens = token_counter(msg.content)

        if total_tokens + msg_tokens <= max_tokens:
            # 完整消息可以加入
            selected.insert(0, msg)
            total_tokens += msg_tokens
        elif allow_partial and total_tokens < max_tokens:
            # 截断消息
            remaining = max_tokens - total_tokens
            truncated = _truncate_message(msg, remaining, token_counter)
            selected.insert(0, truncated)
            break
        else:
            # 无法加入更多消息
            break

    return selected

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")

# 裁剪消息以适应上下文窗口
trimmed = trim_messages(
    long_conversation,
    max_tokens=4000,
    token_counter=model,  # 使用模型的token计数器
    strategy="last",  # 保留最新消息
    include_system=True,  # 始终保留系统消息
)

# 发送裁剪后的消息
response = model.invoke(trimmed)

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant Model as ChatOpenAI
    participant Converter as convert_to_messages
    participant BaseMsg as BaseMessage
    participant AIMsg as AIMessage

    User->>Model: invoke([<br/>  "You are helpful",<br/>  {"role": "user", "content": "Hi"}<br/>])

    Note over Model,Converter: 📥 输入标准化
    Model->>Model: _convert_input(input)
    Model->>Converter: convert_to_messages([...])

    Converter->>Converter: 遍历输入列表

    Note over Converter: 处理第1项: 字符串
    Converter->>Converter: _convert_to_message("You are helpful")
    Converter->>BaseMsg: HumanMessage("You are helpful")
    BaseMsg-->>Converter: HumanMessage实例

    Note over Converter: 处理第2项: 字典
    Converter->>Converter: _convert_to_message({role: "user", ...})
    Converter->>Converter: _create_message_from_message_type(<br/>  "user", "Hi"<br/>)
    Converter->>BaseMsg: HumanMessage("Hi")
    BaseMsg-->>Converter: HumanMessage实例

    Converter-->>Model: [HumanMessage, HumanMessage]

    Note over Model: 🌐 调用API
    Model->>Model: _generate_with_retry(messages)
    Model->>Model: 转换为OpenAI格式
    Model->>Model: HTTP POST /chat/completions

    Note over Model: 📤 响应处理
    Model->>Model: 解析API响应JSON
    Model->>AIMsg: AIMessage(<br/>  content="Hello! How can I help you?",<br/>  response_metadata={...},<br/>  usage_metadata={...}<br/>)
    AIMsg-->>Model: AIMessage实例

    Model-->>User: AIMessage

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant Model as ChatModel
    participant HumanMsg as HumanMessage
    participant ContentBlocks as content_blocks属性
    participant Translator as BlockTranslator

    User->>HumanMsg: HumanMessage(content=[<br/>  {"type": "text", "text": "Describe this"},<br/>  {"type": "image", "source": {...}}<br/>])

    Note over HumanMsg: 💾 存储原始content
    HumanMsg->>HumanMsg: self.content = [dict1, dict2]

    User->>Model: invoke([message])
    Model->>Model: _convert_input([message])

    Note over Model,ContentBlocks: 🔍 内容块解析
    Model->>ContentBlocks: message.content_blocks
    ContentBlocks->>ContentBlocks: 遍历content列表

    ContentBlocks->>ContentBlocks: 检查第1项: {"type": "text"}
    Note over ContentBlocks: 标准v1格式,直接使用
    ContentBlocks->>ContentBlocks: blocks.append(TextContentBlock)

    ContentBlocks->>ContentBlocks: 检查第2项: {"type": "image", "source": ...}
    Note over ContentBlocks: 非标准格式,标记为non_standard
    ContentBlocks->>ContentBlocks: blocks.append({<br/>  "type": "non_standard",<br/>  "value": {...}<br/>})

    Note over ContentBlocks,Translator: 🔄 格式转换
    ContentBlocks->>Translator: _convert_to_v1_from_anthropic_input(blocks)
    Translator->>Translator: 识别Anthropic格式
    Translator->>Translator: 转换为v1标准ImageContentBlock
    Translator-->>ContentBlocks: 标准化的blocks

    ContentBlocks-->>Model: [<br/>  TextContentBlock,<br/>  ImageContentBlock<br/>]

    Note over Model: 🌐 转换为提供商格式
    Model->>Model: convert_to_openai_messages([message])
    Model->>Model: 转换ImageContentBlock → image_url格式
    Model->>Model: 发送API请求

    Model-->>User: AIMessage(response)

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant Agent as Agent/Chain
    participant Model as ChatModel(bind_tools)
    participant ToolExec as Tool执行器
    participant AIMsg as AIMessage
    participant ToolMsg as ToolMessage

    Note over User,Agent: 🚀 初始化工具绑定
    User->>Model: model.bind_tools([get_weather_tool])
    Model->>Model: 保存tool schemas
    Model-->>User: 配置后的model

    Note over User,Agent: 💬 第1轮: 用户请求
    User->>Agent: "What's the weather in Beijing?"
    Agent->>Agent: messages = [HumanMessage(...)]
    Agent->>Model: invoke(messages)

    Note over Model: 📤 发送带工具定义的请求
    Model->>Model: 构造API请求:<br/>{<br/>  messages: [...],<br/>  tools: [weather_tool_schema]<br/>}
    Model->>Model: HTTP POST to API

    Note over Model,AIMsg: 🔧 解析工具调用
    Model->>Model: 接收API响应:<br/>{<br/>  "tool_calls": [{<br/>    "id": "call_123",<br/>    "function": {<br/>      "name": "get_weather",<br/>      "arguments": '{"location":"Beijing"}'<br/>    }<br/>  }]<br/>}

    Model->>Model: default_tool_parser(raw_tool_calls)
    Model->>AIMsg: AIMessage(<br/>  content="",<br/>  tool_calls=[<br/>    ToolCall(<br/>      name="get_weather",<br/>      args={"location": "Beijing"},<br/>      id="call_123"<br/>    )<br/>  ]<br/>)
    AIMsg-->>Model: AIMessage实例
    Model-->>Agent: AIMessage(tool_calls=[...])

    Note over Agent,ToolExec: ⚙️ 执行工具
    Agent->>Agent: 检测到tool_calls
    Agent->>ToolExec: 调用get_weather(location="Beijing")
    ToolExec->>ToolExec: 执行工具逻辑
    ToolExec-->>Agent: "Sunny, 25°C"

    Note over Agent,ToolMsg: 📝 构造工具结果
    Agent->>ToolMsg: ToolMessage(<br/>  content="Sunny, 25°C",<br/>  tool_call_id="call_123"<br/>)
    ToolMsg-->>Agent: ToolMessage实例

    Agent->>Agent: messages.extend([<br/>  AIMessage(tool_calls),<br/>  ToolMessage(result)<br/>])

    Note over Agent,Model: 💬 第2轮: 生成最终答案
    Agent->>Model: invoke(messages)
    Model->>Model: 发送完整对话历史:<br/>[HumanMessage, AIMessage(tool_calls), ToolMessage]
    Model->>Model: HTTP POST to API

    Note over Model: 📤 生成最终响应
    Model->>AIMsg: AIMessage(<br/>  content="The weather in Beijing is sunny, 25°C"<br/>)
    Model-->>Agent: AIMessage
    Agent-->>User: 最终答案

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant Model as ChatModel
    participant Stream as stream迭代器
    participant AIChunk as AIMessageChunk
    participant Merge as BaseMessageChunk.__add__

    User->>Model: stream([HumanMessage("Tell me a story")])

    Note over Model,Stream: 🌊 开启流式响应
    Model->>Stream: _stream(messages)
    Model->>Model: 发送streaming=True的API请求

    Note over Stream,AIChunk: 📦 接收第1个chunk
    Stream->>Stream: 接收SSE数据: {"delta": {"content": "Once"}}
    Stream->>AIChunk: AIMessageChunk(<br/>  content="Once",<br/>  chunk_position=0<br/>)
    AIChunk-->>Stream: chunk1
    Stream-->>User: yield chunk1

    Note over User: 显示: "Once"

    Note over Stream,AIChunk: 📦 接收第2个chunk
    Stream->>Stream: 接收SSE数据: {"delta": {"content": " upon"}}
    Stream->>AIChunk: AIMessageChunk(<br/>  content=" upon",<br/>  chunk_position=1<br/>)
    AIChunk-->>Stream: chunk2
    Stream-->>User: yield chunk2

    Note over User: 显示: " upon"

    Note over User: ... (更多chunks)

    Note over User,Merge: 🔗 用户合并chunks
    User->>User: accumulated = None
    User->>User: for chunk in stream():

    User->>Merge: accumulated = accumulated + chunk1
    Merge->>Merge: merge_content("", "Once")
    Merge->>Merge: merge_dicts(additional_kwargs)
    Merge-->>User: AIMessageChunk(content="Once")

    User->>Merge: accumulated = accumulated + chunk2
    Merge->>Merge: merge_content("Once", " upon")
    Merge-->>User: AIMessageChunk(content="Once upon")

    Note over User: ... (继续累积)

    User->>User: 最终: AIMessageChunk(<br/>  content="Once upon a time...",<br/>  response_metadata={完整元数据}<br/>)

messages = [
    SystemMessage("You are a helpful assistant"),
    HumanMessage("What is 2+2?"),
    AIMessage("2+2 equals 4"),
    HumanMessage("What about 3+3?"),
]

response = model.invoke(messages)

message = HumanMessage(
    content=[
        {"type": "text", "text": "What's in this image?"},
        {"type": "image_url", "image_url": {"url": "https://example.com/image.jpg"}},
    ]
)

response = model.invoke([message])

# 1. 用户请求
messages = [HumanMessage("What's the weather in Beijing?")]

# 2. AI请求工具
response = model_with_tools.invoke(messages)
# AIMessage(tool_calls=[{"name": "get_weather", "args": {"location": "Beijing"}, "id": "call_123"}])
messages.append(response)

# 3. 执行工具
result = get_weather("Beijing")
messages.append(ToolMessage(content=result, tool_call_id="call_123"))

# 4. AI生成最终答案
final_response = model.invoke(messages)
# AIMessage("The weather in Beijing is sunny, 25°C")

用户代码
  └─> ChatModel.invoke(input)
        └─> ChatModel._convert_input(input)
              └─> convert_to_messages(input)  # 入口
                    ├─> _convert_to_message(item1)
                    │     ├─> isinstance检查
                    │     ├─> _create_message_from_message_type(role, content)
                    │     │     ├─> 解析role字段
                    │     │     ├─> 解析tool_calls (如果是AI消息)
                    │     │     └─> 构造对应Message类
                    │     └─> 返回BaseMessage实例
                    ├─> _convert_to_message(item2)
                    └─> 返回 list[BaseMessage]

# 步骤1: 入口函数
def convert_to_messages(
    messages: Iterable[MessageLikeRepresentation] | PromptValue,
) -> list[BaseMessage]:
    # 1.1 处理PromptValue特殊情况
    if isinstance(messages, PromptValue):
        return messages.to_messages()

    # 1.2 遍历转换每个元素
    return [_convert_to_message(m) for m in messages]

# 步骤2: 单条消息转换
def _convert_to_message(message: MessageLikeRepresentation) -> BaseMessage:
    # 2.1 已经是BaseMessage,直接返回
    if isinstance(message, BaseMessage):
        return message

    # 2.2 字符串 → HumanMessage
    elif isinstance(message, str):
        return _create_message_from_message_type("human", message)

    # 2.3 元组 (role, content)
    elif isinstance(message, Sequence) and len(message) == 2:
        message_type_str, template = message
        return _create_message_from_message_type(message_type_str, template)

    # 2.4 字典 {role: ..., content: ...}
    elif isinstance(message, dict):
        msg_kwargs = message.copy()
        # 2.4.1 提取role字段(或type字段)
        try:
            msg_type = msg_kwargs.pop("role")
        except KeyError:
            msg_type = msg_kwargs.pop("type")

        # 2.4.2 提取content字段
        msg_content = msg_kwargs.pop("content") or ""

        # 2.4.3 调用创建函数
        return _create_message_from_message_type(
            msg_type, msg_content, **msg_kwargs
        )

# 步骤3: 根据role创建具体消息类型
def _create_message_from_message_type(
    message_type: str,
    content: str,
    name: str | None = None,
    tool_call_id: str | None = None,
    tool_calls: list[dict[str, Any]] | None = None,
    **additional_kwargs: Any,
) -> BaseMessage:
    # 3.1 准备kwargs
    kwargs = {}
    if name is not None:
        kwargs["name"] = name
    if tool_call_id is not None:
        kwargs["tool_call_id"] = tool_call_id

    # 3.2 处理tool_calls
    if tool_calls is not None:
        kwargs["tool_calls"] = []
        for tool_call in tool_calls:
            # 3.2.1 转换OpenAI格式
            if "function" in tool_call:
                args = tool_call["function"]["arguments"]
                if isinstance(args, str):
                    args = json.loads(args, strict=False)
                kwargs["tool_calls"].append({
                    "name": tool_call["function"]["name"],
                    "args": args,
                    "id": tool_call["id"],
                    "type": "tool_call",
                })

    # 3.3 根据类型创建消息
    if message_type in {"human", "user"}:
        return HumanMessage(content=content, **kwargs)
    elif message_type in {"ai", "assistant"}:
        return AIMessage(content=content, **kwargs)
    elif message_type in {"system", "developer"}:
        return SystemMessage(content=content, **kwargs)
    elif message_type == "tool":
        return ToolMessage(content=content, **kwargs)

用户代码
  └─> message.content_blocks  # @property访问
        ├─> 初始化: blocks = []
        │
        ├─> 第一遍扫描: 识别已知类型
        │     ├─> 遍历 self.content
        │     ├─> 字符串 → TextContentBlock
        │     ├─> dict["type"] in KNOWN_BLOCK_TYPES → 直接append
        │     └─> 未知类型 → NonStandardContentBlock
        │
        ├─> 第二遍转换: 尝试解析非标准块
        │     ├─> _convert_v0_multimodal_input_to_v1(blocks)
        │     │     └─> 识别v0格式(source_type字段)
        │     ├─> _convert_to_v1_from_chat_completions_input(blocks)
        │     │     └─> 识别OpenAI格式(image_url字段)
        │     ├─> _convert_to_v1_from_anthropic_input(blocks)
        │     │     └─> 识别Anthropic格式(source字段)
        │     ├─> _convert_to_v1_from_genai_input(blocks)
        │     │     └─> 识别Google格式(mime_type字段)
        │     └─> _convert_to_v1_from_converse_input(blocks)
        │           └─> 识别Bedrock格式
        │
        └─> 返回: list[ContentBlock]

@property
def content_blocks(self) -> list[types.ContentBlock]:
    # 阶段1: 初始化
    blocks: list[types.ContentBlock] = []

    # 1.1 字符串转列表
    content = (
        [self.content] if isinstance(self.content, str) and self.content
        else self.content
    )

    # 阶段2: 第一遍扫描
    for item in content:
        if isinstance(item, str):
            # 2.1 纯字符串 → text块
            blocks.append({"type": "text", "text": item})

        elif isinstance(item, dict):
            item_type = item.get("type")

            # 2.2 检查是否为v1标准类型
            if item_type not in types.KNOWN_BLOCK_TYPES:
                blocks.append({"type": "non_standard", "value": item})

            # 2.3 检查是否为v0格式(含source_type)
            elif "source_type" in item:
                blocks.append({"type": "non_standard", "value": item})

            # 2.4 v1标准块,直接使用
            else:
                blocks.append(cast("types.ContentBlock", item))

    # 阶段3: 多轮转换尝试
    for parsing_step in [
        _convert_v0_multimodal_input_to_v1,
        _convert_to_v1_from_chat_completions_input,
        _convert_to_v1_from_anthropic_input,
        _convert_to_v1_from_genai_input,
        _convert_to_v1_from_converse_input,
    ]:
        blocks = parsing_step(blocks)

    return blocks

# 示例转换器: Anthropic格式
def _convert_to_v1_from_anthropic_input(
    blocks: list[types.ContentBlock],
) -> list[types.ContentBlock]:
    converted = []
    for block in blocks:
        if block.get("type") != "non_standard":
            converted.append(block)
            continue

        value = block["value"]

        # 检查Anthropic图片格式
        if value.get("type") == "image" and "source" in value:
            source = value["source"]
            # 转换为v1标准ImageContentBlock
            converted.append({
                "type": "image",
                "source": "base64" if source["type"] == "base64" else "url",
                "data": source.get("data"),
                "url": source.get("url"),
            })
        else:
            # 无法识别,保持non_standard
            converted.append(block)

    return converted

trim_messages(messages, max_tokens=1000)
  │
  ├─> 1. 参数验证
  │     ├─> 检查start_on/include_system与strategy兼容性
  │     └─> 提取token_counter函数
  │
  ├─> 2. 策略路由
  │     ├─> strategy="first" → _first_max_tokens()
  │     └─> strategy="last"  → _last_max_tokens()
  │
  └─> 3. _last_max_tokens() 执行 (最常用)
        │
        ├─> 3.1 end_on过滤
        │     └─> 从尾部移除不匹配end_on类型的消息
        │
        ├─> 3.2 system消息分离
        │     ├─> 如果include_system=True且第一条是SystemMessage
        │     ├─> 单独保存system_message
        │     └─> 从列表中移除
        │
        ├─> 3.3 计算剩余token预算
        │     └─> remaining_tokens = max_tokens - system_tokens
        │
        ├─> 3.4 反转消息列表
        │     └─> reversed_messages = messages[::-1]
        │
        ├─> 3.5 调用_first_max_tokens(reversed_messages)
        │     │
        │     ├─> 5.1 二分查找最大消息数
        │     │     ├─> left, right = 0, len(messages)
        │     │     └─> 找到idx使得token_counter(messages[:idx]) <= max_tokens
        │     │
        │     ├─> 5.2 allow_partial处理
        │     │     ├─> 如果content是list,逐个移除块直到符合
        │     │     └─> 如果content是str,使用text_splitter分割
        │     │
        │     └─> 5.3 end_on过滤
        │           └─> 从idx向前移除不匹配的消息
        │
        ├─> 3.6 反转回原顺序
        │     └─> result = reversed_result[::-1]
        │
        ├─> 3.7 start_on过滤
        │     └─> 从头部移除不匹配start_on类型的消息
        │
        └─> 3.8 添加回system消息
              └─> return [system_message] + result

def _first_max_tokens(
    messages: Sequence[BaseMessage],
    *,
    max_tokens: int,
    token_counter: Callable[[list[BaseMessage]], int],
    ...
) -> list[BaseMessage]:
    # 优化: 先检查是否全部消息都符合
    if token_counter(messages) <= max_tokens:
        # 只需应用end_on过滤
        if end_on:
            for _ in range(len(messages)):
                if not _is_message_type(messages[-1], end_on):
                    messages.pop()
                else:
                    break
        return messages

    # 二分查找最大idx
    left, right = 0, len(messages)
    max_iterations = len(messages).bit_length()  # log2(n)

    for _ in range(max_iterations):
        if left >= right:
            break
        mid = (left + right + 1) // 2

        # 检查前mid条消息的token数
        if token_counter(messages[:mid]) <= max_tokens:
            left = mid  # mid条可以放入,尝试更多
            idx = mid
        else:
            right = mid - 1  # mid条超出,减少数量

    idx = left

    # allow_partial: 尝试加入部分内容
    if allow_partial and idx < len(messages):
        excluded = messages[idx].model_copy(deep=True)

        # 处理content是列表的情况
        if isinstance(excluded.content, list):
            for i in range(1, len(excluded.content)):
                excluded.content = excluded.content[:-1]  # 移除最后一块
                if token_counter([*messages[:idx], excluded]) <= max_tokens:
                    messages = [*messages[:idx], excluded]
                    idx += 1
                    break

        # 处理content是字符串的情况
        else:
            text = excluded.content
            split_texts = text_splitter(text)  # 按\n分割

            # 二分查找最多可以包含多少分割
            left, right = 0, len(split_texts)
            for _ in range(len(split_texts).bit_length()):
                if left >= right:
                    break
                mid = (left + right + 1) // 2
                excluded.content = "".join(split_texts[:mid])
                if token_counter([*messages[:idx], excluded]) <= max_tokens:
                    left = mid
                else:
                    right = mid - 1

            if left > 0:
                excluded.content = "".join(split_texts[:left])
                messages = [*messages[:idx], excluded]

    return messages[:idx]

# 1. 使用便捷构造方法
from langchain_core.messages import HumanMessage, AIMessage

# 推荐:直接传递content
msg = HumanMessage("Hello")

# 多模态:使用列表
msg = HumanMessage(content=[
    {"type": "text", "text": "Describe this"},
    {"type": "image_url", "image_url": {"url": "https://..."}}
])

# 2. 设置元数据
msg = AIMessage(
    content="Response",
    response_metadata={
        "model": "gpt-4o",
        "finish_reason": "stop",
        "system_fingerprint": "fp_123"
    },
    usage_metadata={
        "input_tokens": 100,
        "output_tokens": 50,
        "total_tokens": 150
    }
)

# 3. 工具调用消息
from langchain_core.messages import ToolCall

ai_msg = AIMessage(
    content="",
    tool_calls=[
        ToolCall(
            name="calculator",
            args={"expression": "2+2"},
            id="call_abc"
        )
    ]
)

tool_msg = ToolMessage(
    content="4",
    tool_call_id="call_abc"
)

# 1. 限制对话长度
from langchain_core.messages import trim_messages

def manage_conversation(history: list[BaseMessage], new_message: BaseMessage, model):
    # 添加新消息
    history.append(new_message)

    # 裁剪以适应上下文窗口
    trimmed = trim_messages(
        history,
        max_tokens=4000,
        token_counter=model,
        strategy="last",
        include_system=True
    )

    return trimmed

# 2. 过滤调试消息
production_messages = filter_messages(
    all_messages,
    exclude_names=["debug", "test"]
)

# 3. 序列化保存
import json

# 保存对话历史
with open("chat_history.json", "w") as f:
    json.dump(messages_to_dict(messages), f)

# 加载对话历史
with open("chat_history.json") as f:
    messages = messages_from_dict(json.load(f))

# 1. 验证工具调用
def validate_tool_call(tool_call: ToolCall, available_tools: dict) -> bool:
    if tool_call["name"] not in available_tools:
        return False

    # 验证参数schema
    tool = available_tools[tool_call["name"]]
    try:
        tool.args_schema.model_validate(tool_call["args"])
        return True
    except ValidationError:
        return False

# 2. 处理无效工具调用
if response.invalid_tool_calls:
    for invalid_call in response.invalid_tool_calls:
        error_msg = ToolMessage(
            content=f"Invalid tool call: {invalid_call['error']}",
            tool_call_id=invalid_call["id"],
            status="error"
        )
        messages.append(error_msg)

# 3. 并行执行工具
import asyncio

async def execute_tools_parallel(tool_calls: list[ToolCall], tools_map):
    tasks = [
        tools_map[tc["name"]].ainvoke(tc["args"])
        for tc in tool_calls
    ]
    results = await asyncio.gather(*tasks)

    return [
        ToolMessage(content=str(result), tool_call_id=tc["id"])
        for tc, result in zip(tool_calls, results)
    ]

# ❌ 不推荐: 频繁访问content_blocks
for _ in range(1000):
    blocks = message.content_blocks  # 每次都重新解析

# ✅ 推荐: 缓存结果
blocks = message.content_blocks
for _ in range(1000):
    process(blocks)  # 复用缓存

# ❌ 不推荐: 使用精确token计数
from langchain_openai import ChatOpenAI
model = ChatOpenAI()

trimmed = trim_messages(
    long_history,
    max_tokens=4000,
    token_counter=model  # 调用API计数,很慢
)

# ✅ 推荐: 使用近似计数
from langchain_core.messages import count_tokens_approximately

trimmed = trim_messages(
    long_history,
    max_tokens=4000,
    token_counter=lambda msgs: count_tokens_approximately(msgs)
)

import json
from langchain_core.messages import messages_to_dict

# ❌ 不推荐: 重复序列化
for message in messages:
    json_str = json.dumps(messages_to_dict([message]))

# ✅ 推荐: 批量序列化
json_str = json.dumps(messages_to_dict(messages))

# 边界情况1: 空字符串content
msg = AIMessage(content="")
assert msg.text == ""
assert msg.content_blocks == []  # 空字符串被过滤

# 边界情况2: 空列表content
msg = AIMessage(content=[])
assert msg.content_blocks == []

# 边界情况3: 仅包含空字符串的列表
msg = AIMessage(content=["", "", ""])
assert msg.content_blocks == []  # 所有空字符串被过滤

from langchain_core.messages import ToolMessage

# ✅ 正确: 关联到存在的tool_call
ai_msg = AIMessage(tool_calls=[{"name": "tool", "args": {}, "id": "call_1"}])
tool_msg = ToolMessage(content="result", tool_call_id="call_1")

# ⚠️ 警告: tool_call_id不匹配
tool_msg = ToolMessage(content="result", tool_call_id="call_999")
# 大多数模型会拒绝此消息或产生错误