Module 9: Human-in-the-Loop

Editable State: Modifying State During a Pause

Next →

Capsule overview

The agent is paused. You already know how to approve it or reject it. But there's a third option that changes everything: modifying the agent's state before it continues. Not restarting. Not re-running. Opening up the state, fixing what's wrong, and letting the agent carry on with correct information.

Picture this: your agent researched a company and found that the CEO is "John Smith." You know the CEO changed to "Jane Doe" last month. Without editable state, you have to restart the entire investigation. With editable state, you fix that one field in the state and the agent continues — it generates the report with "Jane Doe" without repeating the searches it already ran.

This isn't an advanced feature or an edge case. It's the most practical HITL scenario there is: the agent did useful work, but it has one wrong data point. Correcting and continuing beats starting from zero.

The scenario that makes it essential

Without editable state:
  1. Agent researches company (3 minutes, $0.50 in API calls)
  2. Agent finds CEO = "John Smith" ← incorrect
  3. You catch the error
  4. Option: restart the entire investigation
  5. 3 more minutes, $0.50 more
  6. Total: 6 minutes, $1.00

With editable state:
  1. Agent researches company (3 minutes, $0.50 in API calls)
  2. Agent finds CEO = "John Smith" ← incorrect
  3. You catch the error
  4. You edit: CEO = "Jane Doe"
  5. Agent continues from where it paused
  6. Total: 3 minutes + seconds, $0.50

The difference isn't just efficiency. It's that you preserve all the work the agent already did: the searches, the analysis, the correct data. You only fix what's wrong.

Inspecting state during a pause

Before editing, you need to see what the agent has. You use graph.get_state(config):

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class ResearchState(TypedDict):
    company: str
    ceo: str
    revenue: str
    employees: int
    findings: Annotated[list[str], operator.add]
    report: str

def research_company(state: ResearchState) -> dict:
    return {
        "ceo": "John Smith",
        "revenue": "$5.2B (FY2024)",
        "employees": 12000,
        "findings": [
            "Leader in the enterprise AI segment",
            "40% year-over-year growth",
            "Expansion into the European market in Q2",
        ],
    }

def generate_report(state: ResearchState) -> dict:
    report = f"REPORT: {state['company']}\n"
    report += f"  CEO: {state['ceo']}\n"
    report += f"  Revenue: {state['revenue']}\n"
    report += f"  Employees: {state['employees']}\n"
    report += f"  Findings: {len(state['findings'])}\n"
    for f in state["findings"]:
        report += f"    - {f}\n"
    return {"report": report}

graph_builder = StateGraph(ResearchState)
graph_builder.add_node("research", research_company)
graph_builder.add_node("report", generate_report)

graph_builder.add_edge(START, "research")
graph_builder.add_edge("research", "report")
graph_builder.add_edge("report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["research"],
)

config = {"configurable": {"thread_id": "inspect-demo"}}

graph.invoke(
    {"company": "TechCorp AI", "ceo": "", "revenue": "", "employees": 0,
     "findings": [], "report": ""},
    config,
)

print("=== Inspect state during the pause ===\n")

state = graph.get_state(config)

print(f"Full state:")
for key, value in state.values.items():
    if isinstance(value, list):
        print(f"  {key}: [{len(value)} items]")
        for item in value:
            print(f"    - {item}")
    else:
        print(f"  {key}: {value}")

print(f"\nNext node: {state.next}")
print(f"Checkpoint ID: {state.config['configurable']['checkpoint_id'][:20]}...")
# Expected output:
# === Inspect state during the pause ===
#
# Full state:
#   company: TechCorp AI
#   ceo: John Smith
#   revenue: $5.2B (FY2024)
#   employees: 12000
#   findings: [3 items]
#     - Leader in the enterprise AI segment
#     - 40% year-over-year growth
#     - Expansion into the European market in Q2
#   report:
#
# Next node: ('report',)
# Checkpoint ID: 1ef8a1b2c3d4e5f6ab...

state.values gives you the full state. state.next tells you which node is pending. With that, you can decide what to fix.

Modifying state: update_state

You saw that the CEO is "John Smith" but it should be "Jane Doe." You fix it with graph.update_state():

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class ResearchState(TypedDict):
    company: str
    ceo: str
    revenue: str
    employees: int
    findings: Annotated[list[str], operator.add]
    report: str

def research_company(state: ResearchState) -> dict:
    return {
        "ceo": "John Smith",
        "revenue": "$5.2B (FY2024)",
        "employees": 12000,
        "findings": [
            "Leader in the enterprise AI segment",
            "40% year-over-year growth",
        ],
    }

def generate_report(state: ResearchState) -> dict:
    report = f"REPORT: {state['company']}\n"
    report += f"  CEO: {state['ceo']}\n"
    report += f"  Revenue: {state['revenue']}\n"
    report += f"  Employees: {state['employees']}\n"
    report += f"  Findings ({len(state['findings'])}):\n"
    for f in state["findings"]:
        report += f"    - {f}\n"
    return {"report": report}

graph_builder = StateGraph(ResearchState)
graph_builder.add_node("research", research_company)
graph_builder.add_node("report", generate_report)

graph_builder.add_edge(START, "research")
graph_builder.add_edge("research", "report")
graph_builder.add_edge("report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["research"],
)

config = {"configurable": {"thread_id": "edit-demo"}}

graph.invoke(
    {"company": "TechCorp AI", "ceo": "", "revenue": "", "employees": 0,
     "findings": [], "report": ""},
    config,
)

state = graph.get_state(config)
print(f"BEFORE editing:")
print(f"  CEO: {state.values['ceo']}")
print(f"  Findings: {len(state.values['findings'])}")

print("\n--- Editing state ---\n")
graph.update_state(config, {"ceo": "Jane Doe"})

state = graph.get_state(config)
print(f"AFTER editing:")
print(f"  CEO: {state.values['ceo']}")
print(f"  Findings: {len(state.values['findings'])}  ← untouched")

print("\n--- Resuming execution ---\n")
result = graph.invoke(None, config)

