Answers – Scenario 01: Thumbs Feedback Interface

User clicks 👎 → ECS Fargate API → Kinesis Data Streams (feedback-stream)
                                        ├── Lambda consumer → DynamoDB (MangaAssist_ThumbsFeedback) [hot, 90-day TTL]
                                        ├── Kinesis Firehose → S3 (s3://mangaassist-feedback/raw/) [cold, permanent]
                                        └── Kinesis Analytics → CloudWatch Metrics [real-time aggregation]

z = (p₁ - p₂) / √(p̂(1-p̂)(1/n₁ + 1/n₂))
where p̂ = (x₁ + x₂) / (n₁ + n₂)

n = (Z_α/2 + Z_β)² × (p₁(1-p₁) + p₂(1-p₂)) / (p₁ - p₂)²
n ≈ (1.96 + 0.84)² × (0.72×0.28 + 0.58×0.42) / (0.14)²
n ≈ 7.84 × 0.4452 / 0.0196 ≈ 178 per group

Thumbs events → Kinesis Data Streams → Kinesis Data Analytics (SQL application)
                                                ↓
                                        CloudWatch Custom Metric: "ThumbsDownRate_by_Intent"
                                                ↓
                                        CloudWatch Alarm (threshold breach)
                                                ↓
                                        SNS → PagerDuty / Slack

CREATE OR REPLACE STREAM "FEEDBACK_METRICS" (
    intent VARCHAR(64),
    window_end TIMESTAMP,
    total_feedback INTEGER,
    thumbs_down_count INTEGER,
    thumbs_down_rate DOUBLE
);

CREATE OR REPLACE PUMP "FEEDBACK_PUMP" AS
INSERT INTO "FEEDBACK_METRICS"
SELECT STREAM
    intent,
    STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '5' MINUTE) AS window_end,
    COUNT(*) AS total_feedback,
    SUM(CASE WHEN feedback = 'thumbs_down' THEN 1 ELSE 0 END) AS thumbs_down_count,
    CAST(SUM(CASE WHEN feedback = 'thumbs_down' THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*) AS thumbs_down_rate
FROM "SOURCE_SQL_STREAM_001"
GROUP BY
    intent,
    STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '5' MINUTE);

User clicks 👎 → Modal: "What went wrong?" [text field, 500 char limit]
    → POST /api/feedback {session_id, turn_id, feedback: "thumbs_down", comment: "..."}
        → API Gateway → Lambda (feedback-processor)
            → Step 1: Write raw event to DynamoDB (MangaAssist_ThumbsFeedback)
            → Step 2: Invoke Bedrock Claude 3.5 Sonnet (Haiku for cost) to classify reason
            → Step 3: Write classified category back to DynamoDB (update item)
            → Step 4: Push to Kinesis for downstream analytics

Classify this chatbot feedback into exactly one category:
- wrong_genre: Manga recommendation didn't match user's preferred genre
- wrong_product: Recommended a product the user already owns or isn't interested in
- outdated_info: Price, availability, or shipping info was incorrect
- misunderstood_query: Chatbot didn't understand what the user was asking
- incomplete_answer: Answer was partially correct but missing key details
- tone_issue: Response felt robotic, rude, or inappropriate
- other: Doesn't fit above categories

Feedback: "{user_comment}"
Intent: {intent}

Return JSON: {"category": "...", "confidence": 0.XX}

{
    "comment_raw": "it recommended One Piece but I already read all 100+ volumes",
    "comment_category": "wrong_product",
    "comment_confidence": 0.94,
    "comment_classified_at": "2025-03-15T10:24:01Z"
}

thumbs_down_ratio(response_hash, window) = 
    count(thumbs_down for response_hash in window) / 
    count(all_feedback for response_hash in window)

Cache invalidation triggered
    → Lambda publishes to SQS (regeneration-queue)
        → ECS Fargate worker picks up message
            → Fetches original query context from DynamoDB conversation history
            → Calls Bedrock Claude 3.5 Sonnet with updated prompt + latest product catalog from OpenSearch
            → Validates response (guardrails check, length check)
            → Writes to ElastiCache Redis with new cache key: {intent}:{query_hash}:{model_version}:{timestamp}
            → Logs regeneration event for audit

import redis
import hashlib
import json

class FeedbackAwareCacheManager:
    def __init__(self, redis_client):
        self.redis = redis_client

    def check_and_invalidate(self, response_hash: str):
        feedback_key = f"feedback:aggregate:{response_hash}"
        data = self.redis.hgetall(feedback_key)

        if not data:
            return False

        total = int(data.get(b'total', 0))
        thumbs_down = int(data.get(b'thumbs_down', 0))

        if total < 30:
            return False  # Insufficient sample

        ratio = thumbs_down / total

        if ratio > 0.50:
            # Hard invalidation
            cache_key = f"response:{response_hash}"
            self.redis.delete(cache_key)
            self.redis.publish('cache-invalidation', json.dumps({
                'response_hash': response_hash,
                'reason': 'feedback_threshold',
                'ratio': ratio,
                'total': total
            }))
            return True

        elif ratio > 0.35:
            # Soft invalidation — reduce TTL
            cache_key = f"response:{response_hash}"
            self.redis.expire(cache_key, 3600)  # 1 hour instead of 24h
            return False

        return False

