LLMOps User Stories — Implementation Details & Code Snippets

import mlflow
from mlflow.tracking import MlflowClient

client = MlflowClient(tracking_uri="http://mlflow.internal.manga-assist.amazon.com")

# Register a new intent classifier version after training
with mlflow.start_run(run_name="intent-classifier-v3.2") as run:
    mlflow.log_params({
        "model_type": "distilbert-base-uncased",
        "training_samples": 55000,
        "epochs": 5,
        "learning_rate": 2e-5,
        "dataset_hash": "sha256:a3f8c1...",
    })
    mlflow.log_metrics({
        "accuracy": 0.923,
        "f1_macro": 0.891,
        "f1_recommendation": 0.912,
        "f1_order_tracking": 0.956,
        "p99_latency_ms": 48,
    })
    mlflow.pytorch.log_model(
        model,
        artifact_path="intent-classifier",
        registered_model_name="manga-intent-classifier",
    )

# Promote to staging after offline eval passes
client.transition_model_version_stage(
    name="manga-intent-classifier",
    version=12,
    stage="Staging",
    archive_existing_versions=False,
)

# cdk_stack.py — SageMaker endpoint definition (AWS CDK)
from aws_cdk import aws_sagemaker as sagemaker, Stack
from constructs import Construct

class IntentClassifierStack(Stack):
    def __init__(self, scope: Construct, id: str, model_data_url: str, **kwargs):
        super().__init__(scope, id, **kwargs)

        model = sagemaker.CfnModel(self, "IntentModel",
            model_name="manga-intent-classifier",
            execution_role_arn="arn:aws:iam::role/SageMakerExecRole",
            primary_container=sagemaker.CfnModel.ContainerDefinitionProperty(
                image="763104351884.dkr.ecr.us-east-1.amazonaws.com/pytorch-inference:2.1-gpu",
                model_data_url=model_data_url,
                environment={
                    "SAGEMAKER_PROGRAM": "inference.py",
                    "MODEL_NAME": "distilbert-manga-intent",
                },
            ),
        )

        endpoint_config = sagemaker.CfnEndpointConfig(self, "EndpointConfig",
            production_variants=[sagemaker.CfnEndpointConfig.ProductionVariantProperty(
                variant_name="primary",
                model_name=model.model_name,
                instance_type="ml.inf1.xlarge",
                initial_instance_count=2,
            )],
        )

        sagemaker.CfnEndpoint(self, "Endpoint",
            endpoint_name="manga-intent-classifier-prod",
            endpoint_config_name=endpoint_config.attr_endpoint_config_name,
        )

# buildspec-eval.yml — CodeBuild step that runs golden dataset evaluation
version: 0.2
phases:
  install:
    commands:
      - pip install boto3 mlflow bert-score rouge-score
  build:
    commands:
      - python evaluate_golden_dataset.py
        --endpoint manga-intent-classifier-staging
        --dataset s3://manga-assist-eval/golden-dataset-v4.json
        --output-dir eval-results/
      - python check_eval_gates.py --results-dir eval-results/
  post_build:
    commands:
      - |
        if [ $CODEBUILD_BUILD_SUCCEEDING -eq 1 ]; then
          echo "All evaluation gates passed — promoting to production"
          python promote_model.py --stage Production
        fi
artifacts:
  files:
    - eval-results/**/*

import boto3

sagemaker_client = boto3.client("sagemaker")
sns_client = boto3.client("sns")

def rollback_endpoint(endpoint_name: str, previous_variant: str):
    """Shift 100% traffic to the previous model variant."""

    sagemaker_client.update_endpoint_weights_and_capacities(
        EndpointName=endpoint_name,
        DesiredWeightsAndCapacities=[
            {"VariantName": "current", "DesiredWeight": 0},
            {"VariantName": previous_variant, "DesiredWeight": 1},
        ],
    )

    sns_client.publish(
        TopicArn="arn:aws:sns:us-east-1:ACCOUNT:manga-assist-alerts",
        Subject=f"ROLLBACK: {endpoint_name}",
        Message=(
            f"Endpoint {endpoint_name} rolled back to variant {previous_variant}. "
            "Golden dataset re-evaluation triggered automatically."
        ),
    )

    # Trigger golden dataset re-evaluation on the restored variant
    invoke_evaluation_pipeline(endpoint_name, variant=previous_variant)

import boto3

runtime = boto3.client("sagemaker-runtime")

def invoke_intent_classifier(message: str) -> dict:
    response = runtime.invoke_endpoint(
        EndpointName="manga-shared-ml-endpoint",
        TargetModel="intent-classifier/model.tar.gz",  # Routes to DistilBERT
        ContentType="application/json",
        Body=json.dumps({"text": message}),
    )
    return json.loads(response["Body"].read())

def invoke_reranker(query: str, documents: list[str]) -> list[float]:
    response = runtime.invoke_endpoint(
        EndpointName="manga-shared-ml-endpoint",
        TargetModel="reranker/model.tar.gz",  # Routes to cross-encoder
        ContentType="application/json",
        Body=json.dumps({"query": query, "documents": documents}),
    )
    return json.loads(response["Body"].read())["scores"]

import torch
import torch_neuronx
from transformers import DistilBertForSequenceClassification, DistilBertTokenizer

# Load trained model
model = DistilBertForSequenceClassification.from_pretrained("./manga-intent-v3.2")
tokenizer = DistilBertTokenizer.from_pretrained("./manga-intent-v3.2")
model.eval()

# Create sample input for tracing
sample = tokenizer("recommend dark fantasy manga", return_tensors="pt",
                    max_length=128, padding="max_length", truncation=True)
sample_inputs = (sample["input_ids"], sample["attention_mask"])

# Compile for Inferentia
neuron_model = torch_neuronx.trace(model, sample_inputs)
neuron_model.save("manga_intent_neuron.pt")

# Benchmark: verify latency and accuracy
import time
start = time.perf_counter()
for _ in range(1000):
    neuron_model(*sample_inputs)
avg_ms = (time.perf_counter() - start)  # ~8ms avg vs ~15ms on GPU
print(f"Avg inference: {avg_ms:.1f}ms")

# prompts/recommendation/v2.3.yaml
---
version: "2.3"
author: "sr-prompt-eng"
date: "2026-03-01"
description: "Added explicit length constraint to prevent response inflation after Claude 3.5 upgrade"
eval_run_id: "mlflow-run-abc123"
intent: "recommendation"
---

system: |
  You are MangaAssist, a helpful shopping assistant for the JP Manga store on Amazon.com.
  RULES:
  1. Only recommend products from the PRODUCT DATA section below.
  2. Never invent prices, ASINs, or availability.
  3. Keep recommendations to 2-4 titles with one sentence each explaining why.
  4. Do not use emoji.
  5. Response must be under 200 words.