final = graph.get_state(config)
print(final.values["report"])
# Expected output:
# BEFORE editing:
#   CEO: John Smith
#   Findings: 2
#
# --- Editing state ---
#
# AFTER editing:
#   CEO: Jane Doe
#   Findings: 2  ← untouched
#
# --- Resuming execution ---
#
# REPORT: TechCorp AI
#   CEO: Jane Doe
#   Revenue: $5.2B (FY2024)
#   Employees: 12000
#   Findings (2):
#     - Leader in the enterprise AI segment
#     - 40% year-over-year growth

Three important observations:

  1. update_state(config, {"ceo": "Jane Doe"}) only modifies the ceo field. The other fields aren't touched
  2. The findings are still 2 — the agent's prior work is preserved
  3. The final report says "Jane Doe" — the agent used the corrected data

How update_state works with reducers

update_state follows the same reducer rules as the graph itself:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    name: str
    count: int
    tags: Annotated[list[str], operator.add]

def process(state: State) -> dict:
    return {
        "name": "original",
        "count": 10,
        "tags": ["tag1", "tag2"],
    }

def finalize(state: State) -> dict:
    return {"name": f"final_{state['name']}_count_{state['count']}_tags_{len(state['tags'])}"}

graph_builder = StateGraph(State)
graph_builder.add_node("process", process)
graph_builder.add_node("finalize", finalize)

graph_builder.add_edge(START, "process")
graph_builder.add_edge("process", "finalize")
graph_builder.add_edge("finalize", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["process"],
)

config = {"configurable": {"thread_id": "reducer-demo"}}

graph.invoke({"name": "", "count": 0, "tags": []}, config)

state = graph.get_state(config)
print(f"Before: name={state.values['name']}, count={state.values['count']}, tags={state.values['tags']}")

graph.update_state(config, {
    "name": "edited",
    "count": 99,
    "tags": ["new_tag"],
})

state = graph.get_state(config)
print(f"After: name={state.values['name']}, count={state.values['count']}, tags={state.values['tags']}")
# Expected output:
# Before: name=original, count=10, tags=['tag1', 'tag2']
# After: name=edited, count=99, tags=['tag1', 'tag2', 'new_tag']

Notice the difference:

  • ✅ name (no reducer): it gets replaced — from "original" to "edited"
  • ✅ count (no reducer): it gets replaced — from 10 to 99
  • ✅ tags (with operator.add): it gets appended — "new_tag" is added to the existing list

update_state respects the TypedDict's reducers. If a field uses operator.add, the update is appended, not replaced. If it has no reducer, it's replaced outright.

Practical case: fixing the agent's wrong data

The full CEO scenario, with a realistic HITL flow:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class CompanyResearch(TypedDict):
    company: str
    ceo: str
    founded: int
    hq_location: str
    key_products: Annotated[list[str], operator.add]
    market_position: str
    report: str

def research(state: CompanyResearch) -> dict:
    return {
        "ceo": "John Smith",
        "founded": 2018,
        "hq_location": "San Francisco, CA",
        "key_products": ["Enterprise AI Platform", "AI Analytics Suite", "MLOps Toolkit"],
        "market_position": "Top 5 in enterprise AI, competing with OpenAI and Anthropic",
    }

def human_review(state: CompanyResearch) -> dict:
    review = interrupt({
        "message": "Review the research data before generating the report",
        "data": {
            "company": state["company"],
            "ceo": state["ceo"],
            "founded": state["founded"],
            "hq_location": state["hq_location"],
            "key_products": state["key_products"],
            "market_position": state["market_position"],
        },
        "instructions": "Reply {approved: true} or {corrections: {field: value}}",
    })

    if review.get("approved"):
        return {}
    return {}

def generate_report(state: CompanyResearch) -> dict:
    report = f"{'='*50}\n"
    report += f"COMPANY RESEARCH REPORT: {state['company']}\n"
    report += f"{'='*50}\n"
    report += f"CEO: {state['ceo']}\n"
    report += f"Founded: {state['founded']}\n"
    report += f"HQ: {state['hq_location']}\n"
    report += f"Products: {', '.join(state['key_products'])}\n"
    report += f"Market Position: {state['market_position']}\n"
    report += f"{'='*50}"
    return {"report": report}

graph_builder = StateGraph(CompanyResearch)
graph_builder.add_node("research", research)
graph_builder.add_node("human_review", human_review)
graph_builder.add_node("generate_report", generate_report)

graph_builder.add_edge(START, "research")
graph_builder.add_edge("research", "human_review")
graph_builder.add_edge("human_review", "generate_report")
graph_builder.add_edge("generate_report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "company-research"}}

print("=== STEP 1: Agent researches ===\n")
graph.invoke(
    {"company": "TechCorp AI", "ceo": "", "founded": 0, "hq_location": "",
     "key_products": [], "market_position": "", "report": ""},
    config,
)

state = graph.get_state(config)
print(f"Agent found: CEO = {state.values['ceo']}")
print(f"Waiting at: {state.next}")

print("\n=== STEP 2: Human spots errors and corrects them ===\n")

print("Corrections:")
print("  - CEO: 'John Smith' → 'Jane Doe' (changed last month)")
print("  - HQ: add 'European office in London'\n")

graph.update_state(config, {
    "ceo": "Jane Doe",
    "hq_location": "San Francisco, CA (HQ) + London, UK (European office)",
})

state = graph.get_state(config)
print(f"Corrected state:")
print(f"  CEO: {state.values['ceo']}")
print(f"  HQ: {state.values['hq_location']}")

print("\n=== STEP 3: Resume with corrected data ===\n")
result = graph.invoke(Command(resume={"approved": True}), config)

final = graph.get_state(config)
print(final.values["report"])
# Expected output:
# === STEP 1: Agent researches ===
#
# Agent found: CEO = John Smith
# Waiting at: ('human_review',)
#
# === STEP 2: Human spots errors and corrects them ===
#
# Corrections:
#   - CEO: 'John Smith' → 'Jane Doe' (changed last month)
#   - HQ: add 'European office in London'
#
# Corrected state:
#   CEO: Jane Doe
#   HQ: San Francisco, CA (HQ) + London, UK (European office)
#
# === STEP 3: Resume with corrected data ===
#
# ==================================================
# COMPANY RESEARCH REPORT: TechCorp AI
# ==================================================
# CEO: Jane Doe
# Founded: 2018
# HQ: San Francisco, CA (HQ) + London, UK (European office)
# Products: Enterprise AI Platform, AI Analytics Suite, MLOps Toolkit
# Market Position: Top 5 in enterprise AI, competing with OpenAI and Anthropic
# ==================================================

The final report carries the human's corrections, but preserves everything else the agent researched correctly (products, market position, founding year).

Adding information the agent couldn't find

You don't just fix errors — you can add data the agent had no way of getting:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    topic: str
    sources: Annotated[list[str], operator.add]
    analysis: str
    summary: str

def search(state: State) -> dict:
    return {
        "sources": [
            "Wikipedia: general information about the topic",
            "arXiv: 2 relevant papers found",
        ],
    }

def analyze(state: State) -> dict:
    return {"analysis": f"Analysis of {len(state['sources'])} sources on '{state['topic']}'"}

def summarize(state: State) -> dict:
    return {"summary": f"Summary: {state['analysis']}. Sources: {', '.join(state['sources'][:3])}..."}

graph_builder = StateGraph(State)
graph_builder.add_node("search", search)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("summarize", summarize)

graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "analyze")
graph_builder.add_edge("analyze", "summarize")
graph_builder.add_edge("summarize", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["search"],
)

