Token Efficiency Scenarios and Runbooks

flowchart TD
    A[CloudWatch Alarm: InputTokens > 2x intent budget] --> B{Which intent?}
    B -->|recommendation| C[Check session turn count]
    C --> D{Turns > 10?}
    D -->|Yes| E[Multi-turn budget overrun]
    D -->|No| F[Single-turn prompt bloat — check RAG chunks]

    E --> G[Check compression ratio metric]
    G --> H{Compression achieved target?}
    H -->|Yes| I[Budget itself is too low for long conversations]
    H -->|No| J[Compression pipeline failure]

    J --> K[Check which compression stage failed]
    K --> L{History summarization effective?}
    L -->|No| M["Root cause: summarization not reducing enough"]
    L -->|Yes| N["Root cause: RAG context too large after pruning"]

    style E fill:#fff3e0
    style M fill:#ffcdd2
    style N fill:#ffcdd2
    style I fill:#fff9c4

# Emergency: increase recommendation budget for long sessions
def get_dynamic_budget(intent: str, turn_count: int) -> int:
    """Return a dynamic token budget based on conversation length.

    For recommendation intent, we allow budget to scale with turn count
    up to a hard ceiling. This prevents quality degradation in long
    browsing sessions while still capping maximum cost.
    """
    base_budgets = {
        "recommendation": 2000,
        "product_search": 1200,
        "order_status": 600,
        "manga_qa": 1500,
        "chitchat": 400,
    }
    base = base_budgets.get(intent, 1200)

    if intent == "recommendation" and turn_count > 8:
        # Scale: +200 tokens per 4 turns beyond 8, capped at 3,200
        extra_turns = turn_count - 8
        extra_budget = (extra_turns // 4) * 200
        return min(base + extra_budget, 3200)

    return base

PREFERENCE_EXTRACTION_PROMPT = """
Analyze this conversation and extract the user's manga preferences
into a structured profile. Be concise.

Conversation:
{history}

Output format:
- Preferred genres: [list]
- Liked series: [list with what they liked about each]
- Disliked elements: [list]
- Reading level: [beginner/intermediate/advanced]
- Language preference: [JP only / EN only / bilingual]
- Price sensitivity: [low/medium/high]
- Key request: [one sentence summary of what they want]
"""

flowchart TD
    A[Prevention Measures] --> B[Dynamic Budget Tiers]
    A --> C[Proactive Preference Extraction]
    A --> D[Session Length Guardrails]
    A --> E[Monitoring & Alerting]

    B --> B1["recommendation_short: 2,000 tokens (< 8 turns)"]
    B --> B2["recommendation_long: 2,800 tokens (8-15 turns)"]
    B --> B3["recommendation_extended: 3,200 tokens (15+ turns)"]

    C --> C1[At turn 8: extract preference profile with Haiku]
    C --> C2[Replace raw history with 200-token structured profile]
    C --> C3[Continue adding new turns to profile, not raw history]

    D --> D1[At turn 15: suggest the user check their cart]
    D --> D2[At turn 20: offer to email recommendations]
    D --> D3[Never hard-cut a session — always offer a graceful exit]

    E --> E1["CloudWatch alarm: SessionTurnCount > 12"]
    E --> E2["CloudWatch alarm: ModelDowngrade for recommendation"]
    E --> E3[Weekly review of long-session cost distribution]

    style B1 fill:#c8e6c9
    style C1 fill:#bbdefb
    style D1 fill:#fff9c4

flowchart TD
    A[Quality Monitoring Dashboard] --> B{Accuracy drop detected?}
    B -->|Yes| C[Segment by language]
    C --> D{JP queries affected more than EN?}
    D -->|Yes| E[Language-specific compression issue]
    D -->|No| F[General compression issue — check all components]

    E --> G[Check compression log: what tokens were removed?]
    G --> H{Japanese particles removed?}
    H -->|Yes| I["Root cause: perplexity model not calibrated for CJK"]
    H -->|No| J[Check entity preservation]
    J --> K{Product names corrupted?}
    K -->|Yes| L["Root cause: entity protection not applied to JP titles"]
    K -->|No| M[Investigate specific failing queries]

    style I fill:#ffcdd2
    style L fill:#ffcdd2

import re


def should_apply_llmlingua(text: str) -> bool:
    """Check if LLMLingua compression is safe for this text.

    Returns False if the text contains CJK characters, meaning we
    should use only rule-based compression.
    """
    # CJK Unified Ideographs + Hiragana + Katakana ranges
    cjk_pattern = re.compile(
        r'[\u4e00-\u9fff\u3040-\u309f\u30a0-\u30ff]'
    )
    cjk_char_count = len(cjk_pattern.findall(text))
    total_chars = len(text)

    if total_chars == 0:
        return True

    # If more than 10% of characters are CJK, skip LLMLingua
    cjk_ratio = cjk_char_count / total_chars
    return cjk_ratio < 0.10

def protect_entities(text: str, entities: list[str]) -> tuple[str, dict]:
    """Replace entities with placeholder tokens before compression.

    Returns the modified text and a mapping to restore entities after.
    """
    placeholder_map = {}
    protected_text = text

    for i, entity in enumerate(entities):
        placeholder = f"<<ENTITY_{i}>>"
        placeholder_map[placeholder] = entity
        protected_text = protected_text.replace(entity, placeholder)

    return protected_text, placeholder_map


def restore_entities(text: str, placeholder_map: dict) -> str:
    """Restore entities from placeholders after compression."""
    restored = text
    for placeholder, entity in placeholder_map.items():
        restored = restored.replace(placeholder, entity)
    return restored

flowchart TD
    A[Prevention Strategy] --> B[Language-Aware Compression Policy]
    A --> C[Entity Protection Layer]
    A --> D[Quality Gate Before Deployment]
    A --> E[Monitoring]

    B --> B1["English text: LLMLingua + rule-based (aggressive)"]
    B --> B2["Japanese text: rule-based only (conservative)"]
    B --> B3["Mixed text: segment by language, compress separately"]

    C --> C1[Extract entities before compression]
    C --> C2[Replace with placeholders]
    C --> C3[Compress placeholder-protected text]
    C --> C4[Restore entities after compression]

    D --> D1["Golden dataset: 200 JP queries, 200 EN queries"]
    D --> D2["Pre-deploy eval: compression must preserve >95% accuracy on both"]
    D --> D3["Canary deployment: 5% traffic for 24 hours before full rollout"]

    E --> E1["CloudWatch: AccuracyScore segmented by Language"]
    E --> E2["Automated alert: JP accuracy drops >5% below EN accuracy"]
    E --> E3["Weekly compression audit: sample 100 JP prompts, verify entity preservation"]

    style B2 fill:#c8e6c9
    style C1 fill:#bbdefb
    style D1 fill:#fff9c4

flowchart TD
    A[User Feedback: Answer incomplete] --> B[Check query vs response coverage]
    B --> C{All query entities addressed in response?}
    C -->|No| D[Identify missing entity]
    D --> E[Check RAG chunks returned by OpenSearch]
    E --> F{Was relevant chunk retrieved?}
    F -->|No| G["Root cause: OpenSearch retrieval failure (not a pruning issue)"]
    F -->|Yes| H[Was relevant chunk kept after pruning?]
    H -->|No| I["Root cause: ContextPruner removed relevant chunk"]
    H -->|Yes| J["Root cause: Model ignored available context"]

    I --> K[Check composite score of pruned chunk]
    K --> L{Score above threshold?}
    L -->|Yes| M["Token budget too small — chunk cut by budget cap"]
    L -->|No| N["Relevance scoring underweighted this chunk"]

    style I fill:#ffcdd2
    style M fill:#fff3e0
    style N fill:#ffcdd2

def detect_multi_faceted_query(query: str, intent: str) -> int:
    """Detect how many distinct information needs a query contains.

    Multi-faceted queries get a proportionally larger RAG budget.

    Returns the number of detected facets (minimum 1).
    """
    # Simple heuristic: count question-indicating patterns
    facet_indicators = [
        # Conjunctions joining distinct questions
        r'\band\b.*\?',
        r'\balso\b',
        r'\badditionally\b',
        # Multiple question words
        r'(does|is|can|will|how|what|when|where)',
        # Comma-separated aspects
        r',\s*(does|is|and)',
    ]

    import re
    facet_count = 1  # always at least 1
    query_lower = query.lower()

    for pattern in facet_indicators:
        matches = re.findall(pattern, query_lower)
        facet_count += len(matches)

    # Cap at 4 to prevent budget explosion
    return min(facet_count, 4)


def get_dynamic_rag_budget(intent: str, query: str,
                           base_budget: int = 500) -> int:
    """Scale RAG token budget based on query complexity."""
    facets = detect_multi_faceted_query(query, intent)
    # Each additional facet adds 200 tokens to the RAG budget
    return base_budget + (facets - 1) * 200

def verify_query_coverage(query: str, kept_chunks: list,
                          discarded_chunks: list,
                          estimator) -> list:
    """Verify that kept chunks cover all query aspects.

    If a query aspect is uncovered, pull in the best discarded chunk
    that addresses it.
    """
    import re
    # Extract noun phrases (simplified)
    query_phrases = set(re.findall(r'[A-Za-z]+(?:\s+[A-Za-z]+)*', query.lower()))

    # Check which phrases are covered by kept chunks
    kept_text = " ".join(c.text.lower() for c in kept_chunks)
    uncovered = [p for p in query_phrases
                 if p not in kept_text and len(p) > 3]

    if not uncovered:
        return kept_chunks

    # Find discarded chunks that cover the gap
    for phrase in uncovered:
        best_match = None
        best_score = 0
        for chunk in discarded_chunks:
            if phrase in chunk.text.lower():
                if chunk.vector_score > best_score:
                    best_match = chunk
                    best_score = chunk.vector_score

        if best_match:
            kept_chunks.append(best_match)
            discarded_chunks.remove(best_match)

    return kept_chunks

flowchart TD
    A[Prevention] --> B[Dynamic RAG Budget]
    A --> C[Expanded Entity Extraction]
    A --> D[Query Coverage Verification]
    A --> E[Monitoring]

    B --> B1[Detect query facets: 1-4 information needs]
    B --> B2["Budget = base + (facets - 1) x 200 tokens"]
    B --> B3[Cap at 1,100 tokens for 4-facet queries]

    C --> C1[Extract proper nouns: titles, authors, ISBNs]
    C --> C2[Extract descriptive attributes: color pages, hardcover, limited]
    C --> C3[Extract product categories: box set, collection, series]

    D --> D1[After pruning: check every query noun phrase]
    D --> D2[If uncovered: pull best discarded chunk]
    D --> D3[Log coverage gaps for model improvement]

    E --> E1["Alert: ChunksDiscardedAboveThreshold > 2"]
    E --> E2[Weekly audit: sample 50 multi-faceted queries]
    E --> E3["Track: user follow-up rate after product_search"]

    style B2 fill:#c8e6c9
    style D2 fill:#bbdefb

flowchart TD
    A[User Complaint: Incomplete pricing info] --> B[Check response truncation logs]
    B --> C{Was response truncated?}
    C -->|Yes| D[Check intent classification]
    D --> E{Correct intent assigned?}
    E -->|No| F["Root cause: intent misclassification → wrong max_tokens"]
    E -->|Yes| G[Check if natural response exceeds budget]
    G --> H{Natural response > max_tokens by > 50%?}
    H -->|Yes| I["Root cause: budget too low for this query type"]
    H -->|No| J["Root cause: model being verbose — prompt needs tightening"]

    F --> K[Fix: retrain intent classifier]
    I --> L[Fix: increase budget or add sub-intent]
    J --> M[Fix: add conciseness instruction to prompt]

    C -->|No| N[Different issue — check model output quality]

    style F fill:#ffcdd2
    style I fill:#fff3e0

# Add a "critical information" detection pass before truncation
CRITICAL_PATTERNS = {
    "pricing": [
        r'total[:\s]*[\d,]+\s*(JPY|USD|円|ドル)',
        r'合計[:\s]*[\d,]+',
        r'shipping[:\s]*[\d,]+',
        r'送料[:\s]*[\d,]+',
        r'tax[:\s]*[\d,]+',
    ],
    "delivery": [
        r'(arrive|delivery|届く|到着)[:\s]*(by|on|within)',
        r'\d{1,2}[/-]\d{1,2}',
        r'(business days|営業日)',
    ],
    "order_id": [
        r'ORD-\d{4}-\d{5}',
        r'注文番号[:\s]*\w+',
    ],
}


def contains_critical_info(text: str) -> bool:
    """Check if text contains critical information that must not be truncated."""
    import re
    for category, patterns in CRITICAL_PATTERNS.items():
        for pattern in patterns:
            if re.search(pattern, text, re.IGNORECASE):
                return True
    return False


async def stream_with_critical_protection(
    bedrock_stream,
    intent: str,
    config,
    encoding,
):
    """Stream with truncation protection for critical information.

    If the response is approaching the token budget and we detect
    critical information (prices, dates, order IDs), we extend the
    budget by 50% to ensure this information is delivered.
    """
    token_count = 0
    buffer_chunks = []
    in_critical_zone = False

    async for event in bedrock_stream:
        chunk_text = extract_text(event)
        if not chunk_text:
            continue

        chunk_tokens = len(encoding.encode(chunk_text))
        token_count += chunk_tokens

        # Check if we're in the critical zone (80-100% of budget)
        if token_count >= config.max_output_tokens * 0.8:
            if contains_critical_info(chunk_text):
                in_critical_zone = True

        # If in critical zone, extend budget by 50%
        effective_budget = (
            int(config.max_output_tokens * 1.5)
            if in_critical_zone
            else config.max_output_tokens
        )

        if token_count > effective_budget:
            yield chunk_text
            yield config.truncation_message
            return

        yield chunk_text

When answering purchase or pricing questions, lead with the total cost
and delivery date. Put details and disclaimers after the key numbers.

TRUNCATION_MESSAGES = {
    "purchase_inquiry": (
        "\n\n---\n*Want me to add this to your cart, "
        "or do you have more questions?*"
    ),
    "product_search": (
        "\n\n---\n*I can provide more details. "
        "Which aspect interests you most?*"
    ),
    "order_status": (
        "\n\n---\n*For full tracking details, "
        "check your order history page.*"
    ),
}

flowchart TD
    A[Prevention] --> B[Intent Classifier Improvement]
    A --> C[Critical Info Protection]
    A --> D[Prompt Structure Guidance]
    A --> E[Truncation Message Alignment]

    B --> B1["Add purchase_inquiry sub-intent"]
    B --> B2["Training data: 500 pre-purchase queries"]
    B --> B3["Confusion matrix monitoring: order_status vs purchase_inquiry"]

    C --> C1["Detect critical patterns: prices, dates, IDs"]
    C --> C2["Auto-extend budget 50% when critical info is in-flight"]
    C --> C3["Never truncate mid-price or mid-date"]

    D --> D1["System prompt: lead with key numbers"]
    D --> D2["Structured output format: summary first, details second"]
    D --> D3["Test with truncation simulation: verify key info in first 60% of response"]

    E --> E1["Each intent gets a context-appropriate truncation message"]
    E --> E2["Truncation message never references actions the user cannot take"]
    E --> E3["A/B test truncation messages for user satisfaction"]

    style C2 fill:#c8e6c9
    style D1 fill:#bbdefb

flowchart TD
    A["CloudWatch: EstimationDriftPercent > 10%"] --> B[Check drift direction]
    B --> C{Positive drift — over-estimating?}
    C -->|Yes| D["Safe direction but wasteful — budgets too conservative"]
    B --> E{Negative drift — under-estimating?}
    E -->|Yes| F["Dangerous — actual cost exceeds estimates"]

    F --> G[Check when drift started increasing]
    G --> H{Correlated with model update?}
    H -->|Yes| I["Root cause: tokenizer mismatch after model version change"]
    H -->|No| J[Check prompt template changes]
    J --> K{New prompt templates deployed?}
    K -->|Yes| L["Root cause: new templates have different token characteristics"]
    K -->|No| M[Check input data distribution shift]
    M --> N["Root cause: user behavior changed (longer queries, more JP text)"]

    style I fill:#ffcdd2
    style F fill:#fff3e0

# Emergency config change — no code deployment needed
# Update in DynamoDB config table or SSM Parameter Store

import boto3

ssm = boto3.client('ssm')
ssm.put_parameter(
    Name='/mangaassist/token-estimator/calibration-buffer',
    Value='1.20',
    Type='String',
    Overwrite=True,
)

import numpy as np
from collections import deque


class AdaptiveCalibrationBuffer:
    """Auto-calibrates the token estimation buffer using recent observations.

    Maintains a rolling window of (estimated, actual) token pairs and
    computes the optimal calibration buffer to minimize under-estimation
    while avoiding excessive over-estimation.

    MangaAssist design: the buffer is updated every 1,000 invocations
    (approximately every 1.4 minutes at 1M messages/day). This means
    the system self-corrects within minutes of a model update.
    """

    def __init__(self, window_size: int = 5000,
                 min_buffer: float = 1.02,
                 max_buffer: float = 1.30,
                 target_underestimate_rate: float = 0.05):
        """
        Args:
            window_size: Number of recent observations to consider
            min_buffer: Minimum calibration buffer (2% over)
            max_buffer: Maximum calibration buffer (30% over)
            target_underestimate_rate: Target rate of underestimates
                (we want <=5% of estimates to be below actual)
        """
        self.observations = deque(maxlen=window_size)
        self.min_buffer = min_buffer
        self.max_buffer = max_buffer
        self.target_rate = target_underestimate_rate
        self.current_buffer = 1.10  # start conservative

    def record(self, estimated_tokens: int, actual_tokens: int):
        """Record an (estimated, actual) observation pair."""
        if estimated_tokens > 0 and actual_tokens > 0:
            ratio = actual_tokens / estimated_tokens
            self.observations.append(ratio)

    def recalibrate(self) -> float:
        """Recompute the optimal calibration buffer.

        Uses the (1 - target_rate) percentile of actual/estimated ratios
        to set the buffer such that only target_rate of estimates fall
        below actual values.
        """
        if len(self.observations) < 100:
            return self.current_buffer  # not enough data

        ratios = np.array(self.observations)
        # We want 95th percentile of (actual/estimated) ratios
        # This means 95% of the time, estimated * buffer >= actual
        target_percentile = 1.0 - self.target_rate
        p95_ratio = np.percentile(ratios, target_percentile * 100)

        # The buffer should be at least p95_ratio
        new_buffer = np.clip(p95_ratio, self.min_buffer, self.max_buffer)

        # Smooth the update to avoid oscillation
        self.current_buffer = 0.7 * self.current_buffer + 0.3 * new_buffer
        return self.current_buffer

    def get_buffer(self) -> float:
        """Get the current calibration buffer."""
        return self.current_buffer

    def get_stats(self) -> dict:
        """Get drift statistics for monitoring."""
        if len(self.observations) < 10:
            return {"status": "insufficient_data"}

        ratios = np.array(self.observations)
        return {
            "current_buffer": round(self.current_buffer, 4),
            "mean_ratio": round(float(np.mean(ratios)), 4),
            "p50_ratio": round(float(np.median(ratios)), 4),
            "p95_ratio": round(float(np.percentile(ratios, 95)), 4),
            "p99_ratio": round(float(np.percentile(ratios, 99)), 4),
            "underestimate_rate": round(
                float(np.mean(ratios > self.current_buffer)), 4
            ),
            "observation_count": len(self.observations),
        }

flowchart TD
    A[Prevention] --> B[Auto-Calibration System]
    A --> C[Model Update Response]
    A --> D[Drift Monitoring]
    A --> E[Budgeting Safety]

    B --> B1["Rolling window of 5,000 (estimated, actual) pairs"]
    B --> B2["Recalibrate every 1,000 invocations"]
    B --> B3["Target: < 5% underestimation rate"]
    B --> B4["Buffer range: 1.02 to 1.30"]

    C --> C1["Subscribe to Bedrock model update notifications"]
    C --> C2["On update: auto-increase buffer to 1.25"]
    C --> C3["24-hour intensive recalibration window"]
    C --> C4["Alert on-call engineer for manual review"]

    D --> D1["CloudWatch alarm: EstimationDriftPercent > +/-10%"]
    D --> D2["Daily drift report to Slack"]
    D --> D3["Weekly drift trend analysis"]
    D --> D4["Monthly calibration buffer audit"]

    E --> E1["Budget gates use calibrated estimate, not raw count"]
    E --> E2["Session budgets include 15% safety margin"]
    E --> E3["Monthly cost forecast uses P95 estimate, not mean"]

    style B2 fill:#c8e6c9
    style C2 fill:#bbdefb
    style D1 fill:#fff9c4

flowchart TD
    A["Token Efficiency Alert Fired"] --> B{What type of alert?}

    B -->|Cost spike| C{Per-session or system-wide?}
    C -->|Per-session| D["Scenario 1: Budget exceeded in long conversation"]
    C -->|System-wide| E{Correlated with model update?}
    E -->|Yes| F["Scenario 5: Estimation drift"]
    E -->|No| G{Correlated with new deployment?}
    G -->|Yes| H["Check compression or pruning changes"]
    G -->|No| I["Traffic pattern change — investigate intent distribution"]

    B -->|Quality drop| J{Language-specific?}
    J -->|JP queries affected| K["Scenario 2: Compression degrading Japanese content"]
    J -->|All languages| L{Completeness or accuracy?}
    L -->|Completeness| M["Scenario 3: Context pruning too aggressive"]
    L -->|Accuracy| N["Check if RAG retrieval quality changed"]

    B -->|Truncation spike| O{Which intent?}
    O -->|order_status| P{Misclassified intent?}
    P -->|Yes| Q["Scenario 4: Truncation of critical info (misclassification)"]
    P -->|No| R["Budget too low for this query type"]
    O -->|recommendation| S["Scenario 1: Long conversation + truncation"]
    O -->|other| T["Review intent-specific max_tokens config"]

    B -->|Drift alarm| F

    style D fill:#fff3e0
    style K fill:#ffcdd2
    style M fill:#ffcdd2
    style Q fill:#ffcdd2
    style F fill:#fff9c4