Generic InterruptNode — Roadmap Item 9 (v7.2-bound).
Parallel to elicitation.py: receives a _interrupt_pending state
marker carrying a registered interrupt name + payload, calls
interrupt(), and on resume injects the typed response back into the
graph so the next react step can act on it.
Topology
The node is wired into the LangGraph workflow when
FrameworkConfig.experimental_interrupt=True. It sits beside the
elicitation node (which still handles the built-in ask_user
interrupt unchanged) so the two pause paths share no code: a regression
in the generic path cannot influence ask_user, and vice-versa.
Pending state shape
A tool/node triggers an interrupt by returning::
{"_interrupt_pending": {
"name": "approval_required", # registered interrupt name
"interrupt_id": "i-<12hex>", # uuid4.hex[:12]; client correlation
"payload": {...}, # JSON-serialised payload
"tool_call_id": "...", # optional; only set when called from a tool
}}
The :class:SymfonicAgent.interrupt helper builds this marker so callers
do not assemble it by hand.
build_interrupt_marker
build_interrupt_marker(
*,
name: str,
interrupt_id: str,
payload: dict[str, Any],
tool_call_id: str | None = None,
) -> dict[str, Any]
Construct the state-update dict a tool/node returns to trigger an interrupt.
Public helper used by :meth:SymfonicAgent.interrupt and any tool
that wants to wire a generic interrupt by hand. Keeping this in one
place avoids the marker shape drifting between callers.
Source code in src/symfonic/core/nodes/interrupt.py
| def build_interrupt_marker(
*,
name: str,
interrupt_id: str,
payload: dict[str, Any],
tool_call_id: str | None = None,
) -> dict[str, Any]:
"""Construct the state-update dict a tool/node returns to trigger an interrupt.
Public helper used by :meth:`SymfonicAgent.interrupt` and any tool
that wants to wire a generic interrupt by hand. Keeping this in one
place avoids the marker shape drifting between callers.
"""
return {
"_interrupt_pending": {
"name": name,
"interrupt_id": interrupt_id,
"payload": payload,
"tool_call_id": tool_call_id,
}
}
|
create_interrupt_node
create_interrupt_node() -> Any
Return the generic interrupt node function.
The node reads state["_interrupt_pending"], calls interrupt(),
and on resume injects the typed response back into the messages
channel as a structured ToolMessage (when the interrupt was
triggered from a tool) or as plain state (otherwise).
Source code in src/symfonic/core/nodes/interrupt.py
| def create_interrupt_node() -> Any:
"""Return the generic interrupt node function.
The node reads ``state["_interrupt_pending"]``, calls ``interrupt()``,
and on resume injects the typed response back into the messages
channel as a structured ``ToolMessage`` (when the interrupt was
triggered from a tool) or as plain state (otherwise).
"""
async def interrupt_node(state: dict[str, Any]) -> dict[str, Any]:
from langchain_core.messages import ToolMessage
from langgraph.types import interrupt
pending: dict[str, Any] = state["_interrupt_pending"]
name: str = pending["name"]
interrupt_id: str = pending["interrupt_id"]
payload: dict[str, Any] = pending.get("payload", {})
tool_call_id: str | None = pending.get("tool_call_id") or None
# Call interrupt() — LangGraph surfaces this as an
# ExtensionEvent(type="on_interrupt") in astream_events(v2).
# The engine's stream_typed() intercepts it, sees
# ``interrupt_data["type"] == "interrupt"``, and routes to
# ``_mint_interrupt_token`` (the generic minter) instead of
# ``_mint_pause_token`` (the ask_user-specific one).
resume_value: Any = interrupt(
{
"type": "interrupt",
"name": name,
"interrupt_id": interrupt_id,
"payload": payload,
"tool_call_id": tool_call_id,
}
)
# --- Resume path ---------------------------------------------------
# LangGraph re-runs the node body from the top with the resume
# payload returned by ``interrupt()``. Clear the pending marker
# and inject the response into messages so downstream react
# turns see it as conversational context.
if resume_value is not None:
response_dict: dict[str, Any]
if isinstance(resume_value, dict):
response_dict = resume_value
elif hasattr(resume_value, "model_dump"):
response_dict = resume_value.model_dump()
else:
response_dict = {"value": str(resume_value)}
updates: dict[str, Any] = {"_interrupt_pending": None}
if tool_call_id:
# Tool-call-originated interrupt: emit a ToolMessage so
# the downstream react node sees a normal tool result
# for the originating call. The content is the
# JSON-serialised response.
import json as _json
updates["messages"] = [
ToolMessage(
content=_json.dumps(response_dict, default=str),
tool_call_id=tool_call_id,
name=f"interrupt:{name}",
)
]
else:
# Non-tool interrupt: stash on state for the caller
# to read via ``state.get("_interrupt_response")``.
updates["_interrupt_response"] = {
"name": name,
"interrupt_id": interrupt_id,
"response": response_dict,
}
return updates
# Should not reach here in normal operation; log and fail safely.
logger.warning(
"interrupt_node: interrupt() returned None unexpectedly "
"for name=%s interrupt_id=%s",
name,
interrupt_id,
)
return {"_interrupt_pending": None}
return interrupt_node
|
interrupt_condition
interrupt_condition(state: dict[str, Any]) -> str
Conditional-edge router shared with the elicitation condition.
Returns:
| Type |
Description |
str
|
"interrupt" when _interrupt_pending is set (truthy)
|
str
|
"elicitation" when _ask_user_pending is set (preserves
the existing ask_user routing)
|
str
|
"continue" when normal tool calls are present
|
str
|
|
Wired in by presets.py when experimental_interrupt=True so a
graph that mixes ask_user AND generic interrupts routes through
the right pause node deterministically.
Source code in src/symfonic/core/nodes/interrupt.py
| def interrupt_condition(state: dict[str, Any]) -> str:
"""Conditional-edge router shared with the elicitation condition.
Returns:
- ``"interrupt"`` when ``_interrupt_pending`` is set (truthy)
- ``"elicitation"`` when ``_ask_user_pending`` is set (preserves
the existing ask_user routing)
- ``"continue"`` when normal tool calls are present
- ``"finish"`` otherwise
Wired in by ``presets.py`` when ``experimental_interrupt=True`` so a
graph that mixes ``ask_user`` AND generic interrupts routes through
the right pause node deterministically.
"""
if state.get("_interrupt_pending"):
return "interrupt"
if state.get("_ask_user_pending"):
return "elicitation"
from symfonic.core.edges.tool_condition import tools_condition
return tools_condition(state)
|