Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Nick Hollon (nick-hollon-lc) wants to merge 1 commit into
base: nh/native-content-block-streaming-openai
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions libs/partners/openai/langchain_openai/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -21,6 +23,8 @@
"StreamChunkTimeoutError",
"__version__",
"aconvert_openai_completions_stream",
"aconvert_openai_responses_stream",
"convert_openai_completions_stream",
"convert_openai_responses_stream",
"custom_tool",
]
160 changes: 160 additions & 0 deletions libs/partners/openai/langchain_openai/chat_models/_stream_events.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
103 changes: 87 additions & 16 deletions libs/partners/openai/langchain_openai/chat_models/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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.
Expand All @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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(
Expand Down
Loading
Loading