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 00000000000..4492d7f98ea --- /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 c11143f2ac9..d834ba29d21 100644 --- a/libs/partners/anthropic/langchain_anthropic/chat_models.py +++ b/libs/partners/anthropic/langchain_anthropic/chat_models.py @@ -11,7 +11,7 @@ import warnings 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 @@ from langchain_anthropic._client_utils import ( _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 @@ class ChatAnthropic(BaseChatModel): 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 175fc66f337..b80a78e8295 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 test_anthropic_stream_events_v3_lifecycle() -> None: 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 test_anthropic_stream_events_v3_lifecycle() -> None: 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 00000000000..4a405b36f60 --- /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"