Module 9: Human-in-the-Loop

When to Automate vs When to Pause

Capsule overview

This is the most important design decision for agents in production: what should trigger a human interruption?

Too many interruptions → the agent is useless. If it asks permission for every Google search, every calculation, every intermediate step — you'd be faster doing it yourself. Too few interruptions → the agent is dangerous. If it executes financial transactions, deletes data, or sends emails without asking you — one mistake gets expensive.

The sweet spot is somewhere in the middle, and it isn't intuitive. This capsule gives you a concrete framework with four criteria you can apply to any action of any agent. It's not theory — it's a Python function that evaluates the risk of an action and automatically decides whether to interrupt.

In the previous capsules you learned the mechanics: interrupt() to pause, Command(resume=...) to resume, breakpoints, state editing. Now you have all the tools. The question is no longer "how do I interrupt?" but "when should I interrupt?"

The problem: two extremes that don't work

Extreme 1: Interrupt everything

User: "Research AI agents"

Agent: "I'm going to break the topic into sub-questions. Shall I proceed?"
User: "Yes"
Agent: "I'm going to search Wikipedia. Shall I proceed?"
User: "Yes..."
Agent: "I'm going to search arXiv. Shall I proceed?"
User: "YES"
Agent: "I'm going to merge the results. Shall I proceed?"
User: "JUST DO IT"
Agent: "I'm going to generate the summary. Shall I proceed?"
User: *closes the app*

Result: 5 interruptions for a task that should have been autonomous. The user burns 3 minutes approving the obvious. The experience is worse than having no agent at all.

Extreme 2: Interrupt nothing

User: "Research AI agents and send the report to the team"

Agent: [searches 5 sources] ✅
Agent: [calls premium API — $2.50] ✅
Agent: [generates report with incorrect data] ✅
Agent: [emails 50 people with incorrect data] ✅

User: "... I just sent false data to my team"

Result: the agent acted with total autonomy. One incorrect fact reached 50 people. The cost of fixing it is enormous.

The goal: interrupt at exactly the right point

User: "Research AI agents and send the report to the team"

Agent: [searches free sources] ✅ (auto-approve)
Agent: "I found 3 free sources. I can also search
        arXiv Premium ($2.50). Shall I proceed?"
User: "No, the free ones are enough"
Agent: [generates a draft of the report]
Agent: "Here's the draft. Should I send it to 50 people?"
User: [reviews, fixes one data point] "Now yes, send it"
Agent: [sends the corrected email] ✅

Result: 2 interruptions — both justified. The first avoided an unnecessary expense. The second avoided propagating an error.

The decision framework: 4 criteria

Every action an agent can execute gets evaluated across 4 dimensions. The combination of the 4 produces a risk level that determines whether the agent should pause or keep going.

Criterion 1: Cost

Does the action cost money?

Range	Level	Policy
< $0.01	Negligible	Always auto-approve
$0.01 – $1.00	Low	Approve the first time, auto-approve once the pattern is established
$1.00 – $10.00	Medium	Always show the cost, ask for approval
> $10.00	High	Explicit approval with double confirmation

Cost includes: calls to paid APIs, token consumption on expensive models, external services with billing, and any action that generates a charge.

Criterion 2: Reversibility

Can the action be undone?

Type	Examples	Policy
Fully reversible	Reading, searching, generating a draft, calculations	Auto-approve
Partially reversible	Sending an email (you can send a correction), creating a record (you can delete it)	Warn and ask for approval
Irreversible	Deleting data, an executed financial transaction, publishing public content	Always require approval

The key question: "If this action goes wrong, can I undo it without consequences?"

Criterion 3: Impact

How many people or systems does it affect?

Scope	Examples	Policy
Self-contained	The agent's internal state, temp files	Auto-approve
Externally limited	A notification to one user, updating one record	Approve the first few times, then auto-approve
Externally broad	Email to a team, a change in a shared database, a public post	Always require approval

A Google search only affects the agent (self-contained). An email to a customer affects the relationship with that customer (externally limited). A social media post affects the company's reputation (externally broad).

Criterion 4: Confidence

How sure is the agent that the action is correct?

Level	Range	Policy
High	> 90%	Auto-approve (the agent has enough context)
Medium	60% – 90%	Show the plan and ask for quick approval
Low	< 60%	Pause and ask for explicit guidance

Confidence isn't always an explicit number. It can be inferred: if the agent found multiple sources that agree, confidence is high. If the results are contradictory or ambiguous, it's low.

The decision matrix

Each action gets a score on each criterion. The combined score determines the policy:

Risk score = max(cost_score, reversibility_score, impact_score) + (1 - confidence_score)

If risk < 0.3   →  AUTO-APPROVE (execute without asking)
If risk 0.3-0.7 →  QUICK APPROVE (show the plan, quick approval)
If risk > 0.7   →  FULL REVIEW (show full detail, wait for explicit approval)

The max() operator over the first three criteria means that a single criterion in the red is enough to escalate. A free, self-contained, but irreversible action still requires approval.

Actions classified with the framework

Let's apply the framework to the common actions of a research agent:

