Ludzki nadzór w agentach AI: bramy zatwierdzające, eskalacja, ślady audytowe
W sytuacjach, gdy agent potrzebuje zatwierdzenia przez człowieka, istnieją cztery wzorce postępowania (bramy zatwierdzające, eskalacja, wspólne edycje, pobieranie próbek), możliwość pauzowania i kontynuowania pracy za pomocą LangGraph, ograniczenia czasowe, logi audytowe oraz potencjalne problemy.
Źródła: Agentic Architectures — Article 13: Human-in-the-Loop Patterns
Każdy system agentów ostatecznie zmusza do podjęcia jednej decyzji, a większość zespołów dochodzi do niej przypadkowo: co może zrobić agent, nie pytając najpierw człowieka?
W jednym z wczesnych agentów produkcyjnych nieujawnioną odpowiedzią było „wszystko, na co pozwalają narzędzia”. Samodzielnie czytał dokumenty, wyszukiwał dane w bazach danych, otwierał zgłoszenia i wysyłał powiadomienia. Taka sytuacja trwała, dopóki pętla, której nikt się nie spodziewał, nie sprawiła, że w ciągu około dziewięćdziesięciu sekund utworzył czterdzieści trzy duplikatowe zgłoszenia. Nikt nie był w stanie tego zatrzymać, a gdy ktoś to zauważył, usuwanie skutków pochłonęło większość popołudnia.
Po takim incydencie obecność człowieka w procesie decyzyjnym przestaje być rozwiązaniem awaryjnym na wypadek awarii agenta i staje się kluczowym aspektem projektu architektonicznego. Niektóre decyzje wymagają zatwierdzenia człowieka nie dlatego, że agent jest niestabilny, ale dlatego, że błąd może mieć tak duże konsekwencje, iż decyzja powinna być podejmowana przez osobę ludzką.
Pozostała część opisuje, jak celowo umieszczać te punkty kontrolne, jak wdrożyć je bez niszczenia użyteczności agenta oraz jak stworzyć mechanizmy zatwierdzania na AWS: wybór działań wymagających zatwierdzenia, cztery główne wzorce (bramy zatwierdzania, eskalacja, wspólne edytowanie i próbkowanie nadzoru), wstrzymywanie i wznowianie wykonywania, interfejs zatwierdzania oraz ślad audytowy, a także obsługę przekroczenia czasu.
Gdzie wyznaczyć granicę
Autonomia i kontrola działają przeciwko sobie. Pełnie autonomiczny agent jest szybki i skalowalny, ale może wyrządzić szkody z prędkością maszyny. Pełnie nadzorowany agent jest bezpieczny, ale tak wolny, że automatyzacja traci sens. Dwie kryteria pomagają określić właściwą granicę: stopień odwracalności działania oraz koszt błędu.
+--------------------+------------------------+------------------------+
| | 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) |
+--------------------+------------------------+------------------------+
Czytanie dokumentu jest odwracalne i tanie, więc może odbywać się samodzielnie. Tworzenie e-maila również jest odwracalne, więc możliwa jest autonomiczna realizacja, być może z wykorzystaniem próbkowania. Wysyłanie e-maila do klienta nie da się cofnąć i może być bardzo kosztowne, dlatego wymaga bramy zatwierdzającej. Usuwanie danych produkcyjnych jest nieodwracalne i może mieć katastrofalne konsekwencje, więc wymaga bramy z drugim osobą upoważnioną do zatwierdzenia.
Częstym błędem jest stosowanie jednej polityki dla całego agenta. Polityka powinna być zdefiniowana na poziomie narzędzia lub nawet pojedynczej akcji. Jeden agent może swobodnie czytać, tworzyć treści przy użyciu próbkowania oraz wysyłać je po przejściu przez bramę zatwierdzającą.
# 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
Wzorzec 1: bramy zatwierdzenia
Brama zatwierdzenia zatrzymuje agenta przed wykonaniem działania i czeka, aż osoba ją zaakceptuje lub odrzuci. Jest to najczęstszy i najważniejszy wzorzec. Problem polega na tym, że agent nie może utrzymać procesu w toku podczas oczekiwania, ponieważ ocena może potrwać minuty lub godziny.
LangGraph radzi sobie z tym za pomocą przerwań: wykonywanie zostaje zatrzymane w określonym punkcie, stan jest zachowany, a proces wznowiony po otrzymaniu decyzji. Dokumentacja LangGraph dotycząca udziału człowieka w procesie z wykorzystaniem przerwań znajduje się pod adresem 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"),
}
Wzorzec 2: eskalacja
Eskalacja różni się od bramy decyzyjnej. Brama decyzyjna jest planowana, ponieważ z góry wiadomo, że dana akcja wymaga zatwierdzenia. Eskalacja jest reaktywna: agent napotyka sytuację, której nie może lub nie powinien sam sobie poradzić, i przekazuje ją osobie odpowiedzialnej.
To wymaga od agenta, aby zauważył, gdy przekracza swoje możliwości, co jest trudniejsze, niż się wydaje, ponieważ agenci wolą podejmować decyzje z pewnością siebie niż przyznać do niepewności.
# 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
Wzorzec 3: wspólne edytowanie
Czasami odpowiednim modelem nie jest ani zatwierdzenie, ani odrzucenie, lecz edytowanie. Agent pisze wstępny projekt, osoba go dopracowuje, a wersja ulepszona trafia dalej. Pasuje to do generowania treści, opracowywania dokumentów oraz kodu, który po napisaniu przez agenta trafi w ręce człowieka.
Zasada projektowa zakłada, że zmiany wprowadzane przez człowieka powinny wracać jako sygnał uczenia się. Jeśli recenzenci ciągle zmieniają wynik w ten sam sposób, agent powinien się z tego nauczyć.
# 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)
Wzorzec 4: próbkowanie z nadzorem
W przypadku działań odwracalnych o wysokim koszcie błędu, wymaganie zatwierdzenia za każdym razem jest zbyt wolne. Próbkowanie stanowi kompromis – agent działa samodzielnie, a próbka jego działań jest wysyłana do ludzi w celu późniejszej oceny. Ocena nie blokuje żadnych procesów, ale pomaga wykryć problemy systematyczne i dostarcza dane do procesu oceny.
# 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"
),
}
Czas wygaśnięcia i rozwiązania awaryjne
Każda brama potrzebuje polityki czasu wygaśnięcia, ponieważ nie istnieje bezpieczny standard na to, co się dzieje, gdy nikt nie odpowiada. Działanie automatyczne uniemożliwia prawidłowe funkcjonowanie bramy, natomiast automatyczna odrzucenie może bezterminowo zatrzymać legalną pracę. Odpowiednia reakcja zależy od stopnia ryzyka danej akcji, dlatego powinna być określona w wcześniej zdefiniowanym 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())},
)
Ślad audytowy
Każda decyzja podejmowana przez człowieka wymaga niezmienialnego zapisu: kto co zatwierdził, kiedy, w jakim kontekście i z jakim rezultatem. Służy to przestrzeganiu regulacji oraz naprawianiu błędów, a także wspiera odpowiedzialność, która umożliwia ludziom uczestnictwo w tym procesie.
# 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
Sprawdzenie rzeczywistości z produkcji
Model z udziałem człowieka to miejsce, gdzie ambicje pełnej automatyzacji spotykają się z odpowiedzialnością. Z praktyki wynikają różne wnioski.
Zmęczenie z powodu zatwierdzania jest niebezpieczne. Jeśli prosić ludzi o podejmowanie zbyt wielu decyzji, przestaną je czytać i będą jedynie je zatwierdzać bez analizy. Brama decyzyjna, która zawsze daje zgodę, jest gorsza od braku takiej bramy, ponieważ sprawia wrażenie nadzoru bez rzeczywistej kontroli. Należy śledzić wskaźniki zatwierdzania – brama, która jest akceptowana w 99% przypadków, jest albo niepotrzebna, albo już nie podlega dokładnej ocenie.
Kontekst decyduje o jakości. Osoba, która ma zatwierdzić tylko operację „send_email” bez żadnych dodatkowych informacji, podejmie gorszą decyzję niż osoba, która widzi treść wiadomości, historię klienta oraz powód wyboru przez agenta. Jakość nadzoru jest ograniczona dostępnymi informacjami, dlatego interfejs służący do zatwierdzania wymaga rzeczywistych inwestycji.
Czasochłonne operacje wymagają osoby odpowiedzialnej. Prośba, która wygaśnie i zostanie automatycznie odrzucona, może w cichu zablokować ważne prace. Ktoś musi być odpowiedzialny za kolejkę z prośbami w oczekiwaniu lub już wygasłymi. Stwórz panel sterowania, który będzie sygnalizował starzejące się prośby, oraz określ jasną ścieżkę eskalacji w przypadku, gdy główny osoba decyzyjna jest nieobecna.
Edytowanie to najlepszy sygnał do szkolenia, a zazwyczaj pozostaje niewykorzystane. Za każdym razem, gdy ktoś edytuje wynik działania agenta, pokazuje to dokładnie, w czym agent zawodzi. Funkcja analyze_edit_patterns może przekształcić te edycje w ulepszenia. Jeśli recenzenci ciągle poprawiają to samo, należy to naprawić w instrukcjach systemu. W ten sposób zamyka się pętla między nadzorem a doskonaleniem.
Przemysły regulowane wymagają śledzenia audytowego. W finansach, ochronie zdrowia czy prawie rejestr audytowy oparty na łańcuchu hashów jest obowiązkowy, a nie tylko pożądany. Organizacje nadzorcze będą pytać, kto zatwierdził dane działanie, a odpowiedź musi być dostarczana z pewnością, że zapis nie został zmieniony. Taki system powinien być wbudowany od samego początku.
Opóźnienia całkowicie zmieniają sytuację. Agent, który samodzielnie wykonuje zadanie w cztery sekundy, może potrzebować czterech godzin, gdy na jego drodze znajduje się brama kontrolna, ponieważ te godziny to czas reakcji człowieka. To pasuje do prac asynchronicznych, ale niszczy te w czasie rzeczywistym, dlatego projekt produktu powinien uwzględniać opóźnienia ludzkie, a nie opóźnienia agentów.
Architektura referencyjna
Przepływ rozpoczyna się od momentu, gdy agent podejmuje działanie, przechodzi przez weryfikację zasad, a następnie może prowadzić do bezpośredniej realizacji, bramy zatwierdzenia, eskalacji lub wybranej ścieżki, by w końcu powrócić poprzez obsługę przekroczenia czasu i rejestr audytowy.
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 |
+--------------------------------+
Stack infrastruktury referencyjnej
Komponenty, technologia stojąca za każdym z nich oraz ich rola przedstawione są w poniższej tabeli referencyjnej, obejmującej stan zatwierdzenia, powiadomienia, trwałe przywracanie stanu, przechowywanie danych audytowych oraz panele kontrolne w 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 |
+-----------------------------+---------------------+------------------------------+
Powiązane materiały
- Ocena agentów AI: zbiory referencyjne, oceniające modele LLM oraz mechanizmy regresji CI — Ocena offline i online agentów przy użyciu AgentCore, LangFuse, Ragas i DeepEval, wraz z zestawami danych referencyjnymi, mechanizmami kontrolnymi CI, testami A/B oraz informacjami zwrotnymi z pamięci.
- Orchestracja wielu agentów na Bedrock i LangGraph: cztery wzorce — Wzorce supervisor-worker, pipeline, fan-out i debate z wykorzystaniem LangGraph na AWS Bedrock, wraz z mechanizmami przekazywania zadań, kontrolą budżetu, śledzeniem procesów oraz sposobami radzenia sobie z awariami w produkcji.