Module 9: Human-in-the-Loop

User Feedback Loops

Capsule overview

So far, HITL has been binary: the human approves or rejects. But the most valuable interaction isn't "yes/no" — it's iterative feedback. "This summary is too technical. Simplify it." The agent rewrites. "Better, but add more examples." The agent adds examples. This refinement cycle produces results that neither the human nor the AI could reach alone.

The pattern is simple but powerful: Review → Feedback → Revise → Review again → Approve. Each round of feedback accumulates in the graph state, and the agent uses the entire history of corrections to improve its next attempt. It doesn't start from zero — it learns from what you already told it.

This turns the agent from a "first draft" generator into a collaborator that refines its work with you. And the best part: the mechanics are a direct extension of the interrupts and editable state you already know.

The pattern: Review → Feedback → Revise → Approve

Picture the full flow:

generate → interrupt (show output to the user)
                ↓
        What does the user say?
        ├── "approved" → move on to the next step
        └── "too technical" → loop back to generate
                                    (with the feedback as extra context)
                                    → interrupt (show the new version)
                                    ↓
                            What does the user say?
                            ├── "approved" → continue
                            └── "add examples" → loop back...

Each iteration of the loop is a checkpoint. If the process crashes mid-revision, the agent resumes exactly where it was — with all the accumulated feedback.

Implementing a basic feedback loop

The simplest feedback loop: a node generates content, an interrupt shows it to the user, and depending on the response, the graph continues or regenerates:

from dotenv import load_dotenv
load_dotenv()

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

class FeedbackState(TypedDict):
    topic: str
    draft: str
    feedback: Annotated[list[str], operator.add]
    iteration: int
    max_iterations: int
    approved: bool

def generate_summary(state: FeedbackState) -> dict:
    """Generates or regenerates a summary incorporating previous feedback."""
    topic = state["topic"]
    feedback_history = state.get("feedback", [])
    iteration = state.get("iteration", 0)

    if iteration == 0:
        draft = (
            f"Technical summary on '{topic}': "
            f"Language models use transformer architectures with multi-head "
            f"attention mechanisms to process token sequences. The cross-entropy "
            f"loss function optimizes the probability distribution over "
            f"the vocabulary. Fine-tuning with RLHF aligns the model's outputs "
            f"with human preferences through a learned reward function."
        )
    else:
        latest_feedback = feedback_history[-1] if feedback_history else ""
        all_feedback = " | ".join(feedback_history)

        if "simplify" in latest_feedback.lower() or "technical" in latest_feedback.lower():
            draft = (
                f"Summary on '{topic}': "
                f"Language models are programs that learn to write text "
                f"by reading millions of documents. They work by predicting the most likely "
                f"next word. After training, they're tuned with human feedback so "
                f"their answers are more useful and safe."
            )
        elif "example" in latest_feedback.lower():
            draft = (
                f"Summary on '{topic}': "
                f"Language models learn to write text by predicting the "
                f"next word. For example, if you give them 'The cat sat on the...', "
                f"they predict 'couch' or 'floor'. Then they're tuned with human feedback — "
                f"like an editor reviewing drafts. Example: ChatGPT uses this process "
                f"to answer questions conversationally."
            )
        else:
            draft = (
                f"Revised summary on '{topic}' "
                f"(iteration {iteration + 1}, incorporating: {all_feedback}): "
                f"Improved version of the content based on all the corrections received."
            )

    return {
        "draft": draft,
        "iteration": iteration + 1,
    }

def review_draft(state: FeedbackState) -> dict:
    """Shows the draft to the user and waits for feedback."""
    iteration = state["iteration"]
    draft = state["draft"]

    response = interrupt({
        "action": "review_draft",
        "iteration": iteration,
        "draft": draft,
        "instruction": "Reply 'approved' to accept, or write your feedback to request changes.",
        "feedback_so_far": state.get("feedback", []),
    })

    if response.lower() == "approved":
        return {"approved": True}

    return {
        "feedback": [response],
        "approved": False,
    }

def should_continue(state: FeedbackState) -> str:
    if state.get("approved", False):
        return "deliver"

    if state["iteration"] >= state["max_iterations"]:
        return "deliver"

    return "revise"

def deliver(state: FeedbackState) -> dict:
    """Delivers the final result."""
    return {}

graph_builder = StateGraph(FeedbackState)
graph_builder.add_node("generate", generate_summary)
graph_builder.add_node("review", review_draft)
graph_builder.add_node("deliver", deliver)

graph_builder.add_edge(START, "generate")
graph_builder.add_edge("generate", "review")
graph_builder.add_conditional_edges(
    "review", should_continue,
    {"revise": "generate", "deliver": "deliver"},
)
graph_builder.add_edge("deliver", END)

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

config = {"configurable": {"thread_id": "feedback-loop-001"}}

print("=== Round 1: initial generation ===")
result = graph.invoke(
    {
        "topic": "language models",
        "draft": "",
        "feedback": [],
        "iteration": 0,
        "max_iterations": 3,
        "approved": False,
    },
    config,
)
state = graph.get_state(config)
print(f"Draft: {state.values['draft'][:80]}...")
print(f"Iteration: {state.values['iteration']}")

