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
| 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)
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
| 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
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.
Top comments (0)