LangGraph streaming and interrupting long-running agents

LangGraph streaming and interrupting long-running agents requires understanding three moving pieces: the streaming mode that surfaces progress, the interrupt mechanism that pauses execution, and the checkpointing that lets you resume safely. This guide walks through a complete pattern you can drop into a production workflow, from configuring the graph to handling resume logic in a FastAPI endpoint.

Step 1: Define the agent state and graph structure

Start with a minimal state schema that carries the conversation history, a scratchpad for tool results, and a flag indicating whether the graph is waiting on human input. LangGraph’s StateGraph expects a TypedDict or Pydantic model; the example below uses TypedDict for clarity.

# agent/graph.py
from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    scratchpad: dict
    awaiting_human: bool

# A deliberately slow tool to simulate long-running work
@tool
def slow_lookup(query: str) -> str:
    """Simulate an expensive external call."""
    import time
    time.sleep(3)  # blocks the thread; in prod use async HTTP
    return f"Result for '{query}': 42 items found"

tools = [slow_lookup]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)

def agent_node(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"])
    return {"messages": [response], "awaiting_human": False}

def tool_node(state: AgentState) -> AgentState:
    last = state["messages"][-1]
    results = []
    for call in last.tool_calls:
        result = slow_lookup.invoke(call["args"])
        results.append(ToolMessage(content=result, tool_call_id=call["id"]))
    return {"messages": results, "awaiting_human": False}

Wire the graph with a conditional edge that routes to the interrupt point after any tool call. The interrupt node itself does no work — it only yields control.

# agent/graph.py (continued)
from langgraph.types import interrupt, Command

def interrupt_node(state: AgentState) -> Command:
    # Surface whatever context the human needs to decide
    payload = {
        "last_tool_result": state["messages"][-1].content if state["messages"] else None,
        "scratchpad": state.get("scratchpad", {}),
    }
    # This call pauses the graph and returns control to the caller
    human_input = interrupt(payload)
    return Command(
        update={
            "messages": [HumanMessage(content=human_input)],
            "awaiting_human": False,
        },
        goto="agent",
    )

def should_interrupt(state: AgentState) -> str:
    last = state["messages"][-1]
    if isinstance(last, ToolMessage):
        return "interrupt"
    if last.tool_calls:
        return "tools"
    return END

builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.add_node("interrupt", interrupt_node)
builder.set_entry_point("agent")
builder.add_conditional_edges("agent", should_interrupt)
builder.add_edge("tools", "interrupt")
builder.add_edge("interrupt", "agent")

graph = builder.compile(checkpointer=True)  # enables persistence

The checkpointer=True argument uses the default in-memory checkpointer. For production, swap in SqliteSaver or PostgresSaver so interrupted runs survive process restarts.

Step 2: Stream values with the `values` mode

LangGraph exposes four streaming modes: values, updates, messages, and custom. For a chat-style UI you typically want values — each yield is the full accumulated state after a node completes. The updates mode emits only the delta from each node, which is useful for progress bars but requires client-side reconstruction.

# agent/streaming.py
from agent.graph import graph
from langchain_core.messages import HumanMessage

config = {"configurable": {"thread_id": "session-123"}}

# Initial invocation — streams until the first interrupt or completion
for chunk in graph.stream(
    {"messages": [HumanMessage(content="Find all users named Alice")]},
    config=config,
    stream_mode="values",
):
    # chunk is the full AgentState after each node
    print(f"Node finished. Messages: {len(chunk['messages'])}")
    if chunk.get("awaiting_human"):
        print("Graph paused — waiting for human input")
        break

Run this and you’ll see the agent node emit a tool call, the tool node execute slow_lookup (three-second sleep), then the interrupt node yield with awaiting_human=True. The loop breaks at that point, leaving the graph paused at the interrupt.

Step 3: Resume from interrupt with `Command(resume=...)`

Resuming is a separate stream call on the same thread_id. Pass the human’s response through Command(resume=...). The graph picks up exactly where it left off, re-running the interrupt node with the supplied value.

# agent/streaming.py (continued)
# Simulate a human reviewing the tool result and responding
human_response = "Looks good, summarize the findings."

for chunk in graph.stream(
    Command(resume=human_response),
    config=config,
    stream_mode="values",
):
    print(f"Resumed. Messages: {len(chunk['messages'])}")
    if chunk.get("awaiting_human"):
        print("Paused again")
        break

Verification: the second stream iteration should print the final AI message with the summary and then exit the loop without hitting another interrupt. The thread_id ensures the checkpointer retrieves the correct checkpoint.

Step 4: Expose the pattern over HTTP with FastAPI

A real service needs to accept the initial prompt, stream tokens back to the client, pause at interrupts, and expose a resume endpoint. The following FastAPI app demonstrates the minimal surface area.

# server/main.py
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from agent.graph import graph
from langgraph.types import Command
import uuid
import json

app = FastAPI()

class StartRequest(BaseModel):
    prompt: str

class ResumeRequest(BaseModel):
    thread_id: str
    human_input: str

@app.post("/runs")
async def start_run(req: StartRequest):
    thread_id = str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}
    initial = {"messages": [HumanMessage(content=req.prompt)]}

    async def event_generator():
        # Stream values mode yields full state dicts
        for chunk in graph.stream(initial, config=config, stream_mode="values"):
            # Only send the newest message to the client
            if chunk["messages"]:
                last = chunk["messages"][-1]
                yield f"data: {json.dumps({'type': 'message', 'content': last.content})}\n\n"
            if chunk.get("awaiting_human"):
                yield f"data: {json.dumps({'type': 'interrupt', 'thread_id': thread_id})}\n\n"
                break
        yield "data: [DONE]\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