config = {"configurable": {"thread_id": "add-info"}}

graph.invoke(
    {"topic": "AI regulation in EU", "sources": [], "analysis": "", "summary": ""},
    config,
)

state = graph.get_state(config)
print(f"Sources found by the agent: {len(state.values['sources'])}")
for s in state.values["sources"]:
    print(f"  - {s}")

print("\n--- Human adds sources the agent couldn't find ---\n")

graph.update_state(config, {
    "sources": [
        "INSIDER: Internal European Commission document (not public)",
        "INSIDER: Interview with the lead regulator (off the record)",
    ],
})

state = graph.get_state(config)
print(f"Sources after adding: {len(state.values['sources'])}")
for s in state.values["sources"]:
    print(f"  - {s}")

print("\n--- Resuming with enriched sources ---\n")
graph.invoke(None, config)

final = graph.get_state(config)
print(f"Analysis: {final.values['analysis']}")
print(f"Summary: {final.values['summary']}")
# Expected output:
# Sources found by the agent: 2
#   - Wikipedia: general information about the topic
#   - arXiv: 2 relevant papers found
#
# --- Human adds sources the agent couldn't find ---
#
# Sources after adding: 4
#   - Wikipedia: general information about the topic
#   - arXiv: 2 relevant papers found
#   - INSIDER: Internal European Commission document (not public)
#   - INSIDER: Interview with the lead regulator (off the record)
#
# --- Resuming with enriched sources ---
#
# Analysis: Analysis of 4 sources on 'AI regulation in EU'
# Summary: Summary: Analysis of 4 sources on 'AI regulation in EU'. Sources: Wikipedia: general information about the topic, arXiv: 2 relevant papers found, INSIDER: Internal European Commission document (not public)...

Because sources uses operator.add, the new sources are appended to the existing ones. The agent now works with 4 sources: 2 it found itself and 2 the human contributed. Privileged information no agent could get on its own.

update_state with as_node: controlling which node comes next

When you call update_state, you can specify as_node so LangGraph treats the update as if it came from a specific node. This affects which node runs next:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    data: str
    step_a_result: str
    step_b_result: str
    final: str
    log: Annotated[list[str], operator.add]

def step_a(state: State) -> dict:
    return {"step_a_result": f"A processed: {state['data']}", "log": ["step_a"]}

def step_b(state: State) -> dict:
    return {"step_b_result": f"B analyzed: {state['step_a_result']}", "log": ["step_b"]}

def step_final(state: State) -> dict:
    return {"final": f"Final: {state['step_b_result']}", "log": ["step_final"]}

graph_builder = StateGraph(State)
graph_builder.add_node("step_a", step_a)
graph_builder.add_node("step_b", step_b)
graph_builder.add_node("step_final", step_final)

graph_builder.add_edge(START, "step_a")
graph_builder.add_edge("step_a", "step_b")
graph_builder.add_edge("step_b", "step_final")
graph_builder.add_edge("step_final", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["step_b"],
)

config = {"configurable": {"thread_id": "as-node-demo"}}

graph.invoke(
    {"data": "raw input", "step_a_result": "", "step_b_result": "",
     "final": "", "log": []},
    config,
)

state = graph.get_state(config)
print(f"Paused before: {state.next}")
print(f"step_a_result: {state.values['step_a_result']}")

print("\n--- Option 1: update_state WITHOUT as_node ---")
print("  The update doesn't change which node comes next")
print(f"  Next is still: {state.next}")

print("\n--- Option 2: update_state WITH as_node='step_b' ---")
print("  LangGraph treats the update as if step_b had already run")

graph.update_state(
    config,
    {"step_b_result": "B manual result (human replaced step_b)"},
    as_node="step_b",
)

state = graph.get_state(config)
print(f"  Next is now: {state.next}")
print(f"  step_b_result: {state.values['step_b_result']}")

print("\n--- Resuming: step_final runs directly ---\n")
graph.invoke(None, config)

final = graph.get_state(config)
print(f"Final: {final.values['final']}")
print(f"Log: {final.values['log']}")
# Expected output:
# Paused before: ('step_b',)
# step_a_result: A processed: raw input
#
# --- Option 1: update_state WITHOUT as_node ---
#   The update doesn't change which node comes next
#   Next is still: ('step_b',)
#
# --- Option 2: update_state WITH as_node='step_b' ---
#   LangGraph treats the update as if step_b had already run
#   Next is now: ('step_final',)
#   step_b_result: B manual result (human replaced step_b)
#
# --- Resuming: step_final runs directly ---
#
# Final: Final: B manual result (human replaced step_b)
# Log: ['step_a', 'step_final']

as_node="step_b" tells LangGraph: "this update is as if step_b had run." So the graph moves on to the node after step_b, which is step_final. Without as_node, the graph would still want to run step_b.

Practical uses of as_node:

  • ✅ Completely replace a node's execution with a manual value
  • ✅ Skip a node you don't need (you supply its output directly)
  • ✅ Redirect the flow: make the graph think it came from a different node

Editing messages in the conversation

When the state holds messages (as in chatbots or agents with MessagesState), you can edit them during a pause:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    context: str
    result: str

