Building an AI-Assisted Message Router with SQS, Lambda, and Amazon Bedrock

This is Part 3 of the AI-Assisted Message Queue Patterns series. Part 1 introduced the architecture. Part 2 covered guardrails, confidence gating, deterministic fallback, and failure handling. Now let's connect those ideas to the reference implementation.

Repository: github.com/kallayilsreedharansk/ai-message-queue-patterns

What we're building

The goal is not to replace Amazon SQS with an AI model. The goal is to insert a bounded decision layer into an ordinary message-processing pipeline.

Producer
   ↓
Ingest Queue
   ↓
Classifier / Router
   ↓
Priority Scorer
   ↓
┌───────────────┬───────────────┐
↓                               ↓
High Priority                Standard
↓                               ↓
└───────────────┬───────────────┘
                ↓
             Process
                ↓
             Failure?
             /      \
           No        Yes
           ↓          ↓
        Complete   Retry Queue
                      ↓
               Adaptive Retry
                   /       \
                Retry      DLQ

Amazon Bedrock can assist at three decision points — routing, priority, and retry — but every model-assisted decision remains bounded by application policy.

The AWS building blocks

The reference architecture uses:

• Amazon SQS for durable message buffering and processing paths
• AWS Lambda for event-driven consumers and decision handlers
• Amazon Bedrock for contextual recommendations
• AWS SAM for infrastructure definition and deployment
• Dead-letter queues for terminal failure handling

A simplified resource view:

                 Amazon Bedrock
                    ▲   ▲   ▲
                    │   │   │
Ingest SQS ──► Router ─► Priority Scorer
                           │
                 ┌─────────┴─────────┐
                 ▼                   ▼
          High-Priority SQS      Standard SQS
                 │                   │
                 └─────────┬─────────┘
                           ▼
                       Processing
                           │
                        Failure
                           ▼
                       Retry SQS
                           │
                           ▼
                    Retry Handler
                       /       \
                    Retry      DLQ

The important point is that Bedrock is invoked by the application. Bedrock does not poll SQS. For Lambda consumers, the SQS event source mapping handles polling and invokes the Lambda function.

Step 1: Ingest the message

A producer sends a normal application event to the ingest queue:

{
  "event_id": "evt-4821",
  "type": "service_interruption",
  "customer_tier": "critical",
  "region": "us-west-2",
  "payload": {
    "service": "notification-api",
    "severity": "high"
  }
}

The producer does not need to know which downstream queue will eventually process the message. That decision belongs downstream. This separation is useful because producers remain focused on publishing events rather than encoding increasingly complex routing policy.

Step 2: Build decision context

The router receives the event and constructs the context used for the decision, combining message attributes with operational signals:

{
  "message": {
    "type": "service_interruption",
    "customer_tier": "critical",
    "severity": "high"
  },
  "system": {
    "processor_a_health": "degraded",
    "processor_b_health": "healthy",
    "queue_backlog": 4200
  },
  "attempt": 1
}

This is where architecture matters. Don't send every available metric to the model just because you can — select context that can legitimately affect the decision. Useful signals might include message category, workload/SLA class, message age, retry history, queue depth, consumer health, downstream latency, congestion, and approved processing locations.

Policy constraints such as security or residency restrictions should preferably filter invalid destinations before model scoring.

Step 3: Invoke Bedrock

The Lambda function can invoke a supported model through Amazon Bedrock and request a constrained structured response:

def get_ai_recommendation(context):
    prompt = build_prompt(context)
    response = invoke_bedrock(prompt)
    return parse_response(response)

The prompt should define the action space explicitly. For routing:

You are assisting a message-routing application.

Allowed routes:
- PROCESSOR_A
- PROCESSOR_B

Return JSON only:
{
  "route": "...",
  "confidence": 0.0,
  "reason": "..."
}

The application is not asking the model to invent infrastructure. It is asking the model to recommend among known choices.

Step 4: Validate before routing

Assume Bedrock returns:

{
  "route": "PROCESSOR_B",
  "confidence": 0.92,
  "reason": "Processor A is degraded and Processor B is an approved healthy destination."
}