Action	Cost	Reversibility	Impact	Typical confidence	Decision
Web search (free)	Negligible	Reversible	Self-contained	High	✅ Auto-approve
Premium API search ($0.50)	Low	Reversible	Self-contained	High	⚠️ Approve 1st time
Generate report draft	Negligible	Reversible	Self-contained	Medium	✅ Auto-approve
Send an email to one user	Negligible	Partially reversible	Externally limited	Medium	⚠️ Approve
Email the team (50 people)	Negligible	Partially reversible	Externally broad	Medium	❌ Full review
Call an expensive API ($5.00)	High	Reversible	Self-contained	High	⚠️ Approve
Delete DB records	Negligible	Irreversible	Externally broad	High	❌ Full review
Create a temp file	Negligible	Reversible	Self-contained	High	✅ Auto-approve
Post to social media	Negligible	Partially reversible	Externally broad	Medium	❌ Full review
Execute a payment transaction	High	Irreversible	Externally limited	High	❌ Full review

The table makes it obvious why "interrupt everything" is wrong: half the actions are safe to auto-approve. And why "interrupt nothing" is dangerous: the last three rows always need human supervision.

Implementation: a risk assessment function

Let's move the framework into executable code:

from dotenv import load_dotenv
load_dotenv()

from dataclasses import dataclass
from enum import Enum


class CostLevel(Enum):
    NEGLIGIBLE = 0.0
    LOW = 0.3
    MEDIUM = 0.6
    HIGH = 1.0


class Reversibility(Enum):
    REVERSIBLE = 0.0
    PARTIAL = 0.5
    IRREVERSIBLE = 1.0


class Impact(Enum):
    SELF_CONTAINED = 0.0
    EXTERNAL_LIMITED = 0.4
    EXTERNAL_BROAD = 1.0


class RiskDecision(Enum):
    AUTO_APPROVE = "auto_approve"
    QUICK_APPROVE = "quick_approve"
    FULL_REVIEW = "full_review"


@dataclass
class ActionRisk:
    action_name: str
    cost: CostLevel
    reversibility: Reversibility
    impact: Impact
    confidence: float

    @property
    def risk_score(self) -> float:
        base = max(self.cost.value, self.reversibility.value, self.impact.value)
        confidence_penalty = 1.0 - self.confidence
        return min(round(base + confidence_penalty * 0.5, 2), 1.0)

    @property
    def decision(self) -> RiskDecision:
        score = self.risk_score
        if score < 0.3:
            return RiskDecision.AUTO_APPROVE
        elif score <= 0.7:
            return RiskDecision.QUICK_APPROVE
        else:
            return RiskDecision.FULL_REVIEW


actions = [
    ActionRisk("web_search_free", CostLevel.NEGLIGIBLE, Reversibility.REVERSIBLE, Impact.SELF_CONTAINED, 0.95),
    ActionRisk("api_search_paid", CostLevel.LOW, Reversibility.REVERSIBLE, Impact.SELF_CONTAINED, 0.90),
    ActionRisk("send_email_one", CostLevel.NEGLIGIBLE, Reversibility.PARTIAL, Impact.EXTERNAL_LIMITED, 0.80),
    ActionRisk("send_email_team", CostLevel.NEGLIGIBLE, Reversibility.PARTIAL, Impact.EXTERNAL_BROAD, 0.75),
    ActionRisk("delete_records", CostLevel.NEGLIGIBLE, Reversibility.IRREVERSIBLE, Impact.EXTERNAL_BROAD, 0.95),
    ActionRisk("call_expensive_api", CostLevel.HIGH, Reversibility.REVERSIBLE, Impact.SELF_CONTAINED, 0.90),
    ActionRisk("generate_draft", CostLevel.NEGLIGIBLE, Reversibility.REVERSIBLE, Impact.SELF_CONTAINED, 0.70),
]

print("=== Risk assessment per action ===\n")
print(f"{'Action':<22} {'Risk':>7} {'Decision':<16}")
print(f"{'-'*22} {'-'*7} {'-'*16}")

for a in actions:
    print(f"{a.action_name:<22} {a.risk_score:>6.2f}  {a.decision.value}")
# Expected output:
# === Risk assessment per action ===
#
# Action                    Risk Decision
# ---------------------- ------- ----------------
# web_search_free          0.02  auto_approve
# api_search_paid          0.35  quick_approve
# send_email_one           0.60  quick_approve
# send_email_team          1.00  full_review
# delete_records           1.00  full_review
# call_expensive_api       1.00  full_review
# generate_draft           0.15  auto_approve

The risk_score function uses max() for the first three criteria and adds a penalty for low confidence. That captures the intuition: an irreversible action with high confidence is still risky (the cost of the mistake is high even if the mistake is unlikely).

Wiring it to interrupt(): conditional decisions

Now let's connect the risk assessment to the interrupt() mechanism inside a graph:

from dotenv import load_dotenv
load_dotenv()

from dataclasses import dataclass
from enum import Enum
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 CostLevel(Enum):
    NEGLIGIBLE = 0.0
    LOW = 0.3
    MEDIUM = 0.6
    HIGH = 1.0

class Reversibility(Enum):
    REVERSIBLE = 0.0
    PARTIAL = 0.5
    IRREVERSIBLE = 1.0

class Impact(Enum):
    SELF_CONTAINED = 0.0
    EXTERNAL_LIMITED = 0.4
    EXTERNAL_BROAD = 1.0

