M20: Monitoring & Continuous Improvement

import time
import anthropic
from langfuse import Langfuse

langfuse = Langfuse()
client = anthropic.Anthropic()

def run_agent_with_metrics(user_message: str, trace_id: str) -> dict:
    """Run the agent and record production metrics as Langfuse scores."""
    trace = langfuse.trace(id=trace_id, name="agent-request")
    start = time.time()

    try:
        # --- LLM Call ---
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": user_message}],
        )
        latency_ms = (time.time() - start) * 1000
        input_tokens = response.usage.input_tokens
        output_tokens = response.usage.output_tokens

        # --- Record Metrics as Scores ---
        # WHAT: Attach numeric scores to the trace for dashboard aggregation.
        # WHY: Langfuse scores power dashboard charts, alerting thresholds, and drift detection.
        # GOTCHA: Use try/except so a Langfuse outage never crashes your agent.
        try:
            trace.score(name="latency_ms", value=latency_ms)
            trace.score(name="input_tokens", value=input_tokens)
            trace.score(name="output_tokens", value=output_tokens)
            trace.score(name="total_tokens", value=input_tokens + output_tokens)
            trace.score(name="success", value=1)

            # Cost estimation (Claude Sonnet pricing as of 2025)
            cost = (input_tokens * 3.0 / 1_000_000) + (output_tokens * 15.0 / 1_000_000)
            trace.score(name="cost_usd", value=round(cost, 6))
        except Exception as obs_err:
            # Observability failure must NEVER break the agent
            print(f"[WARN] Failed to record metrics: {obs_err}")

        return {
            "response": response.content[0].text,
            "latency_ms": round(latency_ms, 1),
            "tokens": input_tokens + output_tokens,
            "cost_usd": round(cost, 6),
        }

    except Exception as e:
        latency_ms = (time.time() - start) * 1000
        try:
            trace.score(name="latency_ms", value=latency_ms)
            trace.score(name="success", value=0)
            trace.score(name="error_type", value=0, comment=str(type(e).__name__))
        except Exception:
            pass
        raise

from flask import Flask, request, jsonify
from langfuse import Langfuse
import re

app = Flask(__name__)
langfuse = Langfuse()

# WHAT: Validate trace IDs to prevent injection of fake feedback.
# WHY: Without validation, anyone could POST fake scores to manipulate your metrics.
TRACE_ID_PATTERN = re.compile(r"^[a-zA-Z0-9_-]{8,64}$")

@app.route("/api/feedback", methods=["POST"])
def submit_feedback():
    """Accept user thumbs-up/down feedback and attach to the trace."""
    data = request.get_json()

    # --- Input Validation ---
    trace_id = data.get("trace_id", "")
    score = data.get("score")  # 1 = thumbs up, 0 = thumbs down
    comment = data.get("comment", "")

    if not TRACE_ID_PATTERN.match(trace_id):
        return jsonify({"error": "Invalid trace_id format"}), 400
    if score not in (0, 1):
        return jsonify({"error": "Score must be 0 or 1"}), 400
    if len(comment) > 500:
        return jsonify({"error": "Comment too long (max 500 chars)"}), 400

    # --- Record Feedback ---
    # WHAT: Attach user feedback as a score on the original trace.
    # WHY: This links subjective quality signal to the full trace context.
    # GOTCHA: Always record the feedback even if you suspect it is spam —
    #         filter during analysis, not collection.
    try:
        langfuse.score(
            trace_id=trace_id,
            name="user_feedback",
            value=score,
            comment=comment[:500] if comment else None,
        )
        return jsonify({"status": "recorded"}), 200
    except Exception as e:
        # Log but don't expose internal errors to the client
        app.logger.error(f"Failed to record feedback: {e}")
        return jsonify({"status": "recorded"}), 200  # Still return 200

from dataclasses import dataclass
from enum import Enum
from typing import Callable

class Severity(Enum):
    P1_PAGE = "P1"    # Immediate: PagerDuty / phone call
    P2_TICKET = "P2"  # Next business day: Jira ticket
    P3_INFO = "P3"    # Weekly review: dashboard annotation

@dataclass
class AlertRule:
    name: str
    severity: Severity
    check: Callable[[dict], bool]  # Returns True if alert should fire
    message_template: str

