content_blocks() — langchain Function Reference

Architecture documentation for the content_blocks() function in ai.py from the langchain codebase.

Function python LangChainCore ApiManagement calls 1 called by 1

Entity Profile

LangChainCore→ ApiManagement→ content_blocks() — langchain Function Reference

Dependency Diagram

graph TD
  3153d67d_4a59_c444_07e0_3c7cd1bcbc86["content_blocks()"]
  0998183a_ee20_cc02_d37b_948998ae74b7["AIMessageChunk"]
  3153d67d_4a59_c444_07e0_3c7cd1bcbc86 -->|defined in| 0998183a_ee20_cc02_d37b_948998ae74b7
  c00149ad_6593_d143_66cc_6d95af19fd90["content_blocks()"]
  c00149ad_6593_d143_66cc_6d95af19fd90 -->|calls| 3153d67d_4a59_c444_07e0_3c7cd1bcbc86
  c00149ad_6593_d143_66cc_6d95af19fd90["content_blocks()"]
  3153d67d_4a59_c444_07e0_3c7cd1bcbc86 -->|calls| c00149ad_6593_d143_66cc_6d95af19fd90
  style 3153d67d_4a59_c444_07e0_3c7cd1bcbc86 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/messages/ai.py lines 441–492

    def content_blocks(self) -> list[types.ContentBlock]:
        """Return standard, typed `ContentBlock` dicts from the message."""
        if self.response_metadata.get("output_version") == "v1":
            return cast("list[types.ContentBlock]", self.content)

        model_provider = self.response_metadata.get("model_provider")
        if model_provider:
            from langchain_core.messages.block_translators import (  # noqa: PLC0415
                get_translator,
            )

            translator = get_translator(model_provider)
            if translator:
                try:
                    return translator["translate_content_chunk"](self)
                except NotImplementedError:
                    pass

        # Otherwise, use best-effort parsing
        blocks = super().content_blocks

        if (
            self.tool_call_chunks
            and not self.content
            and self.chunk_position != "last"  # keep tool_calls if aggregated
        ):
            blocks = [
                block
                for block in blocks
                if block["type"] not in {"tool_call", "invalid_tool_call"}
            ]
            for tool_call_chunk in self.tool_call_chunks:
                tc: types.ToolCallChunk = {
                    "type": "tool_call_chunk",
                    "id": tool_call_chunk.get("id"),
                    "name": tool_call_chunk.get("name"),
                    "args": tool_call_chunk.get("args"),
                }
                if (idx := tool_call_chunk.get("index")) is not None:
                    tc["index"] = idx
                blocks.append(tc)

        # Best-effort reasoning extraction from additional_kwargs
        # Only add reasoning if not already present
        # Insert before all other blocks to keep reasoning at the start
        has_reasoning = any(block.get("type") == "reasoning" for block in blocks)
        if not has_reasoning and (
            reasoning_block := _extract_reasoning_from_additional_kwargs(self)
        ):
            blocks.insert(0, reasoning_block)

        return blocks