context_template: |
  USER PROFILE:
  - Prime: {{ user.is_prime }}
  - Locale: {{ user.locale }}
  - Recent browsing: {{ user.recent_asins | join(', ') }}

  PRODUCT DATA:
  {{ product_data_json }}

  RETRIEVED KNOWLEDGE:
  {% for chunk in rag_chunks %}
  [{{ chunk.source_type }}] {{ chunk.content }}
  {% endfor %}

  CONVERSATION HISTORY:
  {% for turn in history[-5:] %}
  {{ turn.role }}: {{ turn.content }}
  {% endfor %}

# Load active prompt version from AppConfig at runtime
import boto3

appconfig = boto3.client("appconfig")

def get_active_prompt(intent: str) -> str:
    response = appconfig.get_configuration(
        Application="manga-assist",
        Environment="production",
        Configuration=f"prompt-{intent}",
        ClientId="orchestrator-node-1",
    )
    return yaml.safe_load(response["Content"].read())

# evaluate_golden_dataset.py
import json
import boto3
from bert_score import score as bert_score
from rouge_score import rouge_scorer

def evaluate(endpoint: str, dataset_path: str) -> dict:
    with open(dataset_path) as f:
        golden = json.load(f)

    results = {"intent_correct": 0, "bertscore_sum": 0, "guardrail_pass": 0,
               "prohibited_violations": 0, "total": len(golden)}
    scorer = rouge_scorer.RougeScorer(["rougeL"], use_stemmer=True)

    for entry in golden:
        # Run through full pipeline
        response = invoke_pipeline(endpoint, entry["query"], entry["context"])

        # Check intent accuracy
        if response["classified_intent"] == entry["expected_intent"]:
            results["intent_correct"] += 1

        # BERTScore against reference
        P, R, F1 = bert_score(
            [response["text"]], [entry["reference_response"]], lang="en"
        )
        results["bertscore_sum"] += F1.item()

        # Prohibited element check
        for prohibited in entry.get("prohibited_elements", []):
            if prohibited.lower() in response["text"].lower():
                results["prohibited_violations"] += 1

        # Guardrail pass
        if response["guardrail_passed"]:
            results["guardrail_pass"] += 1

    # Compute final metrics
    n = results["total"]
    return {
        "intent_accuracy": results["intent_correct"] / n,
        "avg_bertscore": results["bertscore_sum"] / n,
        "guardrail_pass_rate": results["guardrail_pass"] / n,
        "prohibited_violations": results["prohibited_violations"],
    }


def check_gates(metrics: dict) -> bool:
    """Return True if all quality gates pass."""
    gates = {
        "intent_accuracy >= 0.90": metrics["intent_accuracy"] >= 0.90,
        "avg_bertscore >= 0.80": metrics["avg_bertscore"] >= 0.80,
        "guardrail_pass_rate >= 0.95": metrics["guardrail_pass_rate"] >= 0.95,
        "prohibited_violations == 0": metrics["prohibited_violations"] == 0,
    }
    for gate, passed in gates.items():
        status = "PASS" if passed else "FAIL"
        print(f"  [{status}] {gate}")
    return all(gates.values())

import hashlib

# Experiment configuration (loaded from AppConfig)
EXPERIMENT = {
    "id": "exp-prompt-rec-v24",
    "variants": {
        "control": {"prompt_version": "v2.3", "weight": 50},
        "treatment": {"prompt_version": "v2.4", "weight": 50},
    },
    "metrics": ["thumbs_up_rate", "escalation_rate", "conversion_rate"],
}

def assign_variant(session_id: str, experiment: dict) -> str:
    """Deterministic variant assignment based on session_id hash."""
    hash_val = int(hashlib.sha256(session_id.encode()).hexdigest(), 16) % 100
    cumulative = 0
    for variant_name, config in experiment["variants"].items():
        cumulative += config["weight"]
        if hash_val < cumulative:
            return variant_name
    return list(experiment["variants"].keys())[-1]


def get_prompt_for_session(session_id: str, intent: str) -> dict:
    variant = assign_variant(session_id, EXPERIMENT)
    prompt_version = EXPERIMENT["variants"][variant]["prompt_version"]
    return {
        "variant": variant,
        "experiment_id": EXPERIMENT["id"],
        "prompt": load_prompt(intent, prompt_version),
    }


def emit_experiment_metric(session_id: str, variant: str, metric: str, value: float):
    """Send to Kinesis for Redshift aggregation."""
    kinesis.put_record(
        StreamName="manga-assist-experiments",
        Data=json.dumps({
            "experiment_id": EXPERIMENT["id"],
            "variant": variant,
            "session_id": session_id,
            "metric": metric,
            "value": value,
            "timestamp": datetime.utcnow().isoformat(),
        }),
        PartitionKey=session_id,
    )

from jinja2 import Environment, FileSystemLoader

class PromptBuilder:
    def __init__(self, prompts_dir: str = "prompts/"):
        self.env = Environment(loader=FileSystemLoader(prompts_dir))
        self._base = self.env.get_template("base_system.jinja2")

    def build(self, intent: str, context: dict) -> str:
        base_prompt = self._base.render(context)
        intent_template = self.env.get_template(f"{intent}/template.jinja2")
        intent_section = intent_template.render(context)
        return f"{base_prompt}\n\n{intent_section}\n\nUSER: {context['user_message']}"


# Usage in orchestrator
builder = PromptBuilder()
prompt = builder.build(
    intent="recommendation",
    context={
        "user": {"is_prime": True, "locale": "en-US"},
        "product_data_json": catalog_results,
        "rag_chunks": retrieved_chunks,
        "history": conversation_turns[-5:],
        "user_message": "Something like Demon Slayer but darker",
    },
)

import boto3
from datetime import datetime, timedelta

s3 = boto3.client("s3")

def quarterly_refresh(dataset_bucket: str, dataset_key: str):
    """Identify stale golden dataset entries and suggest replacements."""

    dataset = json.loads(
        s3.get_object(Bucket=dataset_bucket, Key=dataset_key)["Body"].read()
    )

    stale_entries = []
    for entry in dataset:
        # Check if referenced products still exist in catalog
        if entry.get("context", {}).get("asin"):
            if not catalog_exists(entry["context"]["asin"]):
                stale_entries.append(entry["query_id"])

    # Pull candidates from production error logs (thumbs-down, escalations)
    candidates = query_error_logs(
        start_date=datetime.utcnow() - timedelta(days=90),
        limit=100,
        min_severity="thumbs_down",
    )

    return {
        "stale_count": len(stale_entries),
        "stale_ids": stale_entries,
        "candidate_replacements": len(candidates),
        "candidates": candidates[:50],  # Top 50 by frequency
    }

// Golden dataset entry schema
{
  "query_id": "GD-042",
  "query": "What dark fantasy manga would you recommend for someone who loved Berserk?",
  "intent": "recommendation",
  "context": {
    "user_profile": {"prime": true, "locale": "en-US"},
    "browsing_history": ["B00GX...", "B01HN..."]
  },
  "expected_intent": "recommendation",
  "reference_response": "Based on your love for Berserk, I'd recommend Vinland Saga and Claymore.",
  "required_elements": ["at least 2 product recommendations", "genre reasoning"],
  "prohibited_elements": ["competitor mentions", "fabricated prices"],
  "tags": ["recommendation", "dark-fantasy", "medium-complexity"]
}

