DEV Community

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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"
  }
}
Enter fullscreen mode Exit fullscreen mode

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
}
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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": "..."
}
Enter fullscreen mode Exit fullscreen mode

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."
}
Enter fullscreen mode Exit fullscreen mode

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"]
Enter fullscreen mode Exit fullscreen mode

Then execute the approved decision:

Bedrock → Recommendation → Validation → Policy → SQS destination
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode
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)
Enter fullscreen mode Exit fullscreen mode

This gives us two useful operating modes:

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

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
}
Enter fullscreen mode Exit fullscreen mode

The model might recommend:

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

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"]
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

The adaptive retry component adds failure context:

{
  "error_type": "downstream_timeout",
  "attempt": 2,
  "downstream_health": "degraded",
  "message_priority": "HIGH"
}
Enter fullscreen mode Exit fullscreen mode

A recommendation might be:

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

Or:

{
  "retry": false,
  "confidence": 0.96,
  "reason": "Failure appears permanent."
}
Enter fullscreen mode Exit fullscreen mode

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
    }
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode
Class Examples
Transient Timeout, temporary throttling, short-lived dependency failure
Permanent Invalid payload, unsupported operation, non-recoverable validation failure
Ambiguous Unknown dependency error, incomplete response, unclassified application failure

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)
Enter fullscreen mode Exit fullscreen mode

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
}
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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

Case Expected result
Known message Expected route
Unknown message Valid recommendation or fallback
Forbidden destination Policy rejection
Low confidence Deterministic route
Bedrock failure Deterministic route

Priority tests

Case Expected result
Critical context High-priority recommendation
Routine context Standard recommendation
Invalid priority Fallback
Low confidence Fallback

Retry tests

Case Expected result
Transient failure Bounded retry
Permanent failure Terminal handling
Attempt limit reached DLQ
Excessive model delay Clamp/reject
Model unavailable Deterministic retry policy

Reliability tests

Case Expected result
Duplicate delivery Idempotent outcome
Malformed model JSON Fallback
Timeout Fallback
DLQ redrive Expected behavior

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

to:

SQS → Consumer → Context → AI Recommendation → Policy → Messaging Action
Enter fullscreen mode Exit fullscreen mode

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.

Top comments (0)