Chapter 8 · Streaming Output and Interrupt Recovery

Chapter goals:

• Master the difference between stream() and invoke()
• Understand different StreamMode patterns
• Implement interrupt recovery (interrupt + resume)
• Build Agent with approval workflow

8.1 stream() Basics

from langgraph.graph import StateGraph, MessagesState, START, END

# Use stream instead of invoke
for event in graph.stream(
    {"messages": [HumanMessage("Hello")]},
    stream_mode="values"
):
    print(event["messages"][-1].content)

8.1.1 Stream Modes

Mode	Description
"values"	Full state after each node
"messages"	Only message updates
"updates"	Node outputs only
"debug"	Detailed execution info

---

8.2 Streaming Token by Token

async for event in graph.astream(
    {"messages": [HumanMessage("Tell me a story")]},
    stream_mode="messages"
):
    msg, metadata = event
    if metadata.get("langgraph_node") == "agent":
        print(msg.content, end="", flush=True)

8.2.1 SSE Response

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat/stream")
async def stream_chat():
    async def event_generator():
        async for event in graph.astream(
            {"messages": [HumanMessage(request.prompt)]},
            stream_mode="values"
        ):
            yield f"data: {json.dumps(event)}\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )

---

8.3 Interrupt Recovery

Resume from interrupted state:

# First invocation - will interrupt
try:
    result = graph.invoke(input, config)
except InterruptedError as e:
    # Save checkpoint_id for later resume
    saved_config = e.config

8.3.1 Resume with Input

# Later, provide human input
result = graph.invoke(
    {"approval": True, "feedback": "Good, but add more details"},
    config=saved_config
)

---

8.4 Approval Workflow

def review_node(state: AgentState) -> dict:
    # Check if review needed
    if needs_review(state):
        # Interrupt for human approval
        review = interrupt({
            "question": "Review this response?",
            "content": state["messages"][-1].content
        })
        
        if not review.get("approved"):
            return {"revision_needed": True}
    
    return {"approved": True}

---

Chapter Summary

1. stream() vs invoke(): stream() returns updates incrementally;
2. Stream Modes: values, messages, updates, debug;
3. Token Streaming: Use astream with messages mode for real-time output;
4. Interrupt Recovery: Save checkpoint and resume with human input;
5. Approval Workflow: interrupt() pauses for human review.

---

🛠️ Hands-on Practice

1. Implement SSE streaming endpoint that streams LLM tokens in real-time.

2. Add interrupt checkpoint before expensive API calls for human approval.

3. Build undo functionality using checkpoint history.

---

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. What is the difference between stream() and invoke()?

2. Which stream_mode returns only message updates?

3. How do you resume from an interrupt?

4. What is the use case for interrupt()?