# WHAT: Define alert rules with severity and threshold.
# WHY: Separating rules from routing makes it easy to add/tune alerts.
# GOTCHA: Start with CONSERVATIVE thresholds. Tighten over time.
ALERT_RULES = [
    AlertRule(
        name="high_error_rate",
        severity=Severity.P1_PAGE,
        check=lambda m: m.get("error_rate", 0) > 0.50,
        message_template="CRITICAL: Error rate {error_rate:.1%} exceeds 50%",
    ),
    AlertRule(
        name="elevated_error_rate",
        severity=Severity.P2_TICKET,
        check=lambda m: 0.10 < m.get("error_rate", 0) <= 0.50,
        message_template="WARNING: Error rate {error_rate:.1%} exceeds 10%",
    ),
    AlertRule(
        name="high_latency",
        severity=Severity.P2_TICKET,
        check=lambda m: m.get("p95_latency_s", 0) > 30,
        message_template="WARNING: p95 latency {p95_latency_s:.1f}s exceeds 30s",
    ),
    AlertRule(
        name="cost_anomaly",
        severity=Severity.P2_TICKET,
        check=lambda m: m.get("cost_per_hour", 0) > 50,
        message_template="WARNING: Cost ${cost_per_hour:.2f}/hr exceeds $50/hr",
    ),
    AlertRule(
        name="drift_detected",
        severity=Severity.P3_INFO,
        check=lambda m: abs(m.get("output_length_drift_pct", 0)) > 15,
        message_template="INFO: Output length drifted {output_length_drift_pct:+.1f}%",
    ),
]

def evaluate_alerts(metrics: dict) -> list[dict]:
    """Evaluate all alert rules against current metrics. Return fired alerts."""
    fired = []
    for rule in ALERT_RULES:
        try:
            if rule.check(metrics):
                fired.append({
                    "name": rule.name,
                    "severity": rule.severity.value,
                    "message": rule.message_template.format(**metrics),
                })
        except Exception as e:
            # Never let a broken alert rule crash the alerting system
            print(f"[WARN] Alert rule '{rule.name}' failed: {e}")
    return fired

def route_alert(alert: dict):
    """Route alert to the appropriate channel based on severity."""
    sev = alert["severity"]
    if sev == "P1":
        send_pagerduty(alert["message"])  # Wake someone up
    elif sev == "P2":
        create_jira_ticket(alert["message"])  # Next business day
    elif sev == "P3":
        annotate_dashboard(alert["message"])  # Weekly review

# Placeholder implementations
def send_pagerduty(msg): print(f"[PAGERDUTY] {msg}")
def create_jira_ticket(msg): print(f"[JIRA] {msg}")
def annotate_dashboard(msg): print(f"[DASHBOARD] {msg}")

import hashlib
from dataclasses import dataclass

@dataclass
class PromptVariant:
    name: str
    system_prompt: str
    model: str

# WHAT: Define the control and canary prompt variants.
# WHY: Explicit variant definitions make it clear what changed.
VARIANTS = {
    "control": PromptVariant(
        name="v2.3-stable",
        system_prompt="You are a helpful customer support agent. Be concise.",
        model="claude-sonnet-4-6",
    ),
    "canary": PromptVariant(
        name="v2.4-canary",
        system_prompt="You are a helpful customer support agent. Be concise and empathetic. "
                       "Acknowledge the customer's frustration before solving.",
        model="claude-sonnet-4-6",
    ),
}

def get_variant(user_id: str, canary_pct: int = 5) -> PromptVariant:
    """Deterministically assign user to control or canary group.

    Uses a hash of user_id so the same user always gets the same variant.
    canary_pct: percentage of traffic routed to canary (1-50).
    """
    # WHAT: Hash the user ID to a number between 0-99.
    # WHY: Deterministic — same user always gets same variant.
    # GOTCHA: Don't use random() — it gives inconsistent UX.
    hash_val = int(hashlib.sha256(user_id.encode()).hexdigest(), 16) % 100

    if hash_val < canary_pct:
        variant = VARIANTS["canary"]
    else:
        variant = VARIANTS["control"]

    return variant