The next step is not send_message(decision["route"], message).

First validate:

ALLOWED_ROUTES = {"PROCESSOR_A", "PROCESSOR_B"}
MIN_CONFIDENCE = 0.7

def select_route(message, decision):
    if decision.get("route") not in ALLOWED_ROUTES:
        return deterministic_route(message)

    if decision.get("confidence", 0) < MIN_CONFIDENCE:
        return deterministic_route(message)

    return decision["route"]

Then execute the approved decision:

Bedrock → Recommendation → Validation → Policy → SQS destination

That boundary is the implementation of:

AI recommends. Policy decides. The messaging platform executes.

Step 5: Deterministic mode

The reference pattern supports running without Bedrock:

USE_BEDROCK=false

def decide(context):
    if not USE_BEDROCK:
        return deterministic_decision(context)

    try:
        decision = get_ai_recommendation(context)
        return validate_and_apply_policy(decision, context)
    except Exception:
        return deterministic_decision(context)

This gives us two useful operating modes:

AI-assisted mode:    Message → Context → Bedrock → Policy → Action
Deterministic mode:  Message → Rules → Action

Both ultimately use the same messaging infrastructure.

Step 6: Score priority

After routing/classification, the next component can evaluate urgency. Suppose the message context is:

{
  "type": "service_interruption",
  "severity": "high",
  "customer_tier": "critical",
  "message_age_seconds": 48,
  "queue_backlog": 4200
}

The model might recommend:

{
  "priority": "HIGH",
  "confidence": 0.94,
  "reason": "Critical workload and elevated service impact."
}

Policy validates the category:

ALLOWED_PRIORITIES = {"HIGH", "STANDARD"}

def select_priority(message, decision):
    if decision.get("priority") not in ALLOWED_PRIORITIES:
        return deterministic_priority(message)

    if decision.get("confidence", 0) < MIN_CONFIDENCE:
        return deterministic_priority(message)

    return decision["priority"]

Then HIGH routes to the High-Priority Queue and STANDARD routes to the Standard Queue. The queue selection represents the priority path — SQS visibility timeout is a separate processing/reliability setting and should not be confused with priority itself.

Step 7: Process the message

At this point the processing consumer works much like any other queue consumer:

Priority Queue → Consumer → Business Operation → Success / Failure

The AI layer does not need to participate in ordinary successful processing unless the workload has a reason for it. That's an important cost and latency optimization: don't invoke a model where deterministic processing already has enough information.

Step 8: Classify failure before retrying

Traditional retry behavior may look like:

Failure → Wait → Retry → Wait longer → Retry again

The adaptive retry component adds failure context:

{
  "error_type": "downstream_timeout",
  "attempt": 2,
  "downstream_health": "degraded",
  "message_priority": "HIGH"
}

A recommendation might be:

{
  "retry": true,
  "retry_delay_seconds": 180,
  "confidence": 0.88,
  "reason": "Failure appears transient but downstream service remains degraded."
}

Or:

{
  "retry": false,
  "confidence": 0.96,
  "reason": "Failure appears permanent."
}

Policy then applies hard boundaries:

MAX_ATTEMPTS = 5
MIN_RETRY_DELAY = 30
MAX_RETRY_DELAY = 900

def retry_policy(message, decision, attempt):
    if attempt >= MAX_ATTEMPTS:
        return {"action": "DLQ"}

    if decision.get("confidence", 0) < MIN_CONFIDENCE:
        return deterministic_retry(message, attempt)

    if not decision.get("retry", False):
        return {"action": "DLQ"}

    delay = decision.get("retry_delay_seconds", MIN_RETRY_DELAY)
    delay = max(MIN_RETRY_DELAY, min(delay, MAX_RETRY_DELAY))

    return {
        "action": "RETRY",
        "delay": delay
    }

The model can recommend. It cannot create attempt number six when policy allows only five.

Step 9: Handle failure classes deliberately

A useful baseline:

              Processing Failure
                     ↓
               Failure Class
          ┌──────────┼──────────┐
          ↓          ↓          ↓
      Transient   Permanent   Ambiguous
          ↓          ↓          ↓
       Retry        DLQ      Bounded Retry
                                ↓
                            Re-evaluate

