Human‑in‑the‑Loop Architecture Primer
A repeatable, auditable workflow turns a proposal into an execution only after a deterministic series of checks. By binding the human‑approval token to the exact payload, the decision becomes a verifiable part of the transaction rather than a post‑hoc disclaimer.
The architecture distinguishes four sources of truth:
- Current knowledge – stored in the context or retrieval component.
- Live values – obtained from authoritative APIs at request time.
- Enforceable business rules – expressed as deterministic code that must be satisfied before any state change.
- Model tuning – performed after systematic evaluation and labeling; it supplements but does not replace rule‑based enforcement or freshness checks.
Quick‑Diagnosis Table
| Symptom | Likely Root Cause | Immediate Mitigation |
|---|---|---|
| Unauthorized data mutation after automation | Approval not cryptographically tied to payload | Introduce payload hash in the approval token |
| Escalating reviewer backlog on high‑risk actions | Flat approval process regardless of impact | Tier risk and assign reviewer capacity |
| Irreversible financial adjustments cannot be rolled back | Execution step runs before final sign‑off | Enforce exact‑action approval as the last gate |
| Audit logs missing key fields | Logging tied to mutable objects | Serialize immutable request snapshots |
Decision Flow
flowchart TD A["Proposal"] --> B["Risk‑Policy Evaluation"] B --> C["Tier Assignment"] C --> D["Exact‑Action Approval"] D --> E["Execution"] E --> F["Immutable Audit Record"] The flow consists of Proposal → Risk‑Policy → Exact‑Action Approval → Execution. The final audit record captures the hash‑bound token, tier, reviewer identifiers and timestamps.
Architecture Responsibilities
| Component | Primary Responsibility | Guarantees (intent) |
|---|---|---|
| Proposal Engine | Emit a JSON payload describing the intended mutation; compute a deterministic SHA‑256 hash | Payload immutability (via hash) |
| Risk‑Policy Service | Apply deterministic rules, assign a risk tier, and expose a risk score | Consistent tier assignment |
| Approval Service | Present the exact payload + hash to a reviewer; on sign‑off, issue a signed token containing the hash | Human decision binding |
| Execution Engine | Verify token signature and payload hash before invoking downstream actions | Tamper‑evidence |
| Audit Logger | Persist an immutable event (e.g., append‑only log) with full payload, hash and approval token | Non‑repudiation |
Python Helper – Decision & Payload Binding
# -*- coding: utf-8 -*- """ Teaching‑only helper that demonstrates: 1. deterministic risk tiering, 2. creation of a hash‑bound approval token, 3. verification before execution. All dependencies are from the Python standard library. """ import json import hashlib import hmac import time from typing import Dict, Literal, TypedDict # ---------------------------------------------------------------------- # Types # ---------------------------------------------------------------------- class Payload(TypedDict): action: str # e.g. "adjust_credit_limit" entity_id: str # unique identifier of the subject parameters: Dict[str, float] # e.g. {"value_change": 25000.0} requested_by: str # service or user ID RiskTier = Literal[1, 2, 3] # ---------------------------------------------------------------------- # Deterministic risk policy # ---------------------------------------------------------------------- def risk_tier(payload: Payload) -> RiskTier: """ Simple deterministic rule set. - Tier 1: value_change < 1 000 and reversible - Tier 2: value_change < 10 000 and reversible - Tier 3: otherwise (potentially irreversible) """ change = payload["parameters"].get("value_change", 0.0) irreversible = payload["parameters"].get("irreversible", False) if change < 1_000 and not irreversible: return 1 if change < 10_000 and not irreversible: return 2 return 3 # ---------------------------------------------------------------------- # Payload hashing # ---------------------------------------------------------------------- def payload_hash(payload: Payload) -> str: """ Returns a hex‑encoded SHA‑256 over the canonical JSON representation. """ canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")) return hashlib.sha256(canonical.encode("utf-8")).hexdigest() # ---------------------------------------------------------------------- # Token creation (illustrative only) # ---------------------------------------------------------------------- SECRET_KEY = b"super‑secret‑key‑rotate‑daily" def create_approval_token(payload: Payload, reviewer: str) -> str: """ Very simple JWT‑like token: header.body.signature. Intended for teaching – a production system should use a robust JWT library. """ header = json.dumps({"alg": "HS256", "typ": "JWT"}).encode() body = json.dumps( { "sub": reviewer, "iat": int(time.time()), "hash": payload_hash(payload), } ).encode() signature = hmac.new(SECRET_KEY, header + b"." + body, hashlib.sha256).hexdigest() token = f"{header.hex()}.{body.hex()}.{signature}" return token # ---------------------------------------------------------------------- # Verification before execution # ---------------------------------------------------------------------- def verify_and_execute(payload: Payload, token: str) -> None: """ Confirms that the token’s hash matches the supplied payload. Raises AssertionError on mismatch – in a real system this would abort the transaction. """ header_hex, body_hex, signature = token.split(".") body = json.loads(bytes.fromhex(body_hex)) expected_hash = payload_hash(payload) assert body["hash"] == expected_hash, "Payload hash mismatch – possible tampering" expected_sig = hmac.new( SECRET_KEY, f"{header_hex}.{body_hex}".encode(), hashlib.sha256 ).hexdigest() assert signature == expected_sig, "Invalid approval signature" # ----- Execution placeholder ----- print(f"Executing {payload['action']} for {payload['entity_id']} …") # Replace with an idempotent, atomic operation in production. # ---------------------------------------------------------------------- # Example usage (not executed in docs) # ---------------------------------------------------------------------- if __name__ == "__main__": example: Payload = { "action": "adjust_credit_limit", "entity_id": "cust-12345", "parameters": {"value_change": 2_500.0, "irreversible": False}, "requested_by": "loan‑engine", } tier = risk_tier(example) assert tier == 2, "Tier assignment mismatch in example" token = create_approval_token(example, reviewer="compliance‑officer") verify_and_execute(example, token) Teaching‑Only – The code uses only the standard library and a fixed secret key for clarity. Replace the token logic with a production‑grade JWT implementation and rotate secrets regularly.
Architectural Decision Records (ADRs)
ADR 001 – Payload‑Hash Binding
Context
Early prototypes stored the human sign‑off flag in a mutable column. Audits later revealed that downstream services could modify the underlying request after approval, weakening accountability.
Decision
Include a cryptographic hash of the exact payload in the approval token. Execution components must verify the hash before performing any state change.
Implications
- Positive – Provides evidence that the approved payload has not been altered. Auditors can recompute the hash from immutable logs.
- Negative – Legitimate changes (e.g., enrichment) require a new approval round, which may increase latency for low‑risk items.
Acceptance Criteria
- Audit logs contain both the payload hash and the approval token.
- Unit tests confirm that modifying a payload after token issuance raises an
AssertionError.
ADR 002 – Tier‑Based Reviewer Load Balancing
Context
A single compliance team was overloaded by a mix of low‑ and high‑risk proposals, causing SLA pressures for high‑impact actions.
Decision
Define three deterministic risk tiers with distinct reviewer requirements:
| Tier | Max Δ Value | Irreversibility | Reviewers Required |
|---|---|---|---|
| 1 | < 1 000 | No | 0 (auto‑approve after sanity check) |
| 2 | < 10 000 | No | 1 (single reviewer) |
| 3 | ≥ 10 000 | Yes | 2 (dual sign‑off) |
Tier 1 proposals are routed to an automated sanity check; Tier 3 actions trigger a dedicated dual‑approval queue.
Implications
- Positive – Reviewer effort aligns with risk, reducing backlog for high‑impact changes.
- Negative – The tiering function must remain deterministic; any change to the rule set requires a versioned policy rollout.
Acceptance Criteria
- Simulated runs confirm that identical payloads are consistently assigned the same tier across service versions.
- Policy version is recorded with each decision; a mismatch between recorded and expected version causes the decision to be rejected.
Evaluation & Maintenance Checklist
- [ ] Deterministic Policy – Verify that the risk‑policy service returns the same tier for identical payloads across successive runs.
- [ ] Hash Integrity – Run a periodic job that recomputes payload hashes from audit logs and compares them to stored token hashes.
- [ ] Signature Rotation – Rotate the signing secret on a defined schedule; ensure services accept both current and previous keys during overlap.
- [ ] Reviewer Load Monitoring – Track average queue time per tier; adjust staffing or automation when thresholds are exceeded.
- [ ] Irreversibility Guardrail – For Tier 3 actions, require an explicit “I understand this is irreversible” acknowledgment, logged as a separate event.
- [ ] Regulatory Change Process – When new regulations appear, update the risk‑policy rule set, increment the policy version, and back‑test against historic proposals before production rollout.
- [ ] Test Coverage – Include unit tests for
risk_tier,payload_hash, and token verification; add integration tests that simulate a full proposal → execution flow.
References
- Microsoft Retrieval‑Augmented Generation concepts: https://learn.microsoft.com/en-us/azure/foundry/concepts/retrieval-augmented-generation
- LoRA: Low‑Rank Adaptation of Large Language Models, arXiv 2106.09685: https://arxiv.org/abs/2106.09685
End of guide.