# Usage
variant = get_variant("user-12345", canary_pct=5)
print(f"User assigned to: {variant.name}")
# Output: User assigned to: v2.3-stable (or v2.4-canary for ~5% of users)

# No external dependencies needed — this lab uses only the Python standard library
mkdir monitoring-lab && cd monitoring-lab

"""metrics.py — Metric collection for the monitoring lab."""
import time
import random
import uuid
from dataclasses import dataclass, field
from typing import List

@dataclass
class MetricRecord:
    trace_id: str
    timestamp: float
    latency_ms: float
    input_tokens: int
    output_tokens: int
    success: bool
    error_type: str = ""
    cost_usd: float = 0.0

# In-memory store (replace with Langfuse in production)
metric_store: List[MetricRecord] = []

def simulate_agent_call(user_message: str) -> MetricRecord:
    """Simulate an agent call and record metrics.

    In production, this wraps a real Claude API call.
    Here we use random values to simulate realistic behavior.
    """
    trace_id = f"tr_{uuid.uuid4().hex[:8]}"
    start = time.time()

    # Simulate varying latency (800ms - 8000ms)
    latency_ms = random.gauss(1500, 500)
    latency_ms = max(300, latency_ms)  # Floor at 300ms
    time.sleep(latency_ms / 10000)  # Brief sleep to simulate work

    # Simulate token usage
    input_tokens = random.randint(50, 300)
    output_tokens = random.randint(100, 800)

    # Simulate 5% error rate
    success = random.random() > 0.05
    error_type = ""
    if not success:
        error_type = random.choice(["timeout", "guardrail_block", "api_error"])

    # Cost estimation (Claude Sonnet pricing)
    cost_usd = (input_tokens * 3.0 / 1_000_000) + (output_tokens * 15.0 / 1_000_000)

    record = MetricRecord(
        trace_id=trace_id,
        timestamp=time.time(),
        latency_ms=round(latency_ms, 1),
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        success=success,
        error_type=error_type,
        cost_usd=round(cost_usd, 6),
    )
    metric_store.append(record)
    return record

def get_summary_metrics() -> dict:
    """Aggregate stored metrics into summary statistics."""
    if not metric_store:
        return {}

    latencies = sorted([m.latency_ms for m in metric_store])
    total = len(metric_store)
    errors = [m for m in metric_store if not m.success]

    def percentile(data, pct):
        idx = int(len(data) * pct / 100)
        return data[min(idx, len(data) - 1)]

    # Group errors by type
    error_breakdown = {}
    for e in errors:
        error_breakdown[e.error_type] = error_breakdown.get(e.error_type, 0) + 1

    return {
        "total_requests": total,
        "error_rate": len(errors) / total if total else 0,
        "error_breakdown": error_breakdown,
        "p50_latency_ms": percentile(latencies, 50),
        "p95_latency_ms": percentile(latencies, 95),
        "p99_latency_ms": percentile(latencies, 99),
        "p95_latency_s": percentile(latencies, 95) / 1000,
        "avg_tokens": sum(m.input_tokens + m.output_tokens for m in metric_store) / total,
        "total_cost_usd": sum(m.cost_usd for m in metric_store),
        "cost_per_hour": sum(m.cost_usd for m in metric_store) * 3600 / max(1, metric_store[-1].timestamp - metric_store[0].timestamp) if len(metric_store) > 1 else 0,
    }

if __name__ == "__main__":
    # Quick test: simulate 20 agent calls
    print("Simulating 20 agent calls...")
    for i in range(20):
        record = simulate_agent_call(f"Test message {i}")
        status = "OK" if record.success else f"FAIL ({record.error_type})"
        print(f"  {record.trace_id}: {record.latency_ms}ms, {record.input_tokens + record.output_tokens} tokens, {status}")

    summary = get_summary_metrics()
    print(f"\nSummary:")
    print(f"  Total requests: {summary['total_requests']}")
    print(f"  Error rate: {summary['error_rate']:.1%}")
    print(f"  p50 latency: {summary['p50_latency_ms']:.0f}ms")
    print(f"  p95 latency: {summary['p95_latency_ms']:.0f}ms")
    print(f"  Total cost: ${summary['total_cost_usd']:.4f}")

