feat(core): add content-block-centric streaming (v2)

[Nick Hollon (nick-hollon-lc)]

Summary

Adds a content-block-centric streaming protocol to BaseChatModel — stream_v2() / astream_v2() — alongside the existing stream() / astream(). Gives consumers explicit lifecycle events (message-start, content-block-start/delta/finish, message-finish) instead of merge-your-own AIMessageChunks, plus typed projections (.text, .reasoning, .tool_calls, .usage, .output) on a ChatModelStream object. Uses shared types from the published langchain-protocol package.

V1 remains unchanged; nothing is deprecated.

Motivation

V1 streaming hands consumers Iterator[AIMessageChunk] with no lifecycle boundaries — you merge chunks yourself, parse partial tool-call JSON yourself, and deal with provider-specific content shapes (Anthropic content_block_start/delta/stop, OpenAI deltas, reasoning under three different keys, etc.) because the chunk is the only streaming primitive. V2 compiles all of that into a normalized lifecycle with explicit start/finish boundaries per content block, finalized tool calls with parsed args (or explicit InvalidToolCallBlock on parse failure), and a shared wire format via langchain-protocol.

The long-term target is for providers to implement _stream_chat_model_events natively. Until they do, the compat bridge in this PR translates legacy AIMessageChunk streams into protocol events transparently — every provider participates in v2 without being rewritten.

Key files

• langchain_core/language_models/_compat_bridge.py — the transitional translation layer. chunks_to_events / achunks_to_events turn AIMessageChunk streams into protocol events; message_to_events / amessage_to_events do the same for finalized AIMessages (cache hits, checkpoint replay, model.invoke() inside a graph node). Shrinks as providers adopt the native hook; the non-streamed replay path survives forever.
• langchain_core/language_models/chat_model_stream.py — ChatModelStream / AsyncChatModelStream with typed projections, replay-buffer-backed raw event iteration, and dispatch(event) entry point for driving the stream from outside.
• langchain_core/language_models/chat_models.py — stream_v2() / astream_v2() on BaseChatModel. Wires callbacks (on_chat_model_start, on_llm_end with the assembled AIMessage, on_llm_error, and the new on_stream_event), the rate limiter, and the compat bridge. An optional provider hook _stream_chat_model_events / _astream_chat_model_events bypasses the bridge — documented inline at the getattr call sites.
• langchain_core/callbacks/base.py and callbacks/manager.py — new on_stream_event callback on LLMManagerMixin / AsyncCallbackHandler plus dispatch on CallbackManagerForLLMRun. Fires once per protocol event from stream_v2 / astream_v2, additive to the existing lifecycle callbacks. See the docstring for contract details (event-granularity vs v1 chunk-granularity, transparency across native/compat providers).
• langchain_core/runnables/base.py — RunnableBinding.stream_v2 / astream_v2 overrides that merge bound kwargs into the delegated call, matching the existing stream / astream overrides. Without these, chains built via bind / bind_tools would silently drop their bound kwargs on v2 calls.

Example

# Sync: typed projections on a ChatModelStream
stream = model.stream_v2("Summarize the book")

# Iterate text deltas
for delta in stream.text:
    print(delta, end="", flush=True)

# Or read accumulated values directly (drains the stream)
full_text: str = str(stream.text)
tool_calls: list[ToolCallBlock] = stream.tool_calls.get()  # finalized, args as dict
usage: UsageInfo | None = stream.usage
message: AIMessage = stream.output  # everything assembled

# Raw protocol event iteration — replay-buffer, works after projections
for event in stream:
    print(event["event"])
    # message-start, content-block-start, content-block-delta,
    # content-block-finish, message-finish

# Async: projections are async-iterable and awaitable
stream = await model.astream_v2("Summarize the book")

async for delta in stream.text:
    print(delta, end="", flush=True)

message: AIMessage = await stream              # assembled message
full_text: str = await stream.text             # awaitable resolves to accumulated
tool_calls: list[ToolCallBlock] = await stream.tool_calls

[codspeed-hq]

Merging this PR will not alter performance

✅ 15 untouched benchmarks

---

_{Comparing nh/content-block-centric-streaming (922a000) with master (fb6ab99)¹}

Footnotes

1. No successful run was found on master (acc5498) during the generation of this report, so fb6ab99 was used instead as the comparison base. There might be some changes unrelated to this pull request in this report. ↩

[ccurme (ccurme)]

(nit) is it possible for _should_stream to be False but _should_stream_v2 to be True? (should we call _should_stream here to deduplicate logic)

[ccurme (ccurme)]

do we need a NotImplementedError on Runnable

[ccurme (ccurme)]

shouldn't use sphinx here