feat(openai): native content-block streaming for the Responses API

Author: nick-hollon-lc

libs/partners/openai/langchain_openai/__init__.py

@@ -4,7 +4,9 @@
 from langchain_openai.chat_models._client_utils import StreamChunkTimeoutError
 from langchain_openai.chat_models._stream_events import (
     aconvert_openai_completions_stream,
+    aconvert_openai_responses_stream,
     convert_openai_completions_stream,
+    convert_openai_responses_stream,
 )
 from langchain_openai.embeddings import AzureOpenAIEmbeddings, OpenAIEmbeddings
 from langchain_openai.llms import AzureOpenAI, OpenAI
@@ -19,6 +21,8 @@
     "OpenAIEmbeddings",
     "StreamChunkTimeoutError",
     "aconvert_openai_completions_stream",
+    "aconvert_openai_responses_stream",
     "convert_openai_completions_stream",
+    "convert_openai_responses_stream",
     "custom_tool",
 ]

libs/partners/openai/langchain_openai/chat_models/_stream_events.py

@@ -34,6 +34,13 @@
 # Bound `BaseChatOpenAI._convert_chunk_to_generation_chunk`.
 MakeChunk = Callable[..., "ChatGenerationChunk | None"]
 
+# Bound `_convert_responses_chunk_to_generation_chunk`:
+# (chunk, idx, out_idx, sub_idx, *, schema, metadata, has_reasoning, output_version)
+#   -> (idx, out_idx, sub_idx, ChatGenerationChunk | None)
+ConvertResponsesChunk = Callable[
+    ..., "tuple[int, int, int, ChatGenerationChunk | None]"
+]
+
 
 def _message_start(
     message_id: str | None, model: str | None, provider: str
@@ -160,3 +167,156 @@ async def aconvert_openai_completions_stream(
     for ev in tracker.finish_all():
         yield ev
     yield build_message_finish(usage=usage, response_metadata=response_metadata)
+
+
+def convert_openai_responses_stream(
+    raw: Iterator[Any],
+    convert_chunk: ConvertResponsesChunk,
+    *,
+    schema: Any = None,
+    output_version: str | None = None,
+    message_id: str | None = None,
+    provider: str = "openai",
+) -> Iterator[MessagesData]:
+    """Convert a raw OpenAI Responses API event stream to protocol events.
+
+    Reuses `_convert_responses_chunk_to_generation_chunk` (injected as
+    `convert_chunk` to avoid a circular import) for per-event content, threading
+    its index state. Emits true `content-block-finish` boundaries by closing the
+    open block when the monotonic `current_index` advances.
+
+    Args:
+        raw: Raw Responses API events.
+        convert_chunk: `_convert_responses_chunk_to_generation_chunk`.
+        schema: `response_format` schema, forwarded to `convert_chunk`.
+        output_version: `self.output_version`, forwarded to `convert_chunk`.
+        message_id: Left empty by default so the v3 stream's seeded run id stands.
+        provider: `model_provider` id for downstream reuse.
+
+    Yields:
+        Protocol `MessagesData` lifecycle events.
+    """
+    tracker = BlockStreamTracker()
+    started = False
+    current_index = current_output_index = current_sub_index = -1
+    has_reasoning = False
+    usage: dict[str, Any] | None = None
+    response_metadata: dict[str, Any] = {"model_provider": provider}
+    model: str | None = None
+    open_index: Any = None
+
+    for chunk in raw:
+        (
+            current_index,
+            current_output_index,
+            current_sub_index,
+            gen,
+        ) = convert_chunk(
+            chunk,
+            current_index,
+            current_output_index,
+            current_sub_index,
+            schema=schema,
+            metadata={},
+            has_reasoning=has_reasoning,
+            output_version=output_version,
+        )
+        if gen is None:
+            continue
+        msg = gen.message
+        if model is None:
+            model = (msg.response_metadata or {}).get("model_name") or (
+                msg.response_metadata or {}
+            ).get("model")
+        if not started:
+            started = True
+            yield _message_start(message_id, model, provider)
+        if "reasoning" in msg.additional_kwargs:
+            has_reasoning = True
+        for key, block in iter_protocol_blocks(msg):
+            if open_index is not None and key != open_index:
+                # Monotonic index advanced: the previous block is complete.
+                yield from tracker.finish_block(open_index)
+            yield from tracker.feed(key, block)
+            open_index = key
+        usage_metadata = getattr(msg, "usage_metadata", None)
+        if usage_metadata:
+            usage = accumulate_usage(usage, usage_metadata)
+        merged = {**(gen.generation_info or {}), **(msg.response_metadata or {})}
+        if merged:
+            response_metadata.update(merged)
+            response_metadata["model_provider"] = provider
+
+    if not started:
+        return
+    yield from tracker.finish_all()
+    yield build_message_finish(usage=usage, response_metadata=response_metadata)
+
+
+async def aconvert_openai_responses_stream(
+    raw: AsyncIterator[Any],
+    convert_chunk: ConvertResponsesChunk,
+    *,
+    schema: Any = None,
+    output_version: str | None = None,
+    message_id: str | None = None,
+    provider: str = "openai",
+) -> AsyncIterator[MessagesData]:
+    """Async twin of `convert_openai_responses_stream`. `convert_chunk` is sync."""
+    tracker = BlockStreamTracker()
+    started = False
+    current_index = current_output_index = current_sub_index = -1
+    has_reasoning = False
+    usage: dict[str, Any] | None = None
+    response_metadata: dict[str, Any] = {"model_provider": provider}
+    model: str | None = None
+    open_index: Any = None
+
+    async for chunk in raw:
+        (
+            current_index,
+            current_output_index,
+            current_sub_index,
+            gen,
+        ) = convert_chunk(
+            chunk,
+            current_index,
+            current_output_index,
+            current_sub_index,
+            schema=schema,
+            metadata={},
+            has_reasoning=has_reasoning,
+            output_version=output_version,
+        )
+        if gen is None:
+            continue
+        msg = gen.message
+        if model is None:
+            model = (msg.response_metadata or {}).get("model_name") or (
+                msg.response_metadata or {}
+            ).get("model")
+        if not started:
+            started = True
+            yield _message_start(message_id, model, provider)
+        if "reasoning" in msg.additional_kwargs:
+            has_reasoning = True
+        for key, block in iter_protocol_blocks(msg):
+            if open_index is not None and key != open_index:
+                for ev in tracker.finish_block(open_index):
+                    yield ev
+            for ev in tracker.feed(key, block):
+                yield ev
+            open_index = key
+        usage_metadata = getattr(msg, "usage_metadata", None)
+        if usage_metadata:
+            usage = accumulate_usage(usage, usage_metadata)
+        merged = {**(gen.generation_info or {}), **(msg.response_metadata or {})}
+        if merged:
+            response_metadata.update(merged)
+            response_metadata["model_provider"] = provider
+
+    if not started:
+        return
+    for ev in tracker.finish_all():
+        yield ev
+    yield build_message_finish(usage=usage, response_metadata=response_metadata)

libs/partners/openai/langchain_openai/chat_models/base.py

@@ -154,7 +154,9 @@
 )
 from langchain_openai.chat_models._stream_events import (
     aconvert_openai_completions_stream,
+    aconvert_openai_responses_stream,
     convert_openai_completions_stream,
+    convert_openai_responses_stream,
 )
 from langchain_openai.data._profiles import _PROFILES
 
@@ -1892,22 +1894,24 @@ def _stream_chat_model_events(
         message_id: str | None = None,
         **kwargs: Any,
     ) -> Iterator[MessagesData]:
-        """Emit OpenAI-native content-block events for the Chat Completions path.
+        """Emit OpenAI-native content-block events for Completions and Responses.
 
-        Defers to the compat bridge for cases this converter does not yet
-        specialize: the Responses API, structured output (`response_format`),
-        and raw-header mode. Detected by core's `_iter_v2_events`.
+        The standard Completions and Responses API paths run through their
+        native converters. Structured output (`response_format`) and raw-header
+        mode still defer to the compat bridge over `_stream`, since those keep
+        the final-completion handling only `_stream` performs. Detected by
+        core's `_iter_v2_events`.
         """
-        # Responses API / structured output / raw headers: bridge over `_stream`,
-        # which (on `ChatOpenAI`) routes to the Responses path when applicable.
+        use_responses = self._use_responses_api({**kwargs, **self.model_kwargs})
         # `response_format` may arrive via call kwargs or be baked into
         # `model_kwargs`; both fold into the payload, so check both.
-        if (
-            self._use_responses_api({**kwargs, **self.model_kwargs})
-            or kwargs.get("response_format") is not None
+        has_response_format = (
+            kwargs.get("response_format") is not None
             or self.model_kwargs.get("response_format") is not None
-            or self.include_response_headers
-        ):
+        )
+        # Structured output and raw-header mode keep the post-loop /
+        # final-completion handling that only `_stream` performs — defer those.
+        if has_response_format or self.include_response_headers:
             # Forward kwargs untouched (as core's `_iter_v2_events` would):
             # `_stream` handles `stream_usage` itself, and the Responses path
             # rejects a stray `stream_usage` kwarg, so we must not inject one.
@@ -1921,6 +1925,35 @@ def _stream_chat_model_events(
                 message_id=message_id,
             )
             return
+        if use_responses:
+            self._ensure_sync_client_available()
+            kwargs["stream"] = True
+            payload = self._get_request_payload(messages, stop=stop, **kwargs)
+            try:
+                with self.root_client.responses.create(**payload) as response:
+                    for event in convert_openai_responses_stream(
+                        response,
+                        _convert_responses_chunk_to_generation_chunk,
+                        # Always None here: the `response_format` (structured
+                        # output) path is handled by the bridge branch above.
+                        schema=None,
+                        output_version=self.output_version,
+                        message_id=message_id,
+                    ):
+                        if (
+                            run_manager is not None
+                            and event["event"] == "content-block-delta"
+                            and event["delta"].get("type") == "text-delta"
+                        ):
+                            run_manager.on_llm_new_token(
+                                str(event["delta"].get("text", ""))
+                            )
+                        yield event
+            except openai.BadRequestError as e:
+                _handle_openai_bad_request(e)
+            except openai.APIError as e:
+                _handle_openai_api_error(e)
+            return
 
         self._ensure_sync_client_available()
         kwargs["stream"] = True
@@ -1964,12 +1997,14 @@ async def _astream_chat_model_events(
         **kwargs: Any,
     ) -> AsyncIterator[MessagesData]:
         """Async twin of `_stream_chat_model_events`."""
-        if (
-            self._use_responses_api({**kwargs, **self.model_kwargs})
-            or kwargs.get("response_format") is not None
+        use_responses = self._use_responses_api({**kwargs, **self.model_kwargs})
+        has_response_format = (
+            kwargs.get("response_format") is not None
             or self.model_kwargs.get("response_format") is not None
-            or self.include_response_headers
-        ):
+        )
+        # Structured output and raw-header mode keep the post-loop /
+        # final-completion handling that only `_astream` performs — defer those.
+        if has_response_format or self.include_response_headers:
             # Forward kwargs untouched (as core's `_aiter_v2_events` would):
             # `_astream` handles `stream_usage` itself, and the Responses path
             # rejects a stray `stream_usage` kwarg, so we must not inject one.
@@ -1984,6 +2019,42 @@ async def _astream_chat_model_events(
             ):
                 yield event
             return
+        if use_responses:
+            kwargs["stream"] = True
+            payload = self._get_request_payload(messages, stop=stop, **kwargs)
+            try:
+                response = await self.root_async_client.responses.create(**payload)
+                async with response as stream:
+                    # Mirror `_astream_responses`: apply per-chunk stall
+                    # protection before the converter consumes the stream.
+                    timed_stream = _astream_with_chunk_timeout(
+                        stream,
+                        self.stream_chunk_timeout,
+                        model_name=self.model_name,
+                    )
+                    async for event in aconvert_openai_responses_stream(
+                        timed_stream,
+                        _convert_responses_chunk_to_generation_chunk,
+                        # Always None here: the `response_format` (structured
+                        # output) path is handled by the bridge branch above.
+                        schema=None,
+                        output_version=self.output_version,
+                        message_id=message_id,
+                    ):
+                        if (
+                            run_manager is not None
+                            and event["event"] == "content-block-delta"
+                            and event["delta"].get("type") == "text-delta"
+                        ):
+                            await run_manager.on_llm_new_token(
+                                str(event["delta"].get("text", ""))
+                            )
+                        yield event
+            except openai.BadRequestError as e:
+                _handle_openai_bad_request(e)
+            except openai.APIError as e:
+                _handle_openai_api_error(e)
+            return
 
         kwargs["stream"] = True
         stream_usage = self._should_stream_usage(