# Evaluation gate definitions — all must pass to merge
EVALUATION_GATES = {
    "intent_accuracy":       {"threshold": 0.90, "comparator": ">="},
    "avg_bertscore_f1":      {"threshold": 0.80, "comparator": ">="},
    "rouge_l_delta":         {"threshold": 0.10, "comparator": "<="},  # Max 10% drop
    "format_compliance":     {"threshold": 0.95, "comparator": ">="},
    "guardrail_pass_rate":   {"threshold": 0.95, "comparator": ">="},
    "response_length_delta": {"threshold": 0.30, "comparator": "<="},  # Max 30% drift
    "prohibited_violations": {"threshold": 0,    "comparator": "=="},
    "min_class_f1":          {"threshold": 0.85, "comparator": ">="},
}

def run_offline_evaluation(staging_endpoint: str, golden_dataset: list[dict]) -> dict:
    """Run all golden queries and compute metrics."""
    predictions, references, intents_pred, intents_true = [], [], [], []

    for entry in golden_dataset:
        result = invoke_staging_pipeline(staging_endpoint, entry)
        predictions.append(result["response_text"])
        references.append(entry["reference_response"])
        intents_pred.append(result["classified_intent"])
        intents_true.append(entry["expected_intent"])

    # Compute aggregate metrics
    P, R, F1 = bert_score(predictions, references, lang="en")

    return {
        "intent_accuracy": accuracy_score(intents_true, intents_pred),
        "avg_bertscore_f1": F1.mean().item(),
        "per_class_f1": classification_report(intents_true, intents_pred, output_dict=True),
        "format_compliance": sum(1 for p in predictions if is_valid_format(p)) / len(predictions),
        "guardrail_pass_rate": sum(1 for r in results if r["guardrail_passed"]) / len(results),
    }

import asyncio
import mlflow

async def handle_message_with_shadow(request: ChatRequest, shadow_enabled: bool):
    """Process request through both prod and candidate when shadow mode is on."""

    # Always run production pipeline (serves the user)
    prod_response = await run_pipeline(request, model_version="production")

    if shadow_enabled:
        # Run candidate in parallel — fire-and-forget, does not block response
        asyncio.create_task(
            run_shadow_pipeline(request, prod_response)
        )

    return prod_response


async def run_shadow_pipeline(request: ChatRequest, prod_response: ChatResponse):
    """Run candidate pipeline and log comparison — never serves to user."""
    try:
        candidate_response = await run_pipeline(request, model_version="candidate")

        # Compute comparison metrics
        P, R, F1 = bert_score(
            [candidate_response.text], [prod_response.text], lang="en"
        )

        comparison = {
            "request_id": request.request_id,
            "intent": request.classified_intent,
            "prod_length": len(prod_response.text.split()),
            "candidate_length": len(candidate_response.text.split()),
            "bertscore_f1": F1.item(),
            "prod_guardrail_passed": prod_response.guardrail_passed,
            "candidate_guardrail_passed": candidate_response.guardrail_passed,
            "timestamp": datetime.utcnow().isoformat(),
        }

        # Log to Kinesis for aggregation
        kinesis.put_record(
            StreamName="manga-assist-shadow",
            Data=json.dumps(comparison),
            PartitionKey=request.request_id,
        )
    except Exception as e:
        logger.warning(f"Shadow pipeline failed (non-blocking): {e}")

import boto3

sm = boto3.client("sagemaker")

def deploy_canary(endpoint_name: str, canary_model_url: str):
    """Add a canary variant at 1% traffic."""

    sm.update_endpoint_weights_and_capacities(
        EndpointName=endpoint_name,
        DesiredWeightsAndCapacities=[
            {"VariantName": "baseline", "DesiredWeight": 99},
            {"VariantName": "canary", "DesiredWeight": 1},
        ],
    )

def promote_canary(endpoint_name: str, target_weight: int):
    """Promote canary: 1% -> 10% -> 50% -> 100%."""
    sm.update_endpoint_weights_and_capacities(
        EndpointName=endpoint_name,
        DesiredWeightsAndCapacities=[
            {"VariantName": "baseline", "DesiredWeight": 100 - target_weight},
            {"VariantName": "canary", "DesiredWeight": target_weight},
        ],
    )

def auto_rollback_handler(event, context):
    """Lambda triggered by CloudWatch alarm on canary metrics."""
    alarm_name = event["detail"]["alarmName"]
    endpoint = event["detail"]["configuration"]["metrics"][0]["dimensions"]["EndpointName"]

    sm.update_endpoint_weights_and_capacities(
        EndpointName=endpoint,
        DesiredWeightsAndCapacities=[
            {"VariantName": "baseline", "DesiredWeight": 100},
            {"VariantName": "canary", "DesiredWeight": 0},
        ],
    )

    sns.publish(
        TopicArn="arn:aws:sns:us-east-1:ACCOUNT:manga-assist-alerts",
        Subject=f"AUTO-ROLLBACK: Canary on {endpoint}",
        Message=f"Alarm {alarm_name} triggered. Canary traffic set to 0%.",
    )

import boto3

cw = boto3.client("cloudwatch")

def emit_response_metrics(response: ChatResponse, trace: dict):
    """Emit per-response metrics to CloudWatch for continuous monitoring."""

    cw.put_metric_data(
        Namespace="MangaAssist/AIQuality",
        MetricData=[
            {
                "MetricName": "HallucinationDetected",
                "Value": 1 if response.hallucination_flag else 0,
                "Unit": "Count",
                "Dimensions": [
                    {"Name": "Intent", "Value": response.intent},
                    {"Name": "ModelVersion", "Value": trace["model_version"]},
                ],
            },
            {
                "MetricName": "EscalationTriggered",
                "Value": 1 if response.escalated else 0,
                "Unit": "Count",
                "Dimensions": [
                    {"Name": "Intent", "Value": response.intent},
                ],
            },
            {
                "MetricName": "GuardrailBlocked",
                "Value": 1 if not response.guardrail_passed else 0,
                "Unit": "Count",
            },
            {
                "MetricName": "ResponseLatencyMs",
                "Value": trace["total_latency_ms"],
                "Unit": "Milliseconds",
            },
        ],
    )

// CloudWatch composite alarm — auto-rollback trigger
{
  "AlarmName": "MangaAssist-CriticalQuality",
  "AlarmRule": "ALARM(HallucinationRateHigh) OR ALARM(EscalationRateHigh) OR ALARM(ErrorRateHigh)",
  "AlarmActions": ["arn:aws:sns:us-east-1:ACCOUNT:auto-rollback-trigger"],
  "AlarmDescription": "Triggers auto-rollback if any critical quality metric breaches threshold"
}

import random
from collections import Counter

