Scenarios and Runbooks — Input Data Formatting

mindmap
  root((Input Data<br/>Formatting Failures))
    Bedrock Request Schema
      Missing anthropic_version
      ValidationException 400
      All Inference Blocked
    Conversation Role Labels
      human/ai vs user/assistant
      Context Ignored by Model
      Wrong Schema Silently Accepted
    SageMaker Payload Contract
      prompt vs inputs key mismatch
      400 After Model Swap
      No Backward Compatibility
    Multi-Turn Ordering
      Consecutive User Turns
      No Assistant Turn Between
      Refusal or Hallucination
    Product Data Serialization
      Python dict str() output
      Single-Quote Keys
      Model Misparses Key-Values

flowchart TD
    A([CloudWatch Alarm: 5xx Rate > 5%]) --> B{Error type?}
    B -->|ValidationException| C[Inspect raw Bedrock request body]
    B -->|Other| D[Investigate separately]
    C --> E{anthropic_version present?}
    E -->|No| F([Root Cause: Missing anthropic_version field])
    E -->|Yes| G{max_tokens present?}
    G -->|No| H([Root Cause: Missing max_tokens field])
    G -->|Yes| I{messages array well-formed?}
    I -->|No| J([Root Cause: Malformed messages array])
    I -->|Yes| K([Escalate — other schema violation])

import json
import boto3
from typing import Any

bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")

REQUIRED_BEDROCK_FIELDS = {"anthropic_version", "max_tokens", "messages"}
VALID_ANTHROPIC_VERSION = "bedrock-2023-05-31"


def build_claude_request(
    messages: list[dict],
    max_tokens: int = 1024,
    system: str | None = None,
    temperature: float = 0.7,
) -> dict[str, Any]:
    """
    Build a validated Bedrock Messages API request body for Claude.
    Raises ValueError early if required fields would be missing.
    """
    if not messages:
        raise ValueError("messages list must not be empty")

    body: dict[str, Any] = {
        "anthropic_version": VALID_ANTHROPIC_VERSION,  # REQUIRED — never omit
        "max_tokens": max_tokens,
        "messages": messages,
        "temperature": temperature,
    }

    if system:
        body["system"] = system

    _validate_bedrock_request(body)
    return body


def _validate_bedrock_request(body: dict[str, Any]) -> None:
    """Fail fast before sending to Bedrock — avoid paying for failed calls."""
    missing = REQUIRED_BEDROCK_FIELDS - body.keys()
    if missing:
        raise ValueError(
            f"Bedrock request body is missing required fields: {missing}. "
            f"anthropic_version must be '{VALID_ANTHROPIC_VERSION}'."
        )

    if body.get("anthropic_version") != VALID_ANTHROPIC_VERSION:
        raise ValueError(
            f"anthropic_version must be '{VALID_ANTHROPIC_VERSION}', "
            f"got: '{body.get('anthropic_version')}'"
        )

    if not isinstance(body.get("messages"), list) or len(body["messages"]) == 0:
        raise ValueError("messages must be a non-empty list")


def invoke_claude(
    model_id: str,
    messages: list[dict],
    system: str | None = None,
    max_tokens: int = 1024,
) -> str:
    """Invoke Bedrock Claude model with a validated request body."""
    request_body = build_claude_request(
        messages=messages,
        max_tokens=max_tokens,
        system=system,
    )

    response = bedrock.invoke_model(
        modelId=model_id,
        body=json.dumps(request_body),
        contentType="application/json",
        accept="application/json",
    )

    response_body = json.loads(response["body"].read())
    return response_body["content"][0]["text"]


# --- Hotfix: usage example ---
if __name__ == "__main__":
    msgs = [{"role": "user", "content": "Recommend a shonen manga under $15."}]
    answer = invoke_claude(
        model_id="anthropic.claude-3-sonnet-20240229-v1:0",
        messages=msgs,
        system="You are MangaAssist, a helpful manga recommendation assistant.",
    )
    print(answer)

flowchart TD
    A([User complaint: bot ignores follow-up questions]) --> B[Pull session history from DynamoDB]
    B --> C{Inspect role field values}
    C -->|human / ai labels| D([Root Cause: LangChain serialization — wrong role schema])
    C -->|user / assistant labels| E[Check message ordering]
    E -->|Consecutive user turns| F([Escalate to Scenario 4])
    E -->|Correct ordering| G[Check content field structure]
    G -->|Content is list not string| H([Root Cause: Structured content blocks not supported on this model])
    G -->|Content is string| I([Escalate — unknown context loss])

