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

libs/partners/openai/langchain_openai/__init__.py

@@ -5,7 +5,9 @@ from langchain_openai.chat_models import AzureChatOpenAI, ChatOpenAI
 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
@@ -21,6 +23,8 @@ __all__ = [
     "StreamChunkTimeoutError",
     "__version__",
     "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 @@ if TYPE_CHECKING:
 # 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

@@ -155,7 +155,9 @@ from langchain_openai.chat_models._compat import (
 )
 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
 
@@ -1929,22 +1931,24 @@ class BaseChatOpenAI(BaseChatModel):
         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.
@@ -1958,6 +1962,35 @@ class BaseChatOpenAI(BaseChatModel):
                 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
@@ -2001,12 +2034,14 @@ class BaseChatOpenAI(BaseChatModel):
         **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.
@@ -2021,6 +2056,42 @@ class BaseChatOpenAI(BaseChatModel):
             ):
                 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(

libs/partners/openai/tests/unit_tests/chat_models/test_responses_stream.py

@@ -46,7 +46,10 @@ from openai.types.shared.reasoning import Reasoning
 from openai.types.shared.response_format_text import ResponseFormatText
 
 from langchain_openai import ChatOpenAI
-from tests.unit_tests.chat_models.test_base import MockSyncContextManager
+from tests.unit_tests.chat_models.test_base import (
+    MockAsyncContextManager,
+    MockSyncContextManager,
+)
 
 MODEL = "gpt-5.4"
 
@@ -783,6 +786,12 @@ def test_responses_stream_events_v3_emits_reasoning_lifecycle() -> None:
 
     assert_valid_event_stream(events)
 
+    # `message-start` carries the stream's LangChain run id (threaded from core),
+    # not the provider response id and not an empty string.
+    assert events[0]["event"] == "message-start"
+    assert events[0]["id"]
+    assert not events[0]["id"].startswith("resp")
+
     reasoning_starts = [
         e
         for e in events
@@ -820,6 +829,46 @@ def test_responses_stream_events_v3_emits_reasoning_lifecycle() -> None:
     ]
 
 
+async def test_aresponses_stream_events_v3_emits_reasoning_lifecycle() -> None:
+    """Async twin of `test_responses_stream_events_v3_emits_reasoning_lifecycle`.
+
+    Drives the native async Responses converter via `astream_events(version="v3")`
+    and asserts the same four reasoning `content-block-finish` events with their
+    accumulated text.
+    """
+    llm = ChatOpenAI(model="o4-mini", use_responses_api=True, output_version="v1")
+    mock_client = MagicMock()
+
+    async def mock_create(*args: Any, **kwargs: Any) -> MockAsyncContextManager:
+        return MockAsyncContextManager(responses_stream)
+
+    mock_client.responses.create = mock_create
+
+    with patch.object(llm, "root_async_client", mock_client):
+        stream = await llm.astream_events("test", version="v3")
+        events = [e async for e in stream]
+
+    assert_valid_event_stream(events)
+
+    reasoning_finishes = [
+        e
+        for e in events
+        if e["event"] == "content-block-finish" and e["content"]["type"] == "reasoning"
+    ]
+    assert len(reasoning_finishes) == 4, (
+        f"expected 4 reasoning finish events, got {len(reasoning_finishes)}"
+    )
+    reasoning_texts = [
+        cast("dict[str, Any]", f["content"])["reasoning"] for f in reasoning_finishes
+    ]
+    assert reasoning_texts == [
+        "reasoning block one",
+        "another reasoning block",
+        "more reasoning",
+        "still more reasoning",
+    ]
+
+
 def test_responses_stream_with_image_generation_multiple_calls() -> None:
     """Test that streaming with image_generation tool works across multiple calls.
 

libs/partners/openai/tests/unit_tests/chat_models/test_responses_stream_events.py

@@ -0,0 +1,103 @@
+"""Unit tests for the OpenAI Responses API native stream-events converter."""
+
+from typing import Any, cast
+
+from langchain_tests.utils.stream_lifecycle import assert_valid_event_stream
+
+from langchain_openai.chat_models._stream_events import (
+    convert_openai_responses_stream,
+)
+from langchain_openai.chat_models.base import (
+    _convert_responses_chunk_to_generation_chunk,
+)
+
+# The shared fixture used by the existing bridge parity test.
+from tests.unit_tests.chat_models.test_responses_stream import responses_stream
+
+
+def test_convert_openai_responses_reasoning_lifecycle() -> None:
+    events: list[Any] = list(
+        convert_openai_responses_stream(
+            iter(responses_stream),
+            _convert_responses_chunk_to_generation_chunk,
+            output_version="v1",
+        )
+    )
+    assert_valid_event_stream(events)
+
+    # message-start must NOT carry the provider response id (consistency with
+    # the bridge / the rule from Phases 1-3): empty id lets core's seeded run id
+    # stand.
+    assert events[0]["event"] == "message-start"
+    assert events[0]["id"] == ""
+    assert events[0]["metadata"]["provider"] == "openai"
+
+    reasoning_finishes = [
+        e
+        for e in events
+        if e["event"] == "content-block-finish" and e["content"]["type"] == "reasoning"
+    ]
+    assert len(reasoning_finishes) == 4
+    assert [
+        cast("dict[str, Any]", f["content"])["reasoning"] for f in reasoning_finishes
+    ] == [
+        "reasoning block one",
+        "another reasoning block",
+        "more reasoning",
+        "still more reasoning",
+    ]
+    assert events[-1]["event"] == "message-finish"
+
+
+def test_convert_openai_responses_true_boundaries() -> None:
+    """A block finishes before the next block's content arrives (true boundary)."""
+    events: list[Any] = list(
+        convert_openai_responses_stream(
+            iter(responses_stream),
+            _convert_responses_chunk_to_generation_chunk,
+            output_version="v1",
+        )
+    )
+    # The first content-block-finish must precede the start of a later index.
+    first_finish_idx = next(
+        i for i, e in enumerate(events) if e["event"] == "content-block-finish"
+    )
+    later_start_idx = next(
+        (
+            i
+            for i, e in enumerate(events)
+            if e["event"] == "content-block-start"
+            and e["index"] > events[first_finish_idx]["index"]
+        ),
+        None,
+    )
+    # If there is a higher-index block, its start comes after the prior finish.
+    if later_start_idx is not None:
+        assert first_finish_idx < later_start_idx
+
+
+async def test_aconvert_openai_responses_reasoning_lifecycle() -> None:
+    async def _araw() -> Any:
+        for c in responses_stream:
+            yield c
+
+    from langchain_openai.chat_models._stream_events import (
+        aconvert_openai_responses_stream,
+    )
+
+    events: list[Any] = [
+        e
+        async for e in aconvert_openai_responses_stream(
+            _araw(),
+            _convert_responses_chunk_to_generation_chunk,
+            output_version="v1",
+        )
+    ]
+    assert_valid_event_stream(events)
+    reasoning_finishes = [
+        e
+        for e in events
+        if e["event"] == "content-block-finish" and e["content"]["type"] == "reasoning"
+    ]
+    assert len(reasoning_finishes) == 4
+    assert events[-1]["event"] == "message-finish"

libs/partners/openai/tests/unit_tests/test_imports.py

@@ -10,7 +10,9 @@ EXPECTED_ALL = [
     "AzureOpenAIEmbeddings",
     "StreamChunkTimeoutError",
     "aconvert_openai_completions_stream",
+    "aconvert_openai_responses_stream",
     "convert_openai_completions_stream",
+    "convert_openai_responses_stream",
     "custom_tool",
 ]