def gather_context(state: State) -> dict:
    last_msg = state["messages"][-1].content
    context = f"Context gathered for: '{last_msg}'"
    return {"context": context}

def respond(state: State) -> dict:
    response = f"Response based on context: {state['context']}"
    return {
        "messages": [AIMessage(content=response)],
        "result": response,
    }

graph_builder = StateGraph(State)
graph_builder.add_node("gather_context", gather_context)
graph_builder.add_node("respond", respond)

graph_builder.add_edge(START, "gather_context")
graph_builder.add_edge("gather_context", "respond")
graph_builder.add_edge("respond", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["gather_context"],
)

config = {"configurable": {"thread_id": "edit-messages"}}

graph.invoke(
    {"messages": [HumanMessage(content="Research LangChain")],
     "context": "", "result": ""},
    config,
)

state = graph.get_state(config)
print(f"Current context: {state.values['context']}")
print(f"Messages: {len(state.values['messages'])}")

print("\n--- Human adds a message with extra instructions ---\n")

graph.update_state(config, {
    "messages": [HumanMessage(content="NOTE: focus on LangGraph, not LangChain in general")],
    "context": "Context gathered for: 'Research LangChain'. HUMAN NOTE: focus on LangGraph",
})

state = graph.get_state(config)
print(f"Updated context: {state.values['context']}")
print(f"Messages: {len(state.values['messages'])}")
for msg in state.values["messages"]:
    print(f"  [{msg.type}] {msg.content[:60]}...")

print("\n--- Resuming with enriched context ---\n")
graph.invoke(None, config)

final = graph.get_state(config)
print(f"Result: {final.values['result']}")
# Expected output:
# Current context: Context gathered for: 'Research LangChain'
# Messages: 1
#
# --- Human adds a message with extra instructions ---
#
# Updated context: Context gathered for: 'Research LangChain'. HUMAN NOTE: focus on LangGraph
# Messages: 2
# Messages:
#   [human] Research LangChain...
#   [human] NOTE: focus on LangGraph, not LangChain in general...
#
# --- Resuming with enriched context ---
#
# Result: Response based on context: Context gathered for: 'Research LangChain'. HUMAN NOTE: focus on LangGraph

Because messages uses operator.add, the new message is appended to the existing list. The agent now has both the original instruction and the human's correction.

Comparison: restarting vs editing state

So it's clear when each approach makes sense:

AspectRestart from scratchEdit state and continue
Prior workAll lostPreserved
API costEverything repeatsOnly from the pause onward
TimeMinutes (re-run the pipeline)Seconds (edit + continue)
The agent's correct dataRecomputedKept
When to useFundamental design errorOne specific wrong data point
ComplexitySimple (a new invoke)Requires understanding the state

Rule of thumb:

  • ✅ Edit when the agent got 80%+ of the work right and you only need to fix one field
  • ✅ Restart when the error is so large that it invalidates all the prior work
  • ⚠️ Edit carefully when fields depend on each other (if you change X, is Y still valid?)

Validating edits before continuing

In a production flow, you want to verify the edit is valid before continuing:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class OrderState(TypedDict):
    customer: str
    items: Annotated[list[str], operator.add]
    total: float
    discount_pct: float
    final_total: float
    status: str

def calculate_order(state: OrderState) -> dict:
    total = len(state["items"]) * 29.99
    return {"total": total, "status": "calculated"}

def apply_discount(state: OrderState) -> dict:
    final = state["total"] * (1 - state["discount_pct"] / 100)
    return {"final_total": round(final, 2), "status": "discount_applied"}

def process_order(state: OrderState) -> dict:
    return {"status": f"processed_for_{state['customer']}_total_{state['final_total']}"}

graph_builder = StateGraph(OrderState)
graph_builder.add_node("calculate", calculate_order)
graph_builder.add_node("discount", apply_discount)
graph_builder.add_node("process", process_order)

graph_builder.add_edge(START, "calculate")
graph_builder.add_edge("calculate", "discount")
graph_builder.add_edge("discount", "process")
graph_builder.add_edge("process", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["calculate"],
)

config = {"configurable": {"thread_id": "validate-edit"}}

graph.invoke(
    {"customer": "ABC Corp", "items": ["AI License", "Premium Support", "Training"],
     "total": 0, "discount_pct": 10, "final_total": 0, "status": ""},
    config,
)

state = graph.get_state(config)
print(f"Calculated total: ${state.values['total']:.2f}")
print(f"Configured discount: {state.values['discount_pct']}%")

print("\n--- Human wants to change the discount to 25% ---\n")

new_discount = 25.0

if not (0 <= new_discount <= 50):
    print(f"❌ Discount {new_discount}% outside the allowed range (0-50%)")
else:
    print(f"✅ Discount {new_discount}% is valid, applying...")
    graph.update_state(config, {"discount_pct": new_discount})

    state = graph.get_state(config)
    print(f"  Discount updated: {state.values['discount_pct']}%")

    print("\n--- Resuming with the modified discount ---\n")
    graph.invoke(None, config)

    final = graph.get_state(config)
    print(f"Original total: ${final.values['total']:.2f}")
    print(f"Discount: {final.values['discount_pct']}%")
    print(f"Final total: ${final.values['final_total']:.2f}")
    print(f"Status: {final.values['status']}")
# Expected output:
# Calculated total: $89.97
# Configured discount: 10%
#
# --- Human wants to change the discount to 25% ---
#
# ✅ Discount 25.0% is valid, applying...
#   Discount updated: 25.0%
#
# --- Resuming with the modified discount ---
#
# Original total: $89.97
# Discount: 25.0%
# Final total: $67.48
# Status: processed_for_ABC Corp_total_67.48

Validating before applying the edit keeps you out of invalid states. In production, you could validate types, ranges, relationships between fields, and any business constraint.

Removing irrelevant data before continuing

Another practical case: the agent collected data but some of it is irrelevant. Instead of letting the agent process it, you strip it out:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    query: str
    results: list[dict]
    filtered_results: list[dict]
    summary: str