from __future__ import annotations
from dataclasses import dataclass
from typing import Literal

# Bedrock-compliant role literals
BedrockRole = Literal["user", "assistant"]

# LangChain / legacy role mappings
ROLE_NORMALIZATION_MAP: dict[str, BedrockRole] = {
    "human": "user",
    "user": "user",
    "ai": "assistant",
    "assistant": "assistant",
    "bot": "assistant",
    "system": "user",  # system messages handled separately via system param
}


@dataclass
class BedrockMessage:
    role: BedrockRole
    content: str

    def to_dict(self) -> dict[str, str]:
        return {"role": self.role, "content": self.content}


def normalize_conversation_history(
    raw_history: list[dict[str, str]],
) -> list[dict[str, str]]:
    """
    Convert conversation history from any known role schema to Bedrock-compliant
    user/assistant format.

    Raises ValueError for unrecognized role labels so failures are loud.
    """
    normalized: list[dict[str, str]] = []

    for turn in raw_history:
        raw_role = turn.get("role", "").lower().strip()
        content = turn.get("content", "").strip()

        if raw_role not in ROLE_NORMALIZATION_MAP:
            raise ValueError(
                f"Unrecognized conversation role '{raw_role}'. "
                f"Accepted values: {list(ROLE_NORMALIZATION_MAP.keys())}. "
                "Check your LangChain memory serializer or session store schema."
            )

        if not content:
            # Skip empty turns — they confuse the model
            continue

        bedrock_role = ROLE_NORMALIZATION_MAP[raw_role]
        normalized.append(BedrockMessage(role=bedrock_role, content=content).to_dict())

    return normalized


def load_and_normalize_session(session_id: str, dynamo_table) -> list[dict[str, str]]:
    """Load conversation history from DynamoDB and normalize roles for Bedrock."""
    response = dynamo_table.get_item(Key={"session_id": session_id})
    item = response.get("Item", {})
    raw_history: list[dict] = item.get("conversation_history", [])

    normalized = normalize_conversation_history(raw_history)
    return normalized


# --- Hotfix: migration script to fix corrupted sessions in DynamoDB ---
def migrate_legacy_roles_in_place(dynamo_table, session_id: str) -> None:
    """One-time fix: rewrite role labels for a session already stored with human/ai."""
    response = dynamo_table.get_item(Key={"session_id": session_id})
    item = response.get("Item", {})
    history = item.get("conversation_history", [])

    fixed_history = normalize_conversation_history(history)

    dynamo_table.update_item(
        Key={"session_id": session_id},
        UpdateExpression="SET conversation_history = :h",
        ExpressionAttributeValues={":h": fixed_history},
    )
    print(f"[MIGRATED] session {session_id}: {len(history)} turns rewritten.")

flowchart TD
    A([CloudWatch Alarm: SageMaker InvocationErrors > 0]) --> B[Check SageMaker endpoint logs in CloudWatch]
    B --> C{HTTP status code?}
    C -->|400| D[Inspect ModelError message]
    D --> E{Error contains payload key mismatch?}
    E -->|prompt key / unexpected field| F([Root Cause: Wrong payload schema for current container])
    E -->|inputs key missing| G([Root Cause: Old text-completion format sent to chat container])
    E -->|Other| H[Check content-type header]
    C -->|500| I[Check container resource limits — OOM or timeout]
    C -->|200| J([False alarm — check output parsing])

import json
import boto3
from enum import Enum
from typing import Any

sagemaker_runtime = boto3.client("sagemaker-runtime", region_name="us-east-1")


class SageMakerPayloadSchema(str, Enum):
    LEGACY_TEXT_GENERATION = "legacy_text_generation"  # {"inputs": "string"}
    TGI_CHAT = "tgi_chat"                              # {"inputs": [{"role":...}]}
    CUSTOM_PROMPT = "custom_prompt"                    # {"prompt": "...", "parameters": {}}


