M17: Output Guardrails & Human-in-the-Loop

// Agent response: "The filing was submitted on March 15, 2024
//   by Acme Corp for $2.3M in equipment collateral."
// Source document says: March 22, 2024; Acme Corp; $2.3M equipment

{
  "claims": [
    { "text": "filed on March 15, 2024", "status": "contradicted",
      "source": "Document shows March 22, 2024" },
    { "text": "by Acme Corp", "status": "supported",
      "source": "filing_2024_NY.txt line 3" },
    { "text": "$2.3M in equipment collateral", "status": "supported",
      "source": "filing_2024_NY.txt line 7" }
  ],
  "overall": "block",       // One contradicted claim = block
  "unsupported_count": 0
}
// Result: Response blocked. Agent told to re-check the date.

// After iteration 3 of an agent loop:
{
  "iteration": 3,
  "input_tokens": 24000,    // 8K per iteration × 3
  "output_tokens": 6000,    // 2K per iteration × 3
  "total_cost": "$0.162",   // (24K × $3/M) + (6K × $15/M)
  "budget": "$0.50",
  "budget_remaining": "$0.338",
  "can_afford_next": true,  // Estimated next call: $0.054
  "status": "ok"
}

// After iteration 8:
{
  "iteration": 8,
  "total_cost": "$0.432",
  "budget_remaining": "$0.068",
  "can_afford_next": false, // Next call ~$0.054 would exceed budget
  "status": "budget_exceeded — returning fallback response"
}

// What the human reviewer sees in their approval queue:
{
  "gate_type": "approval",
  "agent_id": "order-support-agent-7",
  "proposed_action": "Send $450 refund to customer",
  "context": {
    "order_id": "ORD-12345",
    "reason": "Damaged item — photos verified by vision model",
    "return_window": "within 30-day policy",
    "customer_history": "12 orders, 0 previous refunds"
  },
  "state": "waiting_for_human",
  "options": ["approve", "deny", "modify"],
  "created_at": "2024-03-22T14:32:00Z",
  "timeout": "4 hours — auto-escalate to manager if no response"
}

// Request 1: success → state unchanged
{ "state": "closed", "failures": 0, "action": "allowed" }

// Request 2: API 500 error
{ "state": "closed", "failures": 1, "action": "allowed" }

// Request 3: API 500 error
{ "state": "closed", "failures": 2, "action": "allowed" }

// Request 4: API 500 error → threshold reached, circuit TRIPS
{ "state": "open", "failures": 3, "action": "tripped",
  "cooldown_until": "2024-03-22T14:33:00Z" }

// Request 5: arrives during cooldown
{ "state": "open", "action": "blocked",
  "response": "fallback — no API call made" }

// After 60s cooldown expires:
{ "state": "half_open", "action": "testing — 1 request allowed" }

// Test request succeeds:
{ "state": "closed", "failures": 0, "action": "recovered" }

import anthropic
import json

HALLUCINATION_PROMPT = """You are a fact-checking judge. Compare the agent's
response against the provided source documents.

For each factual claim in the response, classify it as:
- "supported": Claim is directly backed by source documents
- "unsupported": Claim is NOT in source documents (possible hallucination)
- "contradicted": Claim CONTRADICTS source documents (definite error)

Respond with ONLY a JSON object:
{
  "claims": [
    {"text": "claim text", "status": "supported|unsupported|contradicted", "source": "which document supports/contradicts, or null"}
  ],
  "overall": "pass|flag|block",
  "unsupported_count": 0
}

Source documents:

{sources}


Agent response to fact-check:

{response}
"""

def check_hallucination(
    client: anthropic.Anthropic,
    response_text: str,
    source_docs: list[str],
) -> dict:
    """Verify agent output against source documents.

    Uses a SEPARATE Claude call (Claude-as-judge pattern).
    The judge has its own clean context and can't be influenced
    by the agent's reasoning that led to the response.
    """
    sources_text = "\n---\n".join(source_docs)
    try:
        result = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": HALLUCINATION_PROMPT.format(
                    sources=sources_text,
                    response=response_text,
                ),
            }],
        )
        parsed = json.loads(result.content[0].text)

        # Decision logic:
        # - Any "contradicted" claim = block immediately
        # - 2+ "unsupported" claims = flag for review
        # - All "supported" = pass
        contradicted = [c for c in parsed["claims"] if c["status"] == "contradicted"]
        unsupported = [c for c in parsed["claims"] if c["status"] == "unsupported"]

        if contradicted:
            return {"result": "block", "reason": f"Contradicted claims: {contradicted}", "claims": parsed["claims"]}
        elif len(unsupported) >= 2:
            return {"result": "flag", "reason": f"{len(unsupported)} unsupported claims", "claims": parsed["claims"]}
        else:
            return {"result": "pass", "claims": parsed["claims"]}

    except Exception as e:
        # Fail closed — if fact-checking fails, flag for review
        return {"result": "flag", "reason": f"Fact-check error: {e}", "claims": []}

