From 535219daf1a8519b52ee0c17d1a226c6a8bb0412 Mon Sep 17 00:00:00 2001 From: Nick Hollon Date: Tue, 9 Jun 2026 21:15:47 -0400 Subject: [PATCH] feat(openai): native content-block streaming for the Responses API --- .../openai/langchain_openai/__init__.py | 4 + .../chat_models/_stream_events.py | 160 ++++++++++++++++++ .../langchain_openai/chat_models/base.py | 103 +++++++++-- .../chat_models/test_responses_stream.py | 51 +++++- .../test_responses_stream_events.py | 103 +++++++++++ .../openai/tests/unit_tests/test_imports.py | 2 + 6 files changed, 406 insertions(+), 17 deletions(-) create mode 100644 libs/partners/openai/tests/unit_tests/chat_models/test_responses_stream_events.py diff --git a/libs/partners/openai/langchain_openai/__init__.py b/libs/partners/openai/langchain_openai/__init__.py index cf2234b09b8cf..62e7ee64419b4 100644 --- a/libs/partners/openai/langchain_openai/__init__.py +++ b/libs/partners/openai/langchain_openai/__init__.py @@ -5,7 +5,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 @@ -21,6 +23,8 @@ "StreamChunkTimeoutError", "__version__", "aconvert_openai_completions_stream", + "aconvert_openai_responses_stream", "convert_openai_completions_stream", + "convert_openai_responses_stream", "custom_tool", ] diff --git a/libs/partners/openai/langchain_openai/chat_models/_stream_events.py b/libs/partners/openai/langchain_openai/chat_models/_stream_events.py index 6af51bc265772..4b2e721cceb17 100644 --- a/libs/partners/openai/langchain_openai/chat_models/_stream_events.py +++ b/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) diff --git a/libs/partners/openai/langchain_openai/chat_models/base.py b/libs/partners/openai/langchain_openai/chat_models/base.py index 6bf42dbac813a..6ab5cbbec3a50 100644 --- a/libs/partners/openai/langchain_openai/chat_models/base.py +++ b/libs/partners/openai/langchain_openai/chat_models/base.py @@ -155,7 +155,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 @@ -1929,22 +1931,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. @@ -1958,6 +1962,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 @@ -2001,12 +2034,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. @@ -2021,6 +2056,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( diff --git a/libs/partners/openai/tests/unit_tests/chat_models/test_responses_stream.py b/libs/partners/openai/tests/unit_tests/chat_models/test_responses_stream.py index baaae29bd9c41..e2b5367bc2712 100644 --- a/libs/partners/openai/tests/unit_tests/chat_models/test_responses_stream.py +++ b/libs/partners/openai/tests/unit_tests/chat_models/test_responses_stream.py @@ -46,7 +46,10 @@ 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 mock_create(*args: Any, **kwargs: Any) -> MockSyncContextManager: 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 mock_create(*args: Any, **kwargs: Any) -> MockSyncContextManager: ] +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. diff --git a/libs/partners/openai/tests/unit_tests/chat_models/test_responses_stream_events.py b/libs/partners/openai/tests/unit_tests/chat_models/test_responses_stream_events.py new file mode 100644 index 0000000000000..1dfaede5afd34 --- /dev/null +++ b/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" diff --git a/libs/partners/openai/tests/unit_tests/test_imports.py b/libs/partners/openai/tests/unit_tests/test_imports.py index 10653d45f50d6..6ac1d2d49017a 100644 --- a/libs/partners/openai/tests/unit_tests/test_imports.py +++ b/libs/partners/openai/tests/unit_tests/test_imports.py @@ -10,7 +10,9 @@ "AzureOpenAIEmbeddings", "StreamChunkTimeoutError", "aconvert_openai_completions_stream", + "aconvert_openai_responses_stream", "convert_openai_completions_stream", + "convert_openai_responses_stream", "custom_tool", ]