Cascade & Fallback Pattern

import anthropic
import re
import json
import time

client = anthropic.Anthropic()

# Tier 1: Regex / rule-based responses (instant, free)
FAQ_PATTERNS = {
    r"(?i)(business|office)\s*hours?": "Our business hours are Monday-Friday, 9 AM - 6 PM EST.",
    r"(?i)reset\s*(my\s*)?password": "To reset your password, visit account.example.com/reset and follow the instructions.",
    r"(?i)(refund|return)\s*policy": "We offer full refunds within 30 days of purchase. Visit example.com/returns to start a return.",
    r"(?i)contact\s*(support|us|help)": "You can reach support at support@example.com or call 1-800-EXAMPLE.",
    r"(?i)(cancel|cancellation)": "To cancel your subscription, go to Settings > Subscription > Cancel. You'll retain access until the billing period ends.",
    r"(?i)^(hi|hello|hey|good\s*(morning|afternoon|evening))": "Hello! How can I help you today?",
}

def tier1_regex(query: str) -> dict | None:
    """Tier 1: Pattern matching for common queries."""
    for pattern, response in FAQ_PATTERNS.items():
        if re.search(pattern, query):
            return {
                "tier": 1,
                "method": "regex",
                "response": response,
                "confidence": 0.95,
                "cost": 0.0,
                "latency_ms": 0
            }
    return None  # Can't handle -> cascade to tier 2

def tier2_small_model(query: str) -> dict | None:
    """Tier 2: Small model for straightforward questions."""
    start = time.time()
    response = client.messages.create(
        model="claude-haiku-4-20250514",
        max_tokens=1000,
        system=(
            "Answer the user's question concisely. "
            "If you're not confident in your answer, start with 'UNSURE:' "
            "and explain why."
        ),
        messages=[{"role": "user", "content": query}]
    )
    latency = (time.time() - start) * 1000
    text = response.content[0].text

    # Check confidence signals
    if text.startswith("UNSURE:"):
        return None  # Not confident -> cascade to tier 3

    # Check for hedging language
    hedging_phrases = ["i'm not sure", "i think", "it depends", "possibly",
                       "i don't have enough", "i cannot determine"]
    hedging_count = sum(1 for phrase in hedging_phrases if phrase in text.lower())

    if hedging_count >= 2:
        return None  # Too much uncertainty -> cascade

    return {
        "tier": 2,
        "method": "haiku",
        "response": text,
        "confidence": max(0.5, 0.9 - hedging_count * 0.15),
        "cost": (response.usage.input_tokens + response.usage.output_tokens) * 0.00000025,
        "latency_ms": round(latency)
    }

def tier3_large_model(query: str) -> dict:
    """Tier 3: Large model for complex questions (always returns)."""
    start = time.time()
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4000,
        system="Provide a thorough, well-structured answer. Be helpful and precise.",
        messages=[{"role": "user", "content": query}]
    )
    latency = (time.time() - start) * 1000

    return {
        "tier": 3,
        "method": "sonnet",
        "response": response.content[0].text,
        "confidence": 0.95,
        "cost": (response.usage.input_tokens + response.usage.output_tokens) * 0.000003,
        "latency_ms": round(latency)
    }

def cascade(query: str) -> dict:
    """Execute the 3-tier cascade."""
    # Try each tier in order
    result = tier1_regex(query)
    if result:
        print(f"[Cascade] Handled at Tier 1 (regex) - {result['latency_ms']}ms")
        return result

    result = tier2_small_model(query)
    if result:
        print(f"[Cascade] Handled at Tier 2 (Haiku) - {result['latency_ms']}ms")
        return result

    result = tier3_large_model(query)
    print(f"[Cascade] Handled at Tier 3 (Sonnet) - {result['latency_ms']}ms")
    return result

# Usage examples
print(cascade("What are your business hours?"))      # Tier 1
print(cascade("How do I sort a list in Python?"))     # Tier 2
print(cascade("Compare CQRS vs event sourcing..."))   # Tier 3

import anthropic
import json

client = anthropic.Anthropic()