print("\n=== Round 1: feedback — 'too technical' ===")
result = graph.invoke(
    Command(resume="Too technical. Simplify it for someone with no ML background."),
    config,
)
state = graph.get_state(config)
print(f"Draft: {state.values['draft'][:80]}...")
print(f"Iteration: {state.values['iteration']}")
print(f"Accumulated feedback: {state.values['feedback']}")

print("\n=== Round 2: feedback — 'add examples' ===")
result = graph.invoke(
    Command(resume="Better, but add concrete examples."),
    config,
)
state = graph.get_state(config)
print(f"Draft: {state.values['draft'][:80]}...")
print(f"Iteration: {state.values['iteration']}")
print(f"Accumulated feedback: {state.values['feedback']}")

print("\n=== Round 3: approved ===")
result = graph.invoke(Command(resume="approved"), config)
state = graph.get_state(config)
print(f"Final draft: {state.values['draft']}")
print(f"Approved: {state.values['approved']}")
print(f"Total iterations: {state.values['iteration']}")
print(f"Feedback history: {state.values['feedback']}")
# Expected output:
# === Round 1: initial generation ===
# Draft: Technical summary on 'language models': Language models use transformer ...
# Iteration: 1
#
# === Round 1: feedback — 'too technical' ===
# Draft: Summary on 'language models': Language models are programs that learn t...
# Iteration: 2
# Accumulated feedback: ['Too technical. Simplify it for someone with no ML background.']
#
# === Round 2: feedback — 'add examples' ===
# Draft: Summary on 'language models': Language models learn to write text by pr...
# Iteration: 3
# Accumulated feedback: ['Too technical. Simplify it for someone with no ML background.', 'Better, but add concrete examples.']
#
# === Round 3: approved ===
# Final draft: Summary on 'language models': Language models learn to write text by predicting the next word. For example, if you give them 'The cat sat on the...', they predict 'couch' or 'floor'. Then they're tuned with human feedback — like an editor reviewing drafts. Example: ChatGPT uses this process to answer questions conversationally.
# Approved: True
# Total iterations: 3
# Feedback history: ['Too technical. Simplify it for someone with no ML background.', 'Better, but add concrete examples.']

Notice the mechanics:

feedback: Annotated[list[str], operator.add] — accumulates all the feedback without losing previous rounds
The generate node reads the previous feedback and adapts its output
The review node uses interrupt() to pause and show the draft
The conditional edge decides: if approved → deliver, if there's feedback → regenerate
max_iterations prevents infinite loops

Accumulating feedback in the state

The key to the pattern is that feedback doesn't replace — it accumulates. Each round adds context:

from dotenv import load_dotenv
load_dotenv()

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

class AccumulativeState(TypedDict):
    task: str
    output: str
    feedback_history: Annotated[list[dict], operator.add]
    revision_count: int

def produce_output(state: AccumulativeState) -> dict:
    """Generates output using the ENTIRE feedback history."""
    feedback_history = state.get("feedback_history", [])
    revision = state.get("revision_count", 0)

    context_parts = [f"Task: {state['task']}"]
    for entry in feedback_history:
        context_parts.append(
            f"Round {entry['round']}: user said '{entry['feedback']}'"
        )

    context = "\n".join(context_parts)

    if revision == 0:
        output = f"[v1] Initial proposal for: {state['task']}"
    else:
        corrections = [e["feedback"] for e in feedback_history]
        output = (
            f"[v{revision + 1}] Revised proposal for: {state['task']}. "
            f"Incorporating {len(corrections)} corrections: {', '.join(corrections)}"
        )

    return {
        "output": output,
        "revision_count": revision + 1,
    }

def collect_feedback(state: AccumulativeState) -> dict:
    response = interrupt({
        "current_output": state["output"],
        "revision": state["revision_count"],
        "history": state.get("feedback_history", []),
        "prompt": "Type 'ok' to approve or write your correction.",
    })

    if response.lower() in ("ok", "approved", "lgtm"):
        return {}

    return {
        "feedback_history": [{
            "round": state["revision_count"],
            "feedback": response,
        }],
    }

def route(state: AccumulativeState) -> str:
    history = state.get("feedback_history", [])
    if not history:
        return "finalize"

    latest_round = max(e["round"] for e in history)
    if latest_round < state["revision_count"]:
        return "finalize"

    if state["revision_count"] >= 5:
        return "finalize"

    return "revise"

def finalize(state: AccumulativeState) -> dict:
    return {}

graph_builder = StateGraph(AccumulativeState)
graph_builder.add_node("produce", produce_output)
graph_builder.add_node("collect", collect_feedback)
graph_builder.add_node("finalize", finalize)

graph_builder.add_edge(START, "produce")
graph_builder.add_edge("produce", "collect")
graph_builder.add_conditional_edges(
    "collect", route,
    {"revise": "produce", "finalize": "finalize"},
)
graph_builder.add_edge("finalize", END)

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

config = {"configurable": {"thread_id": "accumulative-001"}}

graph.invoke(
    {
        "task": "Design a landing page for an analytics SaaS",
        "output": "",
        "feedback_history": [],
        "revision_count": 0,
    },
    config,
)

state = graph.get_state(config)
print(f"v1: {state.values['output']}")

graph.invoke(
    Command(resume="The hero section needs a clearer CTA"),
    config,
)
state = graph.get_state(config)
print(f"v2: {state.values['output']}")

