_merge_messages() — langchain Function Reference

Architecture documentation for the _merge_messages() function in chat_models.py from the langchain codebase.

Function python LangChainCore LanguageModelBase called by 1

Entity Profile

LangChainCore→ LanguageModelBase→ _merge_messages() — langchain Function Reference

Dependency Diagram

graph TD
  005a5002_49e5_d89d_8695_51674ab4fc58["_merge_messages()"]
  a85819c7_917d_4c71_2864_a19e68947340["chat_models.py"]
  005a5002_49e5_d89d_8695_51674ab4fc58 -->|defined in| a85819c7_917d_4c71_2864_a19e68947340
  a1122c23_c68d_f096_a90e_42f0e4d39d61["_format_messages()"]
  a1122c23_c68d_f096_a90e_42f0e4d39d61 -->|calls| 005a5002_49e5_d89d_8695_51674ab4fc58
  style 005a5002_49e5_d89d_8695_51674ab4fc58 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/partners/anthropic/langchain_anthropic/chat_models.py lines 236–281

def _merge_messages(
    messages: Sequence[BaseMessage],
) -> list[SystemMessage | AIMessage | HumanMessage]:
    """Merge runs of human/tool messages into single human messages with content blocks."""  # noqa: E501
    merged: list = []
    for curr in messages:
        if isinstance(curr, ToolMessage):
            if (
                isinstance(curr.content, list)
                and curr.content
                and all(
                    isinstance(block, dict) and block.get("type") == "tool_result"
                    for block in curr.content
                )
            ):
                curr = HumanMessage(curr.content)  # type: ignore[misc]
            else:
                curr = HumanMessage(  # type: ignore[misc]
                    [
                        {
                            "type": "tool_result",
                            "content": curr.content,
                            "tool_use_id": curr.tool_call_id,
                            "is_error": curr.status == "error",
                        },
                    ],
                )
        last = merged[-1] if merged else None
        if any(
            all(isinstance(m, c) for m in (curr, last))
            for c in (SystemMessage, HumanMessage)
        ):
            if isinstance(cast("BaseMessage", last).content, str):
                new_content: list = [
                    {"type": "text", "text": cast("BaseMessage", last).content},
                ]
            else:
                new_content = copy.copy(cast("list", cast("BaseMessage", last).content))
            if isinstance(curr.content, str):
                new_content.append({"type": "text", "text": curr.content})
            else:
                new_content.extend(curr.content)
            merged[-1] = curr.model_copy(update={"content": new_content})
        else:
            merged.append(curr)
    return merged