AWS AI Automation · Project C

Production RAG platform

Q: Why does the ask handler call Bedrock's retrieve_and_generate as one operation instead of retrieving and then generating separately?

retrieve_and_generate does both RAG steps in a single managed Bedrock call — it retrieves matching chunks from the Knowledge Base and has Claude generate the grounded answer (with a Guardrail) — reducing glue code and round-trips versus wiring retrieval and generation by hand.

The capstone: a production RAG platform — KB + OpenSearch + Lambda API on CDK, with CI eval gates and a monitoring/cost dashboard. Everything from W4, W6, W7, W13, W14 in one system.

⏱️ ~6 hours🏗️ Capstone project🎯 Expert

⚙️ To run this for realThe code here is complete and correct as written. To actually execute it you'll need:

AWS credentials, Node.js (CDK needs it), + pip install aws-cdk-lib constructs
AWS credentials (aws configure) + Bedrock model access enabled in your region + pip install boto3

Reading and learning works without any of this — run when you're ready. IDs/ARNs in examples are placeholders; swap in your own.

What you'll buildA production RAG platform: a Knowledge Base over OpenSearch Serverless, a Lambda + API Gateway query API, all provisioned with CDK, plus CI, monitoring, and a cost dashboard. The capstone — everything from W4, W6, W7, W13, and W14 in one system.

Learning objectives

Provision a KB + OpenSearch + query API entirely with CDK.
Expose retrieval-augmented answering behind an authenticated API.
Add CI that runs a retrieval-quality eval on every change.
Monitor latency, cost, and answer quality in one dashboard.

▶ Runnable companionEvery code block in this lesson is also a standalone file under code/proj-aws-rag-platform/ in the course, with a README on how to run it. Read here, run there — no need to copy-paste.

Architecture advanced

Example code for learning — review, test, and adapt it before running against real or production systems. Commands can create, change, or delete resources. See the Terms & Disclaimer.

Architecture

Client ─→ API Gateway ─→ Lambda: /ask
                                └─ Bedrock retrieve_and_generate  (KB + Claude + Guardrail)
   KB ⇄ OpenSearch Serverless (vectors)   ← ingestion from S3 data source
   CI: on push → deploy CDK → run retrieval-quality eval → gate
   Ops: CloudWatch dashboard (latency · cost · recall@k)

▶ How this works

This little map is the whole system on one screen — read it top to bottom before any code. A RAG platform (Retrieval-Augmented Generation) answers questions by first finding the right documents, then asking a model to write an answer from those documents. Each arrow (─→) means "hands the request to".

Client → API Gateway → Lambda: /ask — a user (or app) sends a question to a public web address. API Gateway is AWS's front door for APIs; it forwards the request to a Lambda (a small function that runs on demand, with no server for you to manage). The /ask is the URL path people call.
Bedrock retrieve_and_generate — the Lambda calls one Bedrock operation that does both RAG steps at once: it retrieves matching text from the Knowledge Base (KB) and asks Claude to generate the answer, with a Guardrail filtering unsafe input/output. Bedrock is AWS's managed home for foundation models like Claude.
KB ⇄ OpenSearch Serverless — the KB stores your documents as vectors (lists of numbers that capture meaning) inside OpenSearch, a search engine. The double arrow means the KB both writes vectors in and reads them back out. New documents flow in via ingestion from an S3 bucket (S3 = AWS file storage).
CI and Ops lines — the bottom two rows are not request-time. CI (automation that runs on every code push) redeploys and checks answer quality, and Ops is a CloudWatch dashboard watching speed (latency), spend (cost), and accuracy (recall@k). The rest of the lesson builds these pieces one at a time.

Try this: Trace one question with your finger: start at Client, follow the arrows to Bedrock, into the KB and OpenSearch, and back. That round trip is exactly what the code in Steps 1 and 2 wires up.

Step 1 — the query API (CDK) advanced

Step 1

rag_api_stack.pyfrom aws_cdk import Stack, aws_lambda as _lambda, aws_apigateway as apigw
from constructs import Construct

