DEV Community

Salmon Joy
Salmon Joy

Posted on

Human-in-the-loop architecture

#ai #machinelearning #architecture #enterprise

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"]
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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

  1. Audit logs contain both the payload hash and the approval token.
  2. 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

  1. Simulated runs confirm that identical payloads are consistently assigned the same tier across service versions.
  2. 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

End of guide.

Top comments (0)