def build_sagemaker_payload(
    prompt: str | list[dict],
    schema: SageMakerPayloadSchema,
    parameters: dict[str, Any] | None = None,
) -> str:
    """
    Build a SageMaker invoke_endpoint payload according to the deployed container schema.
    Always pass the schema explicitly — never rely on a global default.
    """
    parameters = parameters or {}

    if schema == SageMakerPayloadSchema.LEGACY_TEXT_GENERATION:
        if not isinstance(prompt, str):
            raise ValueError(
                "LEGACY_TEXT_GENERATION schema expects a plain string prompt, "
                f"got {type(prompt).__name__}. Did you mean TGI_CHAT?"
            )
        payload = {"inputs": prompt, **parameters}

    elif schema == SageMakerPayloadSchema.TGI_CHAT:
        if isinstance(prompt, str):
            # Auto-convert plain string to chat message list
            messages = [{"role": "user", "content": prompt}]
        elif isinstance(prompt, list):
            messages = prompt
        else:
            raise ValueError(f"TGI_CHAT schema expects str or list, got {type(prompt).__name__}")
        payload = {"inputs": messages, **parameters}

    elif schema == SageMakerPayloadSchema.CUSTOM_PROMPT:
        if not isinstance(prompt, str):
            raise ValueError("CUSTOM_PROMPT schema expects a plain string prompt.")
        payload = {"prompt": prompt, "parameters": parameters}

    else:
        raise ValueError(f"Unknown SageMaker payload schema: {schema}")

    return json.dumps(payload)