class RagApiStack(Stack):
    def __init__(self, scope: Construct, cid: str, kb_id: str, **kw):
        super().__init__(scope, cid, **kw)
        fn = _lambda.Function(
            self, "AskFn",
            runtime=_lambda.Runtime.PYTHON_3_12,
            handler="ask.handler",
            code=_lambda.Code.from_asset("lambda"),
            environment={"KB_ID": kb_id},
        )
        api = apigw.LambdaRestApi(self, "RagApi", handler=fn)    # authn via API keys / IAM
        self.url = api.url

▶ How this works

This is CDK code — you describe the cloud infrastructure you want in Python, and CDK creates it on AWS for you. Instead of clicking around the AWS console, you write it down once so it's repeatable. This file defines a Stack: a group of AWS resources deployed together. Here the stack is a Lambda function plus an API in front of it.

class RagApiStack(Stack): defines your stack by inheriting from CDK's Stack. The __init__ is the constructor; kb_id is the ID of the Knowledge Base to talk to, passed in so this stack isn't hard-wired to one KB.
_lambda.Function(...) declares the Lambda. runtime=...PYTHON_3_12 says run it on Python 3.12; handler="ask.handler" means "call the handler function in ask.py" (that's Step 2's file); code=...from_asset("lambda") ships the local lambda/ folder as the function's code.
environment={"KB_ID": kb_id} passes the KB id into the running function as an environment variable — that's how the handler in Step 2 finds the right KB without hard-coding it.
apigw.LambdaRestApi(self, "RagApi", handler=fn) puts an API Gateway in front of the Lambda, turning it into a real HTTPS endpoint. The comment notes auth is via API keys or IAM. self.url = api.url saves the public URL so other code can use it.

What the output means: Running this doesn't print text — it provisions cloud resources. After cdk deploy, AWS has a live Lambda and an API URL you can send questions to.

Try this: Notice nothing here says how to build a server — you only declared what you want. That's the point of CDK: describe the goal, let AWS assemble the pieces.

Step 2 — the ask handler expert

Step 2

ask.pyimport boto3, json, os
rt = boto3.client("bedrock-agent-runtime")

def handler(event, context):
    q = json.loads(event["body"])["question"]
    resp = rt.retrieve_and_generate(
        input={"text": q},
        retrieveAndGenerateConfiguration={
            "type": "KNOWLEDGE_BASE",
            "knowledgeBaseConfiguration": {
                "knowledgeBaseId": os.environ["KB_ID"],
                "modelArn": "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-5-sonnet-20241022-v2:0",
            },
        },
    )
    return {"statusCode": 200, "body": json.dumps({
        "answer": resp["output"]["text"],
        "sources": [r["location"] for c in resp["citations"] for r in c["retrievedReferences"]],
    })}

▶ How this works

This is the code that actually runs each time someone asks a question. AWS calls handler(event, context) automatically when a request hits the API. event carries the incoming request (including the question); the function returns the answer as JSON.