graph.invoke(Command(resume="ok"), config)
state = graph.get_state(config)
print(f"Final: {state.values['output']}")
print(f"History: {state.values['feedback_history']}")
# Expected output:
# v1: [v1] Initial proposal for: Design a landing page for an analytics SaaS
# v2: [v2] Revised proposal for: Design a landing page for an analytics SaaS. Incorporating 1 corrections: The hero section needs a clearer CTA
# Final: [v2] Revised proposal for: Design a landing page for an analytics SaaS. Incorporating 1 corrections: The hero section needs a clearer CTA
# History: [{'round': 1, 'feedback': 'The hero section needs a clearer CTA'}]

The feedback_history: Annotated[list[dict], operator.add] pattern, with dictionaries that include round and feedback, gives you full traceability. You can know exactly what the user asked for in each round and how the output evolved.

Feeding feedback into the prompt, not replacing the output

A common mistake: receiving feedback and generating a completely new response from scratch. The right pattern is to add the feedback as extra context so the model improves on what it already has:

from dotenv import load_dotenv
load_dotenv()

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

class WriterState(TypedDict):
    topic: str
    draft: str
    feedback: Annotated[list[str], operator.add]
    iteration: int
    approved: bool

llm = init_chat_model("openai:gpt-4.1-nano")

def write_draft(state: WriterState) -> dict:
    feedback = state.get("feedback", [])
    iteration = state.get("iteration", 0)

    if iteration == 0:
        prompt = (
            f"Write an executive summary paragraph about: {state['topic']}. "
            f"3 sentences max. Professional tone."
        )
    else:
        prompt = (
            f"Previous version:\n{state['draft']}\n\n"
            f"User feedback this round: {feedback[-1]}\n\n"
            f"Full feedback history: {feedback}\n\n"
            f"Rewrite the paragraph incorporating ALL the feedback. "
            f"Keep what was working. 3 sentences max."
        )

    response = llm.invoke(prompt)

    return {
        "draft": response.content,
        "iteration": iteration + 1,
    }

def human_review(state: WriterState) -> dict:
    response = interrupt({
        "draft": state["draft"],
        "iteration": state["iteration"],
        "feedback_history": state.get("feedback", []),
        "instruction": "'approved' to accept, or write your correction.",
    })

    if response.lower() == "approved":
        return {"approved": True}

    return {"feedback": [response], "approved": False}

def route_review(state: WriterState) -> str:
    if state.get("approved", False):
        return "done"
    if state.get("iteration", 0) >= 3:
        return "done"
    return "rewrite"

def done(state: WriterState) -> dict:
    return {}

graph_builder = StateGraph(WriterState)
graph_builder.add_node("write", write_draft)
graph_builder.add_node("review", human_review)
graph_builder.add_node("done", done)

graph_builder.add_edge(START, "write")
graph_builder.add_edge("write", "review")
graph_builder.add_conditional_edges(
    "review", route_review,
    {"rewrite": "write", "done": "done"},
)
graph_builder.add_edge("done", END)

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

config = {"configurable": {"thread_id": "writer-feedback-001"}}

print("=== Initial generation ===")
graph.invoke(
    {
        "topic": "benefits of microservices vs monoliths",
        "draft": "",
        "feedback": [],
        "iteration": 0,
        "approved": False,
    },
    config,
)
state = graph.get_state(config)
print(f"Draft v1:\n{state.values['draft']}\n")

print("=== Feedback: simplify ===")
graph.invoke(
    Command(resume="Too formal. Make it more conversational, like a blog post."),
    config,
)
state = graph.get_state(config)
print(f"Draft v2:\n{state.values['draft']}\n")

print("=== Approved ===")
graph.invoke(Command(resume="approved"), config)
state = graph.get_state(config)
print(f"Final version:\n{state.values['draft']}")
print(f"Iterations: {state.values['iteration']}")
print(f"Feedback received: {state.values['feedback']}")
# Expected output (content varies by LLM):
# === Initial generation ===
# Draft v1:
# [Professional paragraph about microservices vs monoliths]
#
# === Feedback: simplify ===
# Draft v2:
# [More conversational paragraph incorporating the feedback]
#
# === Approved ===
# Final version:
# [Final version of the paragraph]
# Iterations: 2
# Feedback received: ['Too formal. Make it more conversational, like a blog post.']

The key difference is in the rewrite prompt:

Previous version: [what it already generated]
User feedback: [the specific correction]
Full history: [every correction]
→ Rewrite incorporating ALL of it. Keep what was working.

That tells the model: "don't start from zero, improve what's there." The result is a natural evolution, not a restart.

Multi-point feedback: reviews at different stages

Feedback doesn't have to come only at the end. You can ask for feedback at multiple points in the pipeline — each one about a different aspect of the work:

from dotenv import load_dotenv
load_dotenv()

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

class ResearchState(TypedDict):
    query: str
    research_plan: str
    raw_findings: str
    final_report: str
    stage_feedback: Annotated[list[dict], operator.add]

def create_plan(state: ResearchState) -> dict:
    plan = (
        f"Research plan for '{state['query']}':\n"
        f"1. Search recent academic papers\n"
        f"2. Review official documentation\n"
        f"3. Analyze existing benchmarks\n"
        f"4. Synthesize findings into a report"
    )
    return {"research_plan": plan}