Raw Feedback (DynamoDB/S3)
    → Stage 1: Filtering
    → Stage 2: Label Noise Reduction
    → Stage 3: Class Balancing
    → Stage 4: Feature Engineering
    → Stage 5: SageMaker Training Dataset (S3)

# Confidence-weighted labeling
def compute_label_confidence(feedback_event):
    signals = []

    # Signal 1: Comment alignment (if comment exists)
    if feedback_event.get('comment_category'):
        # If comment says "wrong_genre" and intent is "recommendation" → high confidence
        intent_comment_alignment = check_alignment(
            feedback_event['intent'], 
            feedback_event['comment_category']
        )
        signals.append(('comment_alignment', intent_comment_alignment, 0.4))

    # Signal 2: Behavioral corroboration
    # Thumbs-up on recommendation + user clicked product link = corroborated positive
    if feedback_event['intent'] == 'recommendation' and feedback_event.get('product_click'):
        signals.append(('behavioral', 1.0, 0.3))

    # Signal 3: Consistency — same user, same intent, consistent feedback pattern
    historical_consistency = get_user_intent_consistency(
        feedback_event['user_id_hash'], 
        feedback_event['intent']
    )
    signals.append(('consistency', historical_consistency, 0.3))

    confidence = sum(score * weight for _, score, weight in signals) / sum(w for _, _, w in signals)
    return confidence

50M interactions/month × 6% feedback rate = 3M feedback events/month
3M / 30 days / 24 hours / 3600 seconds = ~1.16 events/second (average)
Peak (Prime Day): 10× average = ~12 events/second
Burst (flash sale start): 50× average = ~58 events/second

Stream: mangaassist-feedback-stream
Shards: 2 (each shard handles 1,000 writes/sec, 1 MB/sec)
  → Sufficient for 58 events/sec peak with 97% headroom
  → Auto-scaling policy: add shard if utilization > 70% for 5 minutes
Retention: 24 hours (default, sufficient for consumer catch-up)

Table: MangaAssist_ThumbsFeedback
Capacity mode: ON-DEMAND
  → Justification: Traffic is bursty (Prime Day, flash sales). Provisioned mode
    requires over-provisioning for peaks or complex auto-scaling. On-demand
    absorbs 10× burst within seconds.
Average item size: ~500 bytes
3M items/month × 500 bytes = 1.5 GB/month
TTL: 90 days → max table size ~4.5 GB (well within DynamoDB limits)

GSIs:
  - intent-timestamp-index: PK=intent, SK=timestamp (read-heavy, ~100 RCU)
  - feedback-timestamp-index: PK=feedback, SK=timestamp (read-heavy, ~50 RCU)

Destination: s3://mangaassist-feedback/raw/year=YYYY/month=MM/day=DD/
Format: Parquet (columnar, compressed)
Buffer: 5 MB or 300 seconds (whichever first)
Compression: Snappy (~70% compression ratio)
Monthly data: 3M × 500 bytes × 0.3 (compressed) = ~450 MB/month
Annual: ~5.4 GB → negligible S3 cost ($0.12/year)

-- Athena table definition
CREATE EXTERNAL TABLE mangaassist_feedback (
    session_id STRING,
    timestamp STRING,
    turn_id INT,
    feedback STRING,
    intent STRING,
    response_id STRING,
    user_id_hash STRING,
    device_type STRING,
    latency_ms INT,
    model_version STRING,
    comment_raw STRING,
    comment_category STRING
)
PARTITIONED BY (year INT, month INT, day INT)
STORED AS PARQUET
LOCATION 's3://mangaassist-feedback/raw/'

import hashlib

def assign_variant(user_id: str, experiment_id: str = "feedback_widget_v1") -> str:
    hash_input = f"{user_id}:{experiment_id}"
    hash_value = int(hashlib.sha256(hash_input.encode()).hexdigest(), 16)
    bucket = hash_value % 100
    if bucket < 50:
        return "control"  # No widget shown
    else:
        return "treatment"  # Widget shown