rt = boto3.client("bedrock-agent-runtime") — boto3 is the AWS SDK for Python. This line opens a client to the Bedrock service that does retrieval + generation.
q = json.loads(event["body"])["question"] reads the request body (a JSON string), turns it into a Python dict, and pulls out the "question" field.
rt.retrieve_and_generate(...) is the one call that does RAG end to end: input={"text": q} is the question; the config says type is KNOWLEDGE_BASE, points at your KB via os.environ["KB_ID"] (the env var set in Step 1), and names the Claude model to write the answer via its modelArn (an ARN is AWS's unique ID for a resource).
The return hands back statusCode 200 (HTTP for "OK") and a JSON body with two fields: answer (the text Claude generated) and sources — the documents it used. The [r["location"] for c in ... for r in ...] is a nested list comprehension that flattens every citation's references into one list of source locations, so the caller can see where the answer came from.

What the output means: A JSON reply like {"answer": "...", "sources": [ ... ]}. The sources list is what makes this trustworthy RAG: answers come with receipts.

Try this: Follow q through the function: it enters as the question, becomes input.text, and Bedrock returns text you read at resp["output"]["text"]. Everything else is plumbing around that one round trip.

Step 3 — CI eval gate expert

Every change redeploys the CDK stack and runs a retrieval-quality eval (recall@k on a golden set, the metric from Ch 5). Below threshold, the pipeline fails — an eval-gated rollout (O3) for RAG.

Step 3

ci_eval.py# ci_eval.py — run in the pipeline after deploy; exit non-zero to block the release.
import sys, urllib.request, json

GOLDEN = [("How do I rotate a key?", "kms-rotation-doc"),
          ("What is the SLA?",        "sla-doc")]

def recall_at_k(api_url, k=5):
    hits = 0
    for q, expected_source in GOLDEN:
        body = json.dumps({"question": q}).encode()
        resp = json.load(urllib.request.urlopen(api_url + "ask", data=body))
        if any(expected_source in s for s in resp["sources"][:k]):
            hits += 1
    return hits / len(GOLDEN)

score = recall_at_k("https://api.example.com/")
print("recall@5:", score)
if score < 0.8:
    sys.exit("BUILD FAILED: retrieval quality below threshold")

▶ How this works

This is the quality gate: an automated test that runs after every deploy and blocks the release if answers get worse. It measures recall@k — of the questions we know the right source for, how often does the true source show up in the top k results the API returns? A build should only ship if retrieval is still finding the right documents.

GOLDEN is a hand-picked list of (question, expected_source) pairs — the "golden set" of questions whose correct answer document we already know. It's the answer key the test grades against.
recall_at_k(api_url, k=5) loops over each golden question, sends it to the live /ask endpoint with urllib.request.urlopen, and reads the JSON reply.
if any(expected_source in s for s in resp["sources"][:k]) checks the top k sources the API returned: if the expected document is among them, that question counts as a hit. hits / len(GOLDEN) is the fraction correct — the score.
if score < 0.8: sys.exit("BUILD FAILED: ...") is the gate. Exiting with a non-zero code tells the CI pipeline the step failed, so the bad version never goes live. Above 0.8 (80%), the pipeline continues.

What the output means: It prints e.g. recall@5: 1.0 (both golden questions found their source). Below 0.8 the process exits with an error and the release is stopped.

Try this: Raise the threshold to 0.95 or add a golden question you know the KB can't answer, and picture the build going red. That red is the safety net catching a quality regression before your users do.

The dashboardBuild one CloudWatch dashboard with three widgets: API/Bedrock latency, cost (token metrics × price), and the CI recall@k over time. That single view is what 'operating a RAG platform' actually looks like.

🪜 Practice ladder beginner → industry

Six graded exercises, easy to real-world. Try each before opening its solution.

Exercise 1 · Milestone 1 — the query API stack in CDK (needs AWS creds)Beginner

Context: Every managed-RAG platform starts with a front door: an HTTP endpoint that answers questions from a Bedrock Knowledge Base. Standing it up as infrastructure-as-code is what makes the whole stack reproducible instead of clicked together in the console.

Your task: Define the entry-point milestone in CDK — a Lambda fronted by API Gateway — with the Knowledge Base id passed in as configuration rather than hard-coded.

Requirements:

A Lambda function declared with a Python runtime and an ask handler
API Gateway placed in front of the Lambda so it is reachable over HTTP
The Knowledge Base id supplied as an environment variable, not a literal
The whole query API expressed in typed CDK constructs, reproducible from source
Label the milestone as needing AWS credentials to actually deploy

💡 Hint: Let the KB id flow through the stack constructor into the Lambda's environment map so the same code deploys against any KB.

Show solution

The CDK stack — needs AWS creds + aws-cdk-lib (documented constructs):

from aws_cdk import Stack, aws_lambda as _lambda, aws_apigateway as apigw
from constructs import Construct

class RagApiStack(Stack):
    def __init__(self, scope: Construct, cid: str, kb_id: str, **kw):
        super().__init__(scope, cid, **kw)
        fn = _lambda.Function(
            self, "AskFn",
            runtime=_lambda.Runtime.PYTHON_3_12,
            handler="ask.handler",
            code=_lambda.Code.from_asset("lambda"),
            environment={"KB_ID": kb_id},           # KB id in as config
        )
        api = apigw.LambdaRestApi(self, "RagApi", handler=fn)
        self.url = api.url

CDK declares the Lambda and fronts it with API Gateway in a few typed constructs, and the Knowledge Base id flows in as an environment variable rather than being hard-coded. Infrastructure-as-code means the whole query API is reproducible from source, not clicked together in the console.

Exercise 2 · Milestone 2 — the ask handler with retrieve_and_generate (needs AWS creds)Intermediate

Context: The point of a Knowledge Base is that one managed call does retrieval and grounded generation together — you never hand-wire the vector search. But an answer without its sources can't be trusted or evaluated.

Your task: Build the ask handler that calls Bedrock retrieve_and_generate against the KB and returns both the answer text and the source citations.

Requirements:

Call bedrock-agent-runtime retrieve_and_generate with a KNOWLEDGE_BASE configuration
Pass the KB id (from config) and the model ARN in the request
Return the generated answer and a list of source citation locations
Read the question out of the request body; return a well-formed HTTP response
Label the milestone as needing AWS creds + a live Knowledge Base

💡 Hint: The citation locations are the unit the recall gate scores next — surface them now so downstream milestones have something to check.

Show solution

The ask handler — needs AWS creds + a Bedrock Knowledge Base (documented boto3):

import boto3, json, os
rt = boto3.client("bedrock-agent-runtime")

def handler(event, context):
    q = json.loads(event["body"])["question"]
    resp = rt.retrieve_and_generate(
        input={"text": q},
        retrieveAndGenerateConfiguration={
            "type": "KNOWLEDGE_BASE",
            "knowledgeBaseConfiguration": {
                "knowledgeBaseId": os.environ["KB_ID"],
                "modelArn": ("arn:aws:bedrock:us-east-1::foundation-model/"
                             "anthropic.claude-3-5-sonnet-20241022-v2:0"),
            },
        },
    )
    return {"statusCode": 200, "body": json.dumps({
        "answer": resp["output"]["text"],
        "sources": [r["location"]
                    for c in resp["citations"]
                    for r in c["retrievedReferences"]],
    })}

retrieve_and_generate does retrieval + grounded generation in one managed call against the KB, so you don't hand-wire the vector search. Returning the citation locations alongside the answer is what makes the response verifiable — and what the eval gate scores next.

Exercise 3 · Milestone 3 — flatten nested citations to sourcesAdvanced

Context: Bedrock returns citations nested: each citation object holds several retrievedReferences. To score retrieval you first need one flat, ordered list of the sources that actually backed the answer.

Your task: Write a pure function that flattens the nested citation structure into a flat source list, and prove it on a sample response modelled offline.

Requirements:

Flatten citations → retrievedReferences into one ordered list
Preserve source order as it appears in the response
Work against a hand-built sample that mirrors a real retrieve_and_generate shape — no live call needed
The output is exactly the list the recall@k gate consumes

💡 Hint: A double list-comprehension over citations then references is the whole job; keep it stdlib so it runs offline in tests.

Show solution

Flatten the nested citation structure (pure stdlib, runnable):

def sources_from_response(resp):
    return [ref["location"]
            for c in resp["citations"]
            for ref in c["retrievedReferences"]]

# shape mirrors a real retrieve_and_generate response
resp = {
    "output": {"text": "Rotate keys in the KMS console."},
    "citations": [
        {"retrievedReferences": [
            {"location": "kms-rotation-doc"},
            {"location": "kms-overview-doc"}]},
        {"retrievedReferences": [
            {"location": "security-faq"}]},
    ],
}
print(sources_from_response(resp))
# ['kms-rotation-doc', 'kms-overview-doc', 'security-faq']

The double comprehension flattens citations -> references into one ordered source list. That flat list is the unit the recall@k gate checks — whether the document that should have answered the question actually appears among the retrieved sources.

Exercise 4 · Milestone 4 — the CI recall@k eval gateExpert

Context: You can't improve retrieval you don't measure. The honest RAG question is: for each query, did the document that should have answered it appear in the top-k sources? Wiring that as a build check turns a silent retrieval regression into a failed deploy.

Your task: Implement a recall@k gate over a golden set of (question, expected-source) pairs that fails the build below a threshold, computed offline against a stubbed endpoint.

Requirements:

A golden set of (question, expected_source_id) pairs
recall@k = fraction of questions whose expected source is in the top-k
Stub the deployed /ask endpoint so the gate runs with no live call
Fail the build (non-zero exit in CI) when recall falls below the bar (e.g. 0.8)
Report the score so the failure is legible in CI logs

💡 Hint: Slice the ranked sources to [:k] and test membership of the expected id; the threshold is a policy dial, not a magic number.

Show solution

The recall@k CI gate (pure stdlib, runnable — API stubbed):

import sys
GOLDEN = [("How do I rotate a key?", "kms-rotation-doc"),
          ("What is the SLA?",        "sla-doc")]

def ask_sources(question):
    # STUB for the deployed /ask endpoint (returns ranked source ids)
    return {"How do I rotate a key?": ["kms-rotation-doc", "kms-overview"],
            "What is the SLA?":        ["pricing", "support"]}[question]  # misses sla-doc

def recall_at_k(golden, k=5):
    hits = sum(1 for q, expected in golden if expected in ask_sources(q)[:k])
    return hits / len(golden)

score = recall_at_k(GOLDEN)
print(f"recall@5: {score:.0%}")
if score < 0.8:
    print("BUILD FAILED: retrieval quality below threshold")   # sys.exit in real CI

recall@k asks the honest RAG question: did the document that should answer each query appear in the top-k sources? Wiring it as a CI gate that fails below 0.8 means a change that quietly breaks retrieval for "How do I reset my password?" is blocked before it ships — retrieval quality becomes a build check, not a user complaint.

Exercise 5 · Milestone 5 — track cost and latency per queryProfessional

Context: A managed RAG stack still bills per token and still has a latency tail. Trending cost-per-query and p95 latency is the operational discipline that catches a prompt or retrieval change that quietly doubled the bill.

Your task: Model per-query cost from Bedrock input/output tokens and summarise cost + latency over a batch of sampled queries so a CloudWatch dashboard can trend them.

Requirements:

A query_cost from input and output token counts at set per-token prices
A summary over samples: average cost, total cost, and p95 latency
Compute p95 from the sorted latency list, not a mean
Runs offline over a list of sampled (in_tok, out_tok, latency_ms) tuples
Frame the output as what a dashboard would trend over time

💡 Hint: Separate the per-request cost function from the aggregation so each is testable; p95 is an index into the sorted latencies.

Show solution

Per-query cost + latency summary (pure stdlib, runnable):

IN_PRICE, OUT_PRICE = 3.00/1_000_000, 15.00/1_000_000   # $ per token (illustrative)

def query_cost(in_tok, out_tok):
    return in_tok * IN_PRICE + out_tok * OUT_PRICE

def summarize(samples):                    # samples: (in_tok, out_tok, latency_ms)
    costs = [query_cost(i, o) for i, o, _ in samples]
    lat = sorted(l for _, _, l in samples)
    p95 = lat[max(0, (95 * len(lat)) // 100 - 1)]
    return {"avg_cost_usd": round(sum(costs)/len(costs), 6),
            "total_usd": round(sum(costs), 4),
            "p95_latency_ms": p95}

samples = [(1200, 300, 820), (2000, 400, 1500), (900, 250, 640)]
print(summarize(samples))

A managed RAG stack still bills per token and has a latency tail, so you trend average cost-per-query and p95 latency on a CloudWatch dashboard. Watching cost-per-query is the operational discipline that catches a prompt or retrieval change that quietly doubled the token bill.

Exercise 6 · Milestone 6 — secure and gate the public API as ownerIndustry scenario

Context: As platform owner, the production risk in an unauthenticated, unbounded RAG endpoint is cost abuse and data scraping. The edge must enforce auth and per-caller rate limits, and the deploy itself must clear a composite gate.

Your task: Model the request-admission gate — API-key/IAM auth plus per-caller rate limiting — and reason about the full deploy gate (auth + recall@k + cost).

Requirements:

Reject requests with a missing or invalid API key (403)
Enforce a per-caller request rate limit, rejecting over-limit callers (429)
Admit only valid, under-limit callers (200)
Model the limiter with per-caller counters, runnable offline
Name the three-part deploy gate: auth configured, recall@k above bar, cost-per-query within budget

💡 Hint: A dict of per-key counters is enough to model the rate limiter; the real thing lives in the API Gateway usage plan.

Show solution

The request-admission gate: auth + rate limit (pure stdlib model of the API-GW policy):

RATE_LIMIT = 60          # requests per minute per caller
VALID_KEYS = {"key-abc", "key-def"}

def admit(request, counters):
    key = request.get("api_key")
    if key not in VALID_KEYS:
        return {"status": 403, "reason": "missing/invalid API key"}
    counters[key] = counters.get(key, 0) + 1
    if counters[key] > RATE_LIMIT:
        return {"status": 429, "reason": "rate limit exceeded"}
    return {"status": 200, "reason": "admitted"}

counters = {}
print(admit({"api_key": "key-abc"}, counters))     # 200
print(admit({"api_key": "nope"}, counters))         # 403

Industry scenario: a support KB of 1,000 docs exposed as a query API. Left open it invites cost abuse and data scraping, so API Gateway enforces API-key/IAM auth and per-caller rate limits at the edge. The full deploy gate is three checks — auth configured, recall@k above threshold, and cost-per-query within budget — before the platform serves real users.

✓ Checkpoint — you can move on when you can…

Provision a KB + query API with CDK end to end.
Serve cited RAG answers behind an authenticated API.
Gate every release on a retrieval-quality eval in CI.
Monitor latency, cost, and recall@k in one dashboard.

📋 Staff-level self-scoring — is this RAG platform production-grade?

Dimension	Meets the bar	Above the bar (staff)
Retrieval quality	Answers cite their sources; recall@k on a golden set clears the threshold you set (e.g. 0.8).	Retrieval quality is tracked over time, chunking/embedding choices are justified by measured recall, and low-confidence answers cite fewer/weaker sources honestly.
Scalability	The KB over OpenSearch Serverless + Lambda API handles your expected query load; the API is authenticated.	Behaviour is understood at burst load (cold starts, OpenSearch limits, throttling), and ingestion of new documents does not degrade query latency.
CI eval gate	Every change redeploys the CDK stack and runs a retrieval-quality eval that blocks the release below threshold.	The gate runs offline before deploy where possible, the golden set is representative and version-controlled, and a regression is diagnosable from the CI output.
Cost dashboard	A CloudWatch dashboard shows latency, cost, and recall@k in one place.	Cost per query is attributed across Bedrock + OpenSearch, alerted on a budget, and you can name the dominant cost driver and how to cut it.
Monitoring & ops	You can see request latency and errors live and trace a failing `/ask` call.	Alerts tie to the SLO and the cost budget, and there is a rollback path (redeploy the last green stack) you have actually exercised.

Score each dimension 0 (missing), 1 (meets bar), or 2 (above bar). 0–4: a working prototype. 5–7: reviewable. 8–10: staff-level — retrieval is measured and gated, it scales, and it is priced and monitored. A 0 on CI eval gate means quality can regress silently — fix before shipping.

Knowledge check check yourself

✓ Knowledge check

Why does the ask handler call Bedrock's retrieve_and_generate as one operation instead of retrieving and then generating separately?

Show answer

retrieve_and_generate does both RAG steps in a single managed Bedrock call — it retrieves matching chunks from the Knowledge Base and has Claude generate the grounded answer (with a Guardrail) — reducing glue code and round-trips versus wiring retrieval and generation by hand.

✓ Knowledge check

Why does the CI pipeline run a retrieval-quality eval as a gate on every push rather than only monitoring quality in production?

Show answer

A push can silently degrade retrieval (bad chunking, embedding change, prompt edit). Gating on an eval catches quality regressions before deploy, so a change that drops recall@k fails CI instead of reaching users — monitoring alone would only detect it after harm.

Educational content, provided as-is and without warranty. Code samples are examples — review, test, and adapt them before using in production. See the Terms of Use & Disclaimer. Use at your own risk.

Learning objectives

Architecture advanced

Step 1 — the query API (CDK) advanced

Step 2 — the ask handler expert

Step 3 — CI eval gate expert

🪜 Practice ladder beginner → industry

✓ Checkpoint — you can move on when you can…

Knowledge check check yourself

More in Projects