def review_plan(state: ResearchState) -> dict:
    """Feedback point 1: the research plan."""
    response = interrupt({
        "stage": "plan_review",
        "content": state["research_plan"],
        "instruction": "'ok' to continue, or suggest changes to the plan.",
    })

    if response.lower() == "ok":
        return {}

    updated_plan = f"{state['research_plan']}\n[Adjustment: {response}]"
    return {
        "research_plan": updated_plan,
        "stage_feedback": [{"stage": "plan", "feedback": response}],
    }

def execute_research(state: ResearchState) -> dict:
    findings = (
        f"Findings for '{state['query']}':\n"
        f"- Paper A: positive results in 85% of cases\n"
        f"- Paper B: limitations on small datasets\n"
        f"- Official docs: new API available in v2.0\n"
        f"- Benchmark: 40% improvement over baseline"
    )
    return {"raw_findings": findings}

def review_findings(state: ResearchState) -> dict:
    """Feedback point 2: raw findings."""
    response = interrupt({
        "stage": "findings_review",
        "content": state["raw_findings"],
        "instruction": "'ok' to synthesize, or point out what's missing or extra.",
    })

    if response.lower() == "ok":
        return {}

    updated = f"{state['raw_findings']}\n[Reviewer note: {response}]"
    return {
        "raw_findings": updated,
        "stage_feedback": [{"stage": "findings", "feedback": response}],
    }

def synthesize_report(state: ResearchState) -> dict:
    report = (
        f"Report: {state['query']}\n"
        f"Based on: {state['raw_findings'][:100]}...\n"
        f"Conclusion: The technology shows promising results with "
        f"identified areas for improvement."
    )
    return {"final_report": report}

def review_report(state: ResearchState) -> dict:
    """Feedback point 3: final report."""
    response = interrupt({
        "stage": "report_review",
        "content": state["final_report"],
        "instruction": "'approved' to deliver, or request adjustments to the report.",
    })

    if response.lower() == "approved":
        return {}

    return {
        "stage_feedback": [{"stage": "report", "feedback": response}],
    }

graph_builder = StateGraph(ResearchState)
graph_builder.add_node("plan", create_plan)
graph_builder.add_node("review_plan", review_plan)
graph_builder.add_node("research", execute_research)
graph_builder.add_node("review_findings", review_findings)
graph_builder.add_node("synthesize", synthesize_report)
graph_builder.add_node("review_report", review_report)

graph_builder.add_edge(START, "plan")
graph_builder.add_edge("plan", "review_plan")
graph_builder.add_edge("review_plan", "research")
graph_builder.add_edge("research", "review_findings")
graph_builder.add_edge("review_findings", "synthesize")
graph_builder.add_edge("synthesize", "review_report")
graph_builder.add_edge("review_report", END)

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

config = {"configurable": {"thread_id": "multi-feedback-001"}}

print("=== Stage 1: Plan review ===")
graph.invoke(
    {
        "query": "State of the art in RAG",
        "research_plan": "",
        "raw_findings": "",
        "final_report": "",
        "stage_feedback": [],
    },
    config,
)
state = graph.get_state(config)
print(f"Plan:\n{state.values['research_plan']}\n")

print("=== Feedback on the plan ===")
graph.invoke(
    Command(resume="Also add a review of popular GitHub repos"),
    config,
)
state = graph.get_state(config)
print(f"Adjusted plan:\n{state.values['research_plan']}\n")

print("=== Stage 2: Findings review ===")
graph.invoke(Command(resume="ok"), config)
state = graph.get_state(config)
print(f"Findings:\n{state.values['raw_findings']}\n")

print("=== Stage 3: Report approval ===")
graph.invoke(Command(resume="approved"), config)
state = graph.get_state(config)
print(f"Final report:\n{state.values['final_report']}")
print(f"\nFeedback by stage: {state.values['stage_feedback']}")
# Expected output:
# === Stage 1: Plan review ===
# Plan:
# Research plan for 'State of the art in RAG':
# 1. Search recent academic papers
# 2. Review official documentation
# 3. Analyze existing benchmarks
# 4. Synthesize findings into a report
#
# === Feedback on the plan ===
# Adjusted plan:
# Research plan for 'State of the art in RAG':
# 1. Search recent academic papers
# 2. Review official documentation
# 3. Analyze existing benchmarks
# 4. Synthesize findings into a report
# [Adjustment: Also add a review of popular GitHub repos]
#
# === Stage 2: Findings review ===
# Findings:
# Findings for 'State of the art in RAG':
# - Paper A: positive results in 85% of cases
# ...
#
# === Stage 3: Report approval ===
# Final report:
# Report: State of the art in RAG
# ...
#
# Feedback by stage: [{'stage': 'plan', 'feedback': 'Also add a review of popular GitHub repos'}]

Three review points, each about a different aspect:

Review point	What the user reviews	Value
Research plan	The sources and the approach	Prevents researching in the wrong direction
Raw findings	The data collected	Catches missing or irrelevant data before synthesis
Final report	The deliverable	Last chance to adjust before shipping

When feedback loops hurt

Not everything deserves a refinement loop. Too many iterations produce diminishing returns:

Iteration 1: "Too technical"        → 70% better  (huge)
Iteration 2: "Add examples"         → 20% better  (significant)
Iteration 3: "Change one word"      → 3% better   (marginal)
Iteration 4: "Hmm, maybe revert"    → -5% better  (counterproductive)