def search(state: State) -> dict:
    return {"results": [
        {"source": "arXiv", "data": "Relevant paper on AI agents", "score": 0.95},
        {"source": "Wikipedia", "data": "General article about AI", "score": 0.80},
        {"source": "Reddit", "data": "Meme about robots", "score": 0.15},
        {"source": "Blog spam", "data": "Buy my AI course!!!", "score": 0.05},
        {"source": "IEEE", "data": "Technical study on LLM safety", "score": 0.88},
    ]}

def process(state: State) -> dict:
    results = state.get("filtered_results") or state["results"]
    sources = [r["source"] for r in results]
    return {"summary": f"Analysis of {len(results)} sources: {', '.join(sources)}"}

graph_builder = StateGraph(State)
graph_builder.add_node("search", search)
graph_builder.add_node("process", process)

graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "process")
graph_builder.add_edge("process", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["search"],
)

config = {"configurable": {"thread_id": "filter-results"}}

graph.invoke(
    {"query": "AI agent safety", "results": [], "filtered_results": [], "summary": ""},
    config,
)

state = graph.get_state(config)
print("Agent results:")
for r in state.values["results"]:
    flag = "⚠️" if r["score"] < 0.5 else "✅"
    print(f"  {flag} {r['source']} (score: {r['score']}): {r['data'][:40]}")

print("\n--- Human filters out irrelevant results ---\n")

good_results = [r for r in state.values["results"] if r["score"] >= 0.5]
print(f"Keeping {len(good_results)} of {len(state.values['results'])} results:")
for r in good_results:
    print(f"  ✅ {r['source']}")

graph.update_state(config, {"filtered_results": good_results})

print("\n--- Resuming with the filtered results ---\n")
graph.invoke(None, config)

final = graph.get_state(config)
print(f"Summary: {final.values['summary']}")
# Expected output:
# Agent results:
#   ✅ arXiv (score: 0.95): Relevant paper on AI agents
#   ✅ Wikipedia (score: 0.80): General article about AI
#   ⚠️ Reddit (score: 0.15): Meme about robots
#   ⚠️ Blog spam (score: 0.05): Buy my AI course!!!
#   ✅ IEEE (score: 0.88): Technical study on LLM safety
#
# --- Human filters out irrelevant results ---
#
# Keeping 3 of 5 results:
#   ✅ arXiv
#   ✅ Wikipedia
#   ✅ IEEE
#
# --- Resuming with the filtered results ---
#
# Summary: Analysis of 3 sources: arXiv, Wikipedia, IEEE

The human removed Reddit and Blog spam before the agent processed them. The final summary only includes quality sources.

Changing the direction of the research

The most powerful case: the agent is investigating a topic, but you want to redirect it midway:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    original_query: str
    active_query: str
    research_data: Annotated[list[str], operator.add]
    analysis: str
    report: str

def initial_research(state: State) -> dict:
    query = state["active_query"] or state["original_query"]
    return {
        "research_data": [
            f"Data point 1 on '{query}': general landscape",
            f"Data point 2 on '{query}': current trends",
        ],
    }

def deep_analysis(state: State) -> dict:
    query = state["active_query"] or state["original_query"]
    data_summary = f"{len(state['research_data'])} data points on '{query}'"
    return {"analysis": f"Deep analysis: {data_summary}"}

def final_report(state: State) -> dict:
    return {"report": f"REPORT: {state['analysis']}"}

graph_builder = StateGraph(State)
graph_builder.add_node("initial_research", initial_research)
graph_builder.add_node("deep_analysis", deep_analysis)
graph_builder.add_node("final_report", final_report)

graph_builder.add_edge(START, "initial_research")
graph_builder.add_edge("initial_research", "deep_analysis")
graph_builder.add_edge("deep_analysis", "final_report")
graph_builder.add_edge("final_report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["initial_research"],
)

config = {"configurable": {"thread_id": "redirect-research"}}

graph.invoke(
    {"original_query": "AI market overview", "active_query": "",
     "research_data": [], "analysis": "", "report": ""},
    config,
)

state = graph.get_state(config)
print(f"Original query: {state.values['original_query']}")
print(f"Data points collected: {len(state.values['research_data'])}")
for d in state.values["research_data"]:
    print(f"  - {d}")

print("\n--- Human redirects: 'focus on AI safety, not market overview' ---\n")

graph.update_state(config, {
    "active_query": "AI safety regulations",
    "research_data": ["HUMAN: Redirected from 'AI market overview' to 'AI safety regulations'"],
})

state = graph.get_state(config)
print(f"Active query now: {state.values['active_query']}")
print(f"Data (preserved + redirect note): {len(state.values['research_data'])}")

print("\n--- Resuming with the new direction ---\n")
graph.invoke(None, config)

final = graph.get_state(config)
print(f"Report: {final.values['report']}")
# Expected output:
# Original query: AI market overview
# Data points collected: 2
#   - Data point 1 on 'AI market overview': general landscape
#   - Data point 2 on 'AI market overview': current trends
#
# --- Human redirects: 'focus on AI safety, not market overview' ---
#
# Active query now: AI safety regulations
# Data (preserved + redirect note): 3
#
# --- Resuming with the new direction ---
#
# Report: REPORT: Deep analysis: 3 data points on 'AI safety regulations'

The human changed the direction of the research without losing the data already collected. The deep analysis and the report are generated with the new focus.

Troubleshooting

Problem 1: "update_state doesn't change the value"

Symptom: You call update_state but the field still holds the original value.

Cause: The field uses a reducer (operator.add). The new value is appended, not replaced.

Fix: For lists with operator.add, the value is appended. If you want to replace the whole list, you need a field without a reducer, or you design your state with a separate field:

# If tags uses operator.add:
graph.update_state(config, {"tags": ["new"]})
# Result: tags = ["existing", "new"]  (it's appended)

# To replace, use a field with no reducer:
class State(TypedDict):
    filtered_tags: list[str]  # no operator.add — it gets replaced

Problem 2: "After update_state, the wrong node runs"

Symptom: You edit the state but the graph runs a node you didn't expect.

Cause: update_state without as_node doesn't change which node comes next.

Fix: Use as_node to indicate which node it should continue from:

graph.update_state(config, {"data": "new"}, as_node="step_b")

Problem 3: "The edited state doesn't persist after resuming"

Symptom: You edit a field, resume, but the next node overwrites your edit.

Cause: The next node writes to the same field you edited.