def sample_for_human_eval(responses_table: str, sample_size: int = 200) -> list[dict]:
    """Stratified sampling of production responses for human evaluation."""

    # Target distribution matches golden dataset
    intent_quotas = {
        "recommendation": 48,   # 24%
        "product_question": 40, # 20%
        "faq": 32,              # 16%
        "order_tracking": 24,   # 12%
        "multi_turn": 20,       # 10%
        "edge_case": 16,        # 8%
        "return_request": 12,   # 6%
        "chitchat": 8,          # 4%
    }

    samples = []
    for intent, count in intent_quotas.items():
        intent_responses = query_responses_by_intent(responses_table, intent, limit=count * 3)
        samples.extend(random.sample(intent_responses, min(count, len(intent_responses))))

    return samples


# Annotation schema per response
ANNOTATION_SCHEMA = {
    "factual_correctness":  {"type": "likert", "scale": [1, 2, 3, 4, 5]},
    "helpfulness":          {"type": "likert", "scale": [1, 2, 3, 4, 5]},
    "tone_appropriateness": {"type": "likert", "scale": [1, 2, 3, 4, 5]},
    "format_quality":       {"type": "likert", "scale": [1, 2, 3, 4, 5]},
    "purchase_likelihood":  {"type": "binary", "options": ["yes", "no"]},
    "failure_category":     {"type": "optional_select", "options": [
        "none", "wrong_intent", "bad_retrieval", "hallucination",
        "unhelpful_tone", "missing_info", "format_issue"
    ]},
}

import mlflow
from mlflow.entities import SpanType

# Enable auto-tracing for Bedrock calls
mlflow.anthropic.autolog()

# Configure async logging + sampling for production
mlflow.config.enable_async_logging(True)
mlflow.config.set_sampling_ratio(0.10)

@mlflow.trace(span_type=SpanType.CHAIN, name="orchestrator.handle_message")
async def handle_message(request: ChatRequest) -> ChatResponse:
    """Full pipeline with automatic tracing for every span."""

    # Each function below is also decorated with @mlflow.trace
    context = await load_conversation_context(request.session_id)
    intent = classify_intent(request.message, request.page_context)
    service_data = await fetch_service_data(intent, context)
    prompt = build_prompt(intent, service_data, context)

    # Bedrock call is auto-traced by mlflow.anthropic.autolog()
    raw_response = await call_bedrock(prompt)

    validated = apply_guardrails(raw_response, service_data)
    await save_turn(request.session_id, request.message, validated.text)

    # Tag trace with searchable metadata
    mlflow.update_current_trace(tags={
        "session_id": request.session_id,
        "intent": intent.type.value,
        "model_version": "claude-3.5-sonnet-v2",
        "cache_hit": str(service_data.get("cache_hit", False)),
    })

    return validated

import numpy as np
from scipy.stats import ks_2samp

# Baseline intent distribution (set at training time)
BASELINE_DISTRIBUTION = {
    "recommendation": 0.35, "product_question": 0.22, "faq": 0.18,
    "order_tracking": 0.12, "chitchat": 0.08, "escalation": 0.05,
}

def check_intent_drift(current_distribution: dict, threshold_pp: float = 5.0) -> list[dict]:
    """Alert if any intent shifts > threshold percentage points from baseline."""
    alerts = []
    for intent, baseline_share in BASELINE_DISTRIBUTION.items():
        current_share = current_distribution.get(intent, 0)
        drift_pp = abs(current_share - baseline_share) * 100
        if drift_pp > threshold_pp:
            alerts.append({
                "intent": intent,
                "baseline": f"{baseline_share:.1%}",
                "current": f"{current_share:.1%}",
                "drift_pp": f"{drift_pp:.1f}pp",
                "action": "Investigate query patterns; may need classifier retraining",
            })
    return alerts


def check_embedding_drift(recent_embeddings: np.ndarray, training_embeddings: np.ndarray) -> dict:
    """KS test on cosine similarity distributions to detect embedding space shift."""
    from sklearn.metrics.pairwise import cosine_similarity

    training_centroid = training_embeddings.mean(axis=0, keepdims=True)
    recent_sims = cosine_similarity(recent_embeddings, training_centroid).flatten()
    training_sims = cosine_similarity(training_embeddings, training_centroid).flatten()

    ks_stat, p_value = ks_2samp(recent_sims, training_sims)
    return {
        "ks_statistic": ks_stat,
        "p_value": p_value,
        "drift_detected": p_value < 0.01,
        "action": "Re-index knowledge base with latest queries" if p_value < 0.01 else "No action needed",
    }

// Grafana dashboard panel definition (JSON model excerpt)
{
  "title": "MangaAssist LLM Pipeline — Real-Time",
  "panels": [
    {
      "title": "Active Sessions",
      "type": "stat",
      "datasource": "CloudWatch",
      "targets": [{"namespace": "MangaAssist/Sessions", "metricName": "ActiveSessions"}]
    },
    {
      "title": "P99 Latency (ms)",
      "type": "timeseries",
      "datasource": "Prometheus",
      "targets": [{"expr": "histogram_quantile(0.99, rate(manga_response_latency_bucket[5m]))"}]
    },
    {
      "title": "Intent Distribution",
      "type": "piechart",
      "datasource": "Prometheus",
      "targets": [{"expr": "sum by (intent) (rate(manga_intent_total[1h]))"}]
    },
    {
      "title": "LLM Cost / Hour",
      "type": "stat",
      "datasource": "CloudWatch",
      "targets": [{"namespace": "MangaAssist/Cost", "metricName": "LLMCostPerHour"}]
    },
    {
      "title": "Guardrail Block Rate",
      "type": "gauge",
      "datasource": "Prometheus",
      "targets": [{"expr": "rate(manga_guardrail_blocked_total[1h]) / rate(manga_requests_total[1h])"}]
    }
  ],
  "refresh": "10s"
}

import time
import boto3

cw = boto3.client("cloudwatch")

async def call_bedrock_with_metrics(prompt: str, intent: str) -> dict:
    """Invoke Bedrock and capture LLM-specific metrics."""

    start = time.perf_counter()
    first_token_time = None

    bedrock = boto3.client("bedrock-runtime")
    response = bedrock.invoke_model_with_response_stream(
        modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "messages": [{"role": "user", "content": prompt}],
            "max_tokens": 800,
            "temperature": 0.3,
        }),
    )

    tokens = []
    for event in response["body"]:
        chunk = json.loads(event["chunk"]["bytes"])
        if chunk["type"] == "content_block_delta":
            if first_token_time is None:
                first_token_time = time.perf_counter()
            tokens.append(chunk["delta"]["text"])

    total_time = time.perf_counter() - start
    ttft = (first_token_time - start) if first_token_time else total_time
    output_text = "".join(tokens)

    # Extract usage from final message
    input_tokens = response.get("inputTokenCount", 0)
    output_tokens = len(tokens)

    # Emit LLM-specific metrics
    cw.put_metric_data(
        Namespace="MangaAssist/LLM",
        MetricData=[
            {"MetricName": "TTFT_ms", "Value": ttft * 1000, "Unit": "Milliseconds",
             "Dimensions": [{"Name": "Intent", "Value": intent}, {"Name": "Model", "Value": "claude-3.5-sonnet"}]},
            {"MetricName": "InputTokens", "Value": input_tokens, "Unit": "Count",
             "Dimensions": [{"Name": "Intent", "Value": intent}]},
            {"MetricName": "OutputTokens", "Value": output_tokens, "Unit": "Count",
             "Dimensions": [{"Name": "Intent", "Value": intent}]},
            {"MetricName": "TotalLatency_ms", "Value": total_time * 1000, "Unit": "Milliseconds"},
        ],
    )

    return {"text": output_text, "input_tokens": input_tokens, "output_tokens": output_tokens,
            "ttft_ms": ttft * 1000, "total_ms": total_time * 1000}

