Skip to content

feat(langgraph): more robust pydantic + dataclass support for StateGraph#6963

Draft
Sydney Runkle (sydney-runkle) wants to merge 8 commits into1.1from
sr/more-robust-pydantic-support
Draft

feat(langgraph): more robust pydantic + dataclass support for StateGraph#6963
Sydney Runkle (sydney-runkle) wants to merge 8 commits into1.1from
sr/more-robust-pydantic-support

Conversation

@sydney-runkle
Copy link
Collaborator

@sydney-runkle Sydney Runkle (sydney-runkle) commented Feb 27, 2026

More robust Pydantic support for v2 streaming

When using stream_version="v2", stream data and invoke results now respect the graph's output/state schema types (Pydantic models, dataclasses, etc.) instead of always returning raw dicts. This makes working with typed state much more natural — no more manual Model(**chunk) calls scattered through your code.

Values stream coercion

values stream parts coerce data through the graph's output schema mapper, so you get Pydantic models (or dataclasses) back directly:

class MyState(BaseModel):
    value: str
    items: Annotated[list[str], operator.add]

graph = StateGraph(MyState).compile()

# v1: you get raw dicts back, have to reconstruct manually
for chunk in graph.stream(inputs, stream_mode="values"):
    state = MyState(**chunk)  # manual, error-prone

# v2: data is already a MyState instance
for part in graph.stream(inputs, stream_mode="values", stream_version="v2"):
    assert isinstance(part["data"], MyState)  # just works
    print(part["data"].value)                 # attribute access, IDE autocomplete

This also works for dataclass-based state schemas. TypedDict state stays as plain dicts (no change needed).

Interrupts on stream parts

values stream parts now carry an interrupts field directly, removing the need to cross-reference the updates stream:

for part in graph.stream(inputs, config, stream_mode="values", stream_version="v2"):
    if part["interrupts"]:
        # handle interrupts inline — no need to check updates stream
        for intr in part["interrupts"]:
            print(intr.value)

Checkpoint/debug coercion

Checkpoint and debug stream payloads also coerce their values through the state schema mapper, so stream_mode="checkpoints" and stream_mode="debug" return typed state too.

Generic stream types

StreamPart, ValuesStreamPart, CheckpointPayload, etc. are now generic over StateT/OutputT, enabling better static type checking across the board.

GraphOutput wrapper (experimental — may not ship)

Note: We're not fully convinced GraphOutput should be part of this PR. It adds a new return type to invoke() which is a meaningful API surface change. We may pull this out and ship it separately, or not at all, depending on feedback.

invoke(stream_version="v2") returns a GraphOutput[OutputT] dataclass with .value and .interrupts fields:

result = graph.invoke({"value": "x", "items": []}, stream_version="v2")

# typed access
assert isinstance(result, GraphOutput)
assert isinstance(result.value, MyState)  # coerced to schema type
assert result.interrupts == ()            # always available

# backward compat dict access still works
assert result["value"] == "x_a"

The concern: this changes the return type of invoke() in a way that existing code patterns like result["key"] still work (via __getitem__), but isinstance(result, dict) checks would break. Worth discussing whether the ergonomic benefit justifies the migration cost.

@sydney-runkle Sydney Runkle (sydney-runkle) changed the title poc: more robust pydantic support Feb 27, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

1 participant