Human-in-the-Loop Basics

python -m venv .venv && source .venv/bin/activate
# stdlib only for the examples below

from dataclasses import dataclass
 
 
@dataclass(frozen=True)
class ToolSpec:
    name: str
    requires_approval: bool
 
 
TOOLS = {
    "search": ToolSpec("search", requires_approval=False),
    "draft_email": ToolSpec("draft_email", requires_approval=False),
    "send_email": ToolSpec("send_email", requires_approval=True),
}
 
for t in TOOLS.values():
    mode = "GATE" if t.requires_approval else "AUTO"
    print(f"{t.name:12} {mode}")

from dataclasses import dataclass
 
 
@dataclass
class PendingAction:
    tool: str
    args: dict
    status: str = "awaiting_approval"
 
 
def dispatch(tool: str, args: dict) -> object:
    spec = TOOLS[tool]
    if spec.requires_approval:
        return PendingAction(tool=tool, args=args)
    return {"ok": True, "tool": tool, "args": args, "result": "done"}
 
 
print(dispatch("search", {"q": "refund policy"}))
print(dispatch("send_email", {"to": "user@example.com", "body": "Hello"}))

def decision_packet(pending: PendingAction) -> dict:
    return {
        "title": f"Approve {pending.tool}?",
        "tool": pending.tool,
        "args": pending.args,
        "risk": "external_message",
        "choices": ["approve", "reject", "edit"],
    }
 
 
pending = PendingAction("send_email", {"to": "user@example.com", "body": "Hello"})
print(decision_packet(pending))

def apply_decision(pending: PendingAction, decision: str, edited_args: dict | None = None) -> dict:
    if decision == "reject":
        return {"status": "rejected", "tool": pending.tool}
    if decision == "edit":
        args = edited_args or pending.args
    elif decision == "approve":
        args = pending.args
    else:
        raise ValueError(f"unknown decision: {decision}")
    # Real code: call the email provider here
    return {"status": "executed", "tool": pending.tool, "args": args}
 
 
print(apply_decision(pending, "approve"))
print(apply_decision(pending, "reject"))

def model_choose_tool(user_goal: str) -> tuple[str, dict]:
    # Stand-in for an LLM tool call
    if "email" in user_goal.lower():
        return "send_email", {"to": "user@example.com", "body": "We processed your request."}
    return "search", {"q": user_goal}
 
 
def agent_step(user_goal: str, auto_approve: bool = False) -> dict:
    tool, args = model_choose_tool(user_goal)
    outcome = dispatch(tool, args)
    if isinstance(outcome, PendingAction):
        if auto_approve:
            return apply_decision(outcome, "approve")
        return {"status": "awaiting_approval", "packet": decision_packet(outcome)}
    return {"status": "ok", "result": outcome}
 
 
print(agent_step("Find the refund policy"))
print(agent_step("Email the customer a confirmation"))

from uuid import uuid4
 
 
RUNS: dict[str, dict] = {}
 
 
def start_run(goal: str) -> dict:
    run_id = str(uuid4())
    tool, args = model_choose_tool(goal)
    outcome = dispatch(tool, args)
    if isinstance(outcome, PendingAction):
        RUNS[run_id] = {"goal": goal, "pending": outcome, "status": "awaiting_approval"}
        return {"run_id": run_id, "status": "awaiting_approval", "packet": decision_packet(outcome)}
    RUNS[run_id] = {"goal": goal, "status": "done", "result": outcome}
    return {"run_id": run_id, "status": "done", "result": outcome}
 
 
def resume_run(run_id: str, decision: str, edited_args: dict | None = None) -> dict:
    run = RUNS[run_id]
    if run["status"] != "awaiting_approval":
        return {"error": "not_waiting"}
    result = apply_decision(run["pending"], decision, edited_args)
    run["status"] = result["status"]
    run["result"] = result
    return {"run_id": run_id, **result}
 
 
paused = start_run("Email the customer a confirmation")
print(paused)
print(resume_run(paused["run_id"], "approve"))

from enum import Enum
 
 
class GateMode(Enum):
    AUTO = "auto"
    NOTIFY = "notify"
    HARD = "hard"
 
 
POLICY = {
    "search": GateMode.AUTO,
    "draft_email": GateMode.AUTO,
    "post_internal_note": GateMode.NOTIFY,
    "send_email": GateMode.HARD,
}
 
 
def plan_execution(tool: str) -> str:
    mode = POLICY[tool]
    if mode is GateMode.AUTO:
        return "execute_now"
    if mode is GateMode.NOTIFY:
        return "execute_and_notify"
    return "wait_for_approval"
 
 
print(plan_execution("post_internal_note"))
print(plan_execution("send_email"))

class KillSwitch:
    def __init__(self) -> None:
        self.tripped = False
 
    def trip(self, reason: str) -> None:
        self.tripped = True
        self.reason = reason
 
    def check(self) -> None:
        if self.tripped:
            raise RuntimeError(f"killed: {self.reason}")
 
 
def bounded_loop(steps: list[str], kill: KillSwitch) -> list[str]:
    out: list[str] = []
    for i, step in enumerate(steps):
        kill.check()
        if i == 3:
            kill.trip("too many steps in demo")
        out.append(step)
    return out
 
 
ks = KillSwitch()
try:
    print(bounded_loop(["a", "b", "c", "d", "e"], ks))
except RuntimeError as e:
    print(e)