class RiskDecision(Enum):
    AUTO_APPROVE = "auto_approve"
    QUICK_APPROVE = "quick_approve"
    FULL_REVIEW = "full_review"

@dataclass
class ActionRisk:
    action_name: str
    cost: CostLevel
    reversibility: Reversibility
    impact: Impact
    confidence: float

    @property
    def risk_score(self) -> float:
        base = max(self.cost.value, self.reversibility.value, self.impact.value)
        confidence_penalty = 1.0 - self.confidence
        return min(round(base + confidence_penalty * 0.5, 2), 1.0)

    @property
    def decision(self) -> RiskDecision:
        score = self.risk_score
        if score < 0.3:
            return RiskDecision.AUTO_APPROVE
        elif score <= 0.7:
            return RiskDecision.QUICK_APPROVE
        else:
            return RiskDecision.FULL_REVIEW


def assess_risk(action_name: str, cost: float, reversible: bool, external: bool, confidence: float) -> ActionRisk:
    if cost < 0.01:
        cost_level = CostLevel.NEGLIGIBLE
    elif cost < 1.0:
        cost_level = CostLevel.LOW
    elif cost < 10.0:
        cost_level = CostLevel.MEDIUM
    else:
        cost_level = CostLevel.HIGH

    rev = Reversibility.REVERSIBLE if reversible else Reversibility.IRREVERSIBLE
    imp = Impact.EXTERNAL_BROAD if external else Impact.SELF_CONTAINED

    return ActionRisk(action_name, cost_level, rev, imp, confidence)


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


def plan_and_search(state: State) -> dict:
    topic = state["topic"]
    actions = []

    free_search = assess_risk("web_search", cost=0.0, reversible=True, external=False, confidence=0.95)
    if free_search.decision == RiskDecision.AUTO_APPROVE:
        actions.append(f"[AUTO] web_search (risk: {free_search.risk_score:.2f})")
        sources = [f"Web: free results about '{topic}'"]
    else:
        response = interrupt({
            "action": "web_search",
            "risk_score": free_search.risk_score,
            "message": f"Search the web about '{topic}'?",
        })
        sources = [f"Web: results about '{topic}'"] if response.get("approved") else []
        actions.append(f"[APPROVED] web_search" if response.get("approved") else "[REJECTED] web_search")

    paid_search = assess_risk("premium_api", cost=2.50, reversible=True, external=False, confidence=0.85)
    if paid_search.decision == RiskDecision.AUTO_APPROVE:
        sources.append(f"Premium: paid data about '{topic}'")
        actions.append(f"[AUTO] premium_api")
    else:
        response = interrupt({
            "action": "premium_api",
            "risk_score": paid_search.risk_score,
            "cost": 2.50,
            "message": f"Search the premium API ($2.50) about '{topic}'. Shall I proceed?",
        })
        if response.get("approved"):
            sources.append(f"Premium: paid data about '{topic}'")
            actions.append(f"[APPROVED] premium_api (${2.50})")
        else:
            actions.append(f"[REJECTED] premium_api (${2.50} saved)")

    return {"sources": sources, "actions_log": actions}


def generate_report(state: State) -> dict:
    source_list = "\n".join(f"  - {s}" for s in state["sources"])
    report = f"Report on '{state['topic']}':\n{source_list}\nSources: {len(state['sources'])}"
    return {"report": report, "actions_log": [f"[AUTO] generate_report"]}


graph_builder = StateGraph(State)
graph_builder.add_node("search", plan_and_search)
graph_builder.add_node("report", generate_report)
graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "report")
graph_builder.add_edge("report", END)

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

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

print("=== Invocation 1: the free search auto-approves ===")
result = graph.invoke(
    {"topic": "AI agents", "sources": [], "report": "", "actions_log": []},
    config,
)

print(f"\nCurrent state: {graph.get_state(config).next}")
if graph.get_state(config).next:
    print("Interrupt detected — the agent is asking for approval on the premium API")
    print("Resuming with approval...")

    result = graph.invoke(
        Command(resume={"approved": True}),
        config,
    )

print(f"\nReport:\n{result['report']}")
print(f"\nAction log:")
for action in result["actions_log"]:
    print(f"  {action}")
# Expected output:
# === Invocation 1: the free search auto-approves ===
#
# Current state: ('search',)
# Interrupt detected — the agent is asking for approval on the premium API
# Resuming with approval...
#
# Report:
# Report on 'AI agents':
#   - Web: free results about 'AI agents'
#   - Premium: paid data about 'AI agents'
# Sources: 2
#
# Action log:
#   [AUTO] web_search (risk: 0.02)
#   [APPROVED] premium_api ($2.5)
#   [AUTO] generate_report

