Repository navigation
LangGraph: interrupt() as a Restate signal, langgraph.tools(), and create_agent - #289
Merged
Merged
Conversation
nodes.wait_for_signal, nodes.workflow and nodes.subagent make nodes for a graph of your own, as the tools do for an agent: each reads its input from the state and writes what it gets back to the state field `output`. The graph runs in a handler of an AgentObject, with ainvoke(context=ctx) and SqlCheckpointSaver(ctx.db). Each node's idempotency key is LangGraph's task namespace, which LangGraph derives from the checkpoint the step started from: a node re-run after a crash gets the same one, so it joins the wait, the workflow or the other agent's answer it started before. A wait is listed by the agent's pending view, and its signal view resolves it. The helpers the tools and the nodes share in _tools.py lose their leading underscore (handler_of, input_type_of, inputs_of, untyped).
LangGraph's default durability, "async", only submits a step's checkpoint
save, and the next step's tasks start right away ("exit" saves nothing
between steps). A crash before the save makes the resumed run create the
checkpoint again with a new id, so the node gets a new task id and a new
idempotency key, and makes its step twice. Only "sync" stores the checkpoint
before the step runs.
examples-agents/intermediate/refund_graph.py runs a graph of your own in a handler of an AgentObject: it asks a policy agent about the refund (nodes.subagent), waits for a manager's approval (nodes.wait_for_signal), routes on the answer with an ordinary conditional edge, and runs the payout workflow (nodes.workflow). tests/examples_agents.py serves it with a scripted policy agent, and checks both an approved refund, which is paid, and a rejected one, which ends unpaid.
The graph's state has LangGraph's messages (MessagesState); a StateGraph or a compiled graph (such as LangChain's create_agent makes) is accepted, with no model, system or tools of ours. Each message is a run of the graph on the key's thread, with the message as a new HumanMessage whose id is the prompt's: - The checkpoints are in the key's database (SqlCheckpointSaver, history "latest"), stored before each step (durability "sync"), and the run's context is the handler's, so the prebuilt nodes work in it. A retried attempt resumes the run it cut off, or takes the finished run's answer. - After each step, the AI and tool messages it added are copied into the agent's conversation by id, so the messages, sync and watch views show the run as it goes. LangGraph streams a step's state only once its checkpoint is stored, so the conversation never gets ahead of the checkpoints; a copy a crash cut off is made at the next step or at the end. A test checks that ordering. - The answer is the run's last AI message; output= validates it. - After a stopped run, the next message starts a run of its own. The refund example becomes a desk you chat with: a clerk agent (output=Refund) reads the request, a policy agent checks it, a manager approves, and the payout workflow runs.
… input Graph state is plain data, while a workflow's handler often takes a Pydantic model (or a dataclass, or a TypedDict). A dict from the state was passed through as it was, so the handler's serde failed on every attempt and Restate retried forever. The node now validates the value with Pydantic's TypeAdapter into the handler's input type; a value of that type already is passed through. A value that does not validate fails the node with a terminal error naming the handler and the field, since a retry would read the same state. The refund example passes the state's dict again.
…nodes The graph agent: when a message starts a run after one that was stopped, refused or failed, the run's input starts with what closed that run in the conversation: the results of its cut-off tool calls, so that the graph's history has no tool call without a result (which a model provider refuses), and the note that ends it, so that the graph's model knows how the turn ended, as the built-in agent's does. The previous message comes too, if the graph never stored it. Each carried item has an id of its own, so it is carried once. nodes.call and nodes.send call or send to a handler, as the tools do. A key is required exactly for a keyed handler (a virtual object's or a workflow's), and plain data from the state is validated into the handler's input, as for nodes.workflow. _tools.keyed is shared with them.
- nodes: a handler input type Pydantic does not know (a msgspec Struct) no longer makes building the node fail: the value goes to the handler's serde as it is. - nodes.wait_for_signal: what the graph waits for is turned into JSON, so a model in the state (a workflow's result, say) no longer fails each attempt. - prompt(ctx, text, graph=...): a handler of your own prompts the graph as chat does; agent(graph=...) answers through it. - The graph agent copies a run's messages looking at the run's own entries only, not the whole conversation at each step. - Docs: nodes lead with langgraph.agent(graph=...); output= with a graph is the graph's own prompts' to ask for; the example does not claim the chat shows steps that add no messages. - Tests wait at most 20 seconds for their events, so a broken one fails rather than hangs; new tests for the above.
A run was found by its prompt's place in the graph's state, and its messages and answer were those after it. A graph that trims its history (a summarizing one) removes the prompt as well, and the run then copied nothing and answered with nothing; and the next message brought the trimmed prompt back. Each checkpoint of a run now names its prompt in its metadata, which is how a retried attempt finds the run it cut off. The messages copied are the graph's AI and tool messages that the conversation does not have (not those it gave the graph), and the answer is the last AI message the run copied, read from the conversation, which is a log. The previous prompt is carried into a new run only if its run copied nothing, so a prompt the graph removed is not brought back.
…d response LangGraph's interrupt() stopped the run, and chat answered with nothing. Each pending interrupt is now a wait named "interrupt", which pending lists with the interrupt's value and signal resolves; the run then resumes with Command(resume=...). The turn waits without holding a process, and a suspended or retried attempt joins the same waits. A node that interrupts twice has the same interrupt id both times, so a wait's id is the interrupt's and the number of resume values its task already has. Interrupts of branches at once are waited for at once. A graph whose state has a structured_response (create_agent with a response_format) answers with it, as JSON, which output= validates. Tests: an interrupt, two in one node, two branches at once, and a turn that Restate suspends while it waits (checked through the admin API).
… as create_agent's The restate.agents tools (tool, call, send, workflow, subagent, wait_for_signal) as LangChain tools, which take the handler's context from the graph's run (LangGraph's run context), so they work in langgraph.agent(graph=...) and in a graph a handler of your own invokes with context=ctx. Their Restate calls are made once per tool call. The package docs show create_agent with these tools, and say what to know about a graph of your own: durability is per step, so side effects belong in Restate nodes or tools; changing a graph that has waiting runs is a migration; long conversations store their messages at each step; there is no token streaming. Tests: create_agent with a Restate tool, and with a response_format.
…d fan-out A subgraph's node rejoins its step after a crash in the parent graph, and the same node sent to three times at once makes a step for each.
restate.agents.langgraph relies on LangGraph's checkpoint and config internals (task ids, durability), which a major version may change.
…tool examples-agents/basic/create_agent.py runs create_agent with a call to an orders service as its tool (langgraph.tools), as langgraph.agent(graph=...). Its test stands a scripted model in for the OpenAI one, and checks that the tool call reaches the service.
- An interrupt inside a subgraph: its resume values are among the pending writes of the subgraph's own checkpoint, so a subgraph's node that interrupted twice joined its first wait again. The pending interrupts are now found down into subgraphs, each counted at the task that raised it. - A structured_response is the answer only of the run that wrote it (the version of its channel changed in the run), not one an earlier run left in the state; the same response written again is still the answer. The version before the run is in the run's checkpoints' metadata, for a retried attempt. - tools(): only the tools that are Restate's (call, send, workflow, subagent, wait_for_signal) make their step once; a tool of your own runs your function again after a crash, as the docs now say. Tests for the first two, which fail without the fixes.
![pullfrog[bot]](https://avatars.githubusercontent.com/in/1768019?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Important
The ids of interrupt waits skip the companion's cleanup for turns that end without closing their waits. A killed turn leaves a stale "interrupt" wait behind, and later by-name signals resolve that stale wait instead of the live one.
Reviewed changes
Initial review of the whole PR: interrupt() as a Restate signal, langgraph.tools(), and create_agent structured responses on langgraph.agent(graph=...).
- Interrupts as waits:
run_ofresumes a stopped run withCommand(resume={interrupt_id: value}). The values come from_signals.waitcalls whose ids include each task's resume count, and the count is read from the subgraph's own checkpoint for an interrupt inside a subgraph. - Structured response:
_answeranswers withstructured_responseas JSON only when its channel version changed during the run. The version from before the run is kept in checkpoint metadata, so retries keep it. langgraph.tools(): wrapsrestate.agentstools as_RestateToolwith no bound context. They take the handler's context at call time fromget_runtime().context.- Version cap: the
langgraphextra is capped below LangGraph 2 andlangchain-core2. - Docs, example, tests: a "Good to know" section,
examples-agents/basic/create_agent.py, and Docker-harness tests for interrupts,create_agent, a subgraph crash,Sendfan-out, and the example.
I checked the resume-count logic against LangGraph 1.1.9's interrupt() and _loop.py. A resume map is applied through CONFIG_KEY_RESUME_MAP, and each interrupt() writes its task's full __resume__ list. So len(values) is the right count, and a retry rejoins the same wait id.
claude-opus-5-5 ｜ 𝕏
The companion closes the waits of a turn that ends without them (killed, or out of retries) by their ids, which start with the turn's invocation id, as the tools' waits do. The waits of interrupts and of nodes.wait_for_signal started with "langgraph:", so a killed turn left them pending, and a later signal by name resolved the stale wait instead of the live one. Their ids now start with the invocation id, which is the same in each attempt and suspension of a turn, so a retried attempt still joins the same wait. The other nodes keep their keys: a workflow or a call must stay once even when a later invocation resumes the graph. A test kills a turn waiting on each, and checks pending is empty; it fails with the old ids.
# Conflicts: # python/restate/agents/langgraph/__init__.py # python/restate/agents/langgraph/_graph.py # python/restate/agents/langgraph/nodes.py # tests/agents_langgraph_graph.py # tests/agents_langgraph_nodes.py
Contributor
There was a problem hiding this comment.
✅ No new issues found. The stale-wait finding from the prior review is fixed.
Reviewed changes
This run covers the changes since the review at b7d0eb0: one fix commit, and a merge of main now that #288 is squash-merged. The merge's combined diff is empty. It only brings in main's own change to tests/agents_langgraph_checkpointer.py.
- Interrupt waits close with their turn: in
_resumed, interrupt wait ids are now{ctx.request.id}:interrupt:{interrupt id}:{resume count}. The id starts with the turn's invocation id, which is what_signals._closedfilters on when_turn.pycloses a turn. The id is still the same in every attempt, so a retry rejoins the same wait. nodes.wait_for_signalwaits close too: its wait id and idempotency key are now{step.ctx.request.id}:{step.idempotency_key}.workflow,call,sendandsubagentkeep theirlanggraph:{key}:{namespace}keys on purpose, so they stay exactly-once when a later invocation resumes the graph.- Test: added
test_a_killed_turn_leaves_no_wait_behind, which runs for an interrupt (ask) and for await_for_signalnode (refund). It kills the turn through the admin API and checks thatpendingempties.
claude-opus-5-5 ｜ 𝕏
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Follows #288, now merged.
This PR makes LangGraph graphs of your own, including LangChain's
create_agent, work onlanggraph.agent(graph=...)without changes.interrupt()waits for a Restate signalBefore this PR, a graph that called
interrupt()stopped, andchatanswered with an empty string. Every later message hit the same interrupt again."interrupt". The agent'spendingview lists it with the interrupt's value, and itssignalview resolves it, by the wait's id or as the oldest"interrupt". The run then resumes withCommand(resume=...).langgraph.tools([...])Our tools as LangChain tools, for a graph of your own or for
create_agent:langgraph.agent(graph=...)passes it; a handler of your own passescontext=ctx.call,send,workflow,subagent,wait_for_signal) make their Restate call once per tool call, because the model's reply, with the tool call ids, is stored before the tools run.toolof your own runs your function, and LangChain tools pass through unchanged, so both run again after a crash.create_agent's structured responsestructured_responseduring the run answers with it, as JSON, andoutput=validates it.Also in this PR
langgraphextra is capped below LangGraph 2 (andlangchain-corebelow 2). The nodes rely on LangGraph's checkpoint and config internals. The lockfile change is just those two lines.create_agentwith our tools, and have a "Good to know" section:examples-agents/basic/create_agent.py.Tests
All run against a Restate server:
create_agent:response_format;Sendfan-out.The subgraph-interrupt and stale-structured-response tests fail without their fixes.
Not covered yet:
create_agenttool call;tools()in a graph that a handler of your own runs;