# CDK alarm definitions
from aws_cdk import aws_cloudwatch as cw, aws_cloudwatch_actions as cw_actions, aws_sns as sns

# Critical: pages on-call, triggers auto-rollback
hallucination_alarm = cw.Alarm(self, "HallucinationRateHigh",
    metric=cw.Metric(
        namespace="MangaAssist/AIQuality",
        metric_name="HallucinationDetected",
        statistic="Average",
        period=Duration.minutes(5),
    ),
    threshold=0.05,  # 5%
    evaluation_periods=3,
    comparison_operator=cw.ComparisonOperator.GREATER_THAN_THRESHOLD,
    alarm_description="Hallucination rate >5% for 15 min — auto-rollback triggered",
)
hallucination_alarm.add_alarm_action(cw_actions.SnsAction(pagerduty_topic))
hallucination_alarm.add_alarm_action(cw_actions.SnsAction(auto_rollback_topic))

# Warning: Slack notification, no auto-rollback
escalation_warning = cw.Alarm(self, "EscalationRateWarning",
    metric=cw.Metric(
        namespace="MangaAssist/AIQuality",
        metric_name="EscalationTriggered",
        statistic="Average",
        period=Duration.minutes(15),
    ),
    threshold=0.17,  # 17%
    evaluation_periods=2,
    alarm_description="Escalation rate >17% for 30 min — investigate",
)
escalation_warning.add_alarm_action(cw_actions.SnsAction(slack_topic))

import boto3
from opensearchpy import OpenSearch, RequestsHttpConnection

bedrock = boto3.client("bedrock-runtime")

def chunk_and_index(document: dict, chunk_size: int = 512, overlap: int = 50):
    """Chunk a document, embed each chunk, and upsert to OpenSearch."""

    chunks = sliding_window_chunk(document["content"], chunk_size, overlap)

    for i, chunk_text in enumerate(chunks):
        # Embed using Titan Embeddings V2
        embed_response = bedrock.invoke_model(
            modelId="amazon.titan-embed-text-v2:0",
            body=json.dumps({"inputText": chunk_text}),
        )
        embedding = json.loads(embed_response["body"].read())["embedding"]

        # Upsert to OpenSearch
        chunk_doc = {
            "chunk_id": f"{document['doc_id']}-{i:03d}",
            "content": chunk_text,
            "embedding": embedding,
            "source_type": document["source_type"],
            "asin": document.get("asin"),
            "category": document.get("category"),
            "last_updated": datetime.utcnow().isoformat(),
            "source_doc_hash": hashlib.sha256(document["content"].encode()).hexdigest(),
        }

        opensearch_client.index(
            index="manga-embeddings",
            id=chunk_doc["chunk_id"],
            body=chunk_doc,
        )


def detect_stale_chunks():
    """Find chunks whose source document no longer exists or has been updated."""
    all_chunks = scroll_all_chunks("manga-embeddings")
    stale = []
    for chunk in all_chunks:
        source_doc = fetch_source_document(chunk["source_type"], chunk["asin"])
        if source_doc is None:
            stale.append({"chunk_id": chunk["chunk_id"], "reason": "source_deleted"})
        elif hashlib.sha256(source_doc["content"].encode()).hexdigest() != chunk["source_doc_hash"]:
            stale.append({"chunk_id": chunk["chunk_id"], "reason": "source_updated"})
    return stale

def upgrade_embedding_model(new_model_id: str, new_index: str, old_index: str):
    """Full re-index and atomic cutover for embedding model upgrades."""

    # Step 1: Re-index all documents with new embedding model
    all_source_docs = fetch_all_source_documents()
    for doc in all_source_docs:
        chunk_and_index(doc, embedding_model=new_model_id, target_index=new_index)

    # Step 2: Run retrieval evaluation
    eval_results = evaluate_retrieval(
        index=new_index,
        eval_dataset="s3://manga-assist-eval/retrieval-eval-100.json",
    )

    if eval_results["recall_at_3"] < 0.80 or eval_results["mrr_at_3"] < 0.60:
        raise EvalFailure(f"New embeddings failed: {eval_results}")

    # Step 3: Atomic alias swap
    opensearch_client.indices.update_aliases(body={
        "actions": [
            {"remove": {"index": old_index, "alias": "manga-embeddings-live"}},
            {"add": {"index": new_index, "alias": "manga-embeddings-live"}},
        ]
    })

    # Step 4: Schedule old index deletion after 7 days
    schedule_index_deletion(old_index, delay_days=7)

def weekly_retrieval_evaluation(eval_dataset_key: str):
    """Evaluate RAG retrieval quality against curated test set."""

    eval_data = load_from_s3(eval_dataset_key)
    recall_hits, mrr_sum, reranker_scores = 0, 0.0, []

    for entry in eval_data:
        query = entry["query"]
        expected_doc_ids = set(entry["relevant_chunk_ids"])

        # Retrieve top 3 via full RAG pipeline
        results = retrieve_and_rerank(query, top_k=3)
        retrieved_ids = [r["chunk_id"] for r in results]

        # Recall@3: did we find the relevant document?
        if expected_doc_ids & set(retrieved_ids):
            recall_hits += 1

        # MRR@3: reciprocal rank of first relevant result
        for rank, rid in enumerate(retrieved_ids, 1):
            if rid in expected_doc_ids:
                mrr_sum += 1.0 / rank
                break

        reranker_scores.append(results[0]["reranker_score"] if results else 0)

    n = len(eval_data)
    metrics = {
        "recall_at_3": recall_hits / n,
        "mrr_at_3": mrr_sum / n,
        "avg_reranker_score": sum(reranker_scores) / n,
        "evaluated_queries": n,
    }

    # Alert if below threshold
    if metrics["recall_at_3"] < 0.75:
        send_alert(f"RAG Recall@3 dropped to {metrics['recall_at_3']:.2%} (threshold: 75%)")

    return metrics

import mlflow

CHUNK_EXPERIMENTS = [
    {"chunk_size": 256, "overlap": 25, "name": "small-chunks"},
    {"chunk_size": 512, "overlap": 50, "name": "medium-chunks"},
    {"chunk_size": 1024, "overlap": 100, "name": "large-chunks"},
    {"chunk_size": 512, "overlap": 0, "name": "no-overlap"},
]