Notice the pattern: the free search went through without asking (risk_score: 0.02). The premium API ($2.50) triggered an interrupt. The report was generated automatically (it's reversible, free, self-contained). Two out of three actions were automatic — only one needed human approval.

Calibration: reducing false positives and false negatives

A risk framework isn't something you configure once and forget. You calibrate it with real data.

False positives: unnecessary interruptions

The agent asks permission, the human always says "yes." That tells you the threshold for that action is too low.

from dotenv import load_dotenv
load_dotenv()


def calculate_auto_approve_rate(approval_history: list[dict]) -> dict:
    """Computes the approval rate per action type."""
    stats = {}

    for entry in approval_history:
        action = entry["action"]
        if action not in stats:
            stats[action] = {"total": 0, "approved": 0}
        stats[action]["total"] += 1
        if entry["decision"] == "approved":
            stats[action]["approved"] += 1

    recommendations = {}
    for action, data in stats.items():
        rate = data["approved"] / data["total"] if data["total"] > 0 else 0
        if rate >= 0.95 and data["total"] >= 10:
            rec = "AUTO_APPROVE — the user always approves. Drop the interrupt."
        elif rate >= 0.80:
            rec = "KEEP — most get approved but there are meaningful rejections."
        elif rate >= 0.50:
            rec = "REVIEW — the user rejects often. Maybe the agent shouldn't attempt this action."
        else:
            rec = "DROP THE ACTION — the user almost always rejects it."
        recommendations[action] = {"rate": rate, "total": data["total"], "recommendation": rec}

    return recommendations


history = [
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "send_email_team", "decision": "rejected"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "send_email_team", "decision": "rejected"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "delete_records", "decision": "rejected"},
    {"action": "delete_records", "decision": "rejected"},
    {"action": "delete_records", "decision": "approved"},
    {"action": "delete_records", "decision": "rejected"},
    {"action": "delete_records", "decision": "rejected"},
]

results = calculate_auto_approve_rate(history)
print("=== Interrupt calibration ===\n")
for action, data in results.items():
    print(f"{action}:")
    print(f"  Approval rate: {data['rate']:.0%} ({data['total']} requests)")
    print(f"  → {data['recommendation']}")
    print()
# Expected output:
# === Interrupt calibration ===
#
# web_search_paid:
#   Approval rate: 100% (12 requests)
#   → AUTO_APPROVE — the user always approves. Drop the interrupt.
#
# send_email_team:
#   Approval rate: 80% (10 requests)
#   → KEEP — most get approved but there are meaningful rejections.
#
# delete_records:
#   Approval rate: 20% (5 requests)
#   → DROP THE ACTION — the user almost always rejects it.

The key data point: web_search_paid has a 100% approval rate across 12 requests. The interrupt is noise — drop it and auto-approve. delete_records has a 20% approval rate — the agent shouldn't be trying to delete records unless the user explicitly asks for it.

False negatives: dangerous actions that slipped through

These are harder to detect because you only find them when something goes wrong. The strategy:

Rule of thumb for false negatives:
  - After every incident, ask: "Was there an interrupt that would have prevented this?"
  - If yes → add the interrupt
  - If no → the action should have been designed differently

Incident log:
  - 2026-01-15: Agent sent a report with incorrect data
    → Missing: an interrupt to review the report before sending
    → Fix: add FULL_REVIEW to "send_report"

  - 2026-01-22: Agent called the premium API 15 times in one session ($37.50)
    → Missing: a per-session spend cap with an interrupt
    → Fix: add an interrupt when accumulated spend > $5.00

Autonomy levels per user

Not every user needs the same level of supervision. Someone with 200 sessions under their belt needs fewer interruptions than someone on their first session.

from dotenv import load_dotenv
load_dotenv()

from dataclasses import dataclass


@dataclass
class UserAutonomy:
    user_id: str
    sessions_completed: int
    approval_rate: float
    trust_score: float

    @property
    def autonomy_level(self) -> str:
        if self.sessions_completed < 5:
            return "supervised"
        elif self.sessions_completed < 50 and self.approval_rate >= 0.85:
            return "standard"
        elif self.sessions_completed >= 50 and self.approval_rate >= 0.90:
            return "autonomous"
        else:
            return "standard"

    @property
    def risk_threshold_adjustment(self) -> float:
        """Adjusts the risk threshold based on autonomy.
        Autonomous users tolerate more risk without an interrupt."""
        adjustments = {
            "supervised": 0.0,
            "standard": 0.15,
            "autonomous": 0.30,
        }
        return adjustments[self.autonomy_level]


def should_interrupt(risk_score: float, user: UserAutonomy) -> bool:
    adjusted_threshold = 0.3 + user.risk_threshold_adjustment
    return risk_score >= adjusted_threshold


users = [
    UserAutonomy("new", sessions_completed=2, approval_rate=1.0, trust_score=0.5),
    UserAutonomy("regular", sessions_completed=30, approval_rate=0.92, trust_score=0.8),
    UserAutonomy("expert", sessions_completed=150, approval_rate=0.95, trust_score=0.95),
]

test_risk_scores = [0.25, 0.40, 0.55, 0.70]

print("=== Autonomy per user ===\n")
print(f"{'User':<12} {'Sessions':>8} {'Level':<14} {'Threshold':>9}")
print(f"{'-'*12} {'-'*8} {'-'*14} {'-'*9}")
for u in users:
    threshold = 0.3 + u.risk_threshold_adjustment
    print(f"{u.user_id:<12} {u.sessions_completed:>8} {u.autonomy_level:<14} {threshold:>8.2f}")

print(f"\n{'Risk':<10}", end="")
for u in users:
    print(f" {u.user_id:>12}", end="")
print()
print(f"{'-'*10}", end="")
for _ in users:
    print(f" {'-'*12}", end="")
print()