┌─────────────────────────────────────────────────────────────────────┐
│                     FEEDBACK FLYWHEEL                                │
│                                                                      │
│  ┌──────────┐    ┌──────────────┐    ┌─────────────┐               │
│  │ 1. COLLECT│───→│ 2. AGGREGATE │───→│ 3. OPTIMIZE │               │
│  │           │    │ & DETECT     │    │   PROMPTS   │               │
│  │ Thumbs    │    │              │    │             │               │
│  │ Widget    │    │ Kinesis +    │    │ Bedrock     │               │
│  │ → Kinesis │    │ Lambda       │    │ Prompt Mgr  │               │
│  └──────────┘    │ Anomaly Det. │    └──────┬──────┘               │
│                  └──────────────┘           │                       │
│                                             ▼                       │
│  ┌──────────┐    ┌──────────────┐    ┌─────────────┐               │
│  │ 5. CLOSE │←───│ 4. A/B       │←───│ Deploy      │               │
│  │   LOOP   │    │  DEPLOY      │    │ Candidate   │               │
│  │           │    │              │    │ Prompts     │               │
│  │ Measure   │    │ 10% traffic  │    │             │               │
│  │ via same  │    │ to new       │    │ ECS Fargate │               │
│  │ thumbs    │    │ prompt       │    │ Config Mgr  │               │
│  └──────────┘    └──────────────┘    └─────────────┘               │
└─────────────────────────────────────────────────────────────────────┘

# Lambda consumer on Kinesis — aggregates every 15 minutes
def aggregate_feedback(intent: str, window_minutes: int = 15):
    """Compute rolling thumbs-up rate and detect anomalies."""
    recent = query_dynamodb_gsi(
        index='intent-timestamp-index',
        pk=intent,
        sk_range=(now - timedelta(minutes=window_minutes), now)
    )

    total = len(recent)
    thumbs_up = sum(1 for r in recent if r['feedback'] == 'thumbs_up')
    rate = thumbs_up / total if total > 0 else None

    # Anomaly detection: compare to 7-day rolling average
    historical_rate = get_historical_rate(intent, days=7)
    z_score = (rate - historical_rate['mean']) / historical_rate['std']

    if z_score < -2.5:  # Significant drop
        trigger_optimization(intent, rate, recent_samples=recent[:50])

    return {'intent': intent, 'rate': rate, 'z_score': z_score}

def optimize_prompt(intent: str, current_rate: float, negative_samples: list):
    """Generate candidate prompt improvements using Bedrock Claude."""

    # Retrieve current prompt template
    current_prompt = get_prompt_template(intent)

    # Analyze negative feedback patterns
    analysis_prompt = f"""
    Analyze these {len(negative_samples)} negative feedback responses for the '{intent}' intent 
    in a manga chatbot. Identify the top 3 patterns causing user dissatisfaction:

    {format_samples(negative_samples)}

    Current prompt template:
    {current_prompt}

    Suggest a revised prompt template that addresses these patterns.
    Return JSON: {{"analysis": "...", "revised_prompt": "...", "expected_improvement": "..."}}
    """

    response = bedrock_client.invoke_model(
        modelId='anthropic.claude-3-5-sonnet-20241022-v2:0',
        body=json.dumps({'prompt': analysis_prompt, 'max_tokens': 2000})
    )

    candidate = json.loads(response['body'].read())

    # Store candidate in DynamoDB prompt registry (not yet deployed)
    store_candidate_prompt(intent, candidate['revised_prompt'], 
                          metadata={'trigger_rate': current_rate, 'analysis': candidate['analysis']})

    return candidate

# ECS Fargate prompt router
def get_prompt_for_request(intent: str, user_id: str) -> str:
    experiment = get_active_experiment(intent)

    if experiment and is_in_treatment_group(user_id, experiment['id']):
        return experiment['candidate_prompt']  # 10% of traffic
    else:
        return get_production_prompt(intent)   # 90% of traffic

┌─────────────────────────────────────────────────────────┐
│  MangaAssist User Satisfaction Dashboard                │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  [USS Score: 66.2%]  ▲ +1.3% vs last week              │
│  ═══════════════════════════════════                    │
│                                                         │
│  ┌─── Intent Breakdown (Treemap) ────────────────┐     │
│  │ ┌──────────┐ ┌────────┐ ┌──────┐ ┌────┐      │     │
│  │ │ recom.   │ │checkout│ │prod_q│ │faq │      │     │
│  │ │ 72% ▲    │ │ 55% ▼  │ │ 65%  │ │58% │      │     │
│  │ └──────────┘ └────────┘ └──────┘ └────┘      │     │
│  └───────────────────────────────────────────────┘     │
│                                                         │
│  ┌─── USS Time Series (30 days) ─────────────────┐     │
│  │  68%│    ╱╲                                    │     │
│  │  66%│───╱──╲──────╱╲──────                    │     │
│  │  64%│  ╱    ╲    ╱  ╲                         │     │
│  │  62%│─╱──────╲──╱────╲───── ← Prime Day dip  │     │
│  │     └────────────────────── time →            │     │
│  └───────────────────────────────────────────────┘     │
│                                                         │
│  ┌─── Alerts ────────────────────────────────────┐     │
│  │ ⚠ checkout_help down 8% (flash sale impact)   │     │
│  │ ✅ recommendation up 2% (new prompt deployed)  │     │
│  └───────────────────────────────────────────────┘     │
└─────────────────────────────────────────────────────────┘

DynamoDB → S3 (Firehose, hourly Parquet) → Athena (daily aggregation view) → QuickSight SPICE dataset (refreshed every 4 hours)