@app.post("/runs/{thread_id}/resume")
async def resume_run(thread_id: str, req: ResumeRequest):
    config = {"configurable": {"thread_id": thread_id}}
    try:
        for chunk in graph.stream(
            Command(resume=req.human_input),
            config=config,
            stream_mode="values",
        ):
            if chunk["messages"]:
                last = chunk["messages"][-1]
                return {"message": last.content, "awaiting_human": chunk.get("awaiting_human", False)}
    except KeyError:
        raise HTTPException(404, "Thread not found or expired")
    return {"message": "Completed", "awaiting_human": False}

Start the server with uvicorn server.main:app --reload. Test the flow:

# 1. Start a run
curl -X POST http://localhost:8000/runs \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Find all users named Alice"}'

# Output (SSE stream):
# data: {"type": "message", "content": "I'll look that up for you."}
# data: {"type": "message", "content": "Result for 'Find all users named Alice': 42 items found"}
# data: {"type": "interrupt", "thread_id": "abc-123"}
# data: [DONE]

# 2. Resume with human input (use the thread_id from above)
curl -X POST http://localhost:8000/runs/abc-123/resume \
  -H "Content-Type: application/json" \
  -d '{"thread_id": "abc-123", "human_input": "Summarize the results"}'

# Output:
# {"message": "Found 42 users named Alice across the directory.", "awaiting_human": false}

The /runs endpoint streams until the first interrupt, then returns the thread_id so the client can call /resume. The resume endpoint runs the graph to the next interrupt or completion and returns the final message as JSON — no SSE needed for the second hop unless you want token-level streaming there too.

Step 5: Add token-level streaming for the LLM response

The values mode only yields after each node finishes. To show the model’s tokens as they arrive, wrap the LLM call in a callback handler or use LangGraph’s messages stream mode, which emits AIMessageChunk objects. The trade-off: messages mode doesn’t include tool results or interrupt payloads, so you typically run two streams in parallel or switch modes after the first interrupt.

# agent/streaming.py (token streaming variant)
from langchain_core.messages import AIMessageChunk

config = {"configurable": {"thread_id": "session-456"}}

# First, run to the interrupt with values mode (as before)
for chunk in graph.stream(
    {"messages": [HumanMessage(content="Find all users named Alice")]},
    config=config,
    stream_mode="values",
):
    if chunk.get("awaiting_human"):
        print("Interrupted, switching to messages mode for resume")
        break