for score in test_risk_scores:
    print(f"{score:<10.2f}", end="")
    for u in users:
        decision = "INTERRUPT" if should_interrupt(score, u) else "auto"
        print(f" {decision:>12}", end="")
    print()
# Expected output:
# === Autonomy per user ===
#
# User         Sessions Level          Threshold
# ------------ -------- -------------- ---------
# new                 2 supervised          0.30
# regular            30 standard             0.45
# expert            150 autonomous           0.60
#
# Risk               new      regular       expert
# ---------- ------------ ------------ ------------
# 0.25               auto         auto         auto
# 0.40          INTERRUPT         auto         auto
# 0.55          INTERRUPT    INTERRUPT         auto
# 0.70          INTERRUPT    INTERRUPT    INTERRUPT

The new user (supervised) gets an interrupt on any action with risk ≥ 0.30. The expert user (autonomous) only gets interrupted at risk ≥ 0.60. Same agent, same actions, an experience adapted to the user.

Compliance considerations

In regulated industries, the risk framework isn't enough. There are actions that always require human approval, no matter what the calculated risk says:

Financial industry (SOX, PCI-DSS):
  - Any transaction > $0 → FULL_REVIEW
  - Access to card data → FULL_REVIEW
  - Modification of accounting records → FULL_REVIEW

Healthcare (HIPAA):
  - Access to patient data → FULL_REVIEW
  - Communication with patients → FULL_REVIEW
  - Modification of medical records → FULL_REVIEW

GDPR:
  - Processing of personal data → FULL_REVIEW
  - Cross-border data transfer → FULL_REVIEW
  - Data deletion (right to be forgotten) → FULL_REVIEW

The implementation is straightforward: add an override on top of the risk framework.

from dotenv import load_dotenv
load_dotenv()


COMPLIANCE_OVERRIDES = {
    "financial": {
        "any_transaction": "full_review",
        "card_data_access": "full_review",
        "ledger_modification": "full_review",
    },
    "healthcare": {
        "patient_data_access": "full_review",
        "patient_communication": "full_review",
        "record_modification": "full_review",
    },
}


def get_final_decision(
    action: str,
    risk_decision: str,
    compliance_domain: str = None,
) -> str:
    if compliance_domain and compliance_domain in COMPLIANCE_OVERRIDES:
        overrides = COMPLIANCE_OVERRIDES[compliance_domain]
        if action in overrides:
            return overrides[action]

    return risk_decision


test_cases = [
    ("web_search", "auto_approve", None),
    ("web_search", "auto_approve", "financial"),
    ("any_transaction", "quick_approve", "financial"),
    ("patient_data_access", "auto_approve", "healthcare"),
    ("generate_report", "auto_approve", "healthcare"),
]

print("=== Compliance overrides ===\n")
print(f"{'Action':<25} {'Risk':<16} {'Domain':<14} {'Final':<16}")
print(f"{'-'*25} {'-'*16} {'-'*14} {'-'*16}")
for action, risk, domain in test_cases:
    final = get_final_decision(action, risk, domain)
    domain_str = domain or "none"
    override = " ⚠️" if final != risk else ""
    print(f"{action:<25} {risk:<16} {domain_str:<14} {final:<16}{override}")
# Expected output:
# === Compliance overrides ===
#
# Action                    Risk             Domain         Final
# ------------------------- ---------------- -------------- ----------------
# web_search                auto_approve     none           auto_approve
# web_search                auto_approve     financial      auto_approve
# any_transaction           quick_approve    financial      full_review      ⚠️
# patient_data_access       auto_approve     healthcare     full_review      ⚠️
# generate_report           auto_approve     healthcare     auto_approve

Compliance overrides take priority over the risk score. patient_data_access computed auto_approve through the risk framework, but the healthcare domain forces full_review.

Troubleshooting

Problem 1: "The agent interrupts too much and users are complaining"

Symptom: User feedback: "the agent asks me for permission for everything."

Cause: The risk threshold is too low, or low-risk actions are misclassified.

Fix: Look at the approval rate per action (the calibration section). If an action has >95% approval with at least 10 samples, switch it to auto-approve.

Problem 2: "risk_score always comes out the same for different actions"

Symptom: Actions with very different risk profiles produce the same score.

Cause: The max() in the formula dominates. If every action has at least one high criterion, they all score high.

Fix: Tune the weights. Instead of max(), use a weighted average that reflects your domain:

def weighted_risk(cost_v, rev_v, impact_v, confidence, weights=(0.3, 0.3, 0.3, 0.1)):
    base = cost_v * weights[0] + rev_v * weights[1] + impact_v * weights[2]
    return min(round(base + (1 - confidence) * weights[3], 2), 1.0)

Problem 3: "I don't know what confidence to assign to each action"

Symptom: The confidence criterion feels subjective.

Cause: It is subjective if you assign it by hand. It shouldn't be.

Fix: Infer confidence from objective signals:

def infer_confidence(sources_found: int, sources_agree: bool, llm_certainty: str) -> float:
    base = min(sources_found * 0.2, 0.6)
    if sources_agree:
        base += 0.2
    if llm_certainty == "high":
        base += 0.2
    elif llm_certainty == "medium":
        base += 0.1
    return min(base, 1.0)

Problem 4: "Autonomy levels create inconsistent experiences"

