From 1e1ca88dad9c7e708263257fa9fc27a3fbdfff68 Mon Sep 17 00:00:00 2001 From: Nick Hollon Date: Fri, 24 Jul 2026 13:16:41 -0400 Subject: [PATCH] feat(langgraph): type v3 stream_events return and native projections (#8389) The `version="v3"` overloads of `stream_events`/`astream_events` returned `Any`, and `GraphRunStream`/`AsyncGraphRunStream` attached native projections via a runtime `setattr` loop invisible to type checkers. - Return `GraphRunStream` / `Awaitable[AsyncGraphRunStream]` from the v3 overloads. - Declare the always-registered native projections (`values`, `messages`, `lifecycle`, `subgraphs`) as typed class attributes on both run streams. - Add `assert_type` checks in `test_stream_events_v3.py`. Opt-in native projections (`updates`, `custom`, `checkpoints`, `debug`, `tasks`) are only present when their transformer is registered, so they remain reached via `extensions[...]` rather than annotated as always-present. --- libs/langgraph/langgraph/pregel/main.py | 8 ++-- libs/langgraph/langgraph/stream/run_stream.py | 28 ++++++++++++- libs/langgraph/tests/test_stream_events_v3.py | 41 +++++++++++++++++++ 3 files changed, 72 insertions(+), 5 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/main.py b/libs/langgraph/langgraph/pregel/main.py index 4e098ba5b..f4ca48024 100644 --- a/libs/langgraph/langgraph/pregel/main.py +++ b/libs/langgraph/langgraph/pregel/main.py @@ -3512,7 +3512,7 @@ class Pregel( control: RunControl | None = None, transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None, **kwargs: Any, - ) -> Any: + ) -> GraphRunStream: """Internal v3 sync streaming implementation. Public entry: stream_events(version='v3'). Extra keyword arguments are forwarded to the underlying ``stream(...)`` @@ -3568,7 +3568,7 @@ class Pregel( control: RunControl | None = None, transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None, **kwargs: Any, - ) -> Any: + ) -> AsyncGraphRunStream: """Internal v3 async streaming implementation. Public entry: astream_events(version='v3'). Extra keyword arguments are forwarded to the underlying ``astream(...)`` @@ -3633,7 +3633,7 @@ class Pregel( control: RunControl | None = None, transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None, **kwargs: Any, - ) -> Any: ... + ) -> GraphRunStream: ... def stream_events( self, @@ -3738,7 +3738,7 @@ class Pregel( control: RunControl | None = None, transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None, **kwargs: Any, - ) -> Awaitable[Any]: ... + ) -> Awaitable[AsyncGraphRunStream]: ... def astream_events( self, diff --git a/libs/langgraph/langgraph/stream/run_stream.py b/libs/langgraph/langgraph/stream/run_stream.py index 5eb14a6fa..37062c64d 100644 --- a/libs/langgraph/langgraph/stream/run_stream.py +++ b/libs/langgraph/langgraph/stream/run_stream.py @@ -12,7 +12,13 @@ from langgraph.stream._mux import StreamMux from langgraph.stream._types import ProtocolEvent if TYPE_CHECKING: - from langgraph.stream.transformers import SubgraphStatus + from langchain_core.language_models.chat_model_stream import ( + AsyncChatModelStream, + ChatModelStream, + ) + + from langgraph.stream.stream_channel import StreamChannel + from langgraph.stream.transformers import LifecyclePayload, SubgraphStatus def _drive_until_done(pump: Callable[[], bool]) -> None: @@ -48,6 +54,16 @@ class GraphRunStream: experimental and may change. """ + # Native projections always registered by `stream_events(version="v3")`. + # Attached dynamically by the `setattr` loop in `__init__`; declared here + # so type checkers see them. Opt-in native projections (`updates`, + # `custom`, `checkpoints`, `debug`, `tasks`) are only present when their + # transformer is registered, so they are reached via `extensions[...]`. + values: StreamChannel[dict[str, Any]] + messages: StreamChannel[ChatModelStream] + lifecycle: StreamChannel[LifecyclePayload] + subgraphs: StreamChannel[SubgraphRunStream] + def __init__( self, graph_iter: Iterator[Any] | None, @@ -329,6 +345,16 @@ class AsyncGraphRunStream: experimental and may change. """ + # Native projections always registered by `astream_events(version="v3")`. + # Attached dynamically by the `setattr` loop in `__init__`; declared here + # so type checkers see them. Opt-in native projections (`updates`, + # `custom`, `checkpoints`, `debug`, `tasks`) are only present when their + # transformer is registered, so they are reached via `extensions[...]`. + values: StreamChannel[dict[str, Any]] + messages: StreamChannel[AsyncChatModelStream] + lifecycle: StreamChannel[LifecyclePayload] + subgraphs: StreamChannel[AsyncSubgraphRunStream] + def __init__( self, graph_aiter: AsyncIterator[Any] | None, diff --git a/libs/langgraph/tests/test_stream_events_v3.py b/libs/langgraph/tests/test_stream_events_v3.py index 6399b3084..71d81325e 100644 --- a/libs/langgraph/tests/test_stream_events_v3.py +++ b/libs/langgraph/tests/test_stream_events_v3.py @@ -12,6 +12,10 @@ from dataclasses import dataclass from typing import Annotated, Any, TypeVar import pytest +from langchain_core.language_models.chat_model_stream import ( + AsyncChatModelStream, + ChatModelStream, +) from langchain_core.messages import AIMessage, BaseMessage from langgraph.checkpoint.memory import InMemorySaver from pydantic import BaseModel, ValidationError @@ -24,6 +28,14 @@ from langgraph.func import entrypoint from langgraph.graph import StateGraph from langgraph.graph.message import MessagesState from langgraph.runtime import RunControl +from langgraph.stream import ( + AsyncGraphRunStream, + AsyncSubgraphRunStream, + GraphRunStream, + LifecyclePayload, + StreamChannel, + SubgraphRunStream, +) from langgraph.types import ( CheckpointPayload, CheckpointStreamPart, @@ -1178,3 +1190,32 @@ def _check_type_narrowing(part: StreamPart[_StateT, _OutputT]) -> None: assert_type(part, DebugStreamPart[_StateT]) assert_type(part["data"], DebugPayload[_StateT]) assert_type(part["ns"], tuple[str, ...]) + + +# --- v3 stream_events return / projection typing checks --- +# These functions are never called at runtime; `ty` validates the +# assert_type calls. They pin the public typing surface of +# stream_events(version="v3") / astream_events(version="v3"): the handle +# type and the always-registered native projections. + + +def _check_stream_events_v3_typing() -> None: + """Compile-time checks for sync v3 typing — never called at runtime.""" + graph = _make_simple_graph().compile() + run = graph.stream_events(_SIMPLE_INPUT, version="v3") + assert_type(run, GraphRunStream) + assert_type(run.values, StreamChannel[dict[str, Any]]) + assert_type(run.messages, StreamChannel[ChatModelStream]) + assert_type(run.lifecycle, StreamChannel[LifecyclePayload]) + assert_type(run.subgraphs, StreamChannel[SubgraphRunStream]) + + +async def _check_astream_events_v3_typing() -> None: + """Compile-time checks for async v3 typing — never called at runtime.""" + graph = _make_simple_graph().compile() + run = await graph.astream_events(_SIMPLE_INPUT, version="v3") + assert_type(run, AsyncGraphRunStream) + assert_type(run.values, StreamChannel[dict[str, Any]]) + assert_type(run.messages, StreamChannel[AsyncChatModelStream]) + assert_type(run.lifecycle, StreamChannel[LifecyclePayload]) + assert_type(run.subgraphs, StreamChannel[AsyncSubgraphRunStream])