import time
from enum import Enum
from dataclasses import dataclass, field

class BreakerState(Enum):
    CLOSED = "closed"       # Normal — requests flow through
    OPEN = "open"           # Tripped — all requests get fallback
    HALF_OPEN = "half_open" # Testing — one request allowed

@dataclass
class CircuitBreaker:
    """Circuit breaker for agent API calls.

    Monitors consecutive failures and trips when threshold
    is reached. Prevents cascading failures and wasted spend
    during outages.
    """
    failure_threshold: int = 3      # Trip after N failures
    cooldown_seconds: float = 60.0  # Wait before testing
    state: BreakerState = field(default=BreakerState.CLOSED)
    failure_count: int = field(default=0)
    last_failure_time: float = field(default=0.0)
    opened_at: float = field(default=0.0)

    def can_execute(self) -> bool:
        """Check if a request should be allowed through."""
        if self.state == BreakerState.CLOSED:
            return True

        if self.state == BreakerState.OPEN:
            # Check if cooldown has elapsed
            elapsed = time.time() - self.opened_at
            if elapsed >= self.cooldown_seconds:
                self.state = BreakerState.HALF_OPEN
                return True  # Allow one test request
            return False  # Still cooling down

        if self.state == BreakerState.HALF_OPEN:
            return False  # Already testing, block others

        return False

    def record_success(self):
        """Record a successful request."""
        if self.state == BreakerState.HALF_OPEN:
            # Test passed — close the circuit
            self.state = BreakerState.CLOSED
            self.failure_count = 0
        elif self.state == BreakerState.CLOSED:
            self.failure_count = 0  # Reset on success

    def record_failure(self):
        """Record a failed request."""
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.state == BreakerState.HALF_OPEN:
            # Test failed — re-open with longer cooldown
            self.state = BreakerState.OPEN
            self.opened_at = time.time()
            self.cooldown_seconds *= 2  # Exponential backoff

        elif self.state == BreakerState.CLOSED:
            if self.failure_count >= self.failure_threshold:
                self.state = BreakerState.OPEN
                self.opened_at = time.time()

    @property
    def fallback_response(self) -> str:
        return (
            "I'm temporarily unable to process your request. "
            "Our systems are experiencing issues. Please try "
            "again in a few minutes."
        )

# Usage with an agent
breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=60)

def call_agent_with_breaker(client, messages):
    """Wrap agent calls with circuit breaker protection."""
    if not breaker.can_execute():
        return {"content": breaker.fallback_response, "circuit_breaker": "open"}

    try:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=messages,
        )
        breaker.record_success()
        return {"content": response.content[0].text, "circuit_breaker": breaker.state.value}
    except Exception as e:
        breaker.record_failure()
        if not breaker.can_execute():
            return {"content": breaker.fallback_response, "circuit_breaker": "open"}
        raise  # Re-raise if breaker hasn't tripped yet

mkdir output-guardrails && cd output-guardrails
python -m venv venv && source venv/bin/activate   # Windows: venv\Scripts\activate
pip install anthropic
export ANTHROPIC_API_KEY=your-key-here             # Windows: set ANTHROPIC_API_KEY=your-key-here

"""Output Guardrails & HITL — M17 Hands-On Lab"""
import anthropic
import json
import time
from dataclasses import dataclass, field
from enum import Enum

client = anthropic.Anthropic()


# ── Hallucination Detector (Claude-as-Judge) ─────────────

HALLUCINATION_PROMPT = """You are a fact-checking judge. Compare the response
against the source documents. For each factual claim, classify it as:
- "supported": Directly backed by sources
- "unsupported": NOT in sources (possible hallucination)
- "contradicted": CONTRADICTS sources (definite error)

Respond with ONLY JSON:
{{"claims": [{{"text": "claim", "status": "supported|unsupported|contradicted"}}],
 "overall": "pass|flag|block", "unsupported_count": 0}}

Sources:

{sources}


Response to check:

{response}
"""