Symptom: Experienced users get confused when they occasionally hit an interrupt they normally never see.

Cause: Actions sitting near the user's threshold flip between auto-approve and interrupt depending on context.

Fix: Add hysteresis — once an action auto-approves for a user, it only goes back to interrupt if the risk rises significantly (for example, 0.15 above the threshold, not right at the line).

Exercises

Exercise 1: Classify actions with the framework (Easy)

Given this customer support agent, classify each action using the 4 criteria and determine the risk decision:

Search the internal knowledge base
Generate a response for the customer
Send the response to the customer by email
Create an escalation ticket
Issue a $50 refund
Close the customer's account

See solution

Action	Cost	Reversibility	Impact	Confidence	Decision
Search the KB	Negligible	Reversible	Self-contained	High	✅ Auto-approve
Generate a response	Negligible	Reversible	Self-contained	Medium	✅ Auto-approve
Email the customer	Negligible	Partial	Externally limited	Medium	⚠️ Quick approve
Create escalation ticket	Negligible	Reversible	Externally limited	High	✅ Auto-approve
$50 refund	Medium	Irreversible	Externally limited	High	❌ Full review
Close the account	Negligible	Irreversible	Externally broad	High	❌ Full review

Notes: the refund is irreversible (the money is already transferred) and carries a medium cost. Closing an account is irreversible with broad impact (it affects every service the customer has). Both need explicit approval.

Exercise 2: Implement a custom risk assessment (Medium)

Create a function assess_action() that takes: the action name, the estimated cost, whether it's reversible, the number of people affected, and a confidence score from the LLM. It returns the decision (auto/quick/full). Test it with at least 5 different actions.

See solution

from dotenv import load_dotenv
load_dotenv()


def assess_action(name: str, cost: float, reversible: bool, people_affected: int, llm_confidence: float) -> dict:
    if cost < 0.01:
        cost_score = 0.0
    elif cost < 1.0:
        cost_score = 0.3
    elif cost < 10.0:
        cost_score = 0.6
    else:
        cost_score = 1.0

    rev_score = 0.0 if reversible else 1.0

    if people_affected == 0:
        impact_score = 0.0
    elif people_affected <= 5:
        impact_score = 0.4
    else:
        impact_score = 1.0

    risk = max(cost_score, rev_score, impact_score) + (1 - llm_confidence) * 0.5
    risk = min(risk, 1.0)

    if risk < 0.3:
        decision = "auto_approve"
    elif risk <= 0.7:
        decision = "quick_approve"
    else:
        decision = "full_review"

    return {"name": name, "risk": round(risk, 2), "decision": decision}


tests = [
    ("web_search", 0.0, True, 0, 0.95),
    ("api_call", 0.50, True, 0, 0.90),
    ("send_notification", 0.0, False, 1, 0.85),
    ("batch_email", 0.0, False, 100, 0.80),
    ("refund", 25.0, False, 1, 0.75),
]

for args in tests:
    result = assess_action(*args)
    print(f"{result['name']:<20} risk={result['risk']:.2f}  → {result['decision']}")
# Expected output:
# web_search           risk=0.02  → auto_approve
# api_call             risk=0.35  → quick_approve
# send_notification    risk=1.00  → full_review
# batch_email          risk=1.00  → full_review
# refund               risk=1.00  → full_review

Exercise 3: Add a conditional interrupt to a graph (Medium)

Create a graph with 3 nodes: search → process → send. The send node should evaluate the risk of sending (based on the number of recipients in the state). With 1 recipient, auto-approve. With more than 5, interrupt to ask for approval. Test with 1 recipient (should run all the way through) and with 10 recipients (should pause).

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
from langgraph.types import interrupt, Command


class State(TypedDict):
    data: str
    recipients: int
    result: str


def search(state: State) -> dict:
    return {"data": f"Results found for the report"}


def process(state: State) -> dict:
    return {"data": f"Report processed: {state['data']}"}


def send(state: State) -> dict:
    if state["recipients"] > 5:
        response = interrupt({
            "message": f"Send to {state['recipients']} recipients. Shall I proceed?",
            "recipients": state["recipients"],
        })
        if not response.get("approved"):
            return {"result": "Send cancelled by the user"}

    return {"result": f"Sent to {state['recipients']} recipients: {state['data'][:50]}"}


graph_builder = StateGraph(State)
graph_builder.add_node("search", search)
graph_builder.add_node("process", process)
graph_builder.add_node("send", send)
graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "process")
graph_builder.add_edge("process", "send")
graph_builder.add_edge("send", END)

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

print("=== Test 1: 1 recipient (auto-approve) ===")
config1 = {"configurable": {"thread_id": "send-1"}}
result1 = graph.invoke({"data": "", "recipients": 1, "result": ""}, config1)
print(f"  Result: {result1['result']}")

print("\n=== Test 2: 10 recipients (interrupt) ===")
config2 = {"configurable": {"thread_id": "send-10"}}
result2 = graph.invoke({"data": "", "recipients": 10, "result": ""}, config2)

state = graph.get_state(config2)
if state.next:
    print(f"  Interrupt active. Resuming with approval...")
    result2 = graph.invoke(Command(resume={"approved": True}), config2)
    print(f"  Result: {result2['result']}")