Fix: Check which fields each node writes. If you edit ceo and the next node also writes ceo, your edit will be lost. Place the breakpoint after the node that writes to that field.

Problem 4: "update_state fails with a type error"

Symptom: Error when calling update_state with a value of the wrong type.

Cause: The value doesn't match the type declared in the TypedDict.

Fix: Respect the state's types:

class State(TypedDict):
    count: int
    name: str

# ❌ Wrong type
graph.update_state(config, {"count": "ten"})

# ✅ Correct type
graph.update_state(config, {"count": 10})

Exercises

Exercise 1: Inspect and edit a simple field (Easy)

Build a 2-node graph: fetch_data (gets a product name like "ProdcutX" — with an intentional typo) and create_listing (generates a listing). Pause after fetch_data, inspect the typo, fix it to "ProductX", and continue.

See solution 
from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    product_id: str
    product_name: str
    listing: str

def fetch_data(state: State) -> dict:
    return {"product_name": "ProdcutX"}

def create_listing(state: State) -> dict:
    return {"listing": f"LISTING: {state['product_name']} — Available now"}

graph_builder = StateGraph(State)
graph_builder.add_node("fetch_data", fetch_data)
graph_builder.add_node("create_listing", create_listing)

graph_builder.add_edge(START, "fetch_data")
graph_builder.add_edge("fetch_data", "create_listing")
graph_builder.add_edge("create_listing", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["fetch_data"],
)

config = {"configurable": {"thread_id": "typo-fix"}}
graph.invoke({"product_id": "P001", "product_name": "", "listing": ""}, config)

state = graph.get_state(config)
print(f"Name retrieved: '{state.values['product_name']}' ← typo!")

graph.update_state(config, {"product_name": "ProductX"})

state = graph.get_state(config)
print(f"Name corrected: '{state.values['product_name']}'")

graph.invoke(None, config)

final = graph.get_state(config)
print(f"Listing: {final.values['listing']}")
# Expected output:
# Name retrieved: 'ProdcutX' ← typo!
# Name corrected: 'ProductX'
# Listing: LISTING: ProductX — Available now

Exercise 2: Add data to a list with a reducer (Easy)

Build a graph where a collect_sources node gathers 2 sources. Pause, add 2 more sources as the human (using update_state with a list — remember that operator.add appends them). Verify that the summarize node works with all 4 sources.

See solution 
from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    topic: str
    sources: Annotated[list[str], operator.add]
    summary: str

def collect_sources(state: State) -> dict:
    return {"sources": [
        f"Source 1: Wikipedia on {state['topic']}",
        f"Source 2: arXiv on {state['topic']}",
    ]}

def summarize(state: State) -> dict:
    return {"summary": f"Summary based on {len(state['sources'])} sources"}

graph_builder = StateGraph(State)
graph_builder.add_node("collect", collect_sources)
graph_builder.add_node("summarize", summarize)

graph_builder.add_edge(START, "collect")
graph_builder.add_edge("collect", "summarize")
graph_builder.add_edge("summarize", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["collect"],
)

config = {"configurable": {"thread_id": "add-sources"}}
graph.invoke({"topic": "HITL patterns", "sources": [], "summary": ""}, config)

state = graph.get_state(config)
print(f"Agent sources: {len(state.values['sources'])}")

graph.update_state(config, {
    "sources": [
        "Source 3: Internal paper (not public)",
        "Source 4: Meeting notes with the LangGraph team",
    ],
})

state = graph.get_state(config)
print(f"Sources after adding: {len(state.values['sources'])}")
for s in state.values["sources"]:
    print(f"  - {s}")

graph.invoke(None, config)

final = graph.get_state(config)
print(f"\n{final.values['summary']}")
assert "4 sources" in final.values["summary"]
print("✅ All 4 sources were processed")
# Expected output:
# Agent sources: 2
# Sources after adding: 4
#   - Source 1: Wikipedia on HITL patterns
#   - Source 2: arXiv on HITL patterns
#   - Source 3: Internal paper (not public)
#   - Source 4: Meeting notes with the LangGraph team
#
# Summary based on 4 sources
# ✅ All 4 sources were processed

Exercise 3: Use as_node to skip a node (Medium)

Build a 3-node graph: fetch → validate → save. Set a breakpoint before validate. Instead of letting validate run, use update_state with as_node="validate" to supply the validation result directly and jump to the save node.

See solution 
from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    data: str
    validated: bool
    validation_msg: str
    saved: bool
    log: Annotated[list[str], operator.add]

def fetch(state: State) -> dict:
    return {"data": "raw data from the API", "log": ["fetched"]}

def validate(state: State) -> dict:
    is_valid = len(state["data"]) > 5
    return {
        "validated": is_valid,
        "validation_msg": "OK" if is_valid else "Insufficient data",
        "log": ["validated_by_node"],
    }

def save(state: State) -> dict:
    if state["validated"]:
        return {"saved": True, "log": ["saved"]}
    return {"saved": False, "log": ["save_skipped"]}

graph_builder = StateGraph(State)
graph_builder.add_node("fetch", fetch)
graph_builder.add_node("validate", validate)
graph_builder.add_node("save", save)

graph_builder.add_edge(START, "fetch")
graph_builder.add_edge("fetch", "validate")
graph_builder.add_edge("validate", "save")
graph_builder.add_edge("save", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["validate"],
)

config = {"configurable": {"thread_id": "skip-node"}}
graph.invoke(
    {"data": "", "validated": False, "validation_msg": "", "saved": False, "log": []},
    config,
)

state = graph.get_state(config)
print(f"Data: {state.values['data']}")
print(f"Next: {state.next}")

print("\n--- Human validates manually and skips the validate node ---\n")

graph.update_state(
    config,
    {"validated": True, "validation_msg": "Validated manually by a human", "log": ["validated_by_human"]},
    as_node="validate",
)

state = graph.get_state(config)
print(f"Validated: {state.values['validated']}")
print(f"Next (skipped validate): {state.next}")

graph.invoke(None, config)

final = graph.get_state(config)
print(f"Saved: {final.values['saved']}")
print(f"Log: {final.values['log']}")
# Expected output:
# Data: raw data from the API
# Next: ('validate',)
#
# --- Human validates manually and skips the validate node ---
#
# Validated: True
# Next (skipped validate): ('save',)
# Saved: True
# Log: ['fetched', 'validated_by_human', 'saved']

