_convert_to_v1_from_converse() — langchain Function Reference

Source Code

libs/core/langchain_core/messages/block_translators/bedrock_converse.py lines 152–280

def _convert_to_v1_from_converse(message: AIMessage) -> list[types.ContentBlock]:
    """Convert Bedrock Converse message content to v1 format."""
    if (
        message.content == ""
        and not message.additional_kwargs
        and not message.tool_calls
    ):
        # Converse outputs multiple chunks containing response metadata
        return []

    if isinstance(message.content, str):
        message.content = [{"type": "text", "text": message.content}]

    def _iter_blocks() -> Iterator[types.ContentBlock]:
        for block in message.content:
            if not isinstance(block, dict):
                continue
            block_type = block.get("type")

            if block_type == "text":
                if citations := block.get("citations"):
                    text_block: types.TextContentBlock = {
                        "type": "text",
                        "text": block.get("text", ""),
                        "annotations": [_convert_citation_to_v1(a) for a in citations],
                    }
                else:
                    text_block = {"type": "text", "text": block["text"]}
                if "index" in block:
                    text_block["index"] = block["index"]
                yield text_block

            elif block_type == "reasoning_content":
                reasoning_block: types.ReasoningContentBlock = {"type": "reasoning"}
                if reasoning_content := block.get("reasoning_content"):
                    if reasoning := reasoning_content.get("text"):
                        reasoning_block["reasoning"] = reasoning
                    if signature := reasoning_content.get("signature"):
                        if "extras" not in reasoning_block:
                            reasoning_block["extras"] = {}
                        reasoning_block["extras"]["signature"] = signature

                if "index" in block:
                    reasoning_block["index"] = block["index"]

                known_fields = {"type", "reasoning_content", "index", "extras"}
                for key in block:
                    if key not in known_fields:
                        if "extras" not in reasoning_block:
                            reasoning_block["extras"] = {}
                        reasoning_block["extras"][key] = block[key]
                yield reasoning_block

            elif block_type == "tool_use":
                if (
                    isinstance(message, AIMessageChunk)
                    and len(message.tool_call_chunks) == 1
                    and message.chunk_position != "last"
                ):
                    # Isolated chunk
                    chunk = message.tool_call_chunks[0]
                    tool_call_chunk = types.ToolCallChunk(
                        name=chunk.get("name"),
                        id=chunk.get("id"),
                        args=chunk.get("args"),
                        type="tool_call_chunk",
                    )
                    index = chunk.get("index")
                    if index is not None:
                        tool_call_chunk["index"] = index
                    yield tool_call_chunk
                else:
                    tool_call_block: types.ToolCall | None = None
                    # Non-streaming or gathered chunk
                    if len(message.tool_calls) == 1:
                        tool_call_block = {
                            "type": "tool_call",
                            "name": message.tool_calls[0]["name"],
                            "args": message.tool_calls[0]["args"],
                            "id": message.tool_calls[0].get("id"),
                        }