# Expected output:
# === Test 1: 1 recipient (auto-approve) ===
#   Result: Sent to 1 recipients: Report processed: Results found for the repo
#
# === Test 2: 10 recipients (interrupt) ===
#   Interrupt active. Resuming with approval...
#   Result: Sent to 10 recipients: Report processed: Results found for the rep

Exercise 4: A calibration system with history (Medium)

Implement a class InterruptCalibrator that records every approval decision and, after N decisions per action, recommends whether to change the policy. Test it with a simulated history of 20 decisions across 3 action types.

See solution

from dotenv import load_dotenv
load_dotenv()

from collections import defaultdict


class InterruptCalibrator:
    def __init__(self, min_samples: int = 5, auto_approve_threshold: float = 0.95):
        self.history = defaultdict(list)
        self.min_samples = min_samples
        self.auto_approve_threshold = auto_approve_threshold

    def record(self, action: str, approved: bool):
        self.history[action].append(approved)

    def recommend(self, action: str) -> str:
        decisions = self.history.get(action, [])
        if len(decisions) < self.min_samples:
            return f"NOT ENOUGH DATA ({len(decisions)}/{self.min_samples})"

        rate = sum(decisions) / len(decisions)
        if rate >= self.auto_approve_threshold:
            return f"→ AUTO_APPROVE (rate: {rate:.0%}, {len(decisions)} samples)"
        elif rate >= 0.70:
            return f"→ KEEP THE INTERRUPT (rate: {rate:.0%})"
        elif rate >= 0.40:
            return f"→ REVIEW THE ACTION (rate: {rate:.0%} — lots of rejections)"
        else:
            return f"→ DROP THE ACTION (rate: {rate:.0%} — almost always rejected)"

    def report(self):
        for action in sorted(self.history.keys()):
            print(f"  {action}: {self.recommend(action)}")


cal = InterruptCalibrator(min_samples=5)

for _ in range(10):
    cal.record("search_paid", True)
cal.record("search_paid", True)
cal.record("search_paid", True)

for approved in [True, True, False, True, True, False, True]:
    cal.record("send_email", approved)

for approved in [False, False, True, False, False, False]:
    cal.record("delete_data", approved)

cal.record("new_action", True)
cal.record("new_action", False)

print("=== Calibration report ===\n")
cal.report()
# Expected output:
# === Calibration report ===
#
#   delete_data: → DROP THE ACTION (rate: 17% — almost always rejected)
#   new_action: NOT ENOUGH DATA (2/5)
#   search_paid: → AUTO_APPROVE (rate: 100%, 12 samples)
#   send_email: → KEEP THE INTERRUPT (rate: 71%)

Exercise 5: A risk framework for a regulated domain (Advanced)

Extend the risk framework to support "compliance domains." Create a function that takes the calculated risk, the domain (finance, healthcare, or none), and the action type. If the domain has an override for that action, the override wins. Test it with 6 different combinations.

See solution

from dotenv import load_dotenv
load_dotenv()


COMPLIANCE_RULES = {
    "finance": {
        "transaction": "full_review",
        "account_modify": "full_review",
        "report_view": "quick_approve",
    },
    "healthcare": {
        "patient_data": "full_review",
        "prescription": "full_review",
        "schedule_view": "quick_approve",
    },
}


def final_decision(action: str, risk_decision: str, domain: str = None) -> dict:
    override = None
    if domain and domain in COMPLIANCE_RULES:
        rules = COMPLIANCE_RULES[domain]
        if action in rules:
            override = rules[action]

    return {
        "action": action,
        "risk_decision": risk_decision,
        "domain": domain or "none",
        "final": override if override else risk_decision,
        "overridden": override is not None and override != risk_decision,
    }


cases = [
    ("web_search", "auto_approve", None),
    ("transaction", "quick_approve", "finance"),
    ("transaction", "quick_approve", None),
    ("patient_data", "auto_approve", "healthcare"),
    ("schedule_view", "auto_approve", "healthcare"),
    ("report_view", "auto_approve", "finance"),
]

print("=== Decisions with compliance ===\n")
print(f"{'Action':<16} {'Risk':<16} {'Domain':<12} {'Final':<16} {'Override?'}")
print(f"{'-'*16} {'-'*16} {'-'*12} {'-'*16} {'-'*9}")
for action, risk, domain in cases:
    r = final_decision(action, risk, domain)
    ov = "YES ⚠️" if r["overridden"] else "no"
    print(f"{r['action']:<16} {r['risk_decision']:<16} {r['domain']:<12} {r['final']:<16} {ov}")
# Expected output:
# === Decisions with compliance ===
#
# Action           Risk             Domain       Final            Override?
# ---------------- ---------------- ------------ ---------------- ---------
# web_search       auto_approve     none         auto_approve     no
# transaction      quick_approve    finance      full_review      YES ⚠️
# transaction      quick_approve    none         quick_approve    no
# patient_data     auto_approve     healthcare   full_review      YES ⚠️
# schedule_view    auto_approve     healthcare   quick_approve    YES ⚠️
# report_view      auto_approve     finance      quick_approve    YES ⚠️

Exercise 6: Simulating an agent with dynamic decisions (Advanced)