AI assistance is most interesting in the ambiguous middle. Deterministic rules should continue handling obvious cases whenever possible.

Step 10: Preserve idempotency

SQS consumers must be designed for possible duplicate delivery — that remains true whether the route was selected by a lookup table or an AI-assisted decision. Use a stable producer event identifier and derive the idempotency key from it:

if already_processed(event_id):
    return

process(message)
record_success(event_id)

Do not generate a new logical identity merely because a message moved between processing paths.

Step 11: Add observability around decisions

Normal messaging metrics are not enough once an AI-assisted decision layer exists. Record both operational outcomes and decision outcomes:

{
  "event_id": "evt-4821",
  "decision_type": "routing",
  "decision_source": "bedrock",
  "selected_route": "PROCESSOR_B",
  "confidence": 0.92,
  "fallback_used": false,
  "policy_version": "v3",
  "decision_latency_ms": 214
}

Useful metrics: model invocation count, model invocation failures, decision latency, fallback rate, low-confidence rate, policy rejection rate, messages per destination, retry count, DLQ count, and end-to-end processing success.

The most useful operational question is: did AI-assisted decisions improve the outcome compared with the deterministic baseline? Without that comparison, adding a model is just adding complexity.

Step 12: Deploy incrementally

Because the architecture is built with separate components, adoption can happen one decision at a time.

Phase 1 — deterministic baseline. USE_BEDROCK=false. Validate the entire messaging path.

Phase 2 — shadow mode. Invoke the model but do not let the recommendation change production behavior.

Message ─┬──► Deterministic Decision ──► Production
         └──► AI Recommendation ───────► Logs / Evaluation

Compare results.

Phase 3 — gated AI assistance. Enable recommendations only for a limited category of messages. Keep confidence thresholds and fallback active.

Phase 4 — expand based on evidence. Only expand to additional routing, priority, or retry scenarios if measured results justify it.

This is much safer than enabling AI across the entire pipeline at once.

AWS SAM as the deployment boundary

The reference project uses AWS SAM so the infrastructure and functions can be versioned together. A simplified template shape looks conceptually like:

Resources:
  IngestQueue:
    Type: AWS::SQS::Queue

  RouterFunction:
    Type: AWS::Serverless::Function

  HighPriorityQueue:
    Type: AWS::SQS::Queue

  StandardQueue:
    Type: AWS::SQS::Queue

  RetryQueue:
    Type: AWS::SQS::Queue

  ProcessingDLQ:
    Type: AWS::SQS::Queue

The actual repository should remain the source of truth for resource names and deployment configuration: ai-message-queue-patterns

Testing the pipeline

A useful test matrix includes both model-assisted and deterministic behavior.

Routing tests

Priority tests

Retry tests

Reliability tests

The interesting test is not whether Bedrock returns an answer. It's whether the whole system remains predictable when Bedrock does not.

What this pattern is — and isn't

This pattern is useful when message-processing decisions genuinely depend on contextual signals that are difficult to express cleanly with static rules alone. It is not an argument for sending every SQS message through an LLM.

A good implementation may look like:

Simple decision       → Deterministic rule
Context-heavy decision → AI recommendation → Deterministic policy

The goal is not to maximize model usage. The goal is to improve selected decisions without surrendering control of the system.

Putting the series together

Across these three articles, the architecture evolved from:

SQS → Fixed Logic → Action

to:

SQS → Consumer → Context → AI Recommendation → Policy → Messaging Action

Part 1 asked where AI might help. Part 2 asked how to constrain it. Part 3 showed how the pieces fit together in an implementation.

The pattern I keep coming back to is intentionally simple:

AI recommends. Policy decides. The messaging platform executes.

That separation lets us experiment with more context-aware routing, prioritization, and retry behavior without discarding the deterministic reliability principles that made message queues useful in the first place.

Explore the implementation: 👉 AI-Assisted Message Queue Patterns on GitHub

The examples in this series are architectural patterns and reference implementations intended for experimentation and learning. Production systems should independently evaluate model behavior, security, latency, cost, failure modes, and workload-specific requirements.