From cc78501296b24840e06e3e25835516729778b4d7 Mon Sep 17 00:00:00 2001 From: Nick Hollon Date: Tue, 9 Jun 2026 17:13:50 -0400 Subject: [PATCH 1/3] feat(core): extract reusable BlockStreamTracker from compat bridge --- .../language_models/_compat_bridge.py | 134 ++++++++++-------- .../language_models/chat_models.py | 18 ++- .../language_models/stream_events.py | 28 ++++ .../test_block_stream_tracker.py | 65 +++++++++ 4 files changed, 187 insertions(+), 58 deletions(-) create mode 100644 libs/core/langchain_core/language_models/stream_events.py create mode 100644 libs/core/tests/unit_tests/language_models/test_block_stream_tracker.py diff --git a/libs/core/langchain_core/language_models/_compat_bridge.py b/libs/core/langchain_core/language_models/_compat_bridge.py index 43490f0717c0e..67626c86cbb17 100644 --- a/libs/core/langchain_core/language_models/_compat_bridge.py +++ b/libs/core/langchain_core/language_models/_compat_bridge.py @@ -563,6 +563,75 @@ def _finalize_and_build_finish( ) +# --------------------------------------------------------------------------- +# Shared block-lifecycle tracker +# --------------------------------------------------------------------------- + + +class BlockStreamTracker: + """Allocate wire indices and emit content-block lifecycle events. + + Single source of truth for the start/delta/accumulate/finalize + mechanics shared by the compat bridge and provider-native converters. + Source-side keys (a block's `index` field, int or str, or a + positional fallback) are mapped to sequential `uint` wire indices. + """ + + def __init__(self) -> None: + self._blocks: dict[Any, tuple[int, CompatBlock]] = {} + self._next_wire_idx = 0 + + def feed(self, key: Any, block: CompatBlock) -> Iterator[MessagesData]: + """Yield lifecycle events for a per-chunk block. + + Yields `content-block-start` the first time `key` is seen, then + `content-block-delta` when the block carries fresh content, + accumulating per-index state for later finalization. + """ + if key not in self._blocks: + wire_idx = self._next_wire_idx + self._next_wire_idx += 1 + self._blocks[key] = (wire_idx, dict(block)) + yield ContentBlockStartData( + event="content-block-start", + index=wire_idx, + content=_start_skeleton(block), + ) + else: + wire_idx, existing = self._blocks[key] + self._blocks[key] = (wire_idx, _accumulate(existing, block)) + if _should_emit_delta(block): + wire_idx, current = self._blocks[key] + is_block_delta = block.get("type") in ( + "tool_call_chunk", + "server_tool_call_chunk", + ) + delta_source = current if is_block_delta else block + yield ContentBlockDeltaData( + event="content-block-delta", + index=wire_idx, + delta=_to_content_delta(delta_source or block), + ) + + def finish_block(self, key: Any) -> Iterator[MessagesData]: + """Finalize and close one block at its true stop boundary. + + Native converters call this on the provider's block-stop signal. + A no-op for unknown / already-closed keys. + """ + entry = self._blocks.pop(key, None) + if entry is None: + return + wire_idx, block = entry + yield _finalize_and_build_finish(wire_idx, block) + + def finish_all(self) -> Iterator[MessagesData]: + """Finalize every still-open block, in wire-index order.""" + for wire_idx, block in self._blocks.values(): + yield _finalize_and_build_finish(wire_idx, block) + self._blocks.clear() + + # --------------------------------------------------------------------------- # Main generators # --------------------------------------------------------------------------- @@ -590,8 +659,7 @@ def chunks_to_events( `MessagesData` lifecycle events. """ started = False - blocks: dict[Any, tuple[int, CompatBlock]] = {} - next_wire_idx = 0 + tracker = BlockStreamTracker() usage: dict[str, Any] | None = None response_metadata: dict[str, Any] = {} additional_kwargs: dict[str, Any] = {} @@ -633,30 +701,7 @@ def chunks_to_events( yield _build_message_start(msg, message_id) for key, block in _iter_protocol_blocks(msg): - if key not in blocks: - wire_idx = next_wire_idx - next_wire_idx += 1 - blocks[key] = (wire_idx, dict(block)) - yield ContentBlockStartData( - event="content-block-start", - index=wire_idx, - content=_start_skeleton(block), - ) - else: - wire_idx, existing = blocks[key] - blocks[key] = (wire_idx, _accumulate(existing, block)) - if _should_emit_delta(block): - wire_idx, current = blocks[key] - is_block_delta = block.get("type") in ( - "tool_call_chunk", - "server_tool_call_chunk", - ) - delta_source = current if is_block_delta else block - yield ContentBlockDeltaData( - event="content-block-delta", - index=wire_idx, - delta=_to_content_delta(delta_source or block), - ) + yield from tracker.feed(key, block) if msg.usage_metadata: usage = _accumulate_usage(usage, msg.usage_metadata) @@ -664,8 +709,7 @@ def chunks_to_events( if not started: return - for wire_idx, block in blocks.values(): - yield _finalize_and_build_finish(wire_idx, block) + yield from tracker.finish_all() yield _build_message_finish( usage=usage, @@ -681,8 +725,7 @@ async def achunks_to_events( ) -> AsyncIterator[MessagesData]: """Async variant of `chunks_to_events`.""" started = False - blocks: dict[Any, tuple[int, CompatBlock]] = {} - next_wire_idx = 0 + tracker = BlockStreamTracker() usage: dict[str, Any] | None = None response_metadata: dict[str, Any] = {} additional_kwargs: dict[str, Any] = {} @@ -713,30 +756,8 @@ async def achunks_to_events( yield _build_message_start(msg, message_id) for key, block in _iter_protocol_blocks(msg): - if key not in blocks: - wire_idx = next_wire_idx - next_wire_idx += 1 - blocks[key] = (wire_idx, dict(block)) - yield ContentBlockStartData( - event="content-block-start", - index=wire_idx, - content=_start_skeleton(block), - ) - else: - wire_idx, existing = blocks[key] - blocks[key] = (wire_idx, _accumulate(existing, block)) - if _should_emit_delta(block): - wire_idx, current = blocks[key] - is_block_delta = block.get("type") in ( - "tool_call_chunk", - "server_tool_call_chunk", - ) - delta_source = current if is_block_delta else block - yield ContentBlockDeltaData( - event="content-block-delta", - index=wire_idx, - delta=_to_content_delta(delta_source or block), - ) + for event in tracker.feed(key, block): + yield event if msg.usage_metadata: usage = _accumulate_usage(usage, msg.usage_metadata) @@ -744,8 +765,8 @@ async def achunks_to_events( if not started: return - for wire_idx, block in blocks.values(): - yield _finalize_and_build_finish(wire_idx, block) + for event in tracker.finish_all(): + yield event yield _build_message_finish( usage=usage, @@ -816,6 +837,7 @@ async def amessage_to_events( __all__ = [ + "BlockStreamTracker", "CompatBlock", "achunks_to_events", "amessage_to_events", diff --git a/libs/core/langchain_core/language_models/chat_models.py b/libs/core/langchain_core/language_models/chat_models.py index 7c21d12110f62..1884417c89aaa 100644 --- a/libs/core/langchain_core/language_models/chat_models.py +++ b/libs/core/langchain_core/language_models/chat_models.py @@ -667,8 +667,16 @@ def _iter_v2_events( getattr(self, "_stream_chat_model_events", None), ) if native is not None: + # Thread the stream's message id (the LangChain run id) into the + # native producer so its `message-start` carries the same id the + # bridge path emits — keeping the two paths consistent for + # consumers that correlate v3 events by `message-start.id`. event_iter: Iterator[MessagesData] = native( - messages, stop=stop, run_manager=run_manager, **kwargs + messages, + stop=stop, + run_manager=run_manager, + message_id=stream.message_id, + **kwargs, ) else: event_iter = chunks_to_events( @@ -698,8 +706,14 @@ async def _aiter_v2_events( getattr(self, "_astream_chat_model_events", None), ) if native is not None: + # See `_iter_v2_events`: thread the stream's message id into the + # native producer for `message-start` consistency with the bridge. event_iter: AsyncIterator[MessagesData] = native( - messages, stop=stop, run_manager=run_manager, **kwargs + messages, + stop=stop, + run_manager=run_manager, + message_id=stream.message_id, + **kwargs, ) else: event_iter = achunks_to_events( diff --git a/libs/core/langchain_core/language_models/stream_events.py b/libs/core/langchain_core/language_models/stream_events.py new file mode 100644 index 0000000000000..b40351ee8f191 --- /dev/null +++ b/libs/core/langchain_core/language_models/stream_events.py @@ -0,0 +1,28 @@ +"""Shared primitives for provider-native streaming-event converters. + +Provider packages implementing `_stream_chat_model_events` import these +to drive raw API events into the same content-block lifecycle the compat +bridge produces, so native and bridged paths emit identical event shapes. +""" + +from langchain_core.language_models._compat_bridge import ( + BlockStreamTracker, + finalize_tool_call_chunk, +) +from langchain_core.language_models._compat_bridge import ( + _accumulate_usage as accumulate_usage, +) +from langchain_core.language_models._compat_bridge import ( + _build_message_finish as build_message_finish, +) +from langchain_core.language_models._compat_bridge import ( + _iter_protocol_blocks as iter_protocol_blocks, +) + +__all__ = [ + "BlockStreamTracker", + "accumulate_usage", + "build_message_finish", + "finalize_tool_call_chunk", + "iter_protocol_blocks", +] diff --git a/libs/core/tests/unit_tests/language_models/test_block_stream_tracker.py b/libs/core/tests/unit_tests/language_models/test_block_stream_tracker.py new file mode 100644 index 0000000000000..4dc47907338c5 --- /dev/null +++ b/libs/core/tests/unit_tests/language_models/test_block_stream_tracker.py @@ -0,0 +1,65 @@ +"""Unit tests for the shared BlockStreamTracker.""" + +from typing import Any + +from langchain_core.language_models.stream_events import BlockStreamTracker + + +def test_feed_text_emits_start_then_delta() -> None: + tracker = BlockStreamTracker() + events: list[Any] = list( + tracker.feed(0, {"type": "text", "text": "Hi", "index": 0}) + ) + assert events[0]["event"] == "content-block-start" + assert events[0]["index"] == 0 + assert events[0]["content"] == {"type": "text", "text": ""} + assert events[1]["event"] == "content-block-delta" + assert events[1]["index"] == 0 + assert events[1]["delta"] == {"type": "text-delta", "text": "Hi"} + + +def test_feed_allocates_sequential_wire_indices() -> None: + tracker = BlockStreamTracker() + list(tracker.feed("a", {"type": "text", "text": "x", "index": "a"})) + second: list[Any] = list( + tracker.feed("b", {"type": "text", "text": "y", "index": "b"}) + ) + assert second[0]["index"] == 1 # second distinct key -> wire index 1 + + +def test_finish_block_finalizes_tool_call_at_boundary() -> None: + tracker = BlockStreamTracker() + list( + tracker.feed( + 0, + { + "type": "tool_call_chunk", + "id": "t1", + "name": "f", + "args": '{"a":', + "index": 0, + }, + ) + ) + list(tracker.feed(0, {"type": "tool_call_chunk", "args": "1}", "index": 0})) + finished: list[Any] = list(tracker.finish_block(0)) + assert len(finished) == 1 + assert finished[0]["event"] == "content-block-finish" + assert finished[0]["content"]["type"] == "tool_call" + assert finished[0]["content"]["args"] == {"a": 1} + # finished block is closed: finish_all must not re-emit it + assert list(tracker.finish_all()) == [] + + +def test_finish_block_unknown_key_is_noop() -> None: + assert list(BlockStreamTracker().finish_block("nope")) == [] + + +def test_finish_all_finalizes_open_blocks_in_wire_order() -> None: + tracker = BlockStreamTracker() + list(tracker.feed(0, {"type": "text", "text": "hello", "index": 0})) + list(tracker.feed(1, {"type": "reasoning", "reasoning": "why", "index": 1})) + finished: list[Any] = list(tracker.finish_all()) + assert [f["index"] for f in finished] == [0, 1] + assert finished[0]["content"]["text"] == "hello" + assert finished[1]["content"]["reasoning"] == "why" From f7d7e0b756a276ee0dd4b15c42c525e607646641 Mon Sep 17 00:00:00 2001 From: Nick Hollon Date: Tue, 9 Jun 2026 17:30:45 -0400 Subject: [PATCH 2/3] feat(anthropic): native content-block streaming events --- .../langchain_anthropic/_stream_events.py | 147 ++++++++++++++ .../langchain_anthropic/chat_models.py | 84 +++++++- .../tests/unit_tests/test_chat_models.py | 77 ++++++-- .../tests/unit_tests/test_stream_events.py | 182 ++++++++++++++++++ 4 files changed, 476 insertions(+), 14 deletions(-) create mode 100644 libs/partners/anthropic/langchain_anthropic/_stream_events.py create mode 100644 libs/partners/anthropic/tests/unit_tests/test_stream_events.py diff --git a/libs/partners/anthropic/langchain_anthropic/_stream_events.py b/libs/partners/anthropic/langchain_anthropic/_stream_events.py new file mode 100644 index 0000000000000..4492d7f98eac9 --- /dev/null +++ b/libs/partners/anthropic/langchain_anthropic/_stream_events.py @@ -0,0 +1,147 @@ +"""Native content-block streaming-event converter for Anthropic. + +Maps the raw Anthropic SDK stream (``message_start`` / +``content_block_start`` / ``content_block_delta`` / ``content_block_stop`` / +``message_delta`` / ``message_stop``) to the protocol ``MessagesData`` +event lifecycle, reusing the shared ``BlockStreamTracker`` for block +mechanics and the existing per-event chunk builder for content extraction. +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import TYPE_CHECKING, Any + +from langchain_core.language_models.stream_events import ( + BlockStreamTracker, + accumulate_usage, + build_message_finish, + iter_protocol_blocks, +) + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Iterator + + from langchain_core.messages import AIMessageChunk + from langchain_protocol.protocol import ( + MessageMetadata, + MessagesData, + MessageStartData, + ) + +# The bound method ``ChatAnthropic._make_message_chunk_from_anthropic_event``. +MakeChunk = Callable[..., "tuple[AIMessageChunk | None, Any]"] + + +def _message_start(event: Any, message_id: str | None) -> MessageStartData: + msg_obj = getattr(event, "message", None) + # Do not use the provider message id (`msg_...`) here: on the v3 path core + # seeds the stream with the LangChain run id, and an empty id lets that + # stand (matching the compat bridge). Only an explicit `message_id` wins. + resolved_id = message_id or "" + metadata: MessageMetadata = {"provider": "anthropic"} + model = getattr(msg_obj, "model", None) + if model: + metadata["model"] = model + return { + "event": "message-start", + "role": "ai", + "id": resolved_id, + "metadata": metadata, + } + + +def convert_anthropic_stream( + raw: Iterator[Any], + make_chunk: MakeChunk, + *, + stream_usage: bool = True, + message_id: str | None = None, +) -> Iterator[MessagesData]: + """Convert a raw Anthropic event stream to protocol events. + + Args: + raw: Raw Anthropic SDK stream events. + make_chunk: ``ChatAnthropic._make_message_chunk_from_anthropic_event``, + injected so the converter stays pure and unit-testable. + stream_usage: Forwarded to ``make_chunk``. + message_id: Overrides the provider message id on ``message-start``. + + Yields: + Protocol ``MessagesData`` lifecycle events. + """ + tracker = BlockStreamTracker() + started = False + block_start_event: Any = None + usage: dict[str, Any] | None = None + response_metadata: dict[str, Any] = {"model_provider": "anthropic"} + + for event in raw: + etype = getattr(event, "type", None) + chunk_msg, block_start_event = make_chunk( + event, + stream_usage=stream_usage, + coerce_content_to_string=False, + block_start_event=block_start_event, + ) + if not started: + started = True + yield _message_start(event, message_id) + if chunk_msg is not None: + for key, block in iter_protocol_blocks(chunk_msg): + yield from tracker.feed(key, block) + if chunk_msg.usage_metadata: + usage = accumulate_usage(usage, chunk_msg.usage_metadata) + if chunk_msg.response_metadata: + response_metadata.update(chunk_msg.response_metadata) + if etype == "content_block_stop": + yield from tracker.finish_block(getattr(event, "index", None)) + + if not started: + return + yield from tracker.finish_all() + yield build_message_finish(usage=usage, response_metadata=response_metadata) + + +async def aconvert_anthropic_stream( + raw: AsyncIterator[Any], + make_chunk: MakeChunk, + *, + stream_usage: bool = True, + message_id: str | None = None, +) -> AsyncIterator[MessagesData]: + """Async twin of `convert_anthropic_stream`. `make_chunk` is sync.""" + tracker = BlockStreamTracker() + started = False + block_start_event: Any = None + usage: dict[str, Any] | None = None + response_metadata: dict[str, Any] = {"model_provider": "anthropic"} + + async for event in raw: + etype = getattr(event, "type", None) + chunk_msg, block_start_event = make_chunk( + event, + stream_usage=stream_usage, + coerce_content_to_string=False, + block_start_event=block_start_event, + ) + if not started: + started = True + yield _message_start(event, message_id) + if chunk_msg is not None: + for key, block in iter_protocol_blocks(chunk_msg): + for ev in tracker.feed(key, block): + yield ev + if chunk_msg.usage_metadata: + usage = accumulate_usage(usage, chunk_msg.usage_metadata) + if chunk_msg.response_metadata: + response_metadata.update(chunk_msg.response_metadata) + if etype == "content_block_stop": + for ev in tracker.finish_block(getattr(event, "index", None)): + yield ev + + 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/anthropic/langchain_anthropic/chat_models.py b/libs/partners/anthropic/langchain_anthropic/chat_models.py index c11143f2ac9ca..d834ba29d2193 100644 --- a/libs/partners/anthropic/langchain_anthropic/chat_models.py +++ b/libs/partners/anthropic/langchain_anthropic/chat_models.py @@ -11,7 +11,7 @@ from collections.abc import AsyncIterator, Callable, Iterator, Mapping, Sequence from functools import cached_property from operator import itemgetter -from typing import Any, Final, Literal, cast +from typing import TYPE_CHECKING, Any, Final, Literal, cast import anthropic from langchain_core.callbacks import ( @@ -64,9 +64,16 @@ _get_default_httpx_client, ) from langchain_anthropic._compat import _convert_from_v1_to_anthropic +from langchain_anthropic._stream_events import ( + aconvert_anthropic_stream, + convert_anthropic_stream, +) from langchain_anthropic.data._profiles import _PROFILES from langchain_anthropic.output_parsers import extract_tool_calls +if TYPE_CHECKING: + from langchain_protocol.protocol import ContentBlockDelta, MessagesData + _message_type_lookups = { "human": "user", "ai": "assistant", @@ -874,6 +881,13 @@ def _handle_anthropic_bad_request(e: anthropic.BadRequestError) -> None: raise +def _delta_text_for_callback(delta: ContentBlockDelta) -> str: + """Text increment to report via `on_llm_new_token` (empty for non-text).""" + if delta.get("type") == "text-delta": + return cast("str", delta.get("text", "")) + return "" + + class ChatAnthropic(BaseChatModel): """Anthropic (Claude) chat models. @@ -1509,6 +1523,74 @@ async def _astream( except anthropic.BadRequestError as e: _handle_anthropic_bad_request(e) + def _stream_chat_model_events( + self, + messages: list[BaseMessage], + stop: list[str] | None = None, + run_manager: CallbackManagerForLLMRun | None = None, + *, + stream_usage: bool | None = None, + message_id: str | None = None, + **kwargs: Any, + ) -> Iterator[MessagesData]: + """Emit Anthropic-native content-block protocol events. + + Detected by `langchain-core`'s `_iter_v2_events`; powers + `stream_events(version="v3")`. Falls through to the compat bridge + only if this method is absent. `message_id` is threaded from the + stream so `message-start` matches the bridge's LangChain run id. + """ + if stream_usage is None: + stream_usage = self.stream_usage + kwargs["stream"] = True + payload = self._get_request_payload(messages, stop=stop, **kwargs) + try: + raw = self._create(payload) + for event in convert_anthropic_stream( + raw, + self._make_message_chunk_from_anthropic_event, + stream_usage=stream_usage, + message_id=message_id, + ): + if run_manager is not None and event["event"] == "content-block-delta": + token = _delta_text_for_callback(event["delta"]) + if token: + run_manager.on_llm_new_token(token) + yield event + except anthropic.BadRequestError as e: + _handle_anthropic_bad_request(e) + + async def _astream_chat_model_events( + self, + messages: list[BaseMessage], + stop: list[str] | None = None, + run_manager: AsyncCallbackManagerForLLMRun | None = None, + *, + stream_usage: bool | None = None, + message_id: str | None = None, + **kwargs: Any, + ) -> AsyncIterator[MessagesData]: + """Async twin of `_stream_chat_model_events`.""" + if stream_usage is None: + stream_usage = self.stream_usage + kwargs["stream"] = True + payload = self._get_request_payload(messages, stop=stop, **kwargs) + try: + raw = await self._acreate(payload) + async for event in aconvert_anthropic_stream( + raw, + self._make_message_chunk_from_anthropic_event, + stream_usage=stream_usage, + message_id=message_id, + ): + if run_manager is not None and event["event"] == "content-block-delta": + token = _delta_text_for_callback(event["delta"]) + if token: + await run_manager.on_llm_new_token(token) + yield event + except anthropic.BadRequestError as e: + _handle_anthropic_bad_request(e) + def _make_message_chunk_from_anthropic_event( self, event: anthropic.types.RawMessageStreamEvent, diff --git a/libs/partners/anthropic/tests/unit_tests/test_chat_models.py b/libs/partners/anthropic/tests/unit_tests/test_chat_models.py index 175fc66f3376f..b80a78e8295d2 100644 --- a/libs/partners/anthropic/tests/unit_tests/test_chat_models.py +++ b/libs/partners/anthropic/tests/unit_tests/test_chat_models.py @@ -3222,19 +3222,12 @@ def test_no_task_budget_no_beta() -> None: assert "task-budgets-2026-03-13" not in betas -def test_anthropic_stream_events_v3_lifecycle() -> None: - """Validate lifecycle events across a thinking + text + tool_use stream. +def _lifecycle_events() -> list[Any]: + """Build a raw thinking + text + tool_use Anthropic stream fixture. - Anthropic emits raw `content_block_start` / `content_block_delta` / - `content_block_stop` events with integer `index` fields, interleaved - with `message_start` and `message_delta`. This test threads a - realistic event sequence through `_stream` via a mocked raw client - and asserts that `stream_events(version="v3")` produces a spec-conformant - event stream: paired start/finish per block, no interleaving, sequential - `uint` wire indices. + Shared by the sync and async v3 lifecycle parity tests so both exercise + the identical event sequence through the native hooks. """ - from unittest.mock import patch - from anthropic.types import ( InputJSONDelta, RawContentBlockDeltaEvent, @@ -3252,7 +3245,6 @@ def test_anthropic_stream_events_v3_lifecycle() -> None: from anthropic.types.raw_message_delta_event import ( MessageDeltaUsage as RawMessageDeltaUsage, ) - from langchain_tests.utils.stream_lifecycle import assert_valid_event_stream msg = Message( id="msg_1", @@ -3265,7 +3257,7 @@ def test_anthropic_stream_events_v3_lifecycle() -> None: type="message", ) - events = [ + return [ RawMessageStartEvent(message=msg, type="message_start"), # thinking block (index=0) RawContentBlockStartEvent( @@ -3337,6 +3329,24 @@ def test_anthropic_stream_events_v3_lifecycle() -> None: RawMessageStopEvent(type="message_stop"), ] + +def test_anthropic_stream_events_v3_lifecycle() -> None: + """Validate lifecycle events across a thinking + text + tool_use stream. + + Anthropic emits raw `content_block_start` / `content_block_delta` / + `content_block_stop` events with integer `index` fields, interleaved + with `message_start` and `message_delta`. This test threads a + realistic event sequence through `_stream` via a mocked raw client + and asserts that `stream_events(version="v3")` produces a spec-conformant + event stream: paired start/finish per block, no interleaving, sequential + `uint` wire indices. + """ + from unittest.mock import patch + + from langchain_tests.utils.stream_lifecycle import assert_valid_event_stream + + events = _lifecycle_events() + # Enable thinking so `coerce_content_to_string=False` in `_stream`, # which gives every content block an integer `index` field — the # structured path the protocol bridge actually exercises. Default @@ -3355,6 +3365,14 @@ def mock_create(_payload: Any) -> list: assert_valid_event_stream(stream_events) + # `message-start` must carry the stream's LangChain run id (threaded from + # core), matching the bridge path — not the provider message id ("msg_1") + # and not an empty string. + message_start = cast("dict[str, Any]", stream_events[0]) + assert message_start["event"] == "message-start" + assert message_start["id"] + assert message_start["id"] != "msg_1" + finishes = [e for e in stream_events if e["event"] == "content-block-finish"] types = [f["content"]["type"] for f in finishes] assert types == ["reasoning", "text", "tool_call"] @@ -3378,3 +3396,36 @@ def mock_create(_payload: Any) -> list: message_finish = cast("dict[str, Any]", stream_events[-1]) assert message_finish["event"] == "message-finish" assert message_finish["metadata"]["stop_reason"] == "tool_use" + + +async def test_anthropic_astream_events_v3_lifecycle() -> None: + """Async twin of `test_anthropic_stream_events_v3_lifecycle`. + + Exercises `_astream_chat_model_events` end-to-end via the + `AsyncChatModelStream` producer task. + """ + from unittest.mock import patch + + events = _lifecycle_events() + llm = ChatAnthropic( + model=MODEL_NAME, thinking={"type": "enabled", "budget_tokens": 1024} + ) + + async def mock_acreate(_payload: Any) -> Any: + async def _gen() -> Any: + for event in events: + yield event + + return _gen() + + with patch.object(llm, "_acreate", mock_acreate): + stream = await llm.astream_events("Test query", version="v3") + collected = [e async for e in stream] + + finishes = [e for e in collected if e["event"] == "content-block-finish"] + assert [f["content"]["type"] for f in finishes] == [ + "reasoning", + "text", + "tool_call", + ] + assert collected[-1]["metadata"]["stop_reason"] == "tool_use" diff --git a/libs/partners/anthropic/tests/unit_tests/test_stream_events.py b/libs/partners/anthropic/tests/unit_tests/test_stream_events.py new file mode 100644 index 0000000000000..4a405b36f60b5 --- /dev/null +++ b/libs/partners/anthropic/tests/unit_tests/test_stream_events.py @@ -0,0 +1,182 @@ +"""Unit tests for the Anthropic native stream-events converter.""" + +from typing import Any, cast + +from anthropic.types import ( + InputJSONDelta, + Message, + RawContentBlockDeltaEvent, + RawContentBlockStartEvent, + RawContentBlockStopEvent, + RawMessageDeltaEvent, + RawMessageStartEvent, + RawMessageStopEvent, + TextBlock, + TextDelta, + ThinkingBlock, + ThinkingDelta, + ToolUseBlock, + Usage, +) +from anthropic.types.raw_message_delta_event import Delta as RawMessageDelta +from anthropic.types.raw_message_delta_event import ( + MessageDeltaUsage as RawMessageDeltaUsage, +) +from langchain_tests.utils.stream_lifecycle import assert_valid_event_stream + +from langchain_anthropic import ChatAnthropic +from langchain_anthropic._stream_events import convert_anthropic_stream + +MODEL_NAME = "claude-haiku-4-5-20251001" + + +def _events() -> list[Any]: + msg = Message( + id="msg_1", + content=[], + model=MODEL_NAME, + role="assistant", + stop_reason=None, + stop_sequence=None, + usage=Usage(input_tokens=10, output_tokens=0), + type="message", + ) + return [ + RawMessageStartEvent(message=msg, type="message_start"), + RawContentBlockStartEvent( + content_block=ThinkingBlock(signature="", thinking="", type="thinking"), + index=0, + type="content_block_start", + ), + RawContentBlockDeltaEvent( + delta=ThinkingDelta(thinking="Let me ", type="thinking_delta"), + index=0, + type="content_block_delta", + ), + RawContentBlockDeltaEvent( + delta=ThinkingDelta(thinking="think.", type="thinking_delta"), + index=0, + type="content_block_delta", + ), + RawContentBlockStopEvent(index=0, type="content_block_stop"), + RawContentBlockStartEvent( + content_block=TextBlock(text="", type="text"), + index=1, + type="content_block_start", + ), + RawContentBlockDeltaEvent( + delta=TextDelta(text="The answer ", type="text_delta"), + index=1, + type="content_block_delta", + ), + RawContentBlockDeltaEvent( + delta=TextDelta(text="is 42.", type="text_delta"), + index=1, + type="content_block_delta", + ), + RawContentBlockStopEvent(index=1, type="content_block_stop"), + RawContentBlockStartEvent( + content_block=ToolUseBlock( + id="toolu_1", input={}, name="search", type="tool_use" + ), + index=2, + type="content_block_start", + ), + RawContentBlockDeltaEvent( + delta=InputJSONDelta(partial_json='{"q":', type="input_json_delta"), + index=2, + type="content_block_delta", + ), + RawContentBlockDeltaEvent( + delta=InputJSONDelta(partial_json=' "weather"}', type="input_json_delta"), + index=2, + type="content_block_delta", + ), + RawContentBlockStopEvent(index=2, type="content_block_stop"), + RawMessageDeltaEvent( + delta=RawMessageDelta(stop_reason="tool_use", stop_sequence=None), + type="message_delta", + usage=RawMessageDeltaUsage( + output_tokens=50, + input_tokens=10, + cache_read_input_tokens=0, + cache_creation_input_tokens=0, + ), + ), + RawMessageStopEvent(type="message_stop"), + ] + + +def test_convert_anthropic_stream_lifecycle() -> None: + llm = ChatAnthropic(model=MODEL_NAME) + events: list[Any] = list( + convert_anthropic_stream( + iter(_events()), llm._make_message_chunk_from_anthropic_event + ) + ) + + assert_valid_event_stream(events) + + assert events[0]["event"] == "message-start" + # The provider message id (`msg_1`) is deliberately NOT used: on the v3 path + # core seeds the stream with the LangChain run id, and an empty id here lets + # that stand (matching the compat bridge). Only an explicit `message_id` + # overrides it. + assert events[0]["id"] == "" + assert events[0]["metadata"]["provider"] == "anthropic" + assert events[0]["metadata"]["model"] == MODEL_NAME + + finishes = [e for e in events if e["event"] == "content-block-finish"] + assert [f["content"]["type"] for f in finishes] == [ + "reasoning", + "text", + "tool_call", + ] + assert [f["index"] for f in finishes] == [0, 1, 2] + reasoning = cast("dict[str, Any]", finishes[0]["content"]) + text = cast("dict[str, Any]", finishes[1]["content"]) + assert reasoning["reasoning"] == "Let me think." + assert text["text"] == "The answer is 42." + tool = cast("dict[str, Any]", finishes[2]["content"]) + assert tool["args"] == {"q": "weather"} + assert tool["name"] == "search" + + message_finish = events[-1] + assert message_finish["event"] == "message-finish" + assert message_finish["metadata"]["stop_reason"] == "tool_use" + # content-block-finish for index 0 arrives before index 1 starts + # (true stop boundaries, not all-at-end). + first_finish = next( + i for i, e in enumerate(events) if e["event"] == "content-block-finish" + ) + first_idx1_start = next( + i + for i, e in enumerate(events) + if e["event"] == "content-block-start" and e["index"] == 1 + ) + assert first_finish < first_idx1_start + + +async def test_aconvert_anthropic_stream_lifecycle() -> None: + llm = ChatAnthropic(model=MODEL_NAME) + + async def _araw() -> Any: + for event in _events(): + yield event + + from langchain_anthropic._stream_events import aconvert_anthropic_stream + + events: list[Any] = [ + e + async for e in aconvert_anthropic_stream( + _araw(), llm._make_message_chunk_from_anthropic_event + ) + ] + assert_valid_event_stream(events) + finishes = [e for e in events if e["event"] == "content-block-finish"] + assert [f["content"]["type"] for f in finishes] == [ + "reasoning", + "text", + "tool_call", + ] + assert events[-1]["metadata"]["stop_reason"] == "tool_use" From 891eae9bd1fd243757efd7575ff0109efeef2f8b Mon Sep 17 00:00:00 2001 From: Nick Hollon Date: Wed, 10 Jun 2026 15:06:32 -0400 Subject: [PATCH 3/3] feat(openrouter): native content-block streaming events --- .../langchain_openrouter/_stream_events.py | 205 +++++++++ .../langchain_openrouter/chat_models.py | 71 +++- .../tests/unit_tests/test_stream_events.py | 392 ++++++++++++++++++ 3 files changed, 667 insertions(+), 1 deletion(-) create mode 100644 libs/partners/openrouter/langchain_openrouter/_stream_events.py create mode 100644 libs/partners/openrouter/tests/unit_tests/test_stream_events.py diff --git a/libs/partners/openrouter/langchain_openrouter/_stream_events.py b/libs/partners/openrouter/langchain_openrouter/_stream_events.py new file mode 100644 index 0000000000000..3b314708e1c94 --- /dev/null +++ b/libs/partners/openrouter/langchain_openrouter/_stream_events.py @@ -0,0 +1,205 @@ +"""Native content-block streaming-event converter for openrouter. + +Builds text, reasoning, and tool-call blocks directly from openrouter's raw +OpenAI-shaped delta (openrouter streams tool-call args incrementally), feeding +the shared `BlockStreamTracker`. Unlike the compat bridge (which leaves +openrouter's `tool_calls`/`reasoning` in `additional_kwargs`), this surfaces +them as blocks. + +Differs from the groq converter in three ways: usage is a top-level +`chunk["usage"]` (carrying openrouter `cost`/`cost_details`), a no-choices +chunk may carry an `error` payload that must be propagated, and the reasoning +field is `delta.reasoning` (not `reasoning_content`). +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from langchain_core.language_models.stream_events import ( + BlockStreamTracker, + accumulate_usage, + build_message_finish, +) + +from langchain_openrouter.chat_models import _create_usage_metadata + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Iterator + + from langchain_protocol.protocol import ( + MessageMetadata, + MessagesData, + MessageStartData, + ) + + +def _message_start(message_id: str | None, model: str | None) -> MessageStartData: + # Do not use a provider completion id here: on the v3 path core seeds the + # stream with the LangChain run id, and an empty id lets that stand + # (matching the compat bridge). Only an explicit `message_id` wins. + metadata: MessageMetadata = {"provider": "openrouter"} + if model: + metadata["model"] = model + return { + "event": "message-start", + "role": "ai", + "id": message_id or "", + "metadata": metadata, + } + + +def _feed_delta( + tracker: BlockStreamTracker, delta: dict[str, Any] +) -> Iterator[MessagesData]: + """Yield block events for one OpenAI-shaped openrouter delta.""" + if reasoning := delta.get("reasoning"): + yield from tracker.feed( + "reasoning", {"type": "reasoning", "reasoning": reasoning} + ) + if content := delta.get("content"): + yield from tracker.feed("text", {"type": "text", "text": content}) + # openrouter structured reasoning fragments (delta["reasoning_details"]) are + # not surfaced as blocks yet; tracked for a follow-up. + for tc in delta.get("tool_calls") or []: + idx = tc.get("index", 0) + fn = tc.get("function") or {} + yield from tracker.feed( + f"tool:{idx}", + { + "type": "tool_call_chunk", + "id": tc.get("id"), + "name": fn.get("name"), + "args": fn.get("arguments") or "", + "index": idx, + }, + ) + + +def _error_message(error: dict[str, Any]) -> str: + """Build the streaming-error message, matching `_stream`'s `ValueError`.""" + return ( + f"OpenRouter API returned an error during streaming: " + f"{error.get('message', str(error))} " + f"(code: {error.get('code', 'unknown')})" + ) + + +class _StreamState: + """Accumulates per-stream state shared by the sync and async converters.""" + + def __init__(self) -> None: + self.tracker = BlockStreamTracker() + self.started = False + self.usage: dict[str, Any] | None = None + self.response_metadata: dict[str, Any] = {"model_provider": "openrouter"} + self.model: str | None = None + + def prepare(self, chunk: Any) -> tuple[dict[str, Any], list[dict[str, Any]]]: + """Coerce a raw chunk, fold in usage/cost, and return it plus choices. + + Raises: + ValueError: If a no-choices chunk carries an openrouter `error`. + """ + if not isinstance(chunk, dict): + chunk = chunk.model_dump(by_alias=True) + if self.model is None: + self.model = chunk.get("model") + if usage_payload := chunk.get("usage"): + self.usage = accumulate_usage( + self.usage, dict(_create_usage_metadata(usage_payload)) + ) + # Surface OpenRouter cost data, mirroring + # `_convert_chunk_to_message_chunk` so the v3 finish metadata keeps + # parity with the compat bridge's `response_metadata`. + if "cost" in usage_payload: + self.response_metadata["cost"] = usage_payload["cost"] + if "cost_details" in usage_payload: + self.response_metadata["cost_details"] = usage_payload["cost_details"] + choices = chunk.get("choices") or [] + if len(choices) == 0 and (error := chunk.get("error")): + raise ValueError(_error_message(error)) + return chunk, choices + + def record_finish(self, chunk: dict[str, Any], choice: dict[str, Any]) -> None: + """Capture finish-chunk response metadata, matching `_stream`. + + Mirrors the `generation_info` that `_stream`/`_astream` attach on the + final chunk so the v3 `message-finish` metadata equals the bridge's. + """ + self.response_metadata["finish_reason"] = choice["finish_reason"] + self.response_metadata["model_name"] = chunk.get("model") or self.model + if system_fingerprint := chunk.get("system_fingerprint"): + self.response_metadata["system_fingerprint"] = system_fingerprint + if native_finish_reason := choice.get("native_finish_reason"): + self.response_metadata["native_finish_reason"] = native_finish_reason + if response_id := chunk.get("id"): + self.response_metadata["id"] = response_id + if created := chunk.get("created"): + self.response_metadata["created"] = int(created) + if object_ := chunk.get("object"): + self.response_metadata["object"] = object_ + + +def convert_openrouter_stream( + raw: Iterator[Any], *, message_id: str | None = None +) -> Iterator[MessagesData]: + """Convert a raw openrouter chat stream to protocol events. + + Args: + raw: Raw openrouter chat-completion stream chunks (dicts or SDK + objects). + message_id: Overrides the provider message id on `message-start`. + + Yields: + Protocol `MessagesData` lifecycle events. + + Raises: + ValueError: If a chunk carries an openrouter `error` payload. + """ + state = _StreamState() + for chunk in raw: + chunk_dict, choices = state.prepare(chunk) + if len(choices) == 0: + continue + if not state.started: + state.started = True + yield _message_start(message_id, state.model) + choice = choices[0] + yield from _feed_delta(state.tracker, choice.get("delta") or {}) + if choice.get("finish_reason"): + state.record_finish(chunk_dict, choice) + + if not state.started: + return + yield from state.tracker.finish_all() + yield build_message_finish( + usage=state.usage, response_metadata=state.response_metadata + ) + + +async def aconvert_openrouter_stream( + raw: AsyncIterator[Any], *, message_id: str | None = None +) -> AsyncIterator[MessagesData]: + """Async twin of `convert_openrouter_stream`.""" + state = _StreamState() + async for chunk in raw: + chunk_dict, choices = state.prepare(chunk) + if len(choices) == 0: + continue + if not state.started: + state.started = True + yield _message_start(message_id, state.model) + choice = choices[0] + for ev in _feed_delta(state.tracker, choice.get("delta") or {}): + yield ev + if choice.get("finish_reason"): + state.record_finish(chunk_dict, choice) + + if not state.started: + return + for ev in state.tracker.finish_all(): + yield ev + yield build_message_finish( + usage=state.usage, response_metadata=state.response_metadata + ) diff --git a/libs/partners/openrouter/langchain_openrouter/chat_models.py b/libs/partners/openrouter/langchain_openrouter/chat_models.py index fcc9828485248..e1a6043810201 100644 --- a/libs/partners/openrouter/langchain_openrouter/chat_models.py +++ b/libs/partners/openrouter/langchain_openrouter/chat_models.py @@ -6,7 +6,7 @@ import warnings from collections.abc import AsyncIterator, Callable, Iterator, Mapping, Sequence from operator import itemgetter -from typing import Any, Literal, cast +from typing import TYPE_CHECKING, Any, Literal, cast from langchain_core.callbacks import ( AsyncCallbackManagerForLLMRun, @@ -72,6 +72,9 @@ from langchain_openrouter._version import __version__ from langchain_openrouter.data._profiles import _PROFILES +if TYPE_CHECKING: + from langchain_protocol.protocol import MessagesData + _MODEL_PROFILES = cast("ModelProfileRegistry", _PROFILES) # LangChain-internal kwargs that must not be forwarded to the SDK. @@ -710,6 +713,72 @@ async def _astream( # noqa: C901, PLR0912 ) yield generation_chunk + def _stream_chat_model_events( + self, + messages: list[BaseMessage], + stop: list[str] | None = None, + run_manager: CallbackManagerForLLMRun | None = None, + *, + message_id: str | None = None, + **kwargs: Any, + ) -> Iterator[MessagesData]: + """Emit openrouter-native content-block protocol events. + + Detected by `langchain-core`'s `_iter_v2_events`; powers + `stream_events(version="v3")`. Falls through to the compat bridge only + if this method is absent. `message_id` is threaded from the stream so + `message-start` matches the bridge's LangChain run id. + """ + from langchain_openrouter._stream_events import ( # noqa: PLC0415 + convert_openrouter_stream, + ) + + message_dicts, params = self._create_message_dicts(messages, stop) + params = {**params, **kwargs, "stream": True} + if self.stream_usage: + params["stream_options"] = {"include_usage": True} + _strip_internal_kwargs(params) + sdk_messages = _wrap_messages_for_sdk(message_dicts) + raw = self.client.chat.send(messages=sdk_messages, **params) + for event in convert_openrouter_stream(raw, 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 + + async def _astream_chat_model_events( + self, + messages: list[BaseMessage], + stop: list[str] | None = None, + run_manager: AsyncCallbackManagerForLLMRun | None = None, + *, + message_id: str | None = None, + **kwargs: Any, + ) -> AsyncIterator[MessagesData]: + """Async twin of `_stream_chat_model_events`.""" + from langchain_openrouter._stream_events import ( # noqa: PLC0415 + aconvert_openrouter_stream, + ) + + message_dicts, params = self._create_message_dicts(messages, stop) + params = {**params, **kwargs, "stream": True} + if self.stream_usage: + params["stream_options"] = {"include_usage": True} + _strip_internal_kwargs(params) + sdk_messages = _wrap_messages_for_sdk(message_dicts) + raw = await self.client.chat.send_async(messages=sdk_messages, **params) + async for event in aconvert_openrouter_stream(raw, 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 + # # Internal methods # diff --git a/libs/partners/openrouter/tests/unit_tests/test_stream_events.py b/libs/partners/openrouter/tests/unit_tests/test_stream_events.py new file mode 100644 index 0000000000000..cdbe1438c41ff --- /dev/null +++ b/libs/partners/openrouter/tests/unit_tests/test_stream_events.py @@ -0,0 +1,392 @@ +"""Unit tests for the openrouter native stream-events converter.""" + +import os +from typing import Any, cast +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from langchain_tests.utils.stream_lifecycle import assert_valid_event_stream + +from langchain_openrouter import ChatOpenRouter +from langchain_openrouter._stream_events import ( + aconvert_openrouter_stream, + convert_openrouter_stream, +) + +if "OPENROUTER_API_KEY" not in os.environ: + os.environ["OPENROUTER_API_KEY"] = "fake-key" + + +def _reasoning_text_tool() -> list[dict]: + m = "anthropic/claude-3.5-sonnet" + return [ + { + "model": m, + "choices": [ + { + "index": 0, + "delta": {"role": "assistant", "reasoning": "Let me "}, + "finish_reason": None, + } + ], + }, + { + "model": m, + "choices": [ + { + "index": 0, + "delta": {"reasoning": "think."}, + "finish_reason": None, + } + ], + }, + { + "model": m, + "choices": [ + { + "index": 0, + "delta": {"content": "Hi "}, + "finish_reason": None, + } + ], + }, + { + "model": m, + "choices": [ + { + "index": 0, + "delta": {"content": "there"}, + "finish_reason": None, + } + ], + }, + { + "model": m, + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": 0, + "id": "t1", + "function": { + "name": "get_weather", + "arguments": '{"city":', + }, + } + ] + }, + "finish_reason": None, + } + ], + }, + { + "model": m, + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + {"index": 0, "function": {"arguments": ' "Paris"}'}} + ] + }, + "finish_reason": "tool_calls", + } + ], + }, + { + "model": m, + "choices": [], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 5, + "total_tokens": 15, + }, + }, + ] + + +def test_convert_openrouter_stream_lifecycle() -> None: + """Reasoning, text, and tool-call deltas surface as finished blocks.""" + events: list[Any] = list(convert_openrouter_stream(iter(_reasoning_text_tool()))) + assert_valid_event_stream(events) + assert events[0]["event"] == "message-start" + assert events[0]["id"] == "" + assert events[0]["metadata"]["provider"] == "openrouter" + + finishes = [e for e in events if e["event"] == "content-block-finish"] + assert [f["content"]["type"] for f in finishes] == [ + "reasoning", + "text", + "tool_call", + ] + assert ( + cast("dict[str, Any]", finishes[0]["content"])["reasoning"] == "Let me think." + ) + assert cast("dict[str, Any]", finishes[1]["content"])["text"] == "Hi there" + tool = cast("dict[str, Any]", finishes[2]["content"]) + assert tool["name"] == "get_weather" + assert tool["args"] == {"city": "Paris"} + + assert events[-1]["event"] == "message-finish" + assert events[-1]["usage"] == { + "input_tokens": 10, + "output_tokens": 5, + "total_tokens": 15, + } + + +def test_convert_openrouter_stream_error_chunk_raises() -> None: + """A no-choices chunk carrying an `error` payload raises `ValueError`.""" + chunks = [{"choices": [], "error": {"message": "rate limited", "code": 429}}] + with pytest.raises(ValueError, match="OpenRouter API returned an error"): + list(convert_openrouter_stream(iter(chunks))) + + +def test_convert_openrouter_stream_surfaces_cost() -> None: + """Cost/native-finish/model metadata reach `message-finish`, like the bridge. + + The final chunk carries both `finish_reason` metadata and usage/cost data; + both must land in the `message-finish` response metadata so switching to the + v3 path does not silently drop OpenRouter's documented cost surfacing. + """ + cost_details = { + "upstream_inference_cost": 7.745e-05, + "upstream_inference_prompt_cost": 8.95e-06, + "upstream_inference_completions_cost": 6.85e-05, + } + chunks: list[dict] = [ + { + "model": "openai/gpt-4o-mini", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "Hi"}}], + }, + { + "model": "openai/gpt-4o-mini", + "id": "gen-cost-stream", + "system_fingerprint": "fp_abc", + "choices": [ + { + "index": 0, + "delta": {}, + "finish_reason": "stop", + "native_finish_reason": "end_turn", + } + ], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 5, + "total_tokens": 15, + "cost": 7.5e-05, + "cost_details": cost_details, + }, + }, + ] + events: list[Any] = list(convert_openrouter_stream(iter(chunks))) + assert_valid_event_stream(events) + + finish = events[-1] + assert finish["event"] == "message-finish" + meta = finish["metadata"] + assert meta["cost"] == 7.5e-05 + assert meta["cost_details"] == cost_details + assert meta["finish_reason"] == "stop" + assert meta["native_finish_reason"] == "end_turn" + assert meta["model_name"] == "openai/gpt-4o-mini" + assert meta["system_fingerprint"] == "fp_abc" + assert meta["id"] == "gen-cost-stream" + assert finish["usage"] == { + "input_tokens": 10, + "output_tokens": 5, + "total_tokens": 15, + } + + +def _parallel_tool_calls() -> list[dict]: + """Two tool calls interleaved at index 0 and 1 across chunks.""" + m = "anthropic/claude-3.5-sonnet" + return [ + { + "model": m, + "choices": [ + { + "index": 0, + "delta": { + "role": "assistant", + "tool_calls": [ + { + "index": 0, + "id": "call_a", + "function": { + "name": "get_weather", + "arguments": '{"city":', + }, + }, + { + "index": 1, + "id": "call_b", + "function": { + "name": "get_time", + "arguments": '{"zone":', + }, + }, + ], + }, + "finish_reason": None, + } + ], + }, + { + "model": m, + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + {"index": 0, "function": {"arguments": ' "Paris"}'}}, + {"index": 1, "function": {"arguments": ' "UTC"}'}}, + ] + }, + "finish_reason": "tool_calls", + } + ], + }, + ] + + +def test_convert_openrouter_stream_parallel_tool_calls() -> None: + """Args at distinct `tool:{idx}` keys finalize as separate tool calls.""" + events: list[Any] = list(convert_openrouter_stream(iter(_parallel_tool_calls()))) + assert_valid_event_stream(events) + + finishes = [e for e in events if e["event"] == "content-block-finish"] + assert [f["content"]["type"] for f in finishes] == ["tool_call", "tool_call"] + + first = cast("dict[str, Any]", finishes[0]["content"]) + second = cast("dict[str, Any]", finishes[1]["content"]) + assert first["name"] == "get_weather" + assert first["args"] == {"city": "Paris"} + assert second["name"] == "get_time" + assert second["args"] == {"zone": "UTC"} + + +async def test_aconvert_openrouter_stream_lifecycle() -> None: + """Async twin of `test_convert_openrouter_stream_lifecycle`.""" + + async def _araw() -> Any: + for chunk in _reasoning_text_tool(): + yield chunk + + events: list[Any] = [e async for e in aconvert_openrouter_stream(_araw())] + assert_valid_event_stream(events) + assert events[0]["event"] == "message-start" + assert events[0]["id"] == "" + assert events[0]["metadata"]["provider"] == "openrouter" + + finishes = [e for e in events if e["event"] == "content-block-finish"] + assert [f["content"]["type"] for f in finishes] == [ + "reasoning", + "text", + "tool_call", + ] + assert ( + cast("dict[str, Any]", finishes[0]["content"])["reasoning"] == "Let me think." + ) + assert cast("dict[str, Any]", finishes[1]["content"])["text"] == "Hi there" + tool = cast("dict[str, Any]", finishes[2]["content"]) + assert tool["name"] == "get_weather" + assert tool["args"] == {"city": "Paris"} + + assert events[-1]["event"] == "message-finish" + assert events[-1]["usage"] == { + "input_tokens": 10, + "output_tokens": 5, + "total_tokens": 15, + } + + +async def test_aconvert_openrouter_stream_error_chunk_raises() -> None: + """Async twin of `test_convert_openrouter_stream_error_chunk_raises`.""" + + async def _araw() -> Any: + yield {"choices": [], "error": {"message": "rate limited", "code": 429}} + + with pytest.raises(ValueError, match="OpenRouter API returned an error"): + [e async for e in aconvert_openrouter_stream(_araw())] + + +def test_openrouter_stream_events_v3_lifecycle() -> None: + """Drive `stream_events(version="v3")` through the native sync hook. + + Confirms the model-level path threads the LangChain run id onto + `message-start` (non-empty, unlike the converter's empty default) and + surfaces reasoning/text/tool blocks with `model_provider == "openrouter"`. + """ + llm = ChatOpenRouter(model="foo") + mock_client = MagicMock() + + def mock_send(*_a: Any, **_k: Any) -> Any: + for chunk in _reasoning_text_tool(): + mock = MagicMock() + mock.model_dump.return_value = chunk + yield mock + + mock_client.chat.send = mock_send + + with patch.object(llm, "client", mock_client): + events = list(llm.stream_events("Test query", version="v3")) + + assert_valid_event_stream(events) + + message_start = cast("dict[str, Any]", events[0]) + assert message_start["event"] == "message-start" + assert message_start["id"] # core seeds a non-empty LangChain run id + assert message_start["metadata"]["provider"] == "openrouter" + + finishes = [e for e in events if e["event"] == "content-block-finish"] + assert [f["content"]["type"] for f in finishes] == [ + "reasoning", + "text", + "tool_call", + ] + tool = cast("dict[str, Any]", finishes[2]["content"]) + assert tool["name"] == "get_weather" + assert tool["args"] == {"city": "Paris"} + + message_finish = cast("dict[str, Any]", events[-1]) + assert message_finish["event"] == "message-finish" + assert message_finish["metadata"]["model_provider"] == "openrouter" + + +async def test_openrouter_astream_events_v3_lifecycle() -> None: + """Async twin of `test_openrouter_stream_events_v3_lifecycle`.""" + llm = ChatOpenRouter(model="foo") + + async def _araw() -> Any: + for chunk in _reasoning_text_tool(): + mock = MagicMock() + mock.model_dump.return_value = chunk + yield mock + + mock_client = MagicMock() + mock_client.chat.send_async = AsyncMock(return_value=_araw()) + + with patch.object(llm, "client", mock_client): + stream = await llm.astream_events("Test query", version="v3") + events = [e async for e in stream] + + assert_valid_event_stream(events) + + message_start = cast("dict[str, Any]", events[0]) + assert message_start["event"] == "message-start" + assert message_start["id"] + assert message_start["metadata"]["provider"] == "openrouter" + + finishes = [e for e in events if e["event"] == "content-block-finish"] + assert [f["content"]["type"] for f in finishes] == [ + "reasoning", + "text", + "tool_call", + ] + message_finish = cast("dict[str, Any]", events[-1]) + assert message_finish["metadata"]["model_provider"] == "openrouter"