Create a graph that simulates a purchasing agent: it takes a list of items to buy, evaluates the risk of each purchase (based on price), and decides whether to auto-approve or interrupt. Items under $10 are auto-approved, items between $10 and $50 require quick approve, items over $50 require full review. The graph must process the whole list, accumulating the approved items, and generate a final purchase order.

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):
    items: list[dict]
    approved_items: Annotated[list[dict], operator.add]
    current_index: int
    total_cost: float
    order_summary: str


def evaluate_item(state: State) -> dict:
    idx = state["current_index"]
    if idx >= len(state["items"]):
        return {"order_summary": "NO_MORE_ITEMS"}

    item = state["items"][idx]
    approved = []

    if item["price"] < 10.0:
        approved.append({**item, "decision": "auto_approved"})
    elif item["price"] <= 50.0:
        response = interrupt({
            "type": "quick_approve",
            "message": f"Buy '{item['name']}' for ${item['price']:.2f}?",
            "item": item,
        })
        if response.get("approved"):
            approved.append({**item, "decision": "human_approved"})
    else:
        response = interrupt({
            "type": "full_review",
            "message": f"⚠️ Large purchase: '{item['name']}' for ${item['price']:.2f}. Review and approve.",
            "item": item,
        })
        if response.get("approved"):
            approved.append({**item, "decision": "human_approved"})

    new_cost = state["total_cost"] + sum(i["price"] for i in approved)
    return {
        "approved_items": approved,
        "current_index": idx + 1,
        "total_cost": new_cost,
    }


def check_more(state: State) -> str:
    if state["current_index"] >= len(state["items"]):
        return "generate_order"
    return "evaluate"


def generate_order(state: State) -> dict:
    if not state["approved_items"]:
        return {"order_summary": "Empty order — no item approved."}
    lines = [f"PURCHASE ORDER — {len(state['approved_items'])} items:"]
    for item in state["approved_items"]:
        lines.append(f"  • {item['name']}: ${item['price']:.2f} ({item['decision']})")
    lines.append(f"  TOTAL: ${state['total_cost']:.2f}")
    return {"order_summary": "\n".join(lines)}


graph_builder = StateGraph(State)
graph_builder.add_node("evaluate", evaluate_item)
graph_builder.add_node("generate_order", generate_order)
graph_builder.add_edge(START, "evaluate")
graph_builder.add_conditional_edges("evaluate", check_more, {"evaluate": "evaluate", "generate_order": "generate_order"})
graph_builder.add_edge("generate_order", END)

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

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

items = [
    {"name": "USB cable", "price": 5.99},
    {"name": "Mechanical keyboard", "price": 45.00},
    {"name": "4K monitor", "price": 350.00},
    {"name": "Mouse pad", "price": 8.50},
]

result = graph.invoke(
    {"items": items, "approved_items": [], "current_index": 0, "total_cost": 0, "order_summary": ""},
    config,
)

approved_count = 0
while graph.get_state(config).next:
    state = graph.get_state(config)
    print(f"  Interrupt — approving...")
    result = graph.invoke(Command(resume={"approved": True}), config)
    approved_count += 1

print(f"\n{result['order_summary']}")
print(f"\n({approved_count} human approvals required)")
# Expected output:
# (The USB cable and the mouse pad auto-approve; the keyboard and the monitor need approval)
#   Interrupt — approving...
#   Interrupt — approving...
#
# PURCHASE ORDER — 4 items:
#   • USB cable: $5.99 (auto_approved)
#   • Mechanical keyboard: $45.00 (human_approved)
#   • 4K monitor: $350.00 (human_approved)
#   • Mouse pad: $8.50 (auto_approved)
#   TOTAL: $409.49
#
# (2 human approvals required)

Summary

In this capsule you learned:

Deciding when to interrupt matters more than the mechanics of how to interrupt. Too many interruptions make the agent useless. Too few make it dangerous. The 4-criteria framework gives you a systematic method for finding the exact point
The 4 criteria are: Cost, Reversibility, Impact, and Confidence. Cost measures the money at stake. Reversibility measures whether you can undo the action. Impact measures how many people or systems are affected. Confidence measures how sure the agent is. A single criterion in the red is enough to escalate
The risk_score function combines the 4 criteria using max() over the first three (one critical criterion dominates) plus a penalty for low confidence. The resulting score determines: auto-approve (< 0.3), quick approve (0.3-0.7), or full review (> 0.7)
The framework gets calibrated with real data. Measure the approval rate per action. If an action has >95% approval, drop the interrupt. If it has <40% approval, the agent shouldn't be attempting that action
Per-user autonomy levels let you adapt the experience: new users get more interrupts, expert users operate with more autonomy. Same agent, different thresholds
In regulated industries, compliance takes priority over the risk score. Certain actions always require human approval, no matter how safe they look

Next capsule: the module project — you'll apply this framework to Research Assistant v3 to build v4, with human approvals, a feedback loop, and state editing.

Additional resources

LangGraph Human-in-the-Loop — Official HITL concepts in LangGraph
How to add human-in-the-loop — Practical HITL guides
interrupt() Reference — API reference for interrupt
LangGraph Command — Using Command for dynamic resume
Trust and Safety in AI Agents — Paper on trust and safety in autonomous agents

Module 9 — LangChain & LangGraph: From Chains to Agents

← PreviousHITL Patterns in Production Next →Evolving Project: Human Approvals (v4)