# Resume with messages mode to get token chunks
for chunk in graph.stream(
    Command(resume="Summarize please"),
    config=config,
    stream_mode="messages",
):
    # chunk is a tuple: (message_chunk, metadata)
    msg_chunk, meta = chunk
    if isinstance(msg_chunk, AIMessageChunk):
        print(msg_chunk.content, end="", flush=True)

This prints the summary token-by-token. In a server you’d multiplex the two stream modes: values for structural events (tool calls, interrupts) and messages for the final LLM response.

Step 6: Persist checkpoints with SQLite for durability

The in-memory checkpointer loses state on restart. Swap it for SqliteSaver (or PostgresSaver) with one line:

# agent/graph.py (checkpointer swap)
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

conn = sqlite3.connect("checkpoints.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)

graph = builder.compile(checkpointer=checkpointer)

The thread_id becomes a row in the checkpoints table. You can inspect it directly:

SELECT thread_id, checkpoint FROM checkpoints WHERE thread_id = 'session-123';

This is essential for any deployment where the graph process may restart between the interrupt and the resume call — which is the common case for human-in-the-loop workflows that wait hours or days.

Step 7: Handle multiple interrupts and branching

A graph can interrupt more than once. Each interrupt() call creates a new checkpoint. The resume payload is matched to the most recent interrupt by thread_id. If you need parallel human reviews, use distinct thread_id values or embed a review ID in the interrupt payload and route conditionally.

# agent/graph.py (branching example)
def interrupt_node(state: AgentState) -> Command:
    review_type = state["scratchpad"].get("review_type", "general")
    payload = {"review_type": review_type, "context": state["messages"][-1].content}
    human_input = interrupt(payload)
    # Route based on what the human said
    if "reject" in human_input.lower():
        return Command(update={"messages": [HumanMessage(content=human_input)]}, goto="agent")
    return Command(update={"messages": [HumanMessage(content=human_input)]}, goto="finalize")

The scratchpad field in state carries routing hints across the interrupt boundary. This pattern scales to approval chains, escalation paths, or any workflow where the graph topology depends on human decisions.

Verification checklist

Run through these steps to confirm the implementation works end-to-end:

Cold start: curl /runs with a fresh prompt → SSE stream ends with interrupt event and thread_id.
Resume: curl /runs/{thread_id}/resume with human input → JSON response with final message, awaiting_human: false.
Restart server: Kill the FastAPI process, restart, then resume the same thread_id → works because SQLite persisted the checkpoint.
Token streaming: Switch the resume call to stream_mode="messages" and verify chunks arrive incrementally.
Double interrupt: Modify the graph to interrupt twice (e.g., after tool result and after draft response). Verify two separate resume calls are required.

Common pitfalls

Forgetting checkpointer=True: Without it, interrupt() raises RuntimeError: No checkpointer configured.
Reusing thread_id across unrelated sessions: Each logical conversation needs its own ID; otherwise resumes apply to the wrong checkpoint.
Blocking in async context: The slow_lookup tool uses time.sleep. In an async FastAPI handler, wrap blocking calls with asyncio.to_thread or use native async clients.
Streaming mode mismatch: values yields AgentState dicts; messages yields (BaseMessageChunk, metadata) tuples. Don’t mix consumers.

Where this fits in a larger system

The pattern above is the backbone of any human-in-the-loop agent: a long-running graph that checkpoints at decision points, exposes a resume API, and streams progress to a frontend. At n4n.ai we use the same primitives — streaming, interrupts, durable checkpoints — to gate model calls behind routing policies and fallback logic without blocking the request thread. The graph doesn’t care whether the LLM call goes to OpenRouter, a local vLLM instance, or a provider that just hit a rate limit; the interrupt surface is identical.

For production, add structured logging around each interrupt (who approved, when, what payload), metrics on time-to-resume, and a dead-letter queue for threads that stall beyond a TTL. The core mechanics, however, stay exactly what you’ve built here.