Files
langgraph/libs/prebuilt
Sydney RunkleandGitHub 3330ccdea4 release(langgraph): 1.1 (#7102)
exciting!

relnotes preview:

# LangGraph 1.1.0 Release Notes

## Type-Safe Streaming & Invoke

LangGraph 1.1 introduces `version="v2"` — a new opt-in streaming format
that brings full type safety to `stream()`, `astream()`, `invoke()`, and
`ainvoke()`.

### What's changing

**v1 (default, unchanged):** `stream()` yields bare tuples like
`(stream_mode, data)` or just `data`. `invoke()` returns a plain `dict`.
Interrupts are mixed into the output dict under `"__interrupt__"`.

**v2 (opt-in):** `stream()` yields strongly-typed `StreamPart` dicts
with `type`, `ns`, `data`, and (for values) `interrupts` fields.
`invoke()` returns a `GraphOutput` object with `.value` and
`.interrupts` attributes. When your state schema is a Pydantic model or
dataclass, outputs are automatically coerced to the correct type.

### `invoke()` / `ainvoke()` with `version="v2"`

```python
from langgraph.types import GraphOutput

result = graph.invoke({"input": "hello"}, version="v2")

# result is a GraphOutput, not a dict
assert isinstance(result, GraphOutput)
result.value       # your output — dict, Pydantic model, or dataclass
result.interrupts  # tuple[Interrupt, ...], empty if none occurred
```

With a non-`"values"` stream mode, `invoke(..., stream_mode="updates",
version="v2")` returns `list[StreamPart]` instead of `list[tuple]`.

### `stream()` / `astream()` with `version="v2"`

```python
for part in graph.stream({"input": "hello"}, version="v2"):
    if part["type"] == "values":
        part["data"]        # OutputT — full state
        part["interrupts"]  # tuple[Interrupt, ...]
    elif part["type"] == "updates":
        part["data"]        # dict[str, Any]
    elif part["type"] == "messages":
        part["data"]        # tuple[BaseMessage, dict]
    elif part["type"] == "custom":
        part["data"]        # Any
    elif part["type"] == "tasks":
        part["data"]        # TaskPayload | TaskResultPayload
    elif part["type"] == "debug":
        part["data"]        # DebugPayload
```

Each stream mode has its own `TypedDict` — `ValuesStreamPart`,
`UpdatesStreamPart`, `MessagesStreamPart`, `CustomStreamPart`,
`CheckpointStreamPart`, `TasksStreamPart`, `DebugStreamPart` — all
importable from `langgraph.types`. The union type `StreamPart` is a
discriminated union on `part["type"]`, enabling full type narrowing in
editors and type checkers.

### Pydantic & dataclass output coercion

When your graph's state schema is a Pydantic model or dataclass,
`version="v2"` automatically coerces outputs to the declared type:

```python
from pydantic import BaseModel

class MyState(BaseModel):
    answer: str
    count: int

graph = StateGraph(MyState)
# ... build graph ...
compiled = graph.compile()

result = compiled.invoke({"answer": "", "count": 0}, version="v2")
assert isinstance(result.value, MyState)  # not a dict!
```

### Backward compatibility

- **Default is still `version="v1"`** — existing code works without
changes.
- To make migration easier, `GraphOutput` supports old-style best-effort
access to graph values and interrupts. Dict-style access
(`result["key"]`, `"key" in result`, `result["__interrupt__"]`) still
works and delegates to `result.value` / `result.interrupts` under the
hood. However, this is **deprecated** and emits a
`LangGraphDeprecatedSinceV11` warning. It will be removed in v3.0 —
migrate to `result.value` and `result.interrupts` at your convenience.

```python
result = graph.invoke({"input": "hello"}, version="v2")

# Old style — still works, but deprecated
result["input"]          # delegates to result.value["input"]
result["__interrupt__"]  # delegates to result.interrupts
"input" in result        # delegates to "input" in result.value

# New style — preferred
result.value["input"]
result.interrupts
```

## Migration Guide

1. **No action required** — `version="v1"` remains the default. All
existing code continues to work.
2. **Adopt v2 incrementally** — Add `version="v2"` to individual
`invoke()`/`stream()` calls to get typed outputs.
3. **Use typed imports** — Import `GraphOutput`, `StreamPart`, and
individual part types from `langgraph.types` for type-safe code.
2026-03-10 12:41:30 +00:00
..
2026-03-10 12:41:30 +00:00

LangGraph Prebuilt

This library defines high-level APIs for creating and executing LangGraph agents and tools.

Important

This library is meant to be bundled with langgraph, don't install it directly

Agents

langgraph-prebuilt provides an implementation of a tool-calling ReAct-style agent - create_react_agent:

pip install langchain-anthropic
from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent

# Define the tools for the agent to use
def search(query: str):
    """Call to surf the web."""
    # This is a placeholder, but don't tell the LLM that...
    if "sf" in query.lower() or "san francisco" in query.lower():
        return "It's 60 degrees and foggy."
    return "It's 90 degrees and sunny."

tools = [search]
model = ChatAnthropic(model="claude-3-7-sonnet-latest")

app = create_react_agent(model, tools)
# run the agent
app.invoke(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
)

Tools

ToolNode

langgraph-prebuilt provides an implementation of a node that executes tool calls - ToolNode:

from langgraph.prebuilt import ToolNode
from langchain_core.messages import AIMessage

def search(query: str):
    """Call to surf the web."""
    # This is a placeholder, but don't tell the LLM that...
    if "sf" in query.lower() or "san francisco" in query.lower():
        return "It's 60 degrees and foggy."
    return "It's 90 degrees and sunny."

tool_node = ToolNode([search])
tool_calls = [{"name": "search", "args": {"query": "what is the weather in sf"}, "id": "1"}]
ai_message = AIMessage(content="", tool_calls=tool_calls)
# execute tool call
tool_node.invoke({"messages": [ai_message]})

ValidationNode

langgraph-prebuilt provides an implementation of a node that validates tool calls against a pydantic schema - ValidationNode:

from pydantic import BaseModel, field_validator
from langgraph.prebuilt import ValidationNode
from langchain_core.messages import AIMessage


class SelectNumber(BaseModel):
    a: int

    @field_validator("a")
    def a_must_be_meaningful(cls, v):
        if v != 37:
            raise ValueError("Only 37 is allowed")
        return v

validation_node = ValidationNode([SelectNumber])
validation_node.invoke({
    "messages": [AIMessage("", tool_calls=[{"name": "SelectNumber", "args": {"a": 42}, "id": "1"}])]
})

Agent Inbox

The library contains schemas for using the Agent Inbox with LangGraph agents. Learn more about how to use Agent Inbox here.

from langgraph.types import interrupt
from langgraph.prebuilt.interrupt import HumanInterrupt, HumanResponse

def my_graph_function():
    # Extract the last tool call from the `messages` field in the state
    tool_call = state["messages"][-1].tool_calls[0]
    # Create an interrupt
    request: HumanInterrupt = {
        "action_request": {
            "action": tool_call['name'],
            "args": tool_call['args']
        },
        "config": {
            "allow_ignore": True,
            "allow_respond": True,
            "allow_edit": False,
            "allow_accept": False
        },
        "description": _generate_email_markdown(state) # Generate a detailed markdown description.
    }
    # Send the interrupt request inside a list, and extract the first response
    response = interrupt([request])[0]
    if response['type'] == "response":
        # Do something with the response
    ...