def run_chunk_experiment(config: dict, test_corpus: list[dict], eval_dataset: list[dict]):
    with mlflow.start_run(run_name=config["name"]):
        mlflow.log_params(config)

        # Create test index with candidate chunk strategy
        test_index = f"chunk-experiment-{config['name']}"
        for doc in test_corpus:
            chunk_and_index(doc, chunk_size=config["chunk_size"],
                           overlap=config["overlap"], target_index=test_index)

        # Evaluate retrieval quality
        retrieval_metrics = evaluate_retrieval(test_index, eval_dataset)
        mlflow.log_metrics(retrieval_metrics)

        # Also evaluate downstream response quality
        response_metrics = evaluate_golden_dataset_with_index(test_index)
        mlflow.log_metrics({"downstream_bertscore": response_metrics["avg_bertscore"]})

        # Cleanup test index
        delete_index(test_index)

SESSION_TOKEN_BUDGET = 5000  # Configurable via AppConfig

class TokenBudgetManager:
    def __init__(self, session_id: str, dynamodb_table):
        self.session_id = session_id
        self.table = dynamodb_table

    def get_remaining(self) -> int:
        item = self.table.get_item(Key={"session_id": self.session_id}).get("Item", {})
        used = item.get("tokens_used", 0)
        return max(0, SESSION_TOKEN_BUDGET - used)

    def record_usage(self, input_tokens: int, output_tokens: int):
        self.table.update_item(
            Key={"session_id": self.session_id},
            UpdateExpression="ADD tokens_used :t",
            ExpressionAttributeValues={":t": input_tokens + output_tokens},
        )

    def get_degradation_strategy(self) -> str:
        remaining = self.get_remaining()
        if remaining > 2000:
            return "full"        # Full pipeline: RAG + Sonnet + full context
        elif remaining > 500:
            return "reduced"     # Skip RAG, use shorter context, switch to Haiku
        else:
            return "minimal"     # Template responses only, no LLM


# Usage in orchestrator
budget = TokenBudgetManager(request.session_id, sessions_table)
strategy = budget.get_degradation_strategy()

if strategy == "full":
    response = await full_pipeline(request)
elif strategy == "reduced":
    response = await reduced_pipeline(request)  # Haiku, no RAG
else:
    response = template_response(request.intent)

budget.record_usage(response.input_tokens, response.output_tokens)

# Cost calculation per request
MODEL_PRICING = {
    "claude-3.5-sonnet": {"input_per_1k": 0.003, "output_per_1k": 0.015},
    "claude-3-haiku":    {"input_per_1k": 0.00025, "output_per_1k": 0.00125},
    "titan-embed-v2":    {"input_per_1k": 0.0001, "output_per_1k": 0.0},
}

def compute_request_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    pricing = MODEL_PRICING[model]
    return (input_tokens / 1000 * pricing["input_per_1k"] +
            output_tokens / 1000 * pricing["output_per_1k"])


def emit_cost_event(request_id: str, intent: str, model: str,
                    input_tokens: int, output_tokens: int, user_type: str):
    cost = compute_request_cost(model, input_tokens, output_tokens)
    kinesis.put_record(
        StreamName="manga-assist-cost",
        Data=json.dumps({
            "request_id": request_id,
            "intent": intent,
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost_usd": cost,
            "user_type": user_type,    # "prime" | "guest"
            "region": "us-east-1",
            "timestamp": datetime.utcnow().isoformat(),
        }),
        PartitionKey=intent,
    )

-- Redshift daily cost attribution query
SELECT
    DATE(timestamp) AS day,
    intent,
    model,
    user_type,
    COUNT(*) AS request_count,
    SUM(cost_usd) AS total_cost,
    AVG(cost_usd) AS avg_cost_per_request,
    SUM(input_tokens + output_tokens) AS total_tokens
FROM manga_assist.cost_events
WHERE timestamp >= DATEADD(day, -7, CURRENT_DATE)
GROUP BY day, intent, model, user_type
ORDER BY total_cost DESC;

# Intent routing configuration (AppConfig)
INTENT_ROUTES = {
    "chitchat":       {"handler": "template", "template_id": "greeting"},
    "order_tracking": {"handler": "api_template", "api": "order-service", "template_id": "order_status"},
    "faq_simple":     {"handler": "cached", "cache_key_pattern": "faq:{query_hash}"},
    "recommendation": {"handler": "llm", "model": "claude-3.5-sonnet"},
    "product_question":{"handler": "llm", "model": "claude-3.5-sonnet"},
    "faq_complex":    {"handler": "llm", "model": "claude-3-haiku"},
}

TEMPLATES = {
    "greeting": "Hi! Welcome to the JP Manga store. How can I help you today?",
    "order_status": "Your order {order_id} shipped on {ship_date} and is expected to arrive by {delivery_date}.",
}

async def route_request(intent: str, request: ChatRequest) -> ChatResponse:
    route = INTENT_ROUTES[intent]
    handler = route["handler"]

    # Track bypass rate
    is_llm = handler == "llm"
    cw.put_metric_data(
        Namespace="MangaAssist/Cost",
        MetricData=[{"MetricName": "LLMBypassed", "Value": 0 if is_llm else 1, "Unit": "Count"}],
    )

    if handler == "template":
        return ChatResponse(text=TEMPLATES[route["template_id"]])

    elif handler == "api_template":
        api_data = await call_api(route["api"], request)
        return ChatResponse(text=TEMPLATES[route["template_id"]].format(**api_data))

    elif handler == "cached":
        cached = redis_client.get(route["cache_key_pattern"].format(query_hash=hash_query(request.message)))
        if cached:
            return ChatResponse(text=cached)
        # Cache miss: fall through to LLM
        return await llm_pipeline(request)

    else:
        return await llm_pipeline(request, model=route.get("model"))

def build_cacheable_prompt(intent: str, context: dict) -> dict:
    """Structure the prompt so the system block is cacheable by Bedrock."""

    # Static system block — identical across requests, highly cacheable
    system_block = {
        "role": "system",
        "content": [
            {
                "type": "text",
                "text": SYSTEM_PROMPT,  # ~500 tokens, stable across requests
                "cache_control": {"type": "ephemeral"},  # Enable caching
            }
        ],
    }

    # Dynamic user block — varies per request
    user_content = (
        f"CONTEXT:\n{json.dumps(context['page_context'])}\n\n"
        f"RETRIEVED KNOWLEDGE:\n{context['rag_chunks']}\n\n"
        f"CONVERSATION:\n{format_history(context['history'])}\n\n"
        f"USER: {context['user_message']}"
    )

    return {
        "anthropic_version": "bedrock-2023-05-31",
        "system": system_block["content"],
        "messages": [{"role": "user", "content": user_content}],
        "max_tokens": 800,
        "temperature": 0.3,
    }

from dataclasses import dataclass
from typing import Optional