Exercise 4: Fix multiple fields at once (Medium)

Build a research graph that collects company data (name, CEO, founding year, HQ). Introduce 2 intentional errors: wrong CEO and wrong founding year. Pause, fix both fields in a single update_state call, and verify that the final report uses the correct data.

See solution 
from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    company: str
    ceo: str
    founded: int
    hq: str
    report: str

def research(state: State) -> dict:
    return {
        "ceo": "Mark Thompson",
        "founded": 2015,
        "hq": "Austin, TX",
    }

def generate_report(state: State) -> dict:
    report = f"{state['company']}: CEO {state['ceo']}, founded in {state['founded']}, HQ in {state['hq']}"
    return {"report": report}

graph_builder = StateGraph(State)
graph_builder.add_node("research", research)
graph_builder.add_node("report", generate_report)

graph_builder.add_edge(START, "research")
graph_builder.add_edge("research", "report")
graph_builder.add_edge("report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["research"],
)

config = {"configurable": {"thread_id": "multi-edit"}}
graph.invoke(
    {"company": "InnovateTech", "ceo": "", "founded": 0, "hq": "", "report": ""},
    config,
)

state = graph.get_state(config)
print(f"Agent data:")
print(f"  CEO: {state.values['ceo']} ← INCORRECT")
print(f"  Founded: {state.values['founded']} ← INCORRECT")
print(f"  HQ: {state.values['hq']} ← correct")

print("\n--- Fixing CEO and founding year ---\n")

graph.update_state(config, {
    "ceo": "Sarah Chen",
    "founded": 2019,
})

state = graph.get_state(config)
print(f"Corrected data:")
print(f"  CEO: {state.values['ceo']}")
print(f"  Founded: {state.values['founded']}")
print(f"  HQ: {state.values['hq']} (unchanged)")

graph.invoke(None, config)

final = graph.get_state(config)
print(f"\nReport: {final.values['report']}")

assert "Sarah Chen" in final.values["report"]
assert "2019" in final.values["report"]
assert "Austin, TX" in final.values["report"]
print("✅ Report with corrected data")
# Expected output:
# Agent data:
#   CEO: Mark Thompson ← INCORRECT
#   Founded: 2015 ← INCORRECT
#   HQ: Austin, TX ← correct
#
# --- Fixing CEO and founding year ---
#
# Corrected data:
#   CEO: Sarah Chen
#   Founded: 2019
#   HQ: Austin, TX (unchanged)
#
# Report: InnovateTech: CEO Sarah Chen, founded in 2019, HQ in Austin, TX
# ✅ Report with corrected data

Exercise 5: Edit state + redirect with as_node (Advanced)

Build an analysis pipeline: collect → analyze → recommend → deliver. Set interrupt_after=["analyze"]. After the analysis, the human decides it's wrong and wants it redone. Use update_state with as_node="collect" to inject new data and make the graph re-run from analyze.

See solution 
from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    query: str
    raw_data: str
    analysis: str
    recommendation: str
    deliverable: str
    log: Annotated[list[str], operator.add]

def collect(state: State) -> dict:
    return {"raw_data": f"Basic data on '{state['query']}'", "log": ["collected"]}

def analyze(state: State) -> dict:
    return {"analysis": f"Analysis: {state['raw_data'][:40]}", "log": ["analyzed"]}

def recommend(state: State) -> dict:
    return {"recommendation": f"Recommendation based on: {state['analysis'][:30]}", "log": ["recommended"]}

def deliver(state: State) -> dict:
    return {"deliverable": f"DELIVERY: {state['recommendation']}", "log": ["delivered"]}

graph_builder = StateGraph(State)
graph_builder.add_node("collect", collect)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("recommend", recommend)
graph_builder.add_node("deliver", deliver)

graph_builder.add_edge(START, "collect")
graph_builder.add_edge("collect", "analyze")
graph_builder.add_edge("analyze", "recommend")
graph_builder.add_edge("recommend", "deliver")
graph_builder.add_edge("deliver", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["analyze"],
)

config = {"configurable": {"thread_id": "redo-analysis"}}

graph.invoke(
    {"query": "AI investment strategy", "raw_data": "", "analysis": "",
     "recommendation": "", "deliverable": "", "log": []},
    config,
)

state = graph.get_state(config)
print(f"Current analysis: {state.values['analysis']}")
print(f"Next: {state.next}")

print("\n--- Human: 'The data is insufficient, I'm injecting better data' ---\n")

graph.update_state(
    config,
    {
        "raw_data": "PREMIUM data: analysis of 50 companies, financial metrics, market forecasts",
        "log": ["human_injected_premium_data"],
    },
    as_node="collect",
)

state = graph.get_state(config)
print(f"Next after as_node='collect': {state.next}")
print(f"Raw data updated: {state.values['raw_data'][:60]}...")

print("\n--- Resuming: analyze re-runs with the premium data ---\n")
graph.invoke(None, config)

final = graph.get_state(config)
print(f"New analysis: {final.values['analysis']}")
print(f"Recommendation: {final.values['recommendation']}")
print(f"Deliverable: {final.values['deliverable']}")
print(f"Log: {final.values['log']}")
# Expected output:
# Current analysis: Analysis: Basic data on 'AI investment st
# Next: ('recommend',)
#
# --- Human: 'The data is insufficient, I'm injecting better data' ---
#
# Next after as_node='collect': ('analyze',)
# Raw data updated: PREMIUM data: analysis of 50 companies, financial metrics...
#
# --- Resuming: analyze re-runs with the premium data ---
#
# New analysis: Analysis: PREMIUM data: analysis of 50 compa
# Recommendation: Recommendation based on: Analysis: PREMIUM data: a
# Deliverable: DELIVERY: Recommendation based on: Analysis: PREMIUM data: a
# Log: ['collected', 'analyzed', 'human_injected_premium_data', 'analyzed', 'recommended', 'delivered']

Exercise 6: Full HITL flow with inspection, editing and approval (Advanced)

Build a research agent with 4 nodes: plan → research → review → report. The review node uses interrupt() to ask the human for feedback. Implement the full flow: (1) the agent plans and runs the research, (2) it pauses at review, (3) the human inspects the data and spots an error in the research, (4) edits the state to fix the error, and (5) approves so the report gets generated.