"""alerts.py — Tiered alerting with configurable thresholds."""
from dataclasses import dataclass
from enum import Enum
from typing import Callable, List

class Severity(Enum):
    P1_PAGE = "P1"    # Immediate: PagerDuty / phone call
    P2_TICKET = "P2"  # Next business day: Jira ticket
    P3_INFO = "P3"    # Weekly review: dashboard annotation

@dataclass
class AlertRule:
    name: str
    severity: Severity
    check: Callable[[dict], bool]
    message_template: str

ALERT_RULES = [
    AlertRule(
        name="high_error_rate",
        severity=Severity.P1_PAGE,
        check=lambda m: m.get("error_rate", 0) > 0.50,
        message_template="CRITICAL: Error rate {error_rate:.1%} exceeds 50%",
    ),
    AlertRule(
        name="elevated_error_rate",
        severity=Severity.P2_TICKET,
        check=lambda m: 0.10 < m.get("error_rate", 0) <= 0.50,
        message_template="WARNING: Error rate {error_rate:.1%} exceeds 10%",
    ),
    AlertRule(
        name="high_latency",
        severity=Severity.P2_TICKET,
        check=lambda m: m.get("p95_latency_s", 0) > 30,
        message_template="WARNING: p95 latency {p95_latency_s:.1f}s exceeds 30s",
    ),
    AlertRule(
        name="cost_anomaly",
        severity=Severity.P2_TICKET,
        check=lambda m: m.get("cost_per_hour", 0) > 50,
        message_template="WARNING: Cost ${cost_per_hour:.2f}/hr exceeds $50/hr",
    ),
    AlertRule(
        name="drift_detected",
        severity=Severity.P3_INFO,
        check=lambda m: abs(m.get("output_length_drift_pct", 0)) > 15,
        message_template="INFO: Output length drifted {output_length_drift_pct:+.1f}%",
    ),
]

def evaluate_alerts(metrics: dict) -> List[dict]:
    """Evaluate all alert rules against current metrics. Return fired alerts."""
    fired = []
    for rule in ALERT_RULES:
        try:
            if rule.check(metrics):
                fired.append({
                    "name": rule.name,
                    "severity": rule.severity.value,
                    "message": rule.message_template.format(**metrics),
                })
        except Exception as e:
            print(f"[WARN] Alert rule '{rule.name}' failed: {e}")
    return fired

def route_alert(alert: dict) -> str:
    """Route alert to the appropriate channel. Returns the action taken."""
    sev = alert["severity"]
    if sev == "P1":
        return f"[PAGERDUTY] {alert['message']}"
    elif sev == "P2":
        return f"[JIRA TICKET] {alert['message']}"
    elif sev == "P3":
        return f"[DASHBOARD NOTE] {alert['message']}"
    return f"[UNKNOWN] {alert['message']}"

if __name__ == "__main__":
    # Test with a scenario that should trigger alerts
    test_metrics = {
        "error_rate": 0.15,
        "p95_latency_s": 4.2,
        "cost_per_hour": 22.50,
        "output_length_drift_pct": 18.5,
    }
    print("Testing alert rules with elevated error rate and drift...")
    print(f"  Metrics: error_rate={test_metrics['error_rate']:.0%}, "
          f"p95={test_metrics['p95_latency_s']}s, "
          f"cost=${test_metrics['cost_per_hour']}/hr, "
          f"drift={test_metrics['output_length_drift_pct']:+.1f}%")
    print()

    alerts = evaluate_alerts(test_metrics)
    if alerts:
        for alert in alerts:
            action = route_alert(alert)
            print(f"  {action}")
    else:
        print("  No alerts fired.")

    print(f"\n  Total alerts fired: {len(alerts)}")

"""feedback.py — User feedback collection and daily aggregation."""
import random
from dataclasses import dataclass
from typing import List, Dict
from collections import Counter

FAILURE_CATEGORIES = [
    "wrong_order_referenced",
    "outdated_shipping_status",
    "hallucinated_tracking_number",
    "incomplete_answer",
    "tone_too_formal",
    "other",
]

@dataclass
class FeedbackRecord:
    trace_id: str
    score: int  # 1 = thumbs up, 0 = thumbs down
    comment: str = ""
    failure_category: str = ""

# In-memory feedback store
feedback_store: List[FeedbackRecord] = []