@dataclass
class GuardrailResult:
    passed: bool
    stage: str
    reason: Optional[str] = None
    modified_text: Optional[str] = None

# Pre-generation guardrails (run on user input)
PRE_GUARDRAILS = [
    ("input_length", lambda text, _: len(text) <= 2000),
    ("prompt_injection", detect_prompt_injection),
    ("input_toxicity", check_input_toxicity),
]

# Post-generation guardrails (run on LLM output)
POST_GUARDRAILS = [
    ("asin_validation", validate_asins_exist),
    ("price_validation", validate_prices_match_catalog),
    ("pii_detection", detect_and_redact_pii),
    ("competitor_filter", check_competitor_mentions),
    ("url_validation", validate_amazon_urls_only),
    ("format_check", validate_response_format),
]

@mlflow.trace(span_type=SpanType.CHAIN, name="guardrails_pipeline")
def apply_post_guardrails(response_text: str, context: dict) -> GuardrailResult:
    current_text = response_text

    for stage_name, check_fn in POST_GUARDRAILS:
        with mlflow.start_span(name=f"guardrail.{stage_name}") as span:
            result = check_fn(current_text, context)
            span.set_attributes({"passed": result.passed, "reason": result.reason or ""})

            if not result.passed:
                return GuardrailResult(passed=False, stage=stage_name, reason=result.reason)
            if result.modified_text:
                current_text = result.modified_text  # e.g., PII redacted

    return GuardrailResult(passed=True, stage="all", modified_text=current_text)


def validate_asins_exist(text: str, context: dict) -> GuardrailResult:
    """Ensure every ASIN mentioned in the response exists in the catalog."""
    import re
    mentioned_asins = re.findall(r"B[0-9A-Z]{9}", text)
    provided_asins = {p["asin"] for p in context.get("products", [])}

    invalid = [a for a in mentioned_asins if a not in provided_asins]
    if invalid:
        return GuardrailResult(passed=False, stage="asin_validation",
                               reason=f"Hallucinated ASINs: {invalid}")
    return GuardrailResult(passed=True, stage="asin_validation")

import re

# Known prompt injection patterns
INJECTION_PATTERNS = [
    r"ignore\s+(all\s+)?previous\s+instructions",
    r"you\s+are\s+now\s+(?:a|an)\s+",
    r"forget\s+(?:everything|all|your)\s+",
    r"system\s*prompt\s*:",
    r"<\s*system\s*>",
    r"IMPORTANT:\s*(?:new|override|ignore)",
    r"(?:base64|rot13|hex)\s*:",                  # Encoded payloads
    r"\[INST\]|\[\/INST\]|<<SYS>>|<\|im_start\|>",  # Model-specific tokens
]

def detect_prompt_injection(user_input: str, context: dict) -> GuardrailResult:
    """Two-layer injection detection: regex + ML classifier."""

    # Layer 1: Regex fast path
    for pattern in INJECTION_PATTERNS:
        if re.search(pattern, user_input, re.IGNORECASE):
            log_security_event("prompt_injection_blocked", user_input, "regex", pattern)
            return GuardrailResult(passed=False, stage="prompt_injection",
                                   reason="Suspected prompt injection detected")

    # Layer 2: ML classifier for novel attacks
    injection_score = invoke_injection_classifier(user_input)
    if injection_score > 0.85:
        log_security_event("prompt_injection_blocked", user_input, "ml_classifier", injection_score)
        return GuardrailResult(passed=False, stage="prompt_injection",
                               reason="ML classifier flagged as injection attempt")

    return GuardrailResult(passed=True, stage="prompt_injection")

async def check_hallucination(response_text: str, context: dict) -> GuardrailResult:
    """Multi-check anti-hallucination validation."""

    # Check 1: ASIN existence
    asin_result = validate_asins_exist(response_text, context)
    if not asin_result.passed:
        return asin_result

    # Check 2: Price accuracy
    price_result = validate_prices(response_text, context)
    if not price_result.passed:
        return price_result

    # Check 3: Faithfulness score via Haiku
    faithfulness = await compute_faithfulness_score(response_text, context)
    if faithfulness < 0.55:
        return GuardrailResult(
            passed=False, stage="hallucination_check",
            reason=f"Faithfulness score {faithfulness:.2f} below threshold 0.55",
        )

    return GuardrailResult(passed=True, stage="hallucination_check")


async def compute_faithfulness_score(response: str, context: dict) -> float:
    """Use Haiku to score whether the response is grounded in provided context."""
    prompt = f"""Score how faithfully this RESPONSE uses only information from the CONTEXT.
Return a single float between 0.0 (completely fabricated) and 1.0 (fully grounded).

CONTEXT:
{context.get('rag_chunks', '')}
{json.dumps(context.get('products', []))}

RESPONSE:
{response}

Score (0.0-1.0):"""

    result = await call_bedrock(prompt, model="anthropic.claude-3-haiku-20240307-v1:0", max_tokens=10)
    return float(result["text"].strip())

import re