def self_evaluate(query: str, response: str, model: str) -> dict:
    """Ask a model to evaluate the quality of a response."""
    eval_result = client.messages.create(
        model=model,
        max_tokens=300,
        messages=[{
            "role": "user",
            "content": (
                "Evaluate this response to the given question.\n\n"
                f"Question: {query}\n\n"
                f"Response: {response}\n\n"
                "Rate on these criteria (1-5 each):\n"
                "- accuracy: Is the information correct?\n"
                "- completeness: Does it fully answer the question?\n"
                "- clarity: Is it well-written and easy to understand?\n\n"
                'Return JSON: {"accuracy": N, "completeness": N, '
                '"clarity": N, "overall": N, "should_escalate": bool}'
            )
        }]
    )
    return json.loads(eval_result.content[0].text)

def quality_cascade(query: str, min_quality: float = 3.5) -> dict:
    """Cascade with quality self-evaluation at each tier."""
    tiers = [
        ("claude-haiku-4-20250514", "fast"),
        ("claude-sonnet-4-20250514", "balanced"),
        ("claude-opus-4-20250514", "powerful")
    ]

    for i, (model, tier_name) in enumerate(tiers):
        # Generate response
        response = client.messages.create(
            model=model,
            max_tokens=4000,
            messages=[{"role": "user", "content": query}]
        )
        text = response.content[0].text

        # Last tier: return without evaluation (no escalation possible)
        if i == len(tiers) - 1:
            return {
                "tier": tier_name, "model": model,
                "response": text, "escalation_reason": None
            }

        # Evaluate quality (use same model for speed)
        evaluation = self_evaluate(query, text, model)

        if evaluation["overall"] >= min_quality and not evaluation["should_escalate"]:
            return {
                "tier": tier_name, "model": model,
                "response": text, "quality": evaluation,
                "escalation_reason": None
            }
        else:
            print(f"[Cascade] {tier_name} scored {evaluation['overall']}/5, "
                  f"escalating to next tier...")

    # Should never reach here, but safety fallback
    return {"error": "All tiers exhausted"}

import time
from dataclasses import dataclass, field

@dataclass
class CircuitBreaker:
    """Circuit breaker for AI model calls."""
    failure_threshold: int = 5       # Failures before opening
    recovery_timeout: int = 60       # Seconds before trying again
    half_open_max_calls: int = 3     # Test calls in half-open state

    _failures: int = field(default=0, init=False)
    _state: str = field(default="closed", init=False)    # closed, open, half-open
    _last_failure_time: float = field(default=0, init=False)
    _half_open_successes: int = field(default=0, init=False)

    def can_execute(self) -> bool:
        """Check if the circuit allows a request."""
        if self._state == "closed":
            return True

        if self._state == "open":
            # Check if recovery timeout has elapsed
            if time.time() - self._last_failure_time > self.recovery_timeout:
                self._state = "half-open"
                self._half_open_successes = 0
                print("[Circuit] Moving to half-open state")
                return True
            return False

        # half-open: allow limited test calls
        return True

    def record_success(self):
        """Record a successful call."""
        if self._state == "half-open":
            self._half_open_successes += 1
            if self._half_open_successes >= self.half_open_max_calls:
                self._state = "closed"
                self._failures = 0
                print("[Circuit] Closed - service recovered")
        else:
            self._failures = max(0, self._failures - 1)

    def record_failure(self):
        """Record a failed call."""
        self._failures += 1
        self._last_failure_time = time.time()
        if self._failures >= self.failure_threshold:
            self._state = "open"
            print(f"[Circuit] OPEN - {self._failures} failures, "
                  f"routing to fallback for {self.recovery_timeout}s")

# Usage with fallback
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30)

def resilient_call(query: str) -> str:
    if breaker.can_execute():
        try:
            result = call_primary_model(query)
            breaker.record_success()
            return result
        except Exception as e:
            breaker.record_failure()
            print(f"[Fallback] Primary failed: {e}")

    # Fallback to secondary model
    return call_fallback_model(query)

import time
import random
import anthropic

def retry_with_backoff(func, max_retries=3, base_delay=1.0):
    """Retry a function with exponential backoff and jitter."""
    for attempt in range(max_retries + 1):
        try:
            return func()
        except anthropic.RateLimitError:
            if attempt == max_retries:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            print(f"[Retry] Rate limited, waiting {delay:.1f}s (attempt {attempt + 1})")
            time.sleep(delay)
        except anthropic.APIStatusError as e:
            if e.status_code >= 500 and attempt < max_retries:
                delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                print(f"[Retry] Server error {e.status_code}, waiting {delay:.1f}s")
                time.sleep(delay)
            else:
                raise