_convert_responses_chunk_to_generation_chunk() — langchain Function Reference

Source Code

libs/partners/openai/langchain_openai/chat_models/base.py lines 4462–4709

def _convert_responses_chunk_to_generation_chunk(
    chunk: Any,
    current_index: int,  # index in content
    current_output_index: int,  # index in Response output
    current_sub_index: int,  # index of content block in output item
    schema: type[_BM] | None = None,
    metadata: dict | None = None,
    has_reasoning: bool = False,
    output_version: str | None = None,
) -> tuple[int, int, int, ChatGenerationChunk | None]:
    def _advance(output_idx: int, sub_idx: int | None = None) -> None:
        """Advance indexes tracked during streaming.

        Example: we stream a response item of the form:

        ```python
        {
            "type": "message",  # output_index 0
            "role": "assistant",
            "id": "msg_123",
            "content": [
                {"type": "output_text", "text": "foo"},  # sub_index 0
                {"type": "output_text", "text": "bar"},  # sub_index 1
            ],
        }
        ```

        This is a single item with a shared `output_index` and two sub-indexes, one
        for each content block.

        This will be processed into an `AIMessage` with two text blocks:

        ```python
        AIMessage(
            [
                {"type": "text", "text": "foo", "id": "msg_123"},  # index 0
                {"type": "text", "text": "bar", "id": "msg_123"},  # index 1
            ]
        )
        ```

        This function just identifies updates in output or sub-indexes and increments
        the current index accordingly.
        """
        nonlocal current_index, current_output_index, current_sub_index
        if sub_idx is None:
            if current_output_index != output_idx:
                current_index += 1
        else:
            if (current_output_index != output_idx) or (current_sub_index != sub_idx):
                current_index += 1
            current_sub_index = sub_idx
        current_output_index = output_idx

    if output_version is None:
        # Sentinel value of None lets us know if output_version is set explicitly.
        # Explicitly setting `output_version="responses/v1"` separately enables the
        # Responses API.
        output_version = "responses/v1"

    content = []
    tool_call_chunks: list = []
    additional_kwargs: dict = {}
    response_metadata = metadata or {}
    response_metadata["model_provider"] = "openai"
    usage_metadata = None
    chunk_position: Literal["last"] | None = None
    id = None
    if chunk.type == "response.output_text.delta":
        _advance(chunk.output_index, chunk.content_index)
        content.append({"type": "text", "text": chunk.delta, "index": current_index})
    elif chunk.type == "response.output_text.annotation.added":
        _advance(chunk.output_index, chunk.content_index)
        if isinstance(chunk.annotation, dict):
            # Appears to be a breaking change in openai==1.82.0
            annotation = chunk.annotation
        else:
            annotation = chunk.annotation.model_dump(exclude_none=True, mode="json")

        content.append(
            {