Rule of thumb: 3 rounds of feedback max. After 3, deliver with a disclaimer:

from dotenv import load_dotenv
load_dotenv()

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

MAX_FEEDBACK_ROUNDS = 3

class BoundedState(TypedDict):
    content: str
    feedback: Annotated[list[str], operator.add]
    round_num: int
    status: str

def generate(state: BoundedState) -> dict:
    round_num = state.get("round_num", 0)
    feedback = state.get("feedback", [])

    if round_num == 0:
        content = "Initial content generated by the agent."
    else:
        content = (
            f"Revised content (round {round_num + 1}). "
            f"Incorporating: {feedback[-1] if feedback else 'N/A'}"
        )

    return {"content": content, "round_num": round_num + 1}

def review(state: BoundedState) -> dict:
    round_num = state["round_num"]

    if round_num >= MAX_FEEDBACK_ROUNDS:
        print(f"  ⚠️  Reached the maximum of {MAX_FEEDBACK_ROUNDS} rounds. Delivering the current version.")
        return {"status": "max_rounds_reached"}

    response = interrupt({
        "content": state["content"],
        "round": round_num,
        "remaining_rounds": MAX_FEEDBACK_ROUNDS - round_num,
        "message": f"You have {MAX_FEEDBACK_ROUNDS - round_num} revision rounds left.",
    })

    if response.lower() in ("approved", "ok"):
        return {"status": "approved"}

    return {
        "feedback": [response],
        "status": "needs_revision",
    }

def route(state: BoundedState) -> str:
    status = state.get("status", "")
    if status in ("approved", "max_rounds_reached"):
        return "deliver"
    return "revise"

def deliver(state: BoundedState) -> dict:
    return {}

graph_builder = StateGraph(BoundedState)
graph_builder.add_node("generate", generate)
graph_builder.add_node("review", review)
graph_builder.add_node("deliver", deliver)

graph_builder.add_edge(START, "generate")
graph_builder.add_edge("generate", "review")
graph_builder.add_conditional_edges(
    "review", route,
    {"revise": "generate", "deliver": "deliver"},
)
graph_builder.add_edge("deliver", END)

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

config = {"configurable": {"thread_id": "bounded-feedback-001"}}

graph.invoke(
    {"content": "", "feedback": [], "round_num": 0, "status": ""},
    config,
)
graph.invoke(Command(resume="Make it shorter"), config)
graph.invoke(Command(resume="Add data"), config)

state = graph.get_state(config)
print(f"Current round: {state.values['round_num']}")
print(f"Status: {state.values['status']}")
print(f"Content: {state.values['content']}")
print(f"Feedback: {state.values['feedback']}")
# Expected output:
#   ⚠️  Reached the maximum of 3 rounds. Delivering the current version.
# Current round: 3
# Status: max_rounds_reached
# Content: Revised content (round 3). Incorporating: Add data
# Feedback: ['Make it shorter', 'Add data']

Protection against infinite loops is mandatory in production. Without it, an indecisive user can rack up unbounded API costs.

Troubleshooting

Problem 1: "Feedback doesn't accumulate, I only see the last one"

Symptom: After 3 rounds of feedback, state.feedback only holds the last correction.

Cause: You used feedback: str instead of feedback: Annotated[list[str], operator.add].

Fix: Use a list with operator.add to accumulate:

# ❌ Only stores the last feedback
class State(TypedDict):
    feedback: str

# ✅ Accumulates all the feedback
class State(TypedDict):
    feedback: Annotated[list[str], operator.add]

Problem 2: "The agent doesn't improve with feedback — it generates something completely different"

Symptom: Each revision throws away the good parts of the previous version.

Cause: The regeneration prompt doesn't include the previous version or the feedback history.

Fix: Include both in the prompt:

prompt = (
    f"Previous version:\n{state['draft']}\n\n"
    f"User feedback: {state['feedback'][-1]}\n\n"
    f"Feedback history: {state['feedback']}\n\n"
    f"Improve the previous version incorporating the feedback. Keep what already works."
)

Problem 3: "The feedback loop never ends"

Symptom: The user keeps giving feedback and the agent keeps regenerating indefinitely.

Cause: There's no iteration limit (max_iterations).

Fix: Add a counter and a ceiling:

def route(state):
    if state.get("approved"):
        return "deliver"
    if state["iteration"] >= state["max_iterations"]:
        return "deliver"
    return "revise"

Problem 4: "The interrupt doesn't show the user the right information"

Symptom: The user gets a pause but doesn't know what they're reviewing.

Cause: The value passed to interrupt() doesn't carry enough context.

Fix: Pass a context-rich dict:

# ❌ No context
interrupt("Waiting for feedback")

# ✅ With full context
interrupt({
    "draft": state["draft"],
    "iteration": state["iteration"],
    "feedback_so_far": state["feedback"],
    "instruction": "'approved' to accept, or write your correction.",
})

Exercises

Exercise 1: Feedback loop for titles (Easy)

Build a graph that generates a title for an article. The user can give feedback up to 3 times. The state must accumulate the feedback. Simulate the interaction without an LLM (use hardcoded strings that change with the feedback).

See solution

from dotenv import load_dotenv
load_dotenv()

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

class TitleState(TypedDict):
    topic: str
    title: str
    feedback: Annotated[list[str], operator.add]
    iteration: int
    approved: bool