def invoke_sagemaker_endpoint(
    endpoint_name: str,
    prompt: str | list[dict],
    schema: SageMakerPayloadSchema,
    parameters: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Invoke a SageMaker endpoint with schema-validated payload."""
    serialized_payload = build_sagemaker_payload(prompt, schema, parameters)

    response = sagemaker_runtime.invoke_endpoint(
        EndpointName=endpoint_name,
        ContentType="application/json",
        Accept="application/json",
        Body=serialized_payload,
    )

    response_body = json.loads(response["Body"].read())
    return response_body


# --- Usage after model swap ---
if __name__ == "__main__":
    result = invoke_sagemaker_endpoint(
        endpoint_name="mangaassist-classifier-tgi-v2",
        prompt="Classify this manga: Demon Slayer Vol 1",
        schema=SageMakerPayloadSchema.TGI_CHAT,  # Updated to match new container
        parameters={"max_new_tokens": 128, "temperature": 0.3},
    )
    print(result)

flowchart TD
    A([Quality alarm: Refusal rate > 1% in 5-min window]) --> B[Sample refused sessions from CloudWatch Logs]
    B --> C[Fetch conversation history for session_id]
    C --> D{Check role alternation pattern}
    D -->|user → user detected| E([Root Cause: Consecutive user turns — missing assistant interleave])
    D -->|Correct alternation| F[Check content of user messages]
    F -->|Injection attempt in content| G([Security: Prompt injection — escalate to security team])
    F -->|Normal content| H[Check model version and system prompt length]
    H -->|System prompt > 8K tokens| I([Root Cause: System prompt truncation squeezing out history])
    H -->|Normal| J([Escalate — unknown refusal cause])

import boto3
from boto3.dynamodb.conditions import Attr
from botocore.exceptions import ClientError

dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
sessions_table = dynamodb.Table("mangaassist-sessions")


def validate_role_alternation(history: list[dict[str, str]]) -> None:
    """
    Enforce strict user/assistant alternation required by Bedrock Messages API.
    Raises ValueError describing the first violation found.
    """
    for i in range(1, len(history)):
        prev_role = history[i - 1].get("role")
        curr_role = history[i].get("role")
        if prev_role == curr_role:
            raise ValueError(
                f"Invalid conversation history: consecutive '{curr_role}' turns at "
                f"positions {i - 1} and {i}. Bedrock Messages API requires strict "
                "user/assistant alternation. Check for race condition in session writer."
            )


def repair_consecutive_turns(history: list[dict[str, str]]) -> list[dict[str, str]]:
    """
    Repair a history with consecutive same-role turns by inserting placeholder
    assistant turns. Use only as a hotfix for existing corrupted sessions.
    """
    if not history:
        return history

    repaired: list[dict[str, str]] = [history[0]]

    for i in range(1, len(history)):
        prev_role = repaired[-1]["role"]
        curr_role = history[i]["role"]

        if prev_role == curr_role == "user":
            # Insert a neutral assistant placeholder to restore alternation
            repaired.append({
                "role": "assistant",
                "content": "[Continuing from your previous message]",
            })

        repaired.append(history[i])

    return repaired


def append_user_message_safe(
    session_id: str,
    user_message: str,
    expected_length: int,
) -> None:
    """
    Append a user message with a conditional write that enforces:
    1. The history length has not changed since we last read it (optimistic lock).
    2. Falls back to repair if a consecutive-turn violation is detected.
    """
    new_turn = {"role": "user", "content": user_message}

    try:
        sessions_table.update_item(
            Key={"session_id": session_id},
            UpdateExpression="SET conversation_history = list_append(conversation_history, :t)",
            # Optimistic lock: only write if history length matches what we read
            ConditionExpression=Attr("history_length").eq(expected_length),
            ExpressionAttributeValues={
                ":t": [new_turn],
            },
        )
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            # Another writer beat us — re-read and validate before retrying
            response = sessions_table.get_item(Key={"session_id": session_id})
            current_history = response["Item"].get("conversation_history", [])
            validate_role_alternation(current_history + [new_turn])
            # If validation passes, retry unconditionally
            sessions_table.update_item(
                Key={"session_id": session_id},
                UpdateExpression="SET conversation_history = list_append(conversation_history, :t)",
                ExpressionAttributeValues={":t": [new_turn]},
            )
        else:
            raise

flowchart TD
    A([Quality alert: product recommendation accuracy < 90%]) --> B[Sample failing recommendations from evaluation dataset]
    B --> C[Inspect raw prompt sent to Bedrock for failing cases]
    C --> D{Product data format in prompt?}
    D -->|Single-quoted Python dict format| E([Root Cause: str dict serialization — model misparses single-quote keys])
    D -->|JSON double-quoted format| F[Check if values contain special characters]
    F -->|Unescaped newlines or quotes in values| G([Root Cause: JSON escaping not applied to product field values])
    F -->|Clean values| H[Check prompt template — product data placement]
    H -->|Data after > 6K tokens| I([Root Cause: Product data truncated by context window])
    H -->|Normal position| J([Escalate — unknown parsing failure])

import json
from decimal import Decimal
from typing import Any


class DecimalEncoder(json.JSONEncoder):
    """Handle DynamoDB Decimal types which are not JSON-serializable by default."""

    def default(self, obj: Any) -> Any:
        if isinstance(obj, Decimal):
            # Preserve numeric precision — return float for LLM consumption
            return float(obj)
        return super().default(obj)


def serialize_product_for_prompt(product: dict[str, Any]) -> str:
    """
    Serialize a manga product dict to a valid JSON string suitable for embedding
    in an LLM prompt. Uses double-quoted keys and handles DynamoDB Decimal types.

    NEVER use str(product) — single quotes are not valid JSON and confuse models.
    """
    return json.dumps(product, cls=DecimalEncoder, ensure_ascii=False, indent=None)


def build_recommendation_prompt(
    user_query: str,
    products: list[dict[str, Any]],
    max_products: int = 5,
) -> str:
    """
    Build a product recommendation prompt with properly formatted product data.
    Limits to max_products to avoid context window bloat.
    """
    # Trim to most relevant products before serializing
    selected = products[:max_products]

    # Serialize each product as valid JSON — never use str()
    products_json_array = json.dumps(
        [
            {
                "title": p.get("title", ""),
                "volume": p.get("volume", ""),
                "price_usd": float(p["price"]) if "price" in p else None,
                "genre": p.get("genre", ""),
                "in_stock": p.get("in_stock", False),
            }
            for p in selected
        ],
        cls=DecimalEncoder,
        ensure_ascii=False,
    )

    prompt = (
        f"You are MangaAssist. A customer asked: \"{user_query}\"\n\n"
        f"Available products (JSON):\n{products_json_array}\n\n"
        "Recommend the most suitable product(s) from this list. "
        "Reference exact titles and prices from the JSON above."
    )

    return prompt


def assert_no_single_quote_keys(prompt: str) -> None:
    """
    Defensive check: scan the built prompt for Python-style single-quoted dict
    patterns and raise if found. Use in CI and staging canaries.
    """
    import re

    # Match patterns like {'key': value} — Python dict serialization artifact
    python_dict_pattern = re.compile(r"\{'\w+'\s*:")
    if python_dict_pattern.search(prompt):
        raise ValueError(
            "Prompt contains Python-style single-quoted dict keys. "
            "Use json.dumps() instead of str() for all product data serialization."
        )


# --- Usage example ---
if __name__ == "__main__":
    # Simulate DynamoDB product records (Decimal from DynamoDB scan)
    sample_products = [
        {"title": "Naruto", "volume": "1", "price": Decimal("9.99"), "genre": "shonen", "in_stock": True},
        {"title": "Attack on Titan", "volume": "1", "price": Decimal("12.99"), "genre": "dark_fantasy", "in_stock": True},
    ]

    prompt = build_recommendation_prompt(
        user_query="I want an action manga under $13",
        products=sample_products,
    )

    # Guard: will raise immediately if str() was used anywhere in the pipeline
    assert_no_single_quote_keys(prompt)

    print(prompt)