Fixes langchain-ai/langgraph#8382 `DeltaChannel` rebuilds its value by replaying ancestor writes through the reducer. Every saver ordered a checkpoint's writes by `(task_id, idx)`, but live execution applies a super-step's writes in task-path order (`apply_writes` sorts tasks by `task_path_str(task.path[:3])`). `task_id` is a hash of the path, so when two or more tasks write one `DeltaChannel` in the same super-step, replay applies them in an arbitrary permutation. Reducers must be batching-invariant, not order-invariant, so the permutation changes the value: `get_state` disagrees with what `invoke` returned, and continuing the thread saves the reordered value as the base for every later write. ## Fix Replay orders by `(task_path, task_id, idx)`, the order `SELECT_PENDING_SENDS_SQL` already uses for sends. `apply_writes` and every saver now sort with one `writes_sort_key` from `langgraph.checkpoint.base`, in Python, so live and replay can't drift apart and don't depend on a database collation. `apply_writes` sorts by the full task path, the string `put_writes` stores, instead of `path[:3]`. The paths that reach `apply_writes` no longer carry task ids, so the live order doesn't change. `BaseCheckpointSaver`'s default `get_delta_channel_history` can't sort by path, since `PendingWrite` has no task path, so it replays each checkpoint's writes in `get_tuple` order. `get_tuple` now has to return `pending_writes` in `writes_sort_key` order, and memory, SQLite and Postgres do, so the default walk is right for any saver that follows that. `AsyncPostgresSaver` also gets the sync `get_delta_channel_history` that `AsyncSqliteSaver` already had; sync reads used to fall back to the default walk. Memory and Postgres already stored `task_path`. SQLite accepted it on `put_writes` and dropped it, so `writes` gains the column and `setup()` adds it to existing databases. Writes without a task path (graph input, `update_state` updates, rows written before the column existed) sort first by `task_id`, which is also where live execution applies input. ## Compatibility - Memory and Postgres already had `task_path` on disk, so existing threads replay in the corrected order after upgrade. - On SQLite, rows from before the upgrade read back as `''` and keep their old order. Only new writes get the corrected order, since there is no stored path to recover. - SQLite has no `ADD COLUMN IF NOT EXISTS`, so `setup()` runs the `ALTER` and treats `duplicate column name` as success, which also covers two processes migrating one file. That error comes while SQLite prepares the statement, before it asks for the write lock, so `setup()` on an up-to-date database doesn't wait on other writers. A read-only database from before the column is read with empty paths instead of failing setup. - Downgrade is safe: older code's `INSERT` omits the column and the column has a default. - `get_tuple` and `list` return `pending_writes` in `writes_sort_key` order instead of task id order (SQLite, Postgres) or put order (memory). Nothing in langgraph depends on the old order. - Postgres reads `task_path` and `idx` with each pending write, so a subclass that pairs `SELECT_SQL` with its own `_load_writes`, or the other way round, has to match the new row shape. - A third-party saver has to replay in `writes_sort_key` order: in its own `get_delta_channel_history` if it has one, otherwise by returning `pending_writes` from `get_tuple` in that order. Otherwise it fails the new conformance tests; the `get_tuple` one is a base test, so it runs for every saver. - langgraph, checkpoint-sqlite and checkpoint-postgres import `writes_sort_key` from langgraph-checkpoint, so this bumps langgraph-checkpoint to 4.3.0 and requires it there. 4.3.0 has to be released first. ## Limits - Several updates in one `bulk_update_state` super-step share a task id, so the saver keeps only the first one's writes and the `DeltaChannel` loses the rest (plain channels are fine). That's on main too; #9128 fixes it and stores a task path per update, so with this PR those updates also replay in the order given. - Exit durability stores its writes differently; resume ordering there is fixed in #9114, stacked on this PR. - langgraphjs applies `DeltaChannel` writes in task id order, live and on replay (langchain-ai/langgraphjs#2544), so its reads match too. Moving JS to task path order is a follow-up. ## Tests Each fails on main and passes here: - `test_delta_channel_parallel_order.py`, across memory, SQLite and Postgres: parallel writers and a `Send` fan-out, comparing `get_state` with the live result, including sync `get_state` on the async savers. - Three conformance tests whose task ids sort opposite to their task paths, so they cannot pass by chance: two on `get_delta_channel_history`, one on `get_tuple`'s `pending_writes` order. - SQLite migration tests from a pre-column database, sync and async, including a read-only file. One more that passes on main too: SQLite `setup()` while another connection holds the write lock, with no busy timeout, sync and async. It keeps the `ALTER` from waiting on that lock. Thanks to @ErenAta16 for the report, the reproduction, and for tracing `task_path` through all three backends, and to @ragnarok268 for the conformance-test design with task ids ordered opposite to their paths. --------- Co-authored-by: ErenAta16 <149434812+ErenAta16@users.noreply.github.com> Co-authored-by: ragnarok268 <58264829+ragnarok268@users.noreply.github.com>
LangGraph Prebuilt
To help you ship LangGraph apps to production faster, check out LangSmith. LangSmith is a unified developer platform for building, testing, and monitoring LLM applications.
Quick Install
uv add langgraph
🤔 What is this?
This library defines high-level APIs for creating and executing LangGraph agents and tools. It includes prebuilt components such as create_react_agent, ToolNode, validation helpers, and Agent Inbox schemas.
📖 Documentation
For full documentation, see the API reference. For conceptual guides and tutorials, see the LangGraph Docs.
Important
This library is bundled with
langgraph; most users should installlanggraphinstead of installinglanggraph-prebuiltdirectly.
Agents
langgraph-prebuilt provides an implementation of a tool-calling ReAct-style agent - create_react_agent:
uv add 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
...
📕 Releases & Versioning
See our Releases and Versioning policies.
💁 Contributing
As an open-source project in a rapidly developing field, we are extremely open to contributions, whether it be in the form of a new feature, improved infrastructure, or better documentation.
For detailed information on how to contribute, see the Contributing Guide.