# Lightweight patterns for streaming check (must be fast: <5ms per chunk)
STREAMING_BLOCKLIST = {
    "competitor": re.compile(r"\b(Barnes\s*&?\s*Noble|RightStuf|Crunchyroll\s*Store)\b", re.I),
    "pii_email": re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"),
    "pii_phone": re.compile(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b"),
    "pii_ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
}

async def stream_with_guardrails(prompt: str, websocket):
    """Stream tokens to user with real-time guardrail checks."""

    accumulated = []
    response_stream = bedrock.invoke_model_with_response_stream(
        modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
        body=json.dumps({"messages": [{"role": "user", "content": prompt}],
                         "max_tokens": 800, "temperature": 0.3,
                         "anthropic_version": "bedrock-2023-05-31"}),
    )

    for event in response_stream["body"]:
        chunk = json.loads(event["chunk"]["bytes"])
        if chunk["type"] != "content_block_delta":
            continue

        token_text = chunk["delta"]["text"]
        accumulated.append(token_text)

        # Real-time streaming check (<5ms)
        recent_window = "".join(accumulated[-20:])  # Check last 20 tokens
        for check_name, pattern in STREAMING_BLOCKLIST.items():
            if pattern.search(recent_window):
                # HALT: replace response and stop streaming
                await websocket.send(json.dumps({
                    "type": "replace",
                    "text": "I can only help with manga-related questions on the JP Manga store.",
                }))
                log_streaming_violation(check_name, recent_window)
                return

        # Safe: forward token to user
        await websocket.send(json.dumps({"type": "token", "text": token_text}))

    # Full response post-generation guardrails
    full_response = "".join(accumulated)
    final_result = apply_post_guardrails(full_response, context)
    if not final_result.passed:
        await websocket.send(json.dumps({
            "type": "replace",
            "text": "Let me rephrase that. " + generate_safe_fallback(context),
        }))

import boto3

dynamodb = boto3.resource("dynamodb")
feedback_table = dynamodb.Table("manga-assist-feedback")

def record_feedback(feedback: dict):
    """Store feedback linked to the full trace context."""

    feedback_table.put_item(Item={
        "feedback_id": str(uuid.uuid4()),
        "session_id": feedback["session_id"],
        "message_id": feedback["message_id"],
        "trace_id": feedback["trace_id"],
        "value": feedback["value"],           # "thumbs_up" | "thumbs_down"
        "free_text": feedback.get("text", ""),
        "intent": feedback["intent"],
        "model_version": feedback["model_version"],
        "prompt_version": feedback["prompt_version"],
        "timestamp": datetime.utcnow().isoformat(),
        "ttl": int((datetime.utcnow() + timedelta(days=365)).timestamp()),
    })

    # Emit to Kinesis for real-time dashboards
    kinesis.put_record(
        StreamName="manga-assist-feedback",
        Data=json.dumps({
            "value": feedback["value"],
            "intent": feedback["intent"],
            "model_version": feedback["model_version"],
            "timestamp": datetime.utcnow().isoformat(),
        }),
        PartitionKey=feedback["session_id"],
    )

    # Update real-time thumbs-up/down rates in CloudWatch
    cw.put_metric_data(
        Namespace="MangaAssist/Feedback",
        MetricData=[{
            "MetricName": "ThumbsUp" if feedback["value"] == "thumbs_up" else "ThumbsDown",
            "Value": 1,
            "Unit": "Count",
            "Dimensions": [{"Name": "Intent", "Value": feedback["intent"]}],
        }],
    )

FAILURE_CATEGORIES = [
    "wrong_intent", "bad_retrieval", "hallucination", "unhelpful_tone",
    "missing_info", "format_issue", "out_of_scope", "slow_response", "other",
]

async def categorize_failures(failures: list[dict]) -> list[dict]:
    """Use Haiku to categorize negative feedback into root cause buckets."""

    categorized = []
    for failure in failures:
        prompt = f"""Categorize this chatbot failure into exactly one category.
Categories: {', '.join(FAILURE_CATEGORIES)}

User query: {failure['user_query']}
Bot response: {failure['bot_response']}
Intent classified as: {failure['intent']}
Feedback: {failure['feedback_value']}
User comment: {failure.get('free_text', 'none')}

Return ONLY the category name, nothing else."""

        result = await call_bedrock(prompt, model="anthropic.claude-3-haiku-20240307-v1:0", max_tokens=20)
        category = result["text"].strip().lower().replace(" ", "_")

        if category not in FAILURE_CATEGORIES:
            category = "other"

        categorized.append({**failure, "failure_category": category})

    return categorized


def generate_weekly_report(categorized_failures: list[dict]) -> str:
    """Generate a summary report for the sprint review."""
    from collections import Counter

    counts = Counter(f["failure_category"] for f in categorized_failures)
    total = len(categorized_failures)

    report = "## Weekly Error Analysis Report\n\n"
    report += f"Total negative signals: {total}\n\n"
    report += "| Category | Count | % | Example |\n|---|---|---|---|\n"

    for category, count in counts.most_common():
        example = next(f for f in categorized_failures if f["failure_category"] == category)
        report += f"| {category} | {count} | {count/total:.0%} | {example['user_query'][:60]}... |\n"

    return report

def check_and_trigger_retraining():
    """Check drift metrics and trigger retraining if needed."""

    # Check intent classifier accuracy (rolling 1K eval)
    accuracy = get_rolling_accuracy(window_days=7, sample_size=1000)

    # Check intent distribution drift
    drift_alerts = check_intent_drift(get_current_distribution())

    should_retrain = (
        accuracy < 0.88 or
        len(drift_alerts) > 0 or
        days_since_last_training() > 90  # Quarterly retrain at minimum
    )

    if should_retrain:
        # Assemble training dataset
        training_data = assemble_training_data(
            base_dataset="s3://manga-assist-data/intent-training-v3.json",
            augment_with_production=True,
            production_window_days=90,
            max_augmented_samples=2000,
        )

        # Trigger SageMaker Training Job
        sagemaker = boto3.client("sagemaker")
        sagemaker.create_training_job(
            TrainingJobName=f"intent-classifier-{datetime.utcnow().strftime('%Y%m%d')}",
            AlgorithmSpecification={
                "TrainingImage": "763104351884.dkr.ecr.us-east-1.amazonaws.com/pytorch-training:2.1-gpu",
                "TrainingInputMode": "File",
            },
            InputDataConfig=[{
                "ChannelName": "training",
                "DataSource": {"S3DataSource": {
                    "S3DataType": "S3Prefix",
                    "S3Uri": training_data["s3_uri"],
                }},
            }],
            OutputDataConfig={"S3OutputPath": "s3://manga-assist-models/intent-classifier/"},
            ResourceConfig={
                "InstanceType": "ml.g4dn.xlarge",
                "InstanceCount": 1,
                "VolumeSizeInGB": 50,
            },
            StoppingCondition={"MaxRuntimeInSeconds": 3600},
            RoleArn="arn:aws:iam::role/SageMakerTrainingRole",
        )

        return {"triggered": True, "reason": f"accuracy={accuracy:.3f}, drift_alerts={len(drift_alerts)}"}

    return {"triggered": False, "accuracy": accuracy}

import boto3

comprehend = boto3.client("comprehend")

def production_data_flywheel():
    """Extract production conversations for training and golden dataset augmentation."""

    # High-confidence: candidates for training data
    high_conf = query_conversations(
        filters={
            "feedback": "thumbs_up",
            "guardrail_passed": True,
            "escalated": False,
            "intent_confidence": {"gte": 0.90},
        },
        window_days=90,
        limit=5000,
    )

    # Low-confidence: candidates for golden dataset
    low_conf = query_conversations(
        filters={
            "feedback": "thumbs_down",
            "escalated": True,
        },
        window_days=90,
        limit=500,
    )

    # Anonymize (strip PII)
    anonymized_training = [anonymize_conversation(c) for c in high_conf]
    anonymized_golden = [anonymize_conversation(c) for c in low_conf]

    # Deduplicate by semantic similarity
    unique_training = deduplicate_by_embedding(anonymized_training, threshold=0.95)

    # Store for human review before inclusion
    upload_to_s3(unique_training, "s3://manga-assist-data/flywheel/training-candidates/")
    upload_to_s3(anonymized_golden, "s3://manga-assist-data/flywheel/golden-candidates/")

    return {"training_candidates": len(unique_training), "golden_candidates": len(anonymized_golden)}


def anonymize_conversation(conversation: dict) -> dict:
    """Strip PII using Amazon Comprehend."""
    pii_result = comprehend.detect_pii_entities(
        Text=conversation["user_message"],
        LanguageCode="en",
    )
    redacted = conversation["user_message"]
    for entity in sorted(pii_result["Entities"], key=lambda e: e["BeginOffset"], reverse=True):
        redacted = redacted[:entity["BeginOffset"]] + f"[{entity['Type']}]" + redacted[entity["EndOffset"]:]

    return {**conversation, "user_message": redacted, "anonymized": True}