Handling Errors Troubleshooting and Edge Cases Part 3

​

Detecting safety blocks (guardrails)

import logging

logger = logging.getLogger(__name__)

def handle_bedrock_response(response, prompt=None):
    """
    Handle a Bedrock response that may include a safety/block indicator.

    Args:
        response (dict): The response returned by Bedrock (InvokeModel or Converse).
        prompt (str, optional): The user prompt sent to Bedrock (used for logging).
    Returns:
        str: A safe message for the end user, or the model output.
    """
    # Example: check for a blocked / safety response
    if response.get("blocked", False):
        # Log for monitoring/troubleshooting — avoid logging raw PII in production!
        logger.warning("Guardrail triggered for prompt: %s", prompt or "<prompt omitted>")

        # Return a concise, non-specific message to the end user
        return "Sorry, I can't help with that request. Please try rephrasing."

    # Normal flow: return the model's output (adjust key to match your integration)
    return response.get("output")

• Return a concise, non-specific message to the user (avoid exposing guardrail internals).
• Log the event with enough context for diagnostics, but do not persist raw prompts or PII in plain text.
• Emit metrics so you can monitor guardrail frequency by route, user population, or department — this helps triage and policy tuning.

​

Addressing hallucinations and improving grounding

​

Key mitigation strategies

• Use Retrieval-Augmented Generation (RAG) to ground answers in your documents. When supplying retrieved chunks, instruct the model to rely primarily on those chunks rather than its pre-trained knowledge.
• Require explicit fallback behavior: ask the model to say “I don’t know” if it cannot verify the answer.
• Ask for source citations and display provenance and confidence where useful.
• Validate output formats (for example, ensure returned JSON is valid and conforms to your schema).
• Always sanitize and validate model outputs before use in downstream systems.

​

Example: validate JSON output against a schema (Python)

import json
from jsonschema import validate, ValidationError

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer"}
    },
    "required": ["name", "age"]
}

def validate_model_json(output_text: str):
    try:
        data = json.loads(output_text)
    except json.JSONDecodeError:
        raise ValueError("Model did not return valid JSON")

    try:
        validate(instance=data, schema=schema)
    except ValidationError as e:
        raise ValueError(f"JSON schema validation failed: {e.message}")

    return data

​

Observability, monitoring, and resilience

• Emit metrics: prompt size, response size, latency, success/error counts, guardrail triggers, retrieved chunk counts (RAG), and throttling/backoff events.
• Track retrieved chunk counts to understand context size vs. latency and token usage.
• Log structured events for guardrail triggers with anonymized/hashed context for triage.
• Implement retries with exponential backoff for transient errors; detect and surface throttling (rate limit) responses so clients can back off elegantly.

• Circuit breakers or client-side rate limiting for protecting downstream services.
• Timeouts and safe defaults for partial or malformed responses.
• Health checks and synthetic tests to validate end-to-end retrieval + generation pipelines.

​

Putting it together

​

Summary

• Detect and handle guardrail blocks gracefully.
• Validate and sanitize model outputs before they reach downstream systems.
• Emit metrics and logs (with privacy-preserving practices) for monitoring and triage.
• Use retries/backoff for transient failures and surface meaningful messages to users.

​

Links and references

• Amazon Bedrock documentation
• jsonschema (Python): https://pypi.org/project/jsonschema/
• Observability patterns for distributed systems: https://martinfowler.com/articles/monitoring.html