See solution 
from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    objective: str
    plan: str
    findings: Annotated[list[dict], operator.add]
    corrections_applied: str
    report: str
    log: Annotated[list[str], operator.add]

def plan(state: State) -> dict:
    return {
        "plan": f"Plan for '{state['objective']}': search 3 sources, analyze trends",
        "log": ["planned"],
    }

def research(state: State) -> dict:
    return {"findings": [
        {"source": "arXiv", "fact": "The AI market will grow 35% in 2025", "confidence": 0.9},
        {"source": "Bloomberg", "fact": "TechCorp trades at $150/share", "confidence": 0.95},
        {"source": "Blog", "fact": "TechCorp's CEO is Robert Lee", "confidence": 0.6},
    ], "log": ["researched"]}

def review(state: State) -> dict:
    decision = interrupt({
        "message": "Review the research findings",
        "findings": state["findings"],
        "instructions": "Reply {approved: true} after making corrections if needed",
    })
    return {"log": ["reviewed"]}

def report(state: State) -> dict:
    lines = [f"REPORT: {state['objective']}"]
    lines.append(f"Plan: {state['plan']}")
    if state["corrections_applied"]:
        lines.append(f"Corrections: {state['corrections_applied']}")
    for f in state["findings"]:
        lines.append(f"  [{f['source']}] {f['fact']} (confidence: {f['confidence']})")
    return {"report": "\n".join(lines), "log": ["reported"]}

graph_builder = StateGraph(State)
graph_builder.add_node("plan", plan)
graph_builder.add_node("research", research)
graph_builder.add_node("review", review)
graph_builder.add_node("report", report)

graph_builder.add_edge(START, "plan")
graph_builder.add_edge("plan", "research")
graph_builder.add_edge("research", "review")
graph_builder.add_edge("review", "report")
graph_builder.add_edge("report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "full-hitl"}}

print("=== STEP 1: Plan + Research + Pause at review ===\n")
graph.invoke(
    {"objective": "TechCorp analysis for investment", "plan": "",
     "findings": [], "corrections_applied": "", "report": "", "log": []},
    config,
)

state = graph.get_state(config)
print(f"Waiting at: {state.next}")
print(f"Findings:")
for f in state.values["findings"]:
    flag = "⚠️" if f["confidence"] < 0.8 else "✅"
    print(f"  {flag} [{f['source']}] {f['fact']} (conf: {f['confidence']})")

print("\n=== STEP 2: Human spots an error and corrects it ===\n")

corrected_findings = []
for f in state.values["findings"]:
    if f["source"] == "Blog" and "Robert Lee" in f["fact"]:
        corrected = {**f, "fact": "TechCorp's CEO is Amanda Reyes (since January 2025)", "confidence": 1.0}
        corrected_findings.append(corrected)
        print(f"  Corrected: '{f['fact']}' → '{corrected['fact']}'")
    else:
        corrected_findings.append(f)

graph.update_state(config, {
    "findings": corrected_findings,
    "corrections_applied": "CEO corrected: Robert Lee → Amanda Reyes",
})

print("\n=== STEP 3: Approve and generate the report ===\n")
graph.invoke(Command(resume={"approved": True}), config)

final = graph.get_state(config)
print(final.values["report"])
print(f"\nLog: {final.values['log']}")
# Expected output:
# === STEP 1: Plan + Research + Pause at review ===
#
# Waiting at: ('review',)
# Findings:
#   ✅ [arXiv] The AI market will grow 35% in 2025 (conf: 0.9)
#   ✅ [Bloomberg] TechCorp trades at $150/share (conf: 0.95)
#   ⚠️ [Blog] TechCorp's CEO is Robert Lee (conf: 0.6)
#
# === STEP 2: Human spots an error and corrects it ===
#
#   Corrected: 'TechCorp's CEO is Robert Lee' → 'TechCorp's CEO is Amanda Reyes (since January 2025)'
#
# === STEP 3: Approve and generate the report ===
#
# REPORT: TechCorp analysis for investment
# Plan: Plan for 'TechCorp analysis for investment': search 3 sources, analyze trends
# Corrections: CEO corrected: Robert Lee → Amanda Reyes
#   [arXiv] The AI market will grow 35% in 2025 (confidence: 0.9)
#   [Bloomberg] TechCorp trades at $150/share (confidence: 0.95)
#   [Blog] TechCorp's CEO is Amanda Reyes (since January 2025) (confidence: 1.0)
#
# Log: ['planned', 'researched', 'reviewed', 'reported']

Summary

In this capsule you learned:

  • Editable state lets you modify the agent's state during a pause. You don't need to restart: you fix the wrong data point and the agent continues with correct information, preserving all the prior work
  • graph.get_state(config) gives you full access to the current state: every value, which node comes next, and the checkpoint ID. Use it to inspect before you edit
  • graph.update_state(config, updates) modifies the state following the reducer rules: plain fields are replaced, fields with operator.add are appended. Understanding this is essential to predicting the result of your edits
  • as_node controls which node comes next. update_state(config, data, as_node="X") makes LangGraph treat the edit as if node X produced it, which determines the next node in the graph
  • The use cases are practical, not theoretical: fixing wrong data, adding information the agent couldn't find, removing irrelevant results, changing the direction of the research
  • Editing beats restarting when the agent got most of the work right. You preserve searches, analysis, and correct data — you only fix what's wrong
  • Validate before editing in production: check types, ranges, and business constraints before applying update_state

Next capsule: UX design for HITL — when to interrupt, when not to, and how to avoid frustrating the user with too many pauses. The balance between autonomy and human control.

Additional resources

  1. LangGraph — Human-in-the-Loop — Official HITL concepts including editable state
  2. How to edit graph state — Practical guide to inspecting and modifying state during pauses
  3. How to update graph state — Full reference for update_state with reducers and as_node
  4. LangGraph — State Reducers — How reducers work and their impact on update_state
  5. How to wait for user input — Combining editable state with interrupt() for interactive flows

Module 9 — LangChain & LangGraph: From Chains to Agents

← PreviousBreakpoints and ApprovalsNext →User Feedback Loops