def record_feedback(trace_id: str, score: int, comment: str = "") -> FeedbackRecord:
    """Record a single piece of user feedback."""
    if score not in (0, 1):
        raise ValueError("Score must be 0 (thumbs down) or 1 (thumbs up)")

    # Auto-classify negative feedback (in production, use an LLM for this)
    category = ""
    if score == 0:
        category = random.choice(FAILURE_CATEGORIES)

    record = FeedbackRecord(
        trace_id=trace_id,
        score=score,
        comment=comment[:500],
        failure_category=category,
    )
    feedback_store.append(record)
    return record

def daily_aggregation() -> Dict:
    """Aggregate feedback into a daily summary report."""
    if not feedback_store:
        return {"total": 0, "positive": 0, "negative": 0}

    total = len(feedback_store)
    positive = sum(1 for f in feedback_store if f.score == 1)
    negative = sum(1 for f in feedback_store if f.score == 0)

    # Group negative feedback by category
    category_counts = Counter(
        f.failure_category for f in feedback_store
        if f.score == 0 and f.failure_category
    )

    return {
        "total": total,
        "positive": positive,
        "negative": negative,
        "satisfaction_rate": positive / total if total else 0,
        "top_failure_categories": category_counts.most_common(5),
    }

def generate_eval_candidates(min_negatives: int = 3) -> List[FeedbackRecord]:
    """Select negative feedback cases that should become eval test cases."""
    negatives = [f for f in feedback_store if f.score == 0]
    # In production, you would de-duplicate and select diverse examples
    return negatives[:min_negatives]

if __name__ == "__main__":
    # Simulate feedback for 50 agent interactions
    print("Simulating feedback for 50 interactions...")
    for i in range(50):
        trace_id = f"tr_{i:04d}"
        # 85% positive, 15% negative
        score = 1 if random.random() > 0.15 else 0
        comment = "Great answer!" if score == 1 else "This was wrong"
        record_feedback(trace_id, score, comment)

    report = daily_aggregation()
    print(f"\nDaily Feedback Summary:")
    print(f"  Total feedback: {report['total']}")
    print(f"  Positive: {report['positive']} ({report['satisfaction_rate']:.0%})")
    print(f"  Negative: {report['negative']}")
    print(f"\n  Top failure categories:")
    for cat, count in report["top_failure_categories"]:
        print(f"    {cat}: {count} cases")

    candidates = generate_eval_candidates()
    print(f"\n  Eval candidates selected: {len(candidates)}")
    for c in candidates:
        print(f"    {c.trace_id} — {c.failure_category}")

"""router.py — Deterministic A/B traffic router."""
import hashlib
from dataclasses import dataclass

@dataclass
class PromptVariant:
    name: str
    system_prompt: str

VARIANTS = {
    "control": PromptVariant(
        name="v2.3-stable",
        system_prompt="You are a helpful customer support agent. Be concise.",
    ),
    "canary": PromptVariant(
        name="v2.4-canary",
        system_prompt="You are a helpful customer support agent. Be concise and empathetic. "
                       "Acknowledge the customer's frustration before solving.",
    ),
}

def get_variant(user_id: str, canary_pct: int = 5) -> PromptVariant:
    """Deterministically assign user to control or canary group."""
    hash_val = int(hashlib.sha256(user_id.encode()).hexdigest(), 16) % 100
    if hash_val < canary_pct:
        return VARIANTS["canary"]
    else:
        return VARIANTS["control"]

if __name__ == "__main__":
    # Test determinism: same user should always get the same variant
    print("Testing deterministic routing (same user = same variant):")
    for _ in range(3):
        v = get_variant("user-alice-123", canary_pct=5)
        print(f"  user-alice-123 -> {v.name}")

    # Test distribution across many users
    print("\nTesting traffic split with 1000 users at 5% canary:")
    canary_count = 0
    for i in range(1000):
        v = get_variant(f"user-{i}", canary_pct=5)
        if v.name == "v2.4-canary":
            canary_count += 1

    control_count = 1000 - canary_count
    print(f"  Control: {control_count} ({control_count/10:.1f}%)")
    print(f"  Canary:  {canary_count} ({canary_count/10:.1f}%)")

    # Test with 50% split (A/B test mode)
    print("\nTesting 50/50 A/B split with 1000 users:")
    canary_count = sum(
        1 for i in range(1000)
        if get_variant(f"user-{i}", canary_pct=50).name == "v2.4-canary"
    )
    print(f"  Control: {1000 - canary_count} ({(1000 - canary_count)/10:.1f}%)")
    print(f"  Canary:  {canary_count} ({canary_count/10:.1f}%)")