def check_hallucination(response_text: str, source_docs: list[str]) -> dict:
    """Verify agent response against source documents."""
    sources = "\n---\n".join(source_docs)
    try:
        result = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": HALLUCINATION_PROMPT.format(
                    sources=sources, response=response_text
                ),
            }],
        )
        parsed = json.loads(result.content[0].text)

        contradicted = [c for c in parsed["claims"]
                       if c["status"] == "contradicted"]
        unsupported = [c for c in parsed["claims"]
                      if c["status"] == "unsupported"]

        if contradicted:
            return {"result": "block",
                    "reason": f"{len(contradicted)} contradicted claim(s)",
                    "claims": parsed["claims"]}
        elif len(unsupported) >= 2:
            return {"result": "flag",
                    "reason": f"{len(unsupported)} unsupported claim(s)",
                    "claims": parsed["claims"]}
        return {"result": "pass", "claims": parsed["claims"]}
    except Exception as e:
        return {"result": "flag", "reason": f"Check failed: {e}", "claims": []}


# ── Cost Tracker with Budget Enforcement ─────────────────

@dataclass
class CostTracker:
    """Track token usage and enforce per-request budget."""
    budget_dollars: float = 0.50
    input_tokens: int = 0
    output_tokens: int = 0

    # Claude Sonnet pricing (per token)
    INPUT_PRICE = 3.0 / 1_000_000    # $3/M input tokens
    OUTPUT_PRICE = 15.0 / 1_000_000  # $15/M output tokens

    @property
    def total_cost(self) -> float:
        return (self.input_tokens * self.INPUT_PRICE +
                self.output_tokens * self.OUTPUT_PRICE)

    @property
    def budget_remaining(self) -> float:
        return self.budget_dollars - self.total_cost

    def record_usage(self, input_toks: int, output_toks: int):
        self.input_tokens += input_toks
        self.output_tokens += output_toks

    def can_afford(self, estimated_input: int = 5000,
                   estimated_output: int = 1000) -> bool:
        """Check if the next call would stay within budget."""
        estimated_cost = (estimated_input * self.INPUT_PRICE +
                         estimated_output * self.OUTPUT_PRICE)
        return self.total_cost + estimated_cost <= self.budget_dollars

    def summary(self) -> str:
        return (f"Tokens: {self.input_tokens} in / {self.output_tokens} out "
                f"| Cost: ${self.total_cost:.4f} / ${self.budget_dollars:.2f}")


# ── Circuit Breaker ──────────────────────────────────────

class BreakerState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

@dataclass
class CircuitBreaker:
    failure_threshold: int = 3
    cooldown_seconds: float = 5.0   # Short for demo
    state: BreakerState = field(default=BreakerState.CLOSED)
    failure_count: int = field(default=0)
    opened_at: float = field(default=0.0)

    def can_execute(self) -> bool:
        if self.state == BreakerState.CLOSED:
            return True
        if self.state == BreakerState.OPEN:
            if time.time() - self.opened_at >= self.cooldown_seconds:
                self.state = BreakerState.HALF_OPEN
                return True
            return False
        return False  # HALF_OPEN blocks while testing

    def record_success(self):
        if self.state == BreakerState.HALF_OPEN:
            self.state = BreakerState.CLOSED
        self.failure_count = 0

    def record_failure(self):
        self.failure_count += 1
        if self.state == BreakerState.HALF_OPEN:
            self.state = BreakerState.OPEN
            self.opened_at = time.time()
            self.cooldown_seconds *= 2  # Exponential backoff
        elif self.failure_count >= self.failure_threshold:
            self.state = BreakerState.OPEN
            self.opened_at = time.time()


# ── Approval Gate (CLI) ──────────────────────────────────

def approval_gate(action: str, context: str,
                  auto_approve: bool = False) -> dict:
    """Pause for human approval before irreversible actions.
    Set auto_approve=True for automated testing."""
    print(f"\n{'='*50}")
    print(f"🔒 APPROVAL REQUIRED")
    print(f"Action: {action}")
    print(f"Context: {context}")
    print(f"{'='*50}")

    if auto_approve:
        print("  [Auto-approved for testing]")
        return {"approved": True, "modified": False}

    response = input("Approve? (y/n/e to edit): ").strip().lower()
    if response == "y":
        return {"approved": True, "modified": False}
    elif response == "e":
        new_action = input("Enter modified action: ").strip()
        return {"approved": True, "modified": True, "new_action": new_action}
    return {"approved": False, "reason": "Human denied"}


