Mensch im Prozess für KI-Agenten: Freigabestufen, Eskalation, Audit-Protokolle
Wenn ein Agent eine menschliche Freigabe benötigt: Vier Verfahrensweisen (Freigabestufen, Eskalation, gemeinsame Bearbeitung, Stichprobenprüfung), Pause und Wiederaufnahme mit LangGraph, Zeitlimits, Audit-Logs sowie häufige Fehlerquellen.
Quellen: Agentic Architectures — Article 13: Human-in-the-Loop Patterns
Jedes Agentensystem zwingt letztendlich zu einer Entscheidung, und die meisten Teams treffen sie zufällig: Was kann der Agent tun, ohne zuerst eine Person um Rat zu fragen?
Bei einem frühen Produktionsagenten lautete die stillschweigende Antwort „alles, was die Tools zulassen“. Er las Dokumente ein, suchte in Datenbanken, eröffnete Tickets und schickte Benachrichtigungen – alles ohne menschliche Einmischung. Das funktionierte, bis eine von niemandem erwartete Schleife dazu führte, dass innerhalb von etwa neunzig Sekunden vierundvierzigdrei doppelte Tickets erstellt wurden. Niemand war in der Lage, dies zu stoppen, und als schließlich ein Mensch es bemerkte, kostete die Beseitigung der Probleme den größten Teil des Nachmittags.
Nach einem solchen Vorfall ist die Einbeziehung eines Menschen nicht mehr nur eine Notlösung, falls der Agent versagt, sondern wird zu einer wichtigen architektonischen Überlegung erster Klasse. Einige Entscheidungen erfordern die menschliche Genehmigung nicht, weil der Agent unzuverlässig ist, sondern weil ein Fehler so teuer ist, dass eine Person die Verantwortung übernehmen sollte.
Der Rest behandelt, wie man diese Kontrollpunkte gezielt platziert, wie man sie implementiert, ohne die Nützlichkeit des Agenten zu beeinträchtigen, sowie wie man die Genehmigungslogik in AWS aufbaut: welche Aktionen eine Genehmigung benötigen, die vier Hauptmuster (Genehmigungsstellen, Eskalation, kollaboratives Bearbeiten und Stichprobenkontrolle), das Anhalten und Wiederaufnehmen der Ausführung, die Genehmigungsinterface sowie die Audit-Trail und die Handhabung von Zeitüberschreitungen.
Wo die Grenze gezogen wird
Autonomie und Kontrolle stehen im Widerspruch zueinander. Ein vollständig autonomer Agent ist schnell und skalierbar, kann aber in Maschinengeschwindigkeit Schäden anrichten. Ein vollständig überwachter Agent ist sicher, doch so langsam, dass die Automatisierung ihren Sinn verliert. Zwei Faktoren bestimmen die richtige Grenze: wie rückgängig eine Aktion ist und wie teuer ein Fehler wäre.
+--------------------+------------------------+------------------------+
| | Low Error Cost | High Error Cost |
+--------------------+------------------------+------------------------+
| Reversible | AUTONOMOUS | OVERSIGHT SAMPLING |
| | (agent acts freely) | (act, log, review some)|
+--------------------+------------------------+------------------------+
| Irreversible | APPROVAL FOR NOVEL | APPROVAL GATE |
| | (approve first time, | (always require human |
| | then autonomous) | approval before act) |
+--------------------+------------------------+------------------------+
Das Lesen eines Dokuments ist rückgängig machbar und kostengünstig, weshalb es autonom ablaufen kann. Auch das Erstellen einer E-Mail ist rückgängig machbar, sodass eine Autonomie möglich ist – eventuell in Kombination mit Stichprobenverfahren. Das Versenden einer E-Mail an einen Kunden lässt sich nicht rückgängig machen und kann teuer sein, weshalb ein Genehmigungsschritt erforderlich ist. Das Löschen von Produktionsdaten ist nicht rückgängig machbar und kann katastrophale Folgen haben, weshalb ein Schritt mit einer zweiten Genehmigung notwendig ist.
Ein häufiger Fehler besteht darin, eine einzige Richtlinie auf den gesamten Agenten anzuwenden. Die Richtlinie sollte vielmehr auf der Ebene des Tools oder sogar der einzelnen Aktion festgelegt werden. Ein Agent kann frei lesen, unter Verwendung von Stichproben entwerfen und die Nachricht erst nach einem Genehmigungsschritt senden.
# harness/hitl/action_policy.py
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Callable
class ApprovalPolicy(Enum):
AUTONOMOUS = "autonomous" # act freely
OVERSIGHT_SAMPLING = "sampling" # act, log, review a sample
APPROVAL_FOR_NOVEL = "approval_novel" # approve first occurrence, then auto
APPROVAL_GATE = "approval_gate" # always require approval
DUAL_APPROVAL = "dual_approval" # require two approvers
@dataclass
class ActionPolicy:
action_name: str
policy: ApprovalPolicy
reason: str
approver_role: Optional[str] = None # required role to approve
timeout_seconds: int = 3600
timeout_action: str = "reject" # "reject" | "escalate" | "proceed"
# Example policy configuration for a customer support agent
SUPPORT_AGENT_POLICIES = {
"read_customer_record": ActionPolicy(
action_name="read_customer_record",
policy=ApprovalPolicy.AUTONOMOUS,
reason="Read-only, reversible, low risk",
),
"draft_response": ActionPolicy(
action_name="draft_response",
policy=ApprovalPolicy.OVERSIGHT_SAMPLING,
reason="Reversible but customer-facing quality matters",
),
"send_customer_email": ActionPolicy(
action_name="send_customer_email",
policy=ApprovalPolicy.APPROVAL_GATE,
reason="Irreversible, customer-facing, reputation risk",
approver_role="support_agent",
timeout_seconds=1800,
timeout_action="reject",
),
"issue_refund": ActionPolicy(
action_name="issue_refund",
policy=ApprovalPolicy.DUAL_APPROVAL,
reason="Financial impact, irreversible",
approver_role="support_manager",
timeout_seconds=7200,
timeout_action="escalate",
),
"delete_account": ActionPolicy(
action_name="delete_account",
policy=ApprovalPolicy.DUAL_APPROVAL,
reason="Catastrophic and irreversible",
approver_role="senior_manager",
timeout_seconds=86400,
timeout_action="reject",
),
}
class ActionPolicyEngine:
"""
Determines whether an action requires human approval and how.
Consulted at the tool execution layer before any action runs.
"""
def __init__(self, policies: dict):
self.policies = policies
def get_policy(self, action_name: str) -> ActionPolicy:
return self.policies.get(
action_name,
# Default to approval gate for unknown actions: fail safe
ActionPolicy(
action_name=action_name,
policy=ApprovalPolicy.APPROVAL_GATE,
reason="Unknown action, defaulting to approval required",
)
)
def requires_approval(self, action_name: str, is_novel: bool = False) -> bool:
policy = self.get_policy(action_name)
if policy.policy == ApprovalPolicy.AUTONOMOUS:
return False
if policy.policy == ApprovalPolicy.OVERSIGHT_SAMPLING:
return False # acts first, reviewed after
if policy.policy == ApprovalPolicy.APPROVAL_FOR_NOVEL:
return is_novel
return True # APPROVAL_GATE and DUAL_APPROVAL always require approval
Muster 1: Genehmigungsschleusen
Eine Genehmigungsschleuse hält den Agenten vor einer Aktion auf und wartet darauf, dass eine Person sie annimmt oder ablehnt. Es handelt sich um das häufigste und wichtigste Muster. Das Problem ist, dass ein Agent den Prozess während des Wartens nicht am Laufen halten kann, da es dem Prüfer Minuten oder Stunden dauern kann.
LangGraph bewältigt dies durch Unterbrechungen: Die Ausführung stoppt an einem definierten Punkt, der Zustand wird gespeichert, und die Ausführung setzt fort, sobald eine Entscheidung vorliegt. Die Dokumentation von LangGraph zu menschlicher Beteiligung mit Unterbrechungen findet sich unter langchain-ai.github.io/langgraph/concepts/human_in_the_loop/.
# harness/hitl/approval_gate.py
import boto3
import time
import uuid
import json
from typing import Optional, Literal
from dataclasses import dataclass
from langgraph.types import interrupt, Command
@dataclass
class ApprovalRequest:
request_id: str
run_id: str
action_name: str
action_args: dict
context_summary: str
requested_at: float
approver_role: str
status: str # PENDING | APPROVED | REJECTED | TIMEOUT
decided_by: Optional[str] = None
decided_at: Optional[float] = None
decision_note: Optional[str] = None
class ApprovalGateManager:
"""
Manages approval requests using LangGraph interrupts for pause/resume
and DynamoDB for durable request state.
"""
def __init__(
self,
table_name: str = "agent-approval-requests",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
self.sns = boto3.client("sns", region_name=region)
def create_approval_request(
self,
run_id: str,
action_name: str,
action_args: dict,
context_summary: str,
approver_role: str,
notification_topic_arn: Optional[str] = None,
) -> ApprovalRequest:
"""
Creates a pending approval request and notifies approvers.
"""
request = ApprovalRequest(
request_id=str(uuid.uuid4()),
run_id=run_id,
action_name=action_name,
action_args=action_args,
context_summary=context_summary,
requested_at=time.time(),
approver_role=approver_role,
status="PENDING",
)
self.table.put_item(Item={
"request_id": request.request_id,
"run_id": request.run_id,
"action_name": request.action_name,
"action_args": json.dumps(request.action_args),
"context_summary": request.context_summary,
"requested_at": int(request.requested_at),
"approver_role": request.approver_role,
"status": "PENDING",
})
# Notify approvers
if notification_topic_arn:
self.sns.publish(
TopicArn=notification_topic_arn,
Subject=f"Approval needed: {action_name}",
Message=json.dumps({
"request_id": request.request_id,
"action": action_name,
"context": context_summary,
"approver_role": approver_role,
}),
)
return request
def record_decision(
self,
request_id: str,
decision: Literal["APPROVED", "REJECTED"],
decided_by: str,
decision_note: Optional[str] = None,
) -> bool:
"""
Records a human's approval decision.
Called by the approval interface when a human responds.
"""
self.table.update_item(
Key={"request_id": request_id},
UpdateExpression=(
"SET #s = :status, decided_by = :by, "
"decided_at = :at, decision_note = :note"
),
ExpressionAttributeNames={"#s": "status"},
ExpressionAttributeValues={
":status": decision,
":by": decided_by,
":at": int(time.time()),
":note": decision_note or "",
},
# Only allow decision on PENDING requests: prevents double-decision
ConditionExpression="#s = :pending",
ExpressionAttributeValues2={":pending": "PENDING"} if False else None,
)
return True
def get_decision(self, request_id: str) -> Optional[str]:
"""Polls the current decision status for a request."""
response = self.table.get_item(Key={"request_id": request_id})
item = response.get("Item")
return item.get("status") if item else None
# The LangGraph node that implements the approval gate
def approval_gate_node(state: dict) -> dict:
"""
A LangGraph node that pauses execution and waits for human approval.
Uses interrupt() to suspend the graph. The graph state is persisted
and can be resumed when the approval decision arrives.
"""
pending_action = state.get("pending_action")
if not pending_action:
return state
# interrupt() pauses the graph and surfaces this data to the caller.
# The caller (your application) presents it to a human and resumes
# the graph with the decision.
decision = interrupt({
"type": "approval_required",
"action": pending_action["name"],
"args": pending_action["args"],
"context": state.get("context_summary", ""),
"run_id": state.get("agent_run_id"),
})
# When resumed, decision contains the human's response
if decision.get("approved"):
return {
**state,
"action_approved": True,
"approved_by": decision.get("approver"),
}
else:
return {
**state,
"action_approved": False,
"rejection_reason": decision.get("reason", "Rejected by human reviewer"),
}
Muster 2: Eskalation
Eskalation unterscheidet sich von einer Kontrollstufe. Eine Kontrollstufe wird im Voraus geplant, weil bereits bekannt ist, dass eine Aktion einer Genehmigung bedarf. Eskalation hingegen ist reaktiv: Der Agent stößt auf etwas, das er nicht oder nicht sollte bewältigen können, und leitet es an eine andere Person weiter.
Dafür muss der Agent erkennen, wann er überfordert ist – was schwieriger ist, als es klingt, da Agenten eher zu selbstbewussten Handlungen neigen, anstatt Unsicherheit einzugestehen.
# harness/hitl/escalation.py
import boto3
import time
import json
from typing import Optional
from dataclasses import dataclass
from langchain_aws import ChatBedrock
from langchain_core.messages import SystemMessage, HumanMessage
ESCALATION_TRIGGERS = {
"low_confidence": "Agent confidence in its solution is below threshold",
"conflicting_information": "Retrieved information contradicts itself",
"policy_ambiguity": "The correct action is genuinely ambiguous under policy",
"high_stakes_uncertainty": "High-stakes decision with insufficient certainty",
"repeated_failure": "Agent has failed the same task multiple times",
"explicit_user_request": "User asked to speak with a human",
}
@dataclass
class EscalationEvent:
escalation_id: str
run_id: str
trigger: str
agent_context: str
agent_attempted_solution: Optional[str]
confidence: float
escalated_at: float
assigned_to: Optional[str] = None
resolution: Optional[str] = None
class EscalationManager:
"""
Handles cases where the agent should hand off to a human.
Distinct from approval gates: escalation is triggered by the agent
recognizing its own limitations.
"""
def __init__(
self,
table_name: str = "agent-escalations",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
bedrock = boto3.client("bedrock-runtime", region_name=region)
self.confidence_model = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-haiku-4-5",
model_kwargs={"temperature": 0, "max_tokens": 256},
)
def should_escalate(
self,
task: str,
proposed_solution: str,
attempts: int,
) -> tuple:
"""
Assesses whether the agent should escalate to a human.
Returns (should_escalate, trigger, confidence).
"""
# Repeated failure is a deterministic trigger
if attempts >= 3:
return True, "repeated_failure", 0.0
# Ask the model to self-assess confidence
response = self.confidence_model.invoke([
SystemMessage(content="""
Assess your confidence in a proposed solution. Be honest about uncertainty.
Consider: Is the information sufficient? Is the answer unambiguous?
Are there conflicting considerations? Is this high-stakes?
Return JSON:
{
"confidence": 0.0 to 1.0,
"should_escalate": true | false,
"trigger": "low_confidence | conflicting_information | policy_ambiguity | high_stakes_uncertainty | none"
}
"""),
HumanMessage(content=f"Task: {task}\n\nProposed solution: {proposed_solution}")
])
try:
assessment = json.loads(response.content)
return (
assessment.get("should_escalate", False),
assessment.get("trigger", "low_confidence"),
assessment.get("confidence", 0.5),
)
except json.JSONDecodeError:
# If we cannot assess confidence, escalate to be safe
return True, "low_confidence", 0.0
def create_escalation(
self,
run_id: str,
trigger: str,
agent_context: str,
attempted_solution: Optional[str],
confidence: float,
notification_topic_arn: Optional[str] = None,
) -> EscalationEvent:
import uuid
event = EscalationEvent(
escalation_id=str(uuid.uuid4()),
run_id=run_id,
trigger=trigger,
agent_context=agent_context,
agent_attempted_solution=attempted_solution,
confidence=confidence,
escalated_at=time.time(),
)
self.table.put_item(Item={
"escalation_id": event.escalation_id,
"run_id": event.run_id,
"trigger": event.trigger,
"agent_context": event.agent_context[:2000],
"attempted_solution": (attempted_solution or "")[:2000],
"confidence": str(confidence),
"escalated_at": int(event.escalated_at),
"status": "OPEN",
})
if notification_topic_arn:
boto3.client("sns").publish(
TopicArn=notification_topic_arn,
Subject=f"Agent escalation: {trigger}",
Message=event.agent_context[:1000],
)
return event
Muster 3: gemeinsame Bearbeitung
Oftmals ist das richtige Vorgehen weder Zustimmung noch Ablehnung, sondern Bearbeitung. Der Agent erstellt einen Entwurf, eine Person überarbeitet ihn, und die finale Fassung wird weiterverwendet. Dies eignet sich für die Erstellung von Inhalten, das Ausarbeiten von Dokumenten sowie Code, den ein Mensch schließlich übernimmt, nachdem der Agent ihn verfasst hat.
Das Designprinzip besagt, dass die Änderungen des Menschen als Lernsignal zurückfließen sollten. Wenn die Überprüfer das Ergebnis ständig auf dieselbe Weise ändern, sollte der Agent daraus lernen.
# harness/hitl/collaborative.py
import boto3
import time
import json
from typing import Optional
from dataclasses import dataclass
@dataclass
class CollaborativeEdit:
edit_id: str
run_id: str
original_draft: str
human_edited: str
edit_distance: float # how much was changed
edit_categories: list # "tone", "factual", "structure", "detail"
edited_by: str
edited_at: float
class CollaborativeEditManager:
"""
Manages the draft-edit-finalize flow where a human refines agent output.
Captures edits as learning signals for future improvement.
"""
def __init__(
self,
table_name: str = "agent-collaborative-edits",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
def record_edit(
self,
run_id: str,
original_draft: str,
human_edited: str,
edited_by: str,
edit_categories: list = None,
) -> CollaborativeEdit:
"""
Records a human edit to agent output.
The edit distance and categories become learning signals.
"""
import uuid
edit_distance = self._compute_edit_distance(original_draft, human_edited)
edit = CollaborativeEdit(
edit_id=str(uuid.uuid4()),
run_id=run_id,
original_draft=original_draft,
human_edited=human_edited,
edit_distance=edit_distance,
edit_categories=edit_categories or [],
edited_by=edited_by,
edited_at=time.time(),
)
self.table.put_item(Item={
"edit_id": edit.edit_id,
"run_id": edit.run_id,
"original_draft": original_draft[:5000],
"human_edited": human_edited[:5000],
"edit_distance": str(edit_distance),
"edit_categories": edit.edit_categories,
"edited_by": edited_by,
"edited_at": int(edit.edited_at),
})
return edit
def analyze_edit_patterns(
self,
run_ids: list = None,
min_edits: int = 20,
) -> dict:
"""
Analyzes accumulated edits to find systematic patterns.
If humans consistently edit for the same reason, the agent's
system prompt or procedure should be updated.
"""
response = self.table.scan(Limit=200)
edits = response.get("Items", [])
if len(edits) < min_edits:
return {"sufficient_data": False, "edit_count": len(edits)}
# Aggregate edit categories
category_counts = {}
total_distance = 0.0
for edit in edits:
for cat in edit.get("edit_categories", []):
category_counts[cat] = category_counts.get(cat, 0) + 1
total_distance += float(edit.get("edit_distance", 0))
avg_distance = total_distance / len(edits)
dominant_category = max(category_counts, key=category_counts.get) if category_counts else None
return {
"sufficient_data": True,
"edit_count": len(edits),
"average_edit_distance": round(avg_distance, 3),
"category_distribution": category_counts,
"dominant_edit_category": dominant_category,
"recommendation": self._recommendation(dominant_category, avg_distance),
}
def _recommendation(self, dominant_category: Optional[str], avg_distance: float) -> str:
if avg_distance < 0.1:
return "Edits are minor. Agent output quality is high."
if dominant_category == "tone":
return "Humans frequently adjust tone. Update system prompt with tone guidance."
if dominant_category == "factual":
return "Humans frequently correct facts. Review agent's information sources."
if dominant_category == "structure":
return "Humans frequently restructure. Add output format guidance to system prompt."
return "Review edit patterns manually for systematic improvements."
def _compute_edit_distance(self, a: str, b: str) -> float:
"""Normalized Levenshtein distance, 0 (identical) to 1 (completely different)."""
if not a and not b:
return 0.0
# Simple word-level distance for efficiency on long text
a_words, b_words = a.split(), b.split()
max_len = max(len(a_words), len(b_words))
if max_len == 0:
return 0.0
# Count matching words in order (simplified)
common = len(set(a_words) & set(b_words))
return 1.0 - (common / max_len)
Muster 4: Überwachungsstichprobenziehung
Für umkehrbare Aktionen mit hohen Fehlerrisiken ist eine Genehmigung bei jeder Ausführung zu langsam. Die Stichprobenziehung stellt einen Kompromiss dar: Der Agent handelt eigenständig, während eine Auswahl seiner Aktionen anschließend zur Überprüfung an Menschen gesendet wird. Die Überprüfung blockiert nichts, erkennt jedoch systematische Probleme und fließt in den Bewertungsprozess ein.
# harness/hitl/oversight_sampling.py
import boto3
import random
import time
import json
from typing import Optional
class OversightSampler:
"""
Routes a sample of autonomous actions to human review after execution.
Does not block the action. Catches quality drift and systematic errors.
Feeds into the Article 8 evaluation pipeline.
"""
def __init__(
self,
base_sample_rate: float = 0.05,
table_name: str = "agent-oversight-samples",
region: str = "us-east-1",
):
self.base_sample_rate = base_sample_rate
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
def should_sample(
self,
action_name: str,
agent_confidence: float = 1.0,
is_novel: bool = False,
) -> bool:
"""
Decides whether this action should be sampled for review.
Samples more aggressively for low-confidence and novel actions.
"""
rate = self.base_sample_rate
# Increase sampling for low confidence
if agent_confidence < 0.7:
rate = min(rate * 3, 1.0)
# Always sample novel actions
if is_novel:
return True
return random.random() < rate
def record_for_review(
self,
run_id: str,
action_name: str,
action_args: dict,
action_result: str,
agent_confidence: float,
):
"""Queues an executed action for human review."""
import uuid
self.table.put_item(Item={
"sample_id": str(uuid.uuid4()),
"run_id": run_id,
"action_name": action_name,
"action_args": json.dumps(action_args)[:2000],
"action_result": action_result[:2000],
"agent_confidence": str(agent_confidence),
"sampled_at": int(time.time()),
"review_status": "PENDING",
})
def record_review_outcome(
self,
sample_id: str,
was_correct: bool,
reviewer: str,
notes: Optional[str] = None,
):
"""
Records the human's assessment of a sampled action.
A pattern of incorrect actions should trigger a policy review.
"""
self.table.update_item(
Key={"sample_id": sample_id},
UpdateExpression=(
"SET review_status = :s, was_correct = :c, "
"reviewed_by = :r, review_notes = :n, reviewed_at = :at"
),
ExpressionAttributeValues={
":s": "REVIEWED",
":c": was_correct,
":r": reviewer,
":n": notes or "",
":at": int(time.time()),
}
)
def get_error_rate(self, action_name: str, days: int = 7) -> dict:
"""
Computes the error rate for a sampled action over a window.
If this exceeds threshold, the action's policy should be tightened.
"""
cutoff = int(time.time()) - (days * 86400)
response = self.table.scan(
FilterExpression="action_name = :a AND sampled_at > :c AND review_status = :s",
ExpressionAttributeValues={
":a": action_name,
":c": cutoff,
":s": "REVIEWED",
}
)
reviews = response.get("Items", [])
if not reviews:
return {"sufficient_data": False}
incorrect = sum(1 for r in reviews if not r.get("was_correct", True))
error_rate = incorrect / len(reviews)
return {
"sufficient_data": len(reviews) >= 10,
"sample_size": len(reviews),
"error_rate": round(error_rate, 3),
"recommendation": (
"TIGHTEN_POLICY" if error_rate > 0.1 else "MAINTAIN"
),
}
Zeitüberschreitungen und Alternativen
Jeder Gate benötigt eine Zeitüberschreitungsrichtlinie, denn es gibt keine sichere Standardlösung dafür, was passiert, wenn niemand antwortet. Automatisches Weitermachen macht den Gate nutzlos. Automatische Ablehnung kann legitime Aufgaben unbegrenzt blockieren. Die richtige Reaktion hängt davon ab, wie riskant die Aktion ist – deshalb gehört sie zur zuvor definierten ActionPolicy.
# harness/hitl/timeout_handler.py
import boto3
import time
from typing import Literal
class ApprovalTimeoutHandler:
"""
Handles approval requests that exceed their timeout.
The timeout action is policy-defined per action type.
Runs as a scheduled Lambda, checking for expired pending requests.
"""
def __init__(
self,
approval_table: str = "agent-approval-requests",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(approval_table)
self.sns = boto3.client("sns", region_name=region)
def process_expired_requests(self, policies: dict) -> dict:
"""
Scans for pending requests past their timeout and applies
the policy-defined timeout action.
"""
now = int(time.time())
response = self.table.scan(
FilterExpression="#s = :pending",
ExpressionAttributeNames={"#s": "status"},
ExpressionAttributeValues={":pending": "PENDING"},
)
results = {"rejected": 0, "escalated": 0, "proceeded": 0}
for item in response.get("Items", []):
requested_at = int(item.get("requested_at", now))
action_name = item.get("action_name", "")
policy = policies.get(action_name)
if not policy:
continue
age = now - requested_at
if age < policy.timeout_seconds:
continue # not expired yet
timeout_action = policy.timeout_action
if timeout_action == "reject":
self._apply_timeout(item["request_id"], "REJECTED",
"Timed out without approval")
results["rejected"] += 1
elif timeout_action == "escalate":
self._escalate_expired(item)
results["escalated"] += 1
elif timeout_action == "proceed":
# Only for low-risk actions where the gate is advisory
self._apply_timeout(item["request_id"], "APPROVED",
"Auto-approved after timeout per policy")
results["proceeded"] += 1
return results
def _apply_timeout(self, request_id: str, status: str, note: str):
self.table.update_item(
Key={"request_id": request_id},
UpdateExpression="SET #s = :status, decision_note = :note, decided_at = :at",
ExpressionAttributeNames={"#s": "status"},
ExpressionAttributeValues={
":status": status,
":note": note,
":at": int(time.time()),
}
)
def _escalate_expired(self, item: dict):
# Notify a higher tier and extend the deadline
self.table.update_item(
Key={"request_id": item["request_id"]},
UpdateExpression="SET escalated = :t, escalated_at = :at",
ExpressionAttributeValues={":t": True, ":at": int(time.time())},
)
Audit-Trail
Jede menschliche Entscheidung benötigt eine unverfälschbare Aufzeichnung: Wer hat was, wann und unter welchem Kontext genehmigt sowie mit welchem Ergebnis. Dies dient der Konformitätskontrolle und dem Fehlerbeheben sowie der Rechenschaftspflicht, die es den Menschen ermöglicht, überhaupt an diesen Prozessen teilzunehmen.
# harness/hitl/audit.py
import boto3
import time
import json
import hashlib
from typing import Optional
class HITLAuditLog:
"""
Immutable audit trail for all human-in-the-loop decisions.
Uses a hash chain so tampering is detectable.
Writes to DynamoDB with a separate append-only access pattern.
"""
def __init__(
self,
table_name: str = "agent-hitl-audit",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
def record_decision(
self,
run_id: str,
decision_type: str, # "approval" | "rejection" | "escalation" | "edit"
action_name: str,
decided_by: str,
decision_details: dict,
previous_hash: Optional[str] = None,
) -> str:
"""
Records a decision in the audit log with a hash chain.
Returns the hash of this entry for chaining the next one.
"""
timestamp = int(time.time())
entry = {
"run_id": run_id,
"decision_type": decision_type,
"action_name": action_name,
"decided_by": decided_by,
"decision_details": json.dumps(decision_details),
"timestamp": timestamp,
"previous_hash": previous_hash or "genesis",
}
# Compute hash of this entry chained to the previous
entry_content = json.dumps(entry, sort_keys=True)
entry_hash = hashlib.sha256(entry_content.encode()).hexdigest()
self.table.put_item(Item={
"audit_id": f"{run_id}#{timestamp}#{entry_hash[:8]}",
**entry,
"entry_hash": entry_hash,
})
return entry_hash
def verify_chain(self, run_id: str) -> bool:
"""
Verifies the hash chain for a run's audit entries.
Returns True if the chain is intact, False if tampering is detected.
"""
response = self.table.query(
KeyConditionExpression="run_id = :rid",
ExpressionAttributeValues={":rid": run_id},
ScanIndexForward=True,
)
entries = sorted(response.get("Items", []), key=lambda x: x["timestamp"])
previous_hash = "genesis"
for entry in entries:
if entry.get("previous_hash") != previous_hash:
return False
# Recompute and verify
check_entry = {
k: entry[k] for k in
["run_id", "decision_type", "action_name", "decided_by",
"decision_details", "timestamp", "previous_hash"]
}
recomputed = hashlib.sha256(
json.dumps(check_entry, sort_keys=True).encode()
).hexdigest()
if recomputed != entry.get("entry_hash"):
return False
previous_hash = entry["entry_hash"]
return True
Realitätscheck aus der Produktion
Human-in-the-loop ist der Punkt, an dem das Streben nach vollständiger Automatisierung auf Verantwortung trifft. Aus der Praxis ergeben sich mehrere Lehren.
Auslastung durch Genehmigungen ist gefährlich. Wenn Menschen zu vielen Entscheidungen gebracht werden, hören sie auf zu lesen und beginnen nur noch zu bestätigen. Eine Kontrollstelle, die immer zustimmt, ist schlimmer als keine, da sie den Anschein von Überwachung ohne echte Kontrolle erweckt. Verfolgen Sie die Genehmigungsraten – eine Kontrollstelle, die zu 99 Prozent zustimmt, ist entweder überflüssig oder wird nicht mehr sorgfältig geprüft.
Der Kontext bestimmt die Qualität. Jemand, der nur gebeten wird, „send_email“ ohne weitere Informationen zu genehmigen, trifft schlechtere Entscheidungen als jemand, der den Nachrichtentext, die Kundenhistorie sowie den Grund des Mitarbeiters für die Auswahl sieht. Die Qualität der Überwachung ist durch die bereitgestellten Informationen begrenzt, weshalb die Genehmigungssoftware eine echte Investition verdient.
Timeouts benötigen Verantwortliche. Eine Anfrage, die einen Timeout erreicht und automatisch abgelehnt wird, kann stillschweigend wichtige Arbeiten blockieren. Jemand muss sich um die Warteschlange der ausstehenden und abgelaufenen Anfragen kümmern. Erstellen Sie ein Dashboard, das auf alternde Anfragen hinweist, und definieren Sie einen klaren Eskalationsweg, falls der Hauptzustimmungsperson nicht zur Verfügung steht.
Änderungen sind das beste Schulungsinstrument und werden in der Regel nicht genutzt. Jedes Mal, wenn jemand die Ausgabe des Assistenten ändert, zeigt das genau an, wo der Assistent versagt. Eine analyze_edit_patterns-Funktion kann diese Änderungen in Verbesserungen umwandeln. Wenn die Prüfer ständig dasselbe beheben, sollten Sie dies in den Systemhinweisen korrigieren. Dadurch wird der Kreislauf zwischen Überwachung und Verbesserung geschlossen.
In regulierten Branchen ist ein Audit-Trail unerlässlich. In der Finanzbranche, im Gesundheitswesen oder im Rechtswesen ist ein mit Hash-Werten versehener Audit-Log nicht nur wünschenswert, sondern sogar vorgeschrieben. Die Aufsichtsbehörden fragen nach, wer eine bestimmte Aktion genehmigt hat, und die Antwort muss mit der Gewissheit erfolgen, dass die Aufzeichnung nicht verändert wurde. Solche Systeme sollten von Anfang an integriert werden.
Die Latenzzeit verändert sich völlig. Ein Agent, der ohne Einschränkungen in vier Sekunden eine Aufgabe erledigt, kann mit der Einbeziehung von Zwischenschritten bis zu vier Stunden benötigen – schließlich handelt es sich dabei um die Reaktionszeit von Menschen. Das eignet sich für asynchrone Arbeitsabläufe, zerstört aber solche in Echtzeit; daher sollte die Produktgestaltung der menschlichen Latenzzeit entsprechen und nicht der Latenzzeit der Agenten.
Referenzarchitektur
Der Ablauf beginnt damit, dass der Agent eine Aktion ausführt, geht über die Überprüfung durch die Policy in eine direkte Ausführung, ein Genehmigungsverfahren, eine Eskalation oder einen stochastischen Pfad über, und endet schließlich durch die Zeitüberschreitungsbearbeitung sowie das Audit-Log.
Agent reaches an action
|
v
+---------------------------+
| ActionPolicyEngine |
| Look up action policy |
+---------------------------+
|
+-----+-----+-----------+-----------+
| | | | |
v v v v v
AUTONO SAMP APPROVAL APPROVAL DUAL
MOUS LING FOR NOVEL GATE APPROVAL
| | | | |
| | v v v
| | +--------------------------------+
| | | ApprovalGateManager |
| | | - create request |
| | | - LangGraph interrupt() |
| | | - notify approvers (SNS) |
| | | - persist state (DynamoDB) |
| | +--------------------------------+
| | |
| | Human decides via interface
| | |
| | +-----+-----+
| | | |
| | APPROVED REJECTED / TIMEOUT
| | | |
v v v v
+--------------------------------+
| Execute or Abort Action |
+--------------------------------+
|
v
+--------------------------------+
| HITLAuditLog (hash chain) |
| Immutable decision record |
+--------------------------------+
|
v
+--------------------------------+
| Feed to Article 8 eval + |
| collaborative edit learning |
+--------------------------------+
Referenz-Infrastruktur-Stack
Die Komponenten, die Technologie hinter jedem von ihnen sowie ihre Funktion sind in der untenstehenden Referenztabelle dargestellt. Sie umfasst den Freigabestatus, Benachrichtigungen, eine dauerhafte Wiederaufnahme, Audit-Speicherung sowie Überwachungs-Dashboards auf AWS.
+-----------------------------+---------------------+------------------------------+
| Component | Technology | Role |
+-----------------------------+---------------------+------------------------------+
| Action Policy Engine | Custom | Per-action approval routing |
| | | based on risk profile |
+-----------------------------+---------------------+------------------------------+
| Pause / Resume | LangGraph interrupt | Durable suspend while |
| | + checkpointing | waiting for human |
+-----------------------------+---------------------+------------------------------+
| Approval Requests | DynamoDB | Pending request state, |
| | | decision records |
+-----------------------------+---------------------+------------------------------+
| Notifications | SNS | Alert approvers when a |
| | | decision is needed |
+-----------------------------+---------------------+------------------------------+
| Escalation | Custom + Haiku | Agent self-assessment of |
| | confidence model | when to hand off to human |
+-----------------------------+---------------------+------------------------------+
| Collaborative Edits | DynamoDB | Draft-edit-finalize flow, |
| | | edit pattern learning |
+-----------------------------+---------------------+------------------------------+
| Oversight Sampling | Custom + DynamoDB | Post-hoc review of a sample |
| | | of autonomous actions |
+-----------------------------+---------------------+------------------------------+
| Timeout Handling | Scheduled Lambda | Policy-defined action on |
| | | expired approval requests |
+-----------------------------+---------------------+------------------------------+
| Audit Trail | DynamoDB hash chain | Immutable, tamper-evident |
| | | record of all decisions |
+-----------------------------+---------------------+------------------------------+
| Auth for Approvers | CredentialManager | Role-based approval |
| | (Article 5) | authorization |
+-----------------------------+---------------------+------------------------------+
| Learning Loop | Article 8 eval | Edits and reviews feed the |
| | pipeline | improvement pipeline |
+-----------------------------+---------------------+------------------------------+
| Local Dev Alternative | Docker Compose + | Full HITL flow testable |
| | mock approver UI | without AWS |
+-----------------------------+---------------------+------------------------------+