"""run_pipeline.py — End-to-end monitoring pipeline."""
from metrics import simulate_agent_call, get_summary_metrics, metric_store
from alerts import evaluate_alerts, route_alert
from feedback import record_feedback, daily_aggregation, generate_eval_candidates
from router import get_variant
import random

def main():
    print("=" * 60)
    print("  MONITORING & CONTINUOUS IMPROVEMENT PIPELINE")
    print("=" * 60)

    # --- Phase 1: Simulate Traffic with A/B Routing ---
    print("\n--- Phase 1: Simulating 100 agent requests with A/B routing ---")
    users = [f"user-{i}" for i in range(100)]
    canary_count = 0
    for user_id in users:
        variant = get_variant(user_id, canary_pct=10)
        if variant.name == "v2.4-canary":
            canary_count += 1
        record = simulate_agent_call(f"Question from {user_id}")

    print(f"  Requests processed: {len(metric_store)}")
    print(f"  Control group: {100 - canary_count} users")
    print(f"  Canary group: {canary_count} users")

    # --- Phase 2: Aggregate Metrics ---
    print("\n--- Phase 2: Aggregating metrics ---")
    summary = get_summary_metrics()
    print(f"  Error rate: {summary['error_rate']:.1%}")
    print(f"  p50 latency: {summary['p50_latency_ms']:.0f}ms")
    print(f"  p95 latency: {summary['p95_latency_ms']:.0f}ms")
    print(f"  Total cost: ${summary['total_cost_usd']:.4f}")

    # Add drift metric for alert evaluation
    summary["output_length_drift_pct"] = random.uniform(-5, 25)
    print(f"  Output drift: {summary['output_length_drift_pct']:+.1f}%")

    # --- Phase 3: Evaluate Alerts ---
    print("\n--- Phase 3: Evaluating alert rules ---")
    alerts = evaluate_alerts(summary)
    if alerts:
        for alert in alerts:
            action = route_alert(alert)
            print(f"  {action}")
    else:
        print("  No alerts fired. All metrics within thresholds.")

    # --- Phase 4: Collect User Feedback ---
    print("\n--- Phase 4: Simulating user feedback ---")
    for record in metric_store:
        # Simulate: 30% of users leave feedback, 85% positive
        if random.random() < 0.30:
            score = 1 if random.random() > 0.15 else 0
            record_feedback(record.trace_id, score)

    report = daily_aggregation()
    print(f"  Feedback collected: {report['total']}")
    print(f"  Satisfaction rate: {report['satisfaction_rate']:.0%}")
    if report["top_failure_categories"]:
        print(f"  Top failures:")
        for cat, count in report["top_failure_categories"][:3]:
            print(f"    - {cat}: {count}")

    # --- Phase 5: Generate Eval Candidates ---
    print("\n--- Phase 5: Selecting eval candidates for weekly review ---")
    candidates = generate_eval_candidates(min_negatives=3)
    print(f"  Candidates for new eval cases: {len(candidates)}")
    for c in candidates:
        print(f"    {c.trace_id} — {c.failure_category}")

    # --- Summary ---
    print("\n" + "=" * 60)
    print("  PIPELINE COMPLETE")
    print("=" * 60)
    print(f"  Requests processed: {len(metric_store)}")
    print(f"  Alerts fired: {len(alerts)}")
    print(f"  Feedback collected: {report['total']}")
    print(f"  Eval candidates: {len(candidates)}")
    print()
    print("  In production, these results would feed into:")
    print("    - A Langfuse/Grafana dashboard (Phase 2)")
    print("    - PagerDuty/Jira/Slack (Phase 3)")
    print("    - Your eval test suite (Phase 5)")
    print("    - A weekly failure review meeting (Phase 5)")

if __name__ == "__main__":
    main()