# ── Test Scenarios ────────────────────────────────────────
if __name__ == "__main__":
    print("=" * 60)
    print("OUTPUT GUARDRAILS — TEST SUITE")
    print("=" * 60)

    # Test 1: Hallucination Detection
    print("\n" + "─" * 60)
    print("TEST 1: Hallucination Detection — Contradicted Claim")
    source_docs = [
        "UCC Filing #2024-NY-0042: Filed March 22, 2024 by Acme Corp. "
        "Collateral: $2.3M in manufacturing equipment. Status: Active.",
        "Amendment filed April 10, 2024: Added collateral description "
        "for warehouse inventory valued at $890K."
    ]
    agent_response = (
        "The UCC filing was submitted on March 15, 2024 by Acme Corp "
        "for $2.3M in equipment collateral. An amendment was filed "
        "on April 10, 2024 adding $890K in warehouse inventory."
    )
    result = check_hallucination(agent_response, source_docs)
    print(f"  Result: {result['result']}")
    print(f"  Reason: {result.get('reason', 'All claims supported')}")
    for claim in result.get("claims", []):
        print(f"    [{claim['status']}] {claim['text']}")

    # Test 2: Hallucination Detection — All Supported
    print("\n" + "─" * 60)
    print("TEST 2: Hallucination Detection — All Supported")
    correct_response = (
        "The UCC filing was submitted on March 22, 2024 by Acme Corp "
        "for $2.3M in equipment collateral."
    )
    result2 = check_hallucination(correct_response, source_docs)
    print(f"  Result: {result2['result']}")

    # Test 3: Cost Tracker
    print("\n" + "─" * 60)
    print("TEST 3: Cost Tracking & Budget Enforcement")
    tracker = CostTracker(budget_dollars=0.10)
    for i in range(5):
        # Estimate matches the actual per-iteration usage below
        if not tracker.can_afford(estimated_input=8000, estimated_output=2000):
            print(f"  Iteration {i+1}: BUDGET EXCEEDED — {tracker.summary()}")
            break
        # Simulate API usage
        tracker.record_usage(input_toks=8000, output_toks=2000)
        print(f"  Iteration {i+1}: {tracker.summary()}")

    # Test 4: Circuit Breaker
    print("\n" + "─" * 60)
    print("TEST 4: Circuit Breaker State Transitions")
    breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=2)

    actions = [
        ("success", "Normal request 1"),
        ("failure", "API error 1"),
        ("failure", "API error 2"),
        ("failure", "API error 3 → TRIPS"),
        ("blocked", "Request during OPEN state"),
    ]
    for action, label in actions:
        can_exec = breaker.can_execute()
        if not can_exec:
            print(f"  {label}: BLOCKED (state={breaker.state.value})")
            continue
        if action == "success":
            breaker.record_success()
            print(f"  {label}: OK (state={breaker.state.value})")
        else:
            breaker.record_failure()
            print(f"  {label}: FAIL (state={breaker.state.value}, "
                  f"failures={breaker.failure_count}/{breaker.failure_threshold})")

    print(f"  Waiting {breaker.cooldown_seconds}s for cooldown...")
    time.sleep(breaker.cooldown_seconds + 0.1)
    can_test = breaker.can_execute()
    print(f"  Half-open test: can_execute={can_test} "
          f"(state={breaker.state.value})")
    breaker.record_success()
    print(f"  Test passed → state={breaker.state.value}")

    # Test 5: Approval Gate
    print("\n" + "─" * 60)
    print("TEST 5: Approval Gate (auto-approved for testing)")
    gate_result = approval_gate(
        action="Send $450 refund to Order #12345",
        context="Customer received damaged item, within return window",
        auto_approve=True,
    )
    print(f"  Result: {gate_result}")

python output_guardrails.py

# Verify key behaviors
python output_guardrails.py 2>&1 | grep -E "(block|BUDGET|TRIPS|APPROVAL)"
# Should show: block, BUDGET EXCEEDED, TRIPS, APPROVAL REQUIRED