def generate_title(state: TitleState) -> dict:
    iteration = state.get("iteration", 0)
    feedback = state.get("feedback", [])

    titles = [
        f"The Complete Guide to {state['topic']}: Everything You Need to Know",
        f"{state['topic']} Simplified: A Practical Guide",
        f"Master {state['topic']} in 2025: A Step-by-Step Guide",
        f"{state['topic']}: From Zero to Production",
    ]

    idx = min(iteration, len(titles) - 1)
    return {"title": titles[idx], "iteration": iteration + 1}

def review_title(state: TitleState) -> dict:
    response = interrupt({
        "title": state["title"],
        "iteration": state["iteration"],
        "instruction": "'approved' or write what to change.",
    })

    if response.lower() == "approved":
        return {"approved": True}
    return {"feedback": [response], "approved": False}

def route(state: TitleState) -> str:
    if state.get("approved"):
        return "done"
    if state["iteration"] >= 3:
        return "done"
    return "regenerate"

def done(state: TitleState) -> dict:
    return {}

graph_builder = StateGraph(TitleState)
graph_builder.add_node("generate", generate_title)
graph_builder.add_node("review", review_title)
graph_builder.add_node("done", done)

graph_builder.add_edge(START, "generate")
graph_builder.add_edge("generate", "review")
graph_builder.add_conditional_edges(
    "review", route,
    {"regenerate": "generate", "done": "done"},
)
graph_builder.add_edge("done", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "title-feedback"}}

graph.invoke(
    {"topic": "Kubernetes", "title": "", "feedback": [], "iteration": 0, "approved": False},
    config,
)
state = graph.get_state(config)
print(f"v1: {state.values['title']}")

graph.invoke(Command(resume="Too generic"), config)
state = graph.get_state(config)
print(f"v2: {state.values['title']}")

graph.invoke(Command(resume="approved"), config)
state = graph.get_state(config)
print(f"Final: {state.values['title']}")
print(f"Feedback: {state.values['feedback']}")
# Expected output:
# v1: The Complete Guide to Kubernetes: Everything You Need to Know
# v2: Kubernetes Simplified: A Practical Guide
# Final: Kubernetes Simplified: A Practical Guide
# Feedback: ['Too generic']

Exercise 2: Feedback with a real LLM (Medium)

Use a real LLM (init_chat_model) to generate a paragraph on a topic. Implement a feedback loop where the regeneration prompt includes the previous version and the entire feedback history. Cap it at 3 rounds.

See solution

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    topic: str
    paragraph: str
    feedback: Annotated[list[str], operator.add]
    iteration: int
    approved: bool

llm = init_chat_model("openai:gpt-4.1-nano")

def write(state: State) -> dict:
    iteration = state.get("iteration", 0)
    feedback = state.get("feedback", [])

    if iteration == 0:
        prompt = f"Write a 3-sentence paragraph about: {state['topic']}"
    else:
        prompt = (
            f"Previous version:\n{state['paragraph']}\n\n"
            f"User feedback: {feedback[-1]}\n"
            f"All previous feedback: {feedback}\n\n"
            f"Rewrite it, improving it based on the feedback. Keep the good parts. 3 sentences max."
        )

    response = llm.invoke(prompt)
    return {"paragraph": response.content, "iteration": iteration + 1}

def review(state: State) -> dict:
    response = interrupt({
        "paragraph": state["paragraph"],
        "round": state["iteration"],
        "remaining": 3 - state["iteration"],
    })

    if response.lower() == "approved":
        return {"approved": True}
    return {"feedback": [response], "approved": False}

def route(state: State) -> str:
    if state.get("approved") or state["iteration"] >= 3:
        return "end"
    return "rewrite"

def end(state: State) -> dict:
    return {}

builder = StateGraph(State)
builder.add_node("write", write)
builder.add_node("review", review)
builder.add_node("end", end)

builder.add_edge(START, "write")
builder.add_edge("write", "review")
builder.add_conditional_edges("review", route, {"rewrite": "write", "end": "end"})
builder.add_edge("end", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "llm-feedback"}}

graph.invoke(
    {"topic": "the advantages of Rust", "paragraph": "", "feedback": [], "iteration": 0, "approved": False},
    config,
)
state = graph.get_state(config)
print(f"v1: {state.values['paragraph'][:100]}...")

graph.invoke(Command(resume="Make it more casual, like you're talking to a friend"), config)
state = graph.get_state(config)
print(f"v2: {state.values['paragraph'][:100]}...")

graph.invoke(Command(resume="approved"), config)
state = graph.get_state(config)
print(f"Final: {state.values['paragraph']}")
print(f"Iterations: {state.values['iteration']}")
# Expected output (content varies by LLM):
# v1: [Formal paragraph about Rust]...
# v2: [More casual paragraph about Rust]...
# Final: [Final version of the paragraph]
# Iterations: 2

Exercise 3: Multi-point feedback pipeline (Medium)

Build a 4-stage pipeline (plan → research → draft → format) with review points after stages 1 and 3. The user can give feedback at each point. Use stage_feedback: Annotated[list[dict], operator.add] to track which stage each piece of feedback came from.

See solution

from dotenv import load_dotenv
load_dotenv()

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

class PipelineState(TypedDict):
    topic: str
    plan: str
    research: str
    draft: str
    formatted: str
    stage_feedback: Annotated[list[dict], operator.add]

def plan(state: PipelineState) -> dict:
    return {"plan": f"Plan for '{state['topic']}': 3 sections, 2 sources per section"}

def review_plan(state: PipelineState) -> dict:
    response = interrupt({"stage": "plan", "content": state["plan"]})
    if response.lower() == "ok":
        return {}
    return {
        "plan": f"{state['plan']} [Adjustment: {response}]",
        "stage_feedback": [{"stage": "plan", "feedback": response}],
    }

def research(state: PipelineState) -> dict:
    return {"research": f"Data on '{state['topic']}': 6 sources found"}

def draft(state: PipelineState) -> dict:
    return {"draft": f"Draft based on plan: {state['plan'][:50]}..."}

def review_draft(state: PipelineState) -> dict:
    response = interrupt({"stage": "draft", "content": state["draft"]})
    if response.lower() == "ok":
        return {}
    return {
        "draft": f"{state['draft']} [Correction: {response}]",
        "stage_feedback": [{"stage": "draft", "feedback": response}],
    }

def format_output(state: PipelineState) -> dict:
    return {"formatted": f"[FORMATTED] {state['draft']}"}

builder = StateGraph(PipelineState)
builder.add_node("plan", plan)
builder.add_node("review_plan", review_plan)
builder.add_node("research", research)
builder.add_node("draft", draft)
builder.add_node("review_draft", review_draft)
builder.add_node("format", format_output)

builder.add_edge(START, "plan")
builder.add_edge("plan", "review_plan")
builder.add_edge("review_plan", "research")
builder.add_edge("research", "draft")
builder.add_edge("draft", "review_draft")
builder.add_edge("review_draft", "format")
builder.add_edge("format", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "pipeline-feedback"}}

graph.invoke(
    {"topic": "GraphQL vs REST", "plan": "", "research": "", "draft": "", "formatted": "", "stage_feedback": []},
    config,
)
graph.invoke(Command(resume="Add a performance comparison"), config)

state = graph.get_state(config)
print(f"Adjusted plan: {state.values['plan']}")

graph.invoke(Command(resume="ok"), config)
state = graph.get_state(config)
print(f"Output: {state.values['formatted']}")
print(f"Feedback: {state.values['stage_feedback']}")
# Expected output:
# Adjusted plan: Plan for 'GraphQL vs REST': 3 sections, 2 sources per section [Adjustment: Add a performance comparison]
# Output: [FORMATTED] Draft based on plan: Plan for 'GraphQL vs REST': 3 sections, 2 s... [Correction: ok]
# Feedback: [{'stage': 'plan', 'feedback': 'Add a performance comparison'}]

Exercise 4: Feedback loop with a cap and a disclaimer (Medium)

Implement a feedback loop that, after 3 rounds without approval, delivers automatically with a message like "Delivered after 3 revisions. Contact support if you need more adjustments." The state must record whether it was approved by the user or auto-delivered.

See solution

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    content: str
    feedback: Annotated[list[str], operator.add]
    iteration: int
    delivery_mode: str

MAX_ROUNDS = 3

def generate(state: State) -> dict:
    iteration = state.get("iteration", 0)
    feedback = state.get("feedback", [])
    content = f"Content v{iteration + 1}"
    if feedback:
        content += f" (incorporating: {feedback[-1]})"
    return {"content": content, "iteration": iteration + 1}

def review(state: State) -> dict:
    if state["iteration"] >= MAX_ROUNDS:
        disclaimer = (
            f"Delivered automatically after {MAX_ROUNDS} revisions. "
            f"Contact support if you need more adjustments."
        )
        return {
            "content": f"{state['content']}\n\n⚠️ {disclaimer}",
            "delivery_mode": "auto_delivered",
        }

    response = interrupt({
        "content": state["content"],
        "round": state["iteration"],
        "remaining": MAX_ROUNDS - state["iteration"],
    })

    if response.lower() == "approved":
        return {"delivery_mode": "user_approved"}
    return {"feedback": [response], "delivery_mode": "needs_revision"}

def route(state: State) -> str:
    mode = state.get("delivery_mode", "")
    if mode in ("user_approved", "auto_delivered"):
        return "deliver"
    return "revise"

def deliver(state: State) -> dict:
    return {}

builder = StateGraph(State)
builder.add_node("generate", generate)
builder.add_node("review", review)
builder.add_node("deliver", deliver)

builder.add_edge(START, "generate")
builder.add_edge("generate", "review")
builder.add_conditional_edges("review", route, {"revise": "generate", "deliver": "deliver"})
builder.add_edge("deliver", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "bounded-delivery"}}

graph.invoke(
    {"content": "", "feedback": [], "iteration": 0, "delivery_mode": ""},
    config,
)
graph.invoke(Command(resume="Shorter"), config)
graph.invoke(Command(resume="More data"), config)

state = graph.get_state(config)
print(f"Content: {state.values['content']}")
print(f"Delivery mode: {state.values['delivery_mode']}")
print(f"Iterations: {state.values['iteration']}")
print(f"Feedback: {state.values['feedback']}")
# Expected output:
# Content: Content v3 (incorporating: More data)
#
# ⚠️ Delivered automatically after 3 revisions. Contact support if you need more adjustments.
# Delivery mode: auto_delivered
# Iterations: 3
# Feedback: ['Shorter', 'More data']

Exercise 5: Feedback loop with response categories (Advanced)

Build a feedback loop where the user doesn't just write free text but can pick a category: "tone" (adjust the tone), "content" (add/remove content), "format" (change the structure), "approve". The generation node must behave differently depending on the feedback category.

See solution

from dotenv import load_dotenv
load_dotenv()

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

class CategorizedState(TypedDict):
    topic: str
    output: str
    feedback_log: Annotated[list[dict], operator.add]
    iteration: int
    approved: bool

def generate(state: CategorizedState) -> dict:
    iteration = state.get("iteration", 0)
    log = state.get("feedback_log", [])

    base = f"Article about {state['topic']}"

    tone_adjustments = [e["detail"] for e in log if e["category"] == "tone"]
    content_adjustments = [e["detail"] for e in log if e["category"] == "content"]
    format_adjustments = [e["detail"] for e in log if e["category"] == "format"]

    parts = [base]
    if tone_adjustments:
        parts.append(f"[Tone adjusted: {', '.join(tone_adjustments)}]")
    if content_adjustments:
        parts.append(f"[Content: {', '.join(content_adjustments)}]")
    if format_adjustments:
        parts.append(f"[Format: {', '.join(format_adjustments)}]")

    return {"output": " ".join(parts), "iteration": iteration + 1}

def review(state: CategorizedState) -> dict:
    if state["iteration"] >= 4:
        return {"approved": True}

    response = interrupt({
        "output": state["output"],
        "round": state["iteration"],
        "instruction": (
            "Reply with the format 'category: detail'. "
            "Categories: tone, content, format, approve. "
            "Example: 'tone: more casual' or 'content: add benchmarks' or 'approve'"
        ),
    })

    if response.lower().strip() == "approve":
        return {"approved": True}

    if ":" in response:
        category, detail = response.split(":", 1)
        category = category.strip().lower()
        detail = detail.strip()
    else:
        category = "content"
        detail = response

    valid_categories = ("tone", "content", "format")
    if category not in valid_categories:
        category = "content"

    return {
        "feedback_log": [{"category": category, "detail": detail, "round": state["iteration"]}],
        "approved": False,
    }

def route(state: CategorizedState) -> str:
    if state.get("approved"):
        return "deliver"
    return "revise"

def deliver(state: CategorizedState) -> dict:
    return {}

builder = StateGraph(CategorizedState)
builder.add_node("generate", generate)
builder.add_node("review", review)
builder.add_node("deliver", deliver)

builder.add_edge(START, "generate")
builder.add_edge("generate", "review")
builder.add_conditional_edges("review", route, {"revise": "generate", "deliver": "deliver"})
builder.add_edge("deliver", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "categorized-feedback"}}

graph.invoke(
    {"topic": "MLOps", "output": "", "feedback_log": [], "iteration": 0, "approved": False},
    config,
)
state = graph.get_state(config)
print(f"v1: {state.values['output']}")

graph.invoke(Command(resume="tone: more technical and direct"), config)
state = graph.get_state(config)
print(f"v2: {state.values['output']}")

graph.invoke(Command(resume="content: add a CI/CD section"), config)
state = graph.get_state(config)
print(f"v3: {state.values['output']}")

graph.invoke(Command(resume="approve"), config)
state = graph.get_state(config)
print(f"Final: {state.values['output']}")
print(f"Log: {state.values['feedback_log']}")
# Expected output:
# v1: Article about MLOps
# v2: Article about MLOps [Tone adjusted: more technical and direct]
# v3: Article about MLOps [Tone adjusted: more technical and direct] [Content: add a CI/CD section]
# Final: Article about MLOps [Tone adjusted: more technical and direct] [Content: add a CI/CD section]
# Log: [{'category': 'tone', 'detail': 'more technical and direct', 'round': 1}, {'category': 'content', 'detail': 'add a CI/CD section', 'round': 2}]

Summary

In this capsule you learned:

Feedback loops turn HITL from binary into iterative — the Review → Feedback → Revise → Approve pattern produces results neither the human nor the AI would reach alone
Feedback accumulates, it doesn't get replaced — use Annotated[list[str], operator.add] to keep the full history of corrections across rounds
Feed the feedback into the prompt, don't restart from zero — "previous version + user feedback → improve what's there" produces a natural evolution of the content
Multi-point feedback gives granular control — reviewing the plan before executing, the data before synthesizing, and the report before delivering avoids wasted work
3 rounds of feedback max — after that, returns are marginal or counterproductive. Auto-delivery with a disclaimer beats an infinite loop
The interrupt's UX matters — pass a dict with the draft, the current round, the rounds remaining, and a clear instruction. The user must know exactly what they're reviewing and what their options are

Next capsule: HITL Patterns in Production — how to implement async approvals, timeouts, escalations and dashboards when "the human" isn't sitting in front of a terminal.

Additional resources

LangGraph Human-in-the-Loop — HITL concepts in LangGraph
How to wait for user input — Implementing interrupts for user input
How to edit graph state — Editing graph state during execution
How to review tool calls — Review patterns for tool calls
LangGraph interrupt() — Reference for the interrupt function
Human-AI Collaboration Patterns — Nielsen Norman Group: human-AI collaboration patterns

Module 9 — LangChain & LangGraph: From Chains to Agents

← PreviousEditable State: Modifying State During a Pause Next →HITL Patterns in Production