Build a Production-Grade RAG Pipeline
Build a full RAG pipeline on AWS with hybrid search and guardrails.
Introduction
30 Second Summary
AI chatbots are impressive until they confidently make things up. Real companies solve this by connecting AI to their actual documents so answers come from verified sources, not hallucinations.
In this project, you will build a production-grade RAG (Retrieval-Augmented Generation) pipeline on AWS that ingests documents, retrieves relevant chunks using hybrid search and reranking, and generates safe answers with guardrails and fallback logic.
What You'll Build
A complete RAG system that processes documents through semantic chunking, stores them in Amazon OpenSearch Serverless for hybrid search, and generates answers using Claude via Amazon Bedrock with production-level safety controls.
By the end of this project, you'll have:
- A document ingestion pipeline that cleans, semantically chunks, and embeds documents using Titan Embeddings V2.
- A hybrid retrieval system combining BM25 keyword search with k-NN vector similarity, plus cross-encoder reranking for precision.
- A guarded generation layer with Bedrock Guardrails, retry loops with exponential backoff, and fallback logic for resilient responses.
- Secret Mission: Add a relevance evaluation loop that automatically scores retrieval quality and adjusts parameters.
Are there any prerequisites?
You'll need an AWS account with Bedrock model access enabled, Python installed (3.11+), and basic familiarity with Python and AWS concepts.
This project uses paid AWS services (OpenSearch Serverless and Bedrock API calls). Expect to spend approximately $2-5 if you complete the project and delete resources promptly.
Before We Start
Most RAG tutorials stop at a basic vector similarity search and call it a day. Production systems need more. They need intelligent chunking that respects document structure, hybrid retrieval that catches both exact keywords and semantic meaning, reranking for precision, and guardrails to keep generated answers safe.
In this step, you will set up the AWS services and Python environment needed to build your full pipeline. This includes the AWS CLI, Amazon Bedrock model access, and all Python dependencies.
In this step, get ready to:
- Install and configure the AWS CLI with credentials.
- Enable Amazon Bedrock model access for Titan Text Embeddings V2 and Claude 3 Haiku.
- Set up a Python project with all required dependencies and configuration.
Install and configure the AWS CLI
The AWS CLI lets you interact with AWS services from your terminal. You will use it to authenticate all the API calls your pipeline makes to Bedrock and OpenSearch Serverless.
- Check if the AWS CLI is already installed by running this command:
aws --version✔️ I see a version number
Your AWS CLI is ready. Move on to configuring your credentials below.
ⓧ Command not found
You need to install the AWS CLI. Follow the official installation guide for your operating system:
After installing, close and reopen your terminal, then run aws --version again to confirm it works.
- Configure the AWS CLI with your credentials by running this command:
aws configureWhat does this command do?
The aws configure command prompts you for four values: your AWS Access Key ID, Secret Access Key, default region, and output format.
For this project, set the default region to us-east-1 and the output format to json. These credentials come from your IAM user in the AWS console.
- Enter your AWS Access Key ID when prompted.
- Enter your AWS Secret Access Key when prompted.
- Enter us-east-1 for the default region.
- Enter json for the default output format.
Don't have an AWS Access Key?
You can create one in the AWS console under IAM > Users > your user > Security credentials > Create access key. Choose the CLI use case when prompted.
Help me set up AWS CLI credentials for my project.
Enable Amazon Bedrock model access
Amazon Bedrock provides access to foundation models on-demand. You need to confirm that two specific models are enabled in your account: Titan Text Embeddings V2 for generating vector embeddings, and Claude 3 Haiku for text generation.
- Open the AWS Management Console and navigate to the Amazon Bedrock service.
- Make sure you are in the US East (N. Virginia) region (us-east-1). You can check and change your region using the dropdown at the top right of the console.
- In the left navigation pane, under Bedrock configurations, choose Model access.
- Choose Enable specific models.
- Select Titan Text Embeddings V2 from the Amazon models.
- Select Claude 3 Haiku from the Anthropic models.
- Choose Next, then Submit.
Why these two models?
Titan Text Embeddings V2 (model ID: amazon.titan-embed-text-v2:0) converts text into 1024-dimensional vectors for semantic search. It costs $0.00002 per 1,000 input tokens.
Claude 3 Haiku (model ID: anthropic.claude-3-haiku-20240307-v1:0) generates natural language answers from retrieved context. It costs $0.00025 per 1,000 input tokens and $0.00125 per 1,000 output tokens. For this project's volume, total API costs will stay under a few dollars.
Anthropic models may need use case details
If this is your first time enabling an Anthropic model, you may be asked to submit use case details. Fill out the form describing your project. Access is typically granted within a few minutes.
Not seeing the models?
Confirm you are in the us-east-1 region. Model availability varies by region, and this project requires us-east-1.
Help me enable Bedrock model access for Titan Embeddings V2 and Claude 3 Haiku.
Set up the Python project and install dependencies
Your pipeline uses four Python libraries that handle AWS communication, OpenSearch indexing, request signing, and cross-encoder reranking. You will create a project folder, install the dependencies, and set up a configuration file.
- Check your Python version by running this command:
python3 --version✔️ I see version 3.11 or higher
Your Python version is ready. Continue to creating your project folder below.
ⓧ I see an older version
This project requires Python 3.11 or higher. Some systems (especially macOS) ship with an older default Python.
- Download and install Python 3.11+ from the official site: python.org/downloads
- After installing, run python3 --version again. If you still see the old version, try python3.11 --version or python3.12 --version to use the specific binary.
Use a version manager
If your system Python is outdated, tools like pyenv let you install and switch between Python versions easily. See github.com/pyenv/pyenv for setup instructions.
ⓧ Command not found
Python is not installed on your system. Install Python 3.11+ from the official site:
- Download from python.org/downloads
- Run the installer and check the box to add Python to your PATH.
- Close and reopen your terminal, then run python3 --version to confirm.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir rag-pipeline
cd rag-pipeline- Create a requirements.txt file with the project dependencies by running this command:
cat > requirements.txt << 'EOF'
boto3>=1.35.0
opensearch-py>=2.8.0
requests-aws4auth>=1.3.1
sentence-transformers>=3.4.0
EOFWhat are these dependencies?
- boto3 is the AWS SDK for Python. It handles all communication with Bedrock and OpenSearch Serverless.
- opensearch-py is the Python client for OpenSearch. You will use it to create indexes and run search queries.
- requests-aws4auth signs HTTP requests with AWS credentials so OpenSearch Serverless accepts them.
- sentence-transformers provides the cross-encoder model for reranking search results locally.
- Install the dependencies by running this command:
pip install -r requirements.txtWhat should I see?
The install will download several packages. The sentence-transformers package is the largest as it includes PyTorch. This may take a few minutes depending on your internet speed.
Install failing?
If you see permission errors, try pip install --user -r requirements.txt. If pip is not found, try pip3 instead.
Help me fix pip install errors for my RAG pipeline dependencies.
- Create the configuration file by running this command:
cat > config.py << 'EOF'
import os
AWS_REGION = os.environ.get("AWS_REGION", "us-east-1")
COLLECTION_NAME = "rag-pipeline"
INDEX_NAME = "documents"
EMBEDDING_MODEL_ID = "amazon.titan-embed-text-v2:0"
GENERATION_MODEL_ID = "anthropic.claude-3-haiku-20240307-v1:0"
EMBEDDING_DIMENSION = 1024
CHUNK_SIZE = 512
CHUNK_OVERLAP = 50
HYBRID_SEMANTIC_WEIGHT = 0.7
HYBRID_LEXICAL_WEIGHT = 0.3
RERANK_MODEL = "cross-encoder/ms-marco-MiniLM-L6-v2"
RERANK_TOP_K = 20
RERANK_FINAL_K = 3
MAX_RETRIES = 3
RETRY_BASE_DELAY = 1.0
EOFWhat does this configuration do?
This file centralizes every tunable parameter your pipeline uses. Instead of scattering magic numbers across multiple files, all modules import from config.py.
- EMBEDDING_DIMENSION is 1024, the default output size for Titan Text Embeddings V2.
- HYBRID_SEMANTIC_WEIGHT and HYBRID_LEXICAL_WEIGHT control the balance between vector similarity and keyword matching in hybrid search.
- RERANK_TOP_K (20) is how many candidates you fetch before reranking. RERANK_FINAL_K (3) is how many you keep after the cross-encoder scores them.
- Confirm the files exist by running:
lsYou should see config.py and requirements.txt in the output.
✔️ Awesome, I've got everything!
Great. Make sure both files are saved in your rag-pipeline folder.
ⓧ I'd like to double check the full code
Here is config.py in full:
import os
AWS_REGION = os.environ.get("AWS_REGION", "us-east-1")
COLLECTION_NAME = "rag-pipeline"
INDEX_NAME = "documents"
EMBEDDING_MODEL_ID = "amazon.titan-embed-text-v2:0"
GENERATION_MODEL_ID = "anthropic.claude-3-haiku-20240307-v1:0"
EMBEDDING_DIMENSION = 1024
CHUNK_SIZE = 512
CHUNK_OVERLAP = 50
HYBRID_SEMANTIC_WEIGHT = 0.7
HYBRID_LEXICAL_WEIGHT = 0.3
RERANK_MODEL = "cross-encoder/ms-marco-MiniLM-L6-v2"
RERANK_TOP_K = 20
RERANK_FINAL_K = 3
MAX_RETRIES = 3
RETRY_BASE_DELAY = 1.0And here is requirements.txt:
boto3>=1.35.0
opensearch-py>=2.8.0
requests-aws4auth>=1.3.1
sentence-transformers>=3.4.0Your environment is locked in. Next up, you will create the OpenSearch Serverless collection that stores your document embeddings.
Create the OpenSearch Serverless Collection
Your environment is set up with all the Python dependencies and configuration constants. Now you need somewhere to store your document embeddings and run searches against them.
In this step, you will create an Amazon OpenSearch Serverless collection. This is a managed vector database that handles scaling automatically. You will write the code that provisions security policies, creates the collection, and sets up an index optimized for hybrid search.
In this step, get ready to:
- Create encryption, network, and data access policies for OpenSearch Serverless.
- Create a VECTORSEARCH collection and wait for it to become ACTIVE.
- Create an index with text and vector fields for hybrid search.
Set up the module and define security policies
Before you can create a collection, OpenSearch Serverless requires three security policies: an encryption policy, a network policy, and a data access policy. You will build a Python module that handles all of this.
- Create a new file called opensearch_setup.py in your project directory.
- Add the imports, module-level client, and authentication helper by pasting this code:
import json
import time
import boto3
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
from config import AWS_REGION, COLLECTION_NAME, INDEX_NAME, EMBEDDING_DIMENSION
aoss_client = boto3.client("opensearchserverless", region_name=AWS_REGION)
def get_aws_auth():
"""Get AWS4Auth for OpenSearch Serverless."""
credentials = boto3.Session().get_credentials()
return AWS4Auth(
credentials.access_key,
credentials.secret_key,
AWS_REGION,
"aoss",
session_token=credentials.token
)What does this code do?
The imports bring in the libraries you installed earlier: boto3 for AWS API calls, opensearchpy for interacting with the OpenSearch index, and AWS4Auth for signing requests to the serverless endpoint.
The aoss_client is a module-level boto3 client for the OpenSearch Serverless control plane. The get_aws_auth() function creates a signed authentication object using aoss as the service name, which is required for OpenSearch Serverless data-plane requests.
- Below the get_aws_auth() function, add the create_policies() function:
def create_policies(collection_name: str, account_id: str):
"""Create encryption, network, and data access policies."""
encryption_policy = {
"Rules": [{"ResourceType": "collection", "Resource": [f"collection/{collection_name}"]}],
"AWSOwnedKey": True
}
aoss_client.create_security_policy(
name=f"{collection_name}-enc",
type="encryption",
policy=json.dumps(encryption_policy)
)
network_policy = [
{
"Rules": [{"ResourceType": "collection", "Resource": [f"collection/{collection_name}"]},
{"ResourceType": "dashboard", "Resource": [f"collection/{collection_name}"]}],
"AllowFromPublic": True
}
]
aoss_client.create_security_policy(
name=f"{collection_name}-net",
type="network",
policy=json.dumps(network_policy)
)
data_access_policy = [
{
"Rules": [
{"ResourceType": "index", "Resource": [f"index/{collection_name}/*"],
"Permission": ["aoss:CreateIndex", "aoss:UpdateIndex", "aoss:DescribeIndex",
"aoss:ReadDocument", "aoss:WriteDocument"]},
{"ResourceType": "collection", "Resource": [f"collection/{collection_name}"],
"Permission": ["aoss:CreateCollectionItems", "aoss:UpdateCollectionItems",
"aoss:DescribeCollectionItems"]}
],
"Principal": [f"arn:aws:iam::{account_id}:root"]
}
]
aoss_client.create_access_policy(
name=f"{collection_name}-access",
type="data",
policy=json.dumps(data_access_policy)
)What do these three policies do?
The encryption policy tells OpenSearch Serverless to encrypt your collection at rest using an AWS-owned key. This is the simplest option and has no additional cost.
The network policy allows public access to the collection endpoint. In a production environment, you would restrict this to specific VPCs or IP ranges.
The data access policy grants your AWS account permission to create indexes, read documents, and write documents. Without this policy, your authenticated requests would be rejected even though you own the collection.
- Save opensearch_setup.py.
Seeing an import error?
If your editor shows a red underline on opensearchpy or requests_aws4auth, make sure your virtual environment is activated and that you ran pip install -r requirements.txt in the previous step.
help me fix import errors in opensearch_setup.py
Add the collection and index creation functions
With the policies defined, you can now add the functions that actually create the collection and configure its index. The collection creation includes a polling loop because it takes several minutes for the collection to become active.
- Below the create_policies() function, add the create_collection() function:
def create_collection():
"""Create OpenSearch Serverless vector search collection."""
response = aoss_client.create_collection(
name=COLLECTION_NAME,
type="VECTORSEARCH"
)
collection_id = response["createCollectionDetail"]["id"]
print(f"Creating collection {COLLECTION_NAME} (ID: {collection_id})...")
while True:
status = aoss_client.batch_get_collection(names=[COLLECTION_NAME])
detail = status["collectionDetails"][0]
if detail["status"] == "ACTIVE":
print(f"Collection active at: {detail['collectionEndpoint']}")
return detail["collectionEndpoint"]
print(" Waiting for collection to become active...")
time.sleep(30)What does this function do?
This function calls the OpenSearch Serverless API to create a VECTORSEARCH collection. After the API call, the collection starts in a CREATING state. The while loop polls every 30 seconds until the status changes to ACTIVE.
Once active, the function returns the collection endpoint URL. This endpoint is what you will use to send search and indexing requests to your collection.
- Below create_collection(), add the create_index() function:
def create_index(host: str):
"""Create index with text and knn_vector fields for hybrid search."""
client = OpenSearch(
hosts=[{"host": host.replace("https://", ""), "port": 443}],
http_auth=get_aws_auth(),
use_ssl=True,
verify_certs=True,
connection_class=RequestsHttpConnection,
timeout=300
)
time.sleep(60) # Wait for data access policies to propagate
index_body = {
"settings": {
"index": {
"knn": True
}
},
"mappings": {
"properties": {
"text": {"type": "text"},
"doc_id": {"type": "keyword"},
"chunk_id": {"type": "keyword"},
"embedding": {
"type": "knn_vector",
"dimension": EMBEDDING_DIMENSION,
"method": {
"name": "hnsw",
"space_type": "cosinesimil",
"engine": "faiss",
"parameters": {
"ef_construction": 256,
"m": 48
}
}
}
}
}
}
response = client.indices.create(index=INDEX_NAME, body=index_body)
print(f"Index created: {response}")
return clientWhat does the index mapping do?
The index has four fields. The text field stores the chunk content for BM25 keyword search. The doc_id and chunk_id fields are keyword fields for exact-match filtering.
The embedding field is a knn_vector with 1024 dimensions (matching Titan Embeddings V2). It uses the HNSW algorithm with cosine similarity via the Faiss engine. The ef_construction and m parameters control the quality of the approximate nearest neighbor graph. Higher values give more accurate search at the cost of more indexing time.
Why the 60-second sleep?
Data access policies can take up to a minute to propagate after creation. Without this wait, the create_index call would fail with an authorization error even though the policy exists.
- Save opensearch_setup.py.
✔️ Awesome, I've got everything!
Great. Double check you have saved your file before moving on.
ⓧ I'd like to double check the full code
import json
import time
import boto3
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
from config import AWS_REGION, COLLECTION_NAME, INDEX_NAME, EMBEDDING_DIMENSION
aoss_client = boto3.client("opensearchserverless", region_name=AWS_REGION)
def get_aws_auth():
"""Get AWS4Auth for OpenSearch Serverless."""
credentials = boto3.Session().get_credentials()
return AWS4Auth(
credentials.access_key,
credentials.secret_key,
AWS_REGION,
"aoss",
session_token=credentials.token
)
def create_policies(collection_name: str, account_id: str):
"""Create encryption, network, and data access policies."""
encryption_policy = {
"Rules": [{"ResourceType": "collection", "Resource": [f"collection/{collection_name}"]}],
"AWSOwnedKey": True
}
aoss_client.create_security_policy(
name=f"{collection_name}-enc",
type="encryption",
policy=json.dumps(encryption_policy)
)
network_policy = [
{
"Rules": [{"ResourceType": "collection", "Resource": [f"collection/{collection_name}"]},
{"ResourceType": "dashboard", "Resource": [f"collection/{collection_name}"]}],
"AllowFromPublic": True
}
]
aoss_client.create_security_policy(
name=f"{collection_name}-net",
type="network",
policy=json.dumps(network_policy)
)
data_access_policy = [
{
"Rules": [
{"ResourceType": "index", "Resource": [f"index/{collection_name}/*"],
"Permission": ["aoss:CreateIndex", "aoss:UpdateIndex", "aoss:DescribeIndex",
"aoss:ReadDocument", "aoss:WriteDocument"]},
{"ResourceType": "collection", "Resource": [f"collection/{collection_name}"],
"Permission": ["aoss:CreateCollectionItems", "aoss:UpdateCollectionItems",
"aoss:DescribeCollectionItems"]}
],
"Principal": [f"arn:aws:iam::{account_id}:root"]
}
]
aoss_client.create_access_policy(
name=f"{collection_name}-access",
type="data",
policy=json.dumps(data_access_policy)
)
def create_collection():
"""Create OpenSearch Serverless vector search collection."""
response = aoss_client.create_collection(
name=COLLECTION_NAME,
type="VECTORSEARCH"
)
collection_id = response["createCollectionDetail"]["id"]
print(f"Creating collection {COLLECTION_NAME} (ID: {collection_id})...")
while True:
status = aoss_client.batch_get_collection(names=[COLLECTION_NAME])
detail = status["collectionDetails"][0]
if detail["status"] == "ACTIVE":
print(f"Collection active at: {detail['collectionEndpoint']}")
return detail["collectionEndpoint"]
print(" Waiting for collection to become active...")
time.sleep(30)
def create_index(host: str):
"""Create index with text and knn_vector fields for hybrid search."""
client = OpenSearch(
hosts=[{"host": host.replace("https://", ""), "port": 443}],
http_auth=get_aws_auth(),
use_ssl=True,
verify_certs=True,
connection_class=RequestsHttpConnection,
timeout=300
)
time.sleep(60) # Wait for data access policies to propagate
index_body = {
"settings": {
"index": {
"knn": True
}
},
"mappings": {
"properties": {
"text": {"type": "text"},
"doc_id": {"type": "keyword"},
"chunk_id": {"type": "keyword"},
"embedding": {
"type": "knn_vector",
"dimension": EMBEDDING_DIMENSION,
"method": {
"name": "hnsw",
"space_type": "cosinesimil",
"engine": "faiss",
"parameters": {
"ef_construction": 256,
"m": 48
}
}
}
}
}
}
response = client.indices.create(index=INDEX_NAME, body=index_body)
print(f"Index created: {response}")
return clientRun the setup to create your collection
Now you will run the functions you just wrote to actually provision the OpenSearch Serverless resources in your AWS account. This process takes several minutes because the collection needs to fully spin up before it can accept index requests.
This will take several minutes
Collection creation typically takes several minutes. If this is the first OpenSearch Serverless collection in your account, it may take over 10 minutes. The script will print status updates every 30 seconds. After the collection is active, there is an additional 60-second wait for data access policies to propagate before creating the index.
- Run the setup script by executing this command in your terminal:
python3 -c "
import boto3
from opensearch_setup import create_policies, create_collection, create_index
from config import AWS_REGION, COLLECTION_NAME
sts = boto3.client('sts', region_name=AWS_REGION)
account_id = sts.get_caller_identity()['Account']
print(f'Account ID: {account_id}')
print('Creating policies...')
create_policies(COLLECTION_NAME, account_id)
print('Policies created.')
print('Creating collection (this takes several minutes)...')
host = create_collection()
print('Creating index...')
client = create_index(host)
print('Setup complete!')
"What should I see?
You will see output similar to this as the script runs:
- Account ID: 123456789012
- Creating policies...
- Policies created.
- Creating collection rag-pipeline (ID: abc123def456)...
- Waiting for collection to become active... (repeated several times)
- Collection active at: https://abc123def456.us-east-1.aoss.amazonaws.com
- Index created: {'acknowledged': True, ...}
- Setup complete!
Getting an AccessDenied or authorization error?
Make sure your AWS CLI credentials have permissions to create OpenSearch Serverless collections and policies. Your IAM user or role needs the aoss:CreateCollection, aoss:CreateSecurityPolicy, and aoss:CreateAccessPolicy permissions.
If you see a ConflictException, a collection or policy with the same name already exists. You may have run this script before. Delete the existing resources in the AWS console first, or choose a different COLLECTION_NAME in config.py.
help me troubleshoot OpenSearch Serverless access errors
Your OpenSearch Serverless collection is live and ready to accept data. Next up, you will build the document ingestion pipeline that cleans, chunks, and embeds your documents before storing them in this collection.
Build the Document Ingestion Pipeline
Your OpenSearch Serverless collection is live and your index is ready to accept documents. Now you need a way to get documents into that index.
Raw documents are messy. They have page headers, inconsistent spacing, and no logical boundaries for a language model to work with. In this step, you will build an ingestion pipeline that cleans text, splits it into semantically meaningful chunks, and generates vector embeddings for each chunk.
In this step, get ready to:
- Create a text cleaning function that removes noise from extracted documents.
- Build a semantic chunking function that splits text on paragraph boundaries with overlap.
- Generate vector embeddings using Titan Text Embeddings V2 and tie everything into a single processing function.
Create the text cleaning function
Documents extracted from PDFs or other sources carry artifacts like page numbers, excessive whitespace, and inconsistent formatting. Cleaning these before chunking prevents noise from polluting your embeddings and search results.
- Create a new file called ingestion.py in your project directory.
- Add the imports, Bedrock client, and text cleaning function by pasting this code:
import re
import json
import boto3
from config import AWS_REGION, EMBEDDING_MODEL_ID, EMBEDDING_DIMENSION, CHUNK_SIZE, CHUNK_OVERLAP
bedrock_runtime = boto3.client("bedrock-runtime", region_name=AWS_REGION)
def clean_text(text: str) -> str:
"""Clean extracted text: normalize whitespace, remove headers/footers patterns."""
text = re.sub(r"Page \d+ of \d+", "", text)
text = re.sub(r"\n{3,}", "\n\n", text)
text = re.sub(r"[ \t]+", " ", text)
text = text.strip()
return textWhat does this code do?
- The bedrock_runtime client connects to Amazon Bedrock in us-east-1 for generating embeddings later.
- The first regex removes "Page X of Y" patterns that commonly appear in PDF extractions.
- The second regex collapses three or more consecutive newlines down to two, preserving paragraph structure without excessive gaps.
- The third regex normalizes horizontal whitespace (multiple spaces or tabs become a single space).
- Save ingestion.py.
- You won't see output yet. The next chunks add the embedding and chunking logic that this function feeds into.
Add embedding generation and semantic chunking
Your pipeline needs two more core functions. The first turns text into a vector embedding using Titan Text Embeddings V2. The second splits cleaned text into chunks that respect paragraph boundaries instead of cutting mid-sentence.
- In ingestion.py, add the embedding function below clean_text() by pasting this code:
def get_embedding(text: str) -> list[float]:
"""Generate embedding using Titan Text Embeddings V2."""
response = bedrock_runtime.invoke_model(
modelId=EMBEDDING_MODEL_ID,
body=json.dumps({
"inputText": text,
"dimensions": EMBEDDING_DIMENSION,
"normalize": True
}),
contentType="application/json",
accept="application/json"
)
result = json.loads(response["body"].read())
return result["embedding"]What does this code do?
This function sends text to the amazon.titan-embed-text-v2:0 model and gets back a 1024-dimensional vector. The normalize: True parameter ensures all vectors have unit length, which is required for cosine similarity comparisons in OpenSearch.
- Now add the semantic chunking function below get_embedding() by pasting this code:
def semantic_chunk(text: str, max_tokens: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> list[str]:
"""Split text into semantic chunks based on paragraph boundaries and size limits."""
paragraphs = text.split("\n\n")
chunks = []
current_chunk = ""
for paragraph in paragraphs:
paragraph = paragraph.strip()
if not paragraph:
continue
if len(current_chunk.split()) + len(paragraph.split()) > max_tokens:
if current_chunk:
chunks.append(current_chunk.strip())
overlap_words = current_chunk.split()[-overlap:]
current_chunk = " ".join(overlap_words) + " " + paragraph
else:
chunks.append(paragraph[:max_tokens * 5])
current_chunk = ""
else:
current_chunk += "\n\n" + paragraph if current_chunk else paragraph
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunksWhy chunk on paragraph boundaries?
Naive chunking splits text at a fixed character count, often breaking mid-sentence. This function respects paragraph boundaries so each chunk contains a complete thought.
The overlap parameter (50 words by default) carries context from the end of one chunk into the beginning of the next. This prevents information loss at chunk boundaries during retrieval.
- Save ingestion.py.
- You won't see output yet. The final function in the next chunk ties everything together into a testable pipeline.
Seeing a syntax error?
Make sure get_embedding() is placed after clean_text() and semantic_chunk() is placed after get_embedding(). Python reads top-to-bottom, so the order matters.
help me debug a syntax error in ingestion.py
Build the document processing pipeline
The final function orchestrates the entire ingestion flow: clean, chunk, embed. It returns a list of dictionaries ready to be indexed into OpenSearch.
- In ingestion.py, add this function below semantic_chunk():
def process_document(text: str, doc_id: str) -> list[dict]:
"""Full ingestion pipeline: clean, chunk, embed."""
cleaned = clean_text(text)
chunks = semantic_chunk(cleaned)
documents = []
for i, chunk in enumerate(chunks):
embedding = get_embedding(chunk)
documents.append({
"doc_id": doc_id,
"chunk_id": f"{doc_id}_chunk_{i}",
"text": chunk,
"embedding": embedding,
})
return documentsWhat does this code do?
- process_document() calls clean_text() to strip noise, then semantic_chunk() to split into paragraphs.
- Each chunk gets its own embedding via get_embedding() and a unique chunk_id combining the document ID and chunk index.
- The returned list of dicts contains doc_id, chunk_id, text, and embedding fields. These map directly to the OpenSearch index fields you created in the previous step.
- Save ingestion.py.
- Verify the module loads without errors by running this in your terminal:
python3 -c "from ingestion import process_document; print('ingestion module loaded successfully')"You should see ingestion module loaded successfully printed to your terminal.
Seeing an import error?
- Make sure config.py is in the same directory as ingestion.py. The import on line 4 pulls constants from that file.
- Check that your virtual environment is activated. The boto3 import will fail if you are running from the system Python.
- If you see a sentence-transformers related warning, that is fine. This module does not use it directly.
help me fix an import error in my ingestion pipeline
✔️ Awesome, I've got everything!
Great. Double check you have saved ingestion.py and the import test passed.
ⓧ I'd like to double check the full code
Here is the complete ingestion.py file. Make sure yours matches exactly:
import re
import json
import boto3
from config import AWS_REGION, EMBEDDING_MODEL_ID, EMBEDDING_DIMENSION, CHUNK_SIZE, CHUNK_OVERLAP
bedrock_runtime = boto3.client("bedrock-runtime", region_name=AWS_REGION)
def clean_text(text: str) -> str:
"""Clean extracted text: normalize whitespace, remove headers/footers patterns."""
text = re.sub(r"Page \d+ of \d+", "", text)
text = re.sub(r"\n{3,}", "\n\n", text)
text = re.sub(r"[ \t]+", " ", text)
text = text.strip()
return text
def get_embedding(text: str) -> list[float]:
"""Generate embedding using Titan Text Embeddings V2."""
response = bedrock_runtime.invoke_model(
modelId=EMBEDDING_MODEL_ID,
body=json.dumps({
"inputText": text,
"dimensions": EMBEDDING_DIMENSION,
"normalize": True
}),
contentType="application/json",
accept="application/json"
)
result = json.loads(response["body"].read())
return result["embedding"]
def semantic_chunk(text: str, max_tokens: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> list[str]:
"""Split text into semantic chunks based on paragraph boundaries and size limits."""
paragraphs = text.split("\n\n")
chunks = []
current_chunk = ""
for paragraph in paragraphs:
paragraph = paragraph.strip()
if not paragraph:
continue
if len(current_chunk.split()) + len(paragraph.split()) > max_tokens:
if current_chunk:
chunks.append(current_chunk.strip())
overlap_words = current_chunk.split()[-overlap:]
current_chunk = " ".join(overlap_words) + " " + paragraph
else:
chunks.append(paragraph[:max_tokens * 5])
current_chunk = ""
else:
current_chunk += "\n\n" + paragraph if current_chunk else paragraph
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunks
def process_document(text: str, doc_id: str) -> list[dict]:
"""Full ingestion pipeline: clean, chunk, embed."""
cleaned = clean_text(text)
chunks = semantic_chunk(cleaned)
documents = []
for i, chunk in enumerate(chunks):
embedding = get_embedding(chunk)
documents.append({
"doc_id": doc_id,
"chunk_id": f"{doc_id}_chunk_{i}",
"text": chunk,
"embedding": embedding,
})
return documentsYour ingestion pipeline is ready. It can take raw document text, clean it, split it intelligently, and produce embeddings that map directly to your OpenSearch index schema. Next up, you will configure hybrid search so your pipeline can retrieve chunks using both keyword matching and vector similarity together.
Configure Hybrid Search
In the previous step, you built the ingestion pipeline that cleans, chunks, and embeds documents into your OpenSearch Serverless collection. The index is ready with both a text field for keyword matching and an embedding field for vector similarity.
Now you need a retrieval system that uses both fields together. Keyword search alone misses synonyms and paraphrases. Vector search alone can drift toward semantically similar but factually wrong passages. Hybrid search combines both approaches and lets a cross-encoder reranker pick the best final results.
In this step, get ready to:
- Build a search pipeline with score normalization for hybrid scoring.
- Implement hybrid queries that combine BM25 keyword search with k-NN vector search.
- Add cross-encoder reranking for two-stage retrieval precision.
Create the retrieval module and search pipeline
The retrieval module needs to convert a query into an embedding, then run that embedding alongside the raw query text through a search pipeline that normalizes and combines scores from both search strategies.
- Create a new file called retrieval.py in your project folder and add the following imports and module-level setup:
import json
import boto3
from sentence_transformers import CrossEncoder
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
from config import (
AWS_REGION, INDEX_NAME, EMBEDDING_MODEL_ID, EMBEDDING_DIMENSION,
HYBRID_SEMANTIC_WEIGHT, HYBRID_LEXICAL_WEIGHT,
RERANK_MODEL, RERANK_TOP_K, RERANK_FINAL_K
)
# Initialize Bedrock client for generating query embeddings
bedrock_runtime = boto3.client("bedrock-runtime", region_name=AWS_REGION)
# Load the cross-encoder model for reranking (downloads ~90MB on first run)
reranker = CrossEncoder(RERANK_MODEL)What does this setup do?
The module imports everything it needs from your config.py file: search weights, model IDs, and retrieval parameters.
The bedrock_runtime client handles embedding queries at search time. The CrossEncoder loads a local reranking model that will rescore results later. The first time this runs, it downloads the model weights from Hugging Face (about 90MB).
- Below the module-level setup, add the query embedding function and the search pipeline creation function:
def get_query_embedding(query: str) -> list[float]:
"""Generate query embedding."""
response = bedrock_runtime.invoke_model(
modelId=EMBEDDING_MODEL_ID,
body=json.dumps({
"inputText": query,
"dimensions": EMBEDDING_DIMENSION,
"normalize": True
}),
contentType="application/json",
accept="application/json"
)
result = json.loads(response["body"].read())
return result["embedding"]
def create_search_pipeline(client: OpenSearch):
"""Create the hybrid search pipeline with normalization."""
pipeline_body = {
"description": "Hybrid search pipeline with score normalization",
"phase_results_processors": [
{
"normalization-processor": {
"normalization": {"technique": "min_max"},
"combination": {
"technique": "arithmetic_mean",
"parameters": {
"weights": [HYBRID_LEXICAL_WEIGHT, HYBRID_SEMANTIC_WEIGHT]
}
}
}
}
]
}
client.http.put(
url="/_search/pipeline/hybrid-search-pipeline",
body=pipeline_body
)
print("Search pipeline created: hybrid-search-pipeline")What does the search pipeline do?
BM25 and k-NN return scores on completely different scales. The search pipeline normalizes both score sets using min_max (mapping each to 0-1), then combines them with arithmetic_mean using a 0.3/0.7 lexical/semantic weight split.
This means vector similarity gets 70% influence and keyword matching gets 30%. The pipeline lives inside OpenSearch itself, so every hybrid query automatically gets normalized scoring.
- Save retrieval.py.
You won't see a change yet. The next chunk adds the hybrid search function that uses this pipeline.
Implement the hybrid search function
The hybrid query sends both a keyword (BM25) subquery and a vector (k-NN) subquery in a single request. OpenSearch runs them in parallel and feeds the results through your normalization pipeline.
- In retrieval.py, add the hybrid_search function below create_search_pipeline:
def hybrid_search(client: OpenSearch, query: str, top_k: int = RERANK_TOP_K) -> list[dict]:
"""Execute hybrid search combining BM25 and k-NN."""
query_embedding = get_query_embedding(query)
search_body = {
"size": top_k,
"query": {
"hybrid": {
"queries": [
{"match": {"text": {"query": query}}},
{"knn": {"embedding": {"vector": query_embedding, "k": top_k}}}
]
}
}
}
response = client.search(
index=INDEX_NAME,
body=search_body,
params={"search_pipeline": "hybrid-search-pipeline"}
)
results = []
for hit in response["hits"]["hits"]:
results.append({
"chunk_id": hit["_source"]["chunk_id"],
"text": hit["_source"]["text"],
"score": hit["_score"]
})
return resultsHow does the hybrid query work?
The hybrid query type is specific to OpenSearch Serverless with Neural Search enabled. It takes an array of subqueries. The first is a standard BM25 match on the text field (keyword search). The second is a knn query on the embedding field (vector search).
The search_pipeline parameter tells OpenSearch to route results through your normalization pipeline before returning them. This ensures the final _score on each hit is already a weighted blend of both signals.
- Save retrieval.py.
You won't see a change yet. The next chunk adds reranking to further refine these results.
Add cross-encoder reranking
Hybrid search returns the top 20 candidates based on a fast approximate score. A cross-encoder is a slower but more accurate model that reads the query and each chunk together to produce a precise relevance score. This two-stage pattern gives you speed from hybrid search and precision from the cross-encoder.
- In retrieval.py, add the rerank function below hybrid_search:
def rerank(query: str, results: list[dict], top_k: int = RERANK_FINAL_K) -> list[dict]:
"""Rerank results using cross-encoder."""
if not results:
return []
pairs = [(query, result["text"]) for result in results]
scores = reranker.predict(pairs)
for i, result in enumerate(results):
result["rerank_score"] = float(scores[i])
reranked = sorted(results, key=lambda x: x["rerank_score"], reverse=True)
return reranked[:top_k]What does the reranker do?
The CrossEncoder model (cross-encoder/ms-marco-MiniLM-L6-v2) reads each (query, chunk) pair and produces a relevance score. Unlike the embedding model which encodes query and document separately, the cross-encoder processes them together for higher accuracy.
The function sorts all candidates by their new rerank score and returns only the top 3. This gives you high-precision results without paying the cross-encoder cost on your entire index.
- Below rerank, add the retrieve function that ties the two-stage pattern together:
def retrieve(client: OpenSearch, query: str) -> list[dict]:
"""Full two-stage retrieval: hybrid search + cross-encoder reranking."""
candidates = hybrid_search(client, query, top_k=RERANK_TOP_K)
reranked = rerank(query, candidates, top_k=RERANK_FINAL_K)
return rerankedWhat does retrieve() do?
This is the single entry point other modules will call. It runs hybrid search to get the top 20 candidates, then passes them through the cross-encoder to return the 3 most relevant chunks. The values 20 and 3 come from RERANK_TOP_K and RERANK_FINAL_K in your config.
- Save retrieval.py.
- Verify the module loads correctly by running:
python -c "import retrieval; print('retrieval module loaded successfully')"You should see retrieval module loaded successfully printed to your terminal. The first run may take a moment as the cross-encoder model downloads.
Seeing an import error?
- If you see ModuleNotFoundError: No module named 'sentence_transformers', re-run pip install -r requirements.txt to install the missing dependency.
- If you see a syntax error, check that each function definition starts with def at the correct indentation level and that all parentheses are balanced.
help me debug my retrieval.py import error
✔️ Awesome, I've got everything!
Great. Double check you have saved retrieval.py with all five functions.
ⓧ I'd like to double check the full code
import json
import boto3
from sentence_transformers import CrossEncoder
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
from config import (
AWS_REGION, INDEX_NAME, EMBEDDING_MODEL_ID, EMBEDDING_DIMENSION,
HYBRID_SEMANTIC_WEIGHT, HYBRID_LEXICAL_WEIGHT,
RERANK_MODEL, RERANK_TOP_K, RERANK_FINAL_K
)
bedrock_runtime = boto3.client("bedrock-runtime", region_name=AWS_REGION)
reranker = CrossEncoder(RERANK_MODEL)
def get_query_embedding(query: str) -> list[float]:
"""Generate query embedding."""
response = bedrock_runtime.invoke_model(
modelId=EMBEDDING_MODEL_ID,
body=json.dumps({
"inputText": query,
"dimensions": EMBEDDING_DIMENSION,
"normalize": True
}),
contentType="application/json",
accept="application/json"
)
result = json.loads(response["body"].read())
return result["embedding"]
def hybrid_search(client: OpenSearch, query: str, top_k: int = RERANK_TOP_K) -> list[dict]:
"""Execute hybrid search combining BM25 and k-NN."""
query_embedding = get_query_embedding(query)
search_body = {
"size": top_k,
"query": {
"hybrid": {
"queries": [
{"match": {"text": {"query": query}}},
{"knn": {"embedding": {"vector": query_embedding, "k": top_k}}}
]
}
}
}
response = client.search(
index=INDEX_NAME,
body=search_body,
params={"search_pipeline": "hybrid-search-pipeline"}
)
results = []
for hit in response["hits"]["hits"]:
results.append({
"chunk_id": hit["_source"]["chunk_id"],
"text": hit["_source"]["text"],
"score": hit["_score"]
})
return results
def rerank(query: str, results: list[dict], top_k: int = RERANK_FINAL_K) -> list[dict]:
"""Rerank results using cross-encoder."""
if not results:
return []
pairs = [(query, result["text"]) for result in results]
scores = reranker.predict(pairs)
for i, result in enumerate(results):
result["rerank_score"] = float(scores[i])
reranked = sorted(results, key=lambda x: x["rerank_score"], reverse=True)
return reranked[:top_k]
def retrieve(client: OpenSearch, query: str) -> list[dict]:
"""Full two-stage retrieval: hybrid search + cross-encoder reranking."""
candidates = hybrid_search(client, query, top_k=RERANK_TOP_K)
reranked = rerank(query, candidates, top_k=RERANK_FINAL_K)
return reranked
def create_search_pipeline(client: OpenSearch):
"""Create the hybrid search pipeline with normalization."""
pipeline_body = {
"description": "Hybrid search pipeline with score normalization",
"phase_results_processors": [
{
"normalization-processor": {
"normalization": {"technique": "min_max"},
"combination": {
"technique": "arithmetic_mean",
"parameters": {
"weights": [HYBRID_LEXICAL_WEIGHT, HYBRID_SEMANTIC_WEIGHT]
}
}
}
}
]
}
client.http.put(
url="/_search/pipeline/hybrid-search-pipeline",
body=pipeline_body
)
print("Search pipeline created: hybrid-search-pipeline")Your retrieval module is ready. It combines the speed of hybrid search with the precision of cross-encoder reranking. Next up, you will add guardrails and resilient generation to protect the final answer step.
Add Cross-Encoder Reranking
Your hybrid search retrieves candidates using both keyword matching and vector similarity. The results are good, but the scoring from each method is approximate.
A cross-encoder fixes this by reading the query and each candidate together as a single input. This joint attention produces much more accurate relevance scores. In this step, you will build a two-stage retrieval system that uses hybrid search to cast a wide net, then a cross-encoder to pick the best results.
In this step, get ready to:
- Understand the cross-encoder model loaded in your retrieval module.
- Build a reranking function that rescores hybrid search results.
- Implement the full two-stage retrieval pattern: retrieve 20 candidates, rerank to the best 3.
Understand the cross-encoder model
Your retrieval.py already initializes a cross-encoder at the module level. This line loads the model once when the file is imported, so every reranking call reuses it without re-downloading.
- In retrieval.py, find this line near the top of the file:
reranker = CrossEncoder(RERANK_MODEL)What is a cross-encoder?
A bi-encoder (like Titan Embeddings) processes the query and document separately, producing independent vectors. It is fast but misses nuanced interactions between query and document tokens.
A cross-encoder concatenates the query and document into one input and reads them together with full self-attention. This captures negation, entity relationships, and subtle meaning. The tradeoff is speed. Cross-encoders are too slow to scan thousands of documents, so you use them to rescore a small shortlist from the fast first stage.
Why this model?
The cross-encoder/ms-marco-MiniLM-L6-v2 model is trained on the MS MARCO passage ranking dataset. It processes about 1800 documents per second, making it fast enough to rerank 20 candidates in real time.
Build the reranking function
The reranking function takes the list of candidate results from hybrid search, pairs each with the original query, and asks the cross-encoder to score relevance.
- In retrieval.py, find the stub for the rerank() function and replace it with the following:
def rerank(query: str, results: list[dict], top_k: int = RERANK_FINAL_K) -> list[dict]:
"""Rerank results using cross-encoder."""
if not results:
return []
# Create (query, document) pairs for the cross-encoder
pairs = [(query, result["text"]) for result in results]
# Score all pairs in one batch
scores = reranker.predict(pairs)
# Attach rerank scores to each result
for i, result in enumerate(results):
result["rerank_score"] = float(scores[i])
# Sort by cross-encoder score (highest first) and return top results
reranked = sorted(results, key=lambda x: x["rerank_score"], reverse=True)
return reranked[:top_k]What does this code do?
- The early return handles the edge case where hybrid search found nothing.
- reranker.predict(pairs) sends all query-document pairs through the cross-encoder in one batch. It returns a raw relevance score for each pair.
- The function attaches a rerank_score to each result dict, sorts descending, and slices to top_k (defaults to 3 from RERANK_FINAL_K).
- Save retrieval.py.
Seeing a NameError for reranker?
Make sure the reranker = CrossEncoder(RERANK_MODEL) line is above the rerank() function definition. The function references reranker as a module-level variable.
help me debug a NameError for reranker in my retrieval.py
Implement the two-stage retrieval pattern
The final piece ties hybrid search and reranking into a single call. The retrieve() function fetches a broad set of 20 candidates with hybrid search, then narrows to the 3 most relevant using the cross-encoder.
- In retrieval.py, find the stub for the retrieve() function and replace it with:
def retrieve(client: OpenSearch, query: str) -> list[dict]:
"""Full two-stage retrieval: hybrid search + cross-encoder reranking."""
# Stage 1: Cast a wide net with hybrid search
candidates = hybrid_search(client, query, top_k=RERANK_TOP_K)
# Stage 2: Narrow to the best matches with the cross-encoder
reranked = rerank(query, candidates, top_k=RERANK_FINAL_K)
return rerankedWhy two stages?
Hybrid search is fast. It uses pre-built indexes (BM25 inverted index and HNSW vector graph) to scan thousands of chunks in milliseconds. But its scoring is approximate.
The cross-encoder is accurate but slow. By running it only on the top 20 candidates instead of the entire corpus, you get the best of both worlds. This retrieve-then-rerank pattern is standard in production search systems.
- Save retrieval.py.
- Verify the function is syntactically valid by running:
python -c "from retrieval import retrieve; print('Two-stage retrieval ready')"You should see Two-stage retrieval ready printed to the terminal. The first run may take a moment while the cross-encoder model downloads (~90MB).
Import hanging or failing?
If the import hangs, the cross-encoder model is downloading. Wait for it to finish. If you see a connection error, check that you have internet access.
If you see a ModuleNotFoundError for sentence_transformers, confirm you installed dependencies with pip install -r requirements.txt and that you are in the correct virtual environment.
help me debug my retrieval.py import
✔️ Awesome, I've got everything!
Great. Double check you have saved retrieval.py and that the import test passed.
ⓧ I'd like to double check the full code
import json
import boto3
from sentence_transformers import CrossEncoder
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
from config import (
AWS_REGION, INDEX_NAME, EMBEDDING_MODEL_ID, EMBEDDING_DIMENSION,
HYBRID_SEMANTIC_WEIGHT, HYBRID_LEXICAL_WEIGHT,
RERANK_MODEL, RERANK_TOP_K, RERANK_FINAL_K
)
bedrock_runtime = boto3.client("bedrock-runtime", region_name=AWS_REGION)
reranker = CrossEncoder(RERANK_MODEL)
def get_query_embedding(query: str) -> list[float]:
response = bedrock_runtime.invoke_model(
modelId=EMBEDDING_MODEL_ID,
body=json.dumps({"inputText": query, "dimensions": EMBEDDING_DIMENSION, "normalize": True}),
contentType="application/json",
accept="application/json"
)
result = json.loads(response["body"].read())
return result["embedding"]
def hybrid_search(client: OpenSearch, query: str, top_k: int = RERANK_TOP_K) -> list[dict]:
query_embedding = get_query_embedding(query)
search_body = {
"size": top_k,
"query": {
"hybrid": {
"queries": [
{"match": {"text": {"query": query}}},
{"knn": {"embedding": {"vector": query_embedding, "k": top_k}}}
]
}
}
}
response = client.search(index=INDEX_NAME, body=search_body, params={"search_pipeline": "hybrid-search-pipeline"})
results = []
for hit in response["hits"]["hits"]:
results.append({"chunk_id": hit["_source"]["chunk_id"], "text": hit["_source"]["text"], "score": hit["_score"]})
return results
def rerank(query: str, results: list[dict], top_k: int = RERANK_FINAL_K) -> list[dict]:
"""Rerank results using cross-encoder."""
if not results:
return []
pairs = [(query, result["text"]) for result in results]
scores = reranker.predict(pairs)
for i, result in enumerate(results):
result["rerank_score"] = float(scores[i])
reranked = sorted(results, key=lambda x: x["rerank_score"], reverse=True)
return reranked[:top_k]
def retrieve(client: OpenSearch, query: str) -> list[dict]:
"""Full two-stage retrieval: hybrid search + cross-encoder reranking."""
candidates = hybrid_search(client, query, top_k=RERANK_TOP_K)
reranked = rerank(query, candidates, top_k=RERANK_FINAL_K)
return reranked
def create_search_pipeline(client: OpenSearch):
pipeline_body = {
"description": "Hybrid search pipeline with score normalization",
"phase_results_processors": [{"normalization-processor": {"normalization": {"technique": "min_max"}, "combination": {"technique": "arithmetic_mean", "parameters": {"weights": [HYBRID_LEXICAL_WEIGHT, HYBRID_SEMANTIC_WEIGHT]}}}}]
}
client.http.put(url="/_search/pipeline/hybrid-search-pipeline", body=pipeline_body)
print("Search pipeline created: hybrid-search-pipeline")Your retrieval module now has a complete two-stage pipeline. Hybrid search casts a wide net over the full index, and the cross-encoder zeroes in on the most relevant chunks. Next, you will build the generation layer that turns these retrieved chunks into guarded, resilient answers using Claude.
Implement Guardrails and Resilient Generation
Your retrieval pipeline can now find and rank the most relevant chunks for any query. The next challenge is generating a safe, reliable answer from those chunks.
Production systems face two risks at the generation stage: the model might produce harmful content, and transient API errors can crash the entire pipeline. In this step, you will build a generation module that addresses both by attaching Amazon Bedrock Guardrails for content safety and wrapping the call in retry logic with graceful fallbacks.
In this step, get ready to:
- Create a Bedrock Guardrail with content filters and a denied topic.
- Build a generation function that calls Claude 3 Haiku via the Converse API with the guardrail attached.
- Add retry logic with exponential backoff and fallback responses for error scenarios.
Create the generation module and guardrail function
Amazon Bedrock exposes two separate boto3 service clients. The bedrock client handles control-plane operations like creating guardrails. The bedrock-runtime client handles data-plane operations like running inference with the Converse API. Your generation module needs both.
- Create a new file called generation.py in your project folder.
- Add the following imports, client initializations, and guardrail function:
import json
import time
import boto3
from botocore.exceptions import ClientError
from config import AWS_REGION, GENERATION_MODEL_ID, MAX_RETRIES, RETRY_BASE_DELAY
# Control-plane client for creating guardrails
bedrock_client = boto3.client("bedrock", region_name=AWS_REGION)
# Data-plane client for running inference
bedrock_runtime = boto3.client("bedrock-runtime", region_name=AWS_REGION)
# Module-level state for the active guardrail
GUARDRAIL_ID = None
GUARDRAIL_VERSION = None
def create_guardrail() -> tuple[str, str]:
"""Create a Bedrock Guardrail with content filters and denied topics."""
global GUARDRAIL_ID, GUARDRAIL_VERSION
response = bedrock_client.create_guardrail(
name="rag-pipeline-guardrail",
description="Guardrail for RAG pipeline generation",
contentPolicyConfig={
"filtersConfig": [
{"type": "HATE", "inputStrength": "HIGH", "outputStrength": "HIGH"},
{"type": "VIOLENCE", "inputStrength": "HIGH", "outputStrength": "HIGH"},
{"type": "SEXUAL", "inputStrength": "HIGH", "outputStrength": "HIGH"},
{"type": "MISCONDUCT", "inputStrength": "HIGH", "outputStrength": "HIGH"},
{"type": "INSULTS", "inputStrength": "MEDIUM", "outputStrength": "HIGH"},
]
},
topicPolicyConfig={
"topicsConfig": [
{
"name": "PersonalMedicalAdvice",
"definition": "Providing specific medical diagnoses, treatment plans, or medication recommendations for individual health conditions",
"type": "DENY",
"examples": [
"What medication should I take for my headache?",
"Do I have diabetes based on these symptoms?"
]
}
]
},
blockedInputMessaging="I cannot process this request as it contains content that violates our safety policies.",
blockedOutputsMessaging="I cannot provide this response as it contains content that violates our safety policies."
)
GUARDRAIL_ID = response["guardrailId"]
GUARDRAIL_VERSION = response["version"]
print(f"Guardrail created: {GUARDRAIL_ID} (version {GUARDRAIL_VERSION})")
return GUARDRAIL_ID, GUARDRAIL_VERSIONWhat does this code do?
The module starts by initializing two boto3 clients. The bedrock client creates the guardrail resource itself. The bedrock-runtime client calls Claude through the Converse API later.
The create_guardrail() function configures five content filters (HATE, VIOLENCE, SEXUAL, MISCONDUCT at HIGH strength, INSULTS at MEDIUM input/HIGH output) plus a denied topic that blocks personal medical advice. The returned ID and version are stored as module globals so the generation function can reference them.
- Save generation.py.
Why two separate clients?
Amazon Bedrock splits its API surface into a control plane and a data plane. The control plane (bedrock) manages resources like guardrails, custom models, and provisioned throughput. The data plane (bedrock-runtime) handles inference requests and model invocations.
This separation follows the same pattern as other AWS services. For example, s3 manages buckets while s3 also serves objects, but Bedrock makes the split explicit with separate service names.
Getting an access denied error?
Make sure your IAM user or role has the bedrock:CreateGuardrail permission. If you are using the same credentials that enabled model access in Step 1, you likely already have full Bedrock access.
help me debug a Bedrock guardrail access denied error
Build the generation function with guardrails and retry logic
The generation function takes a user query and the retrieved context chunks, sends them to Claude 3 Haiku via the Converse API, and handles three scenarios: successful generation, guardrail intervention, and transient API errors.
- Add the following function below create_guardrail() in generation.py:
def generate_with_guardrails(query: str, context_chunks: list[dict]) -> dict:
"""Generate a response using Claude with guardrails, retry logic, and fallback."""
# Join retrieved chunk texts into a single context block
context = "\n\n---\n\n".join([chunk["text"] for chunk in context_chunks])
# Build the prompt with context and question
messages = [
{
"role": "user",
"content": [{"text": f"""Based on the following context, answer the question.
If the context does not contain enough information to answer, say so clearly.
Context:
{context}
Question: {query}
Answer:"""}]
}
]
for attempt in range(MAX_RETRIES):
try:
# Build the Converse API request
kwargs = {
"modelId": GENERATION_MODEL_ID,
"messages": messages,
"inferenceConfig": {"maxTokens": 1024, "temperature": 0.1}
}
# Attach guardrail if one has been created
if GUARDRAIL_ID and GUARDRAIL_VERSION:
kwargs["guardrailConfig"] = {
"guardrailIdentifier": GUARDRAIL_ID,
"guardrailVersion": GUARDRAIL_VERSION
}
response = bedrock_runtime.converse(**kwargs)
# Check if the guardrail blocked the request
stop_reason = response.get("stopReason", "")
if stop_reason == "guardrail_intervened":
return {"answer": "I cannot answer this question as it was flagged by our safety system.", "status": "guardrail_blocked", "attempt": attempt + 1}
# Extract the generated text on success
output_text = response["output"]["message"]["content"][0]["text"]
return {"answer": output_text, "status": "success", "attempt": attempt + 1}
except ClientError as e:
error_code = e.response["Error"]["Code"]
if error_code == "ThrottlingException" and attempt < MAX_RETRIES - 1:
# Exponential backoff on throttling
delay = RETRY_BASE_DELAY * (2 ** attempt)
print(f" Throttled. Retrying in {delay}s (attempt {attempt + 1}/{MAX_RETRIES})...")
time.sleep(delay)
continue
elif error_code == "ValidationException":
return {"answer": "I encountered an issue processing your request. Please rephrase your question.", "status": "validation_error", "attempt": attempt + 1}
else:
if attempt < MAX_RETRIES - 1:
delay = RETRY_BASE_DELAY * (2 ** attempt)
time.sleep(delay)
continue
return {"answer": "I'm temporarily unable to process your request. Please try again later.", "status": "error", "error": str(e), "attempt": attempt + 1}
# Fallback if all retries are exhausted
return {"answer": "I'm temporarily unable to process your request. Please try again later.", "status": "max_retries_exhausted", "attempt": MAX_RETRIES}What does this function do?
The function implements three production resilience patterns in a single loop:
- Guardrail enforcement: if the Converse API returns a stopReason of guardrail_intervened, the function returns a safe canned response instead of the blocked content.
- Retry with exponential backoff: on a ThrottlingException, the function waits RETRY_BASE_DELAY * 2^attempt seconds before retrying. This gives the service time to recover without hammering it.
- Graceful fallback: if all retries are exhausted or an unrecoverable error occurs, the function returns a safe default message rather than crashing.
- Save generation.py.
- Verify the file has no syntax errors by running:
python -c "import generation; print('generation.py loaded successfully')"You should see generation.py loaded successfully printed to the terminal with no errors.
Seeing an ImportError or ModuleNotFoundError?
Make sure you are running the command from your project directory where config.py and generation.py both live. The import of config will fail if Python cannot find config.py in the current directory.
help me fix an import error when loading generation.py
Understand the retry and fallback strategy
The retry loop uses exponential backoff, meaning each successive wait doubles. With MAX_RETRIES = 3 and RETRY_BASE_DELAY = 1.0, the waits are 1 second, then 2 seconds, then the final attempt fails without further retry.
Why exponential backoff instead of fixed delays?
Fixed-interval retries can overwhelm a throttled service because every client retries at the same cadence. Exponential backoff spreads retries over increasing intervals, giving the service progressively more breathing room.
In production, you would also add jitter (a random offset) to prevent synchronized retry storms across multiple clients. For this project, the doubling pattern demonstrates the core concept.
How does the fallback decision tree work?
- ThrottlingException: retry with backoff until attempts are exhausted.
- ValidationException: return immediately with a "rephrase your question" message. Retrying would produce the same error.
- Any other ClientError: retry with backoff. If retries exhaust, return a generic safe fallback.
- Guardrail intervention: return a safe canned message immediately. No retry needed because the guardrail will block the same input every time.
✔️ Awesome, I've got everything!
Great. Double check you have saved generation.py and move on to the next step.
ⓧ I'd like to double check the full code
import json
import time
import boto3
from botocore.exceptions import ClientError
from config import AWS_REGION, GENERATION_MODEL_ID, MAX_RETRIES, RETRY_BASE_DELAY
bedrock_client = boto3.client("bedrock", region_name=AWS_REGION)
bedrock_runtime = boto3.client("bedrock-runtime", region_name=AWS_REGION)
GUARDRAIL_ID = None
GUARDRAIL_VERSION = None
def create_guardrail() -> tuple[str, str]:
"""Create a Bedrock Guardrail with content filters and denied topics."""
global GUARDRAIL_ID, GUARDRAIL_VERSION
response = bedrock_client.create_guardrail(
name="rag-pipeline-guardrail",
description="Guardrail for RAG pipeline generation",
contentPolicyConfig={
"filtersConfig": [
{"type": "HATE", "inputStrength": "HIGH", "outputStrength": "HIGH"},
{"type": "VIOLENCE", "inputStrength": "HIGH", "outputStrength": "HIGH"},
{"type": "SEXUAL", "inputStrength": "HIGH", "outputStrength": "HIGH"},
{"type": "MISCONDUCT", "inputStrength": "HIGH", "outputStrength": "HIGH"},
{"type": "INSULTS", "inputStrength": "MEDIUM", "outputStrength": "HIGH"},
]
},
topicPolicyConfig={
"topicsConfig": [
{
"name": "PersonalMedicalAdvice",
"definition": "Providing specific medical diagnoses, treatment plans, or medication recommendations for individual health conditions",
"type": "DENY",
"examples": [
"What medication should I take for my headache?",
"Do I have diabetes based on these symptoms?"
]
}
]
},
blockedInputMessaging="I cannot process this request as it contains content that violates our safety policies.",
blockedOutputsMessaging="I cannot provide this response as it contains content that violates our safety policies."
)
GUARDRAIL_ID = response["guardrailId"]
GUARDRAIL_VERSION = response["version"]
print(f"Guardrail created: {GUARDRAIL_ID} (version {GUARDRAIL_VERSION})")
return GUARDRAIL_ID, GUARDRAIL_VERSION
def generate_with_guardrails(query: str, context_chunks: list[dict]) -> dict:
"""Generate a response using Claude with guardrails, retry logic, and fallback."""
context = "\n\n---\n\n".join([chunk["text"] for chunk in context_chunks])
messages = [
{
"role": "user",
"content": [{"text": f"""Based on the following context, answer the question.
If the context does not contain enough information to answer, say so clearly.
Context:
{context}
Question: {query}
Answer:"""}]
}
]
for attempt in range(MAX_RETRIES):
try:
kwargs = {
"modelId": GENERATION_MODEL_ID,
"messages": messages,
"inferenceConfig": {"maxTokens": 1024, "temperature": 0.1}
}
if GUARDRAIL_ID and GUARDRAIL_VERSION:
kwargs["guardrailConfig"] = {
"guardrailIdentifier": GUARDRAIL_ID,
"guardrailVersion": GUARDRAIL_VERSION
}
response = bedrock_runtime.converse(**kwargs)
stop_reason = response.get("stopReason", "")
if stop_reason == "guardrail_intervened":
return {"answer": "I cannot answer this question as it was flagged by our safety system.", "status": "guardrail_blocked", "attempt": attempt + 1}
output_text = response["output"]["message"]["content"][0]["text"]
return {"answer": output_text, "status": "success", "attempt": attempt + 1}
except ClientError as e:
error_code = e.response["Error"]["Code"]
if error_code == "ThrottlingException" and attempt < MAX_RETRIES - 1:
delay = RETRY_BASE_DELAY * (2 ** attempt)
print(f" Throttled. Retrying in {delay}s (attempt {attempt + 1}/{MAX_RETRIES})...")
time.sleep(delay)
continue
elif error_code == "ValidationException":
return {"answer": "I encountered an issue processing your request. Please rephrase your question.", "status": "validation_error", "attempt": attempt + 1}
else:
if attempt < MAX_RETRIES - 1:
delay = RETRY_BASE_DELAY * (2 ** attempt)
time.sleep(delay)
continue
return {"answer": "I'm temporarily unable to process your request. Please try again later.", "status": "error", "error": str(e), "attempt": attempt + 1}
return {"answer": "I'm temporarily unable to process your request. Please try again later.", "status": "max_retries_exhausted", "attempt": MAX_RETRIES}Your generation module is locked down with content filters, denied topics, and resilient error handling. Next up, you will wire all the modules together into an end-to-end pipeline that takes a question and produces a grounded answer.
Wire the End-to-End Pipeline
You have all the individual modules working: ingestion cleans and chunks documents, retrieval combines hybrid search with reranking, and generation produces guarded responses with retry logic. The missing piece is an orchestrator that connects them into a single executable pipeline.
In this step, you will wire everything together into pipeline.py, test the full flow with different query types, and create a cleanup script for tearing down resources when you are finished.
In this step, get ready to:
- Connect all modules into an end-to-end pipeline orchestrator.
- Test with sample queries demonstrating keyword-heavy, semantic, and adversarial inputs.
- Create a cleanup script for tearing down all AWS resources.
Create the pipeline orchestrator
The orchestrator script imports every module and exposes three functions: one to initialize infrastructure, one to ingest documents, and one to run end-to-end queries.
- Create a new file called pipeline.py in your project folder.
- Add the imports and the initialize_pipeline() function by pasting this code:
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
from config import AWS_REGION, COLLECTION_NAME
from ingestion import process_document
from retrieval import retrieve, create_search_pipeline
from generation import generate_with_guardrails, create_guardrail
from opensearch_setup import create_policies, create_collection, create_index, get_aws_auth
import boto3
import json
def initialize_pipeline():
"""Set up all infrastructure and return the OpenSearch client."""
# Get the current AWS account ID for policy creation
sts = boto3.client("sts", region_name=AWS_REGION)
account_id = sts.get_caller_identity()["Account"]
print("Step 1: Creating OpenSearch Serverless policies...")
create_policies(COLLECTION_NAME, account_id)
print("Step 2: Creating collection...")
host = create_collection()
print("Step 3: Creating index...")
client = create_index(host)
print("Step 4: Creating search pipeline...")
create_search_pipeline(client)
print("Step 5: Creating Bedrock Guardrail...")
create_guardrail()
return clientWhat does this code do?
The imports pull in every module you built in previous steps. Each module handles one responsibility: ingestion, retrieval, generation, or OpenSearch setup.
The initialize_pipeline() function orchestrates the full infrastructure setup in order. It uses STS to retrieve your AWS account ID, then creates policies, the collection, the index, the search pipeline, and the guardrail sequentially.
- Below initialize_pipeline(), add the ingest_documents() function:
def ingest_documents(client: OpenSearch, documents: list[dict]):
"""Ingest processed documents into OpenSearch."""
from config import INDEX_NAME
# Index each document chunk individually
for doc in documents:
client.index(
index=INDEX_NAME,
body={
"text": doc["text"],
"doc_id": doc["doc_id"],
"chunk_id": doc["chunk_id"],
"embedding": doc["embedding"]
},
id=doc["chunk_id"]
)
print(f"Ingested {len(documents)} chunks")What does this code do?
This function takes the processed documents (already cleaned, chunked, and embedded by the ingestion module) and indexes them into OpenSearch one by one. Each document is stored with its text, metadata, and embedding vector using the chunk_id as the document ID to prevent duplicates.
- Below ingest_documents(), add the query_pipeline() function:
def query_pipeline(client: OpenSearch, query: str) -> dict:
"""Execute the full RAG pipeline: retrieve, rerank, generate."""
print(f"\nQuery: {query}")
# Stage 1: Two-stage retrieval (hybrid search + reranking)
print(" Retrieving relevant chunks...")
chunks = retrieve(client, query)
# Handle empty results
if not chunks:
return {
"query": query,
"answer": "No relevant information found in the knowledge base.",
"status": "no_results",
"chunks": []
}
print(f" Found {len(chunks)} relevant chunks after reranking")
print(f" Top chunk score: {chunks[0]['rerank_score']:.4f}")
# Low-confidence check: if the best chunk scores poorly, don't generate
if chunks[0]["rerank_score"] < 0.01:
return {
"query": query,
"answer": "I found some documents but they don't appear relevant enough to answer your question confidently.",
"status": "low_confidence",
"chunks": chunks
}
# Stage 2: Guarded generation
print(" Generating response with guardrails...")
result = generate_with_guardrails(query, chunks)
result["query"] = query
result["chunks"] = chunks
return resultWhat does this code do?
This is the heart of the RAG pipeline. It runs two stages: retrieval (hybrid search + cross-encoder reranking) followed by guarded generation.
- If no chunks are returned at all, it returns a no_results status immediately.
- If the top reranker score is below 0.01, it returns a low_confidence status rather than generating from weak evidence.
- Otherwise, it passes the chunks to generate_with_guardrails() which handles guardrail checks, retries, and fallback.
- Save pipeline.py.
Seeing import errors?
Make sure pipeline.py is in the same directory as your other module files (config.py, ingestion.py, retrieval.py, etc.). Python resolves imports relative to the working directory.
help me debug import errors in my pipeline.py
Test with sample queries
To validate the full pipeline, you need a test harness that ingests a sample document and runs queries against it. The __main__ block provides this: it initializes infrastructure, ingests a sample text about AWS services, waits for the index to refresh, and runs a query.
- At the bottom of pipeline.py, add the __main__ block:
if __name__ == "__main__":
sample_text = """
Amazon Web Services (AWS) provides cloud computing services that enable
organizations to build and deploy applications at scale. AWS offers over 200
services including compute, storage, database, analytics, and machine learning.
Amazon Bedrock is a fully managed service that offers access to foundation models
from leading AI companies. It supports models from Anthropic, Meta, Mistral AI,
and Amazon. Bedrock enables you to build generative AI applications with security
and privacy controls.
Amazon OpenSearch Serverless removes the operational complexity of provisioning
and managing OpenSearch clusters. It provides automatic scaling based on workload
demands and supports both search and vector workloads for building AI applications.
"""
print("Initializing RAG pipeline...")
client = initialize_pipeline()
print("\nIngesting sample document...")
docs = process_document(sample_text, "aws-overview")
ingest_documents(client, docs)
import time
print("\nWaiting for index refresh (60s)...")
time.sleep(60)
result = query_pipeline(client, "What is Amazon Bedrock?")
print(f"\nAnswer: {result['answer']}")
print(f"Status: {result['status']}")What does this code do?
This test harness demonstrates the full pipeline end-to-end. It defines a sample document about AWS services, processes it through the ingestion pipeline (cleaning, chunking, embedding), and indexes it into OpenSearch.
The 60-second wait is required because OpenSearch Serverless vector collections have a 60-second refresh interval before newly indexed documents become searchable. After the wait, it runs a semantic query to verify the full retrieve-rerank-generate flow.
- Save pipeline.py.
- Run the end-to-end pipeline by executing this command:
python pipeline.pyWhat should I see?
The script prints progress as it creates infrastructure, ingests the sample document, waits 60 seconds for the index refresh, then runs the query. You should see output ending with an answer describing Amazon Bedrock and a status of success.
The total runtime is around 6-8 minutes due to collection creation (5-10 minutes) plus the 60-second index refresh wait.
Getting a resource already exists error?
If you already created the OpenSearch collection and guardrail in earlier steps, the script will fail when trying to recreate them. You can either delete existing resources first (using the cleanup script from the next substep), or skip initialize_pipeline() and connect directly to your existing collection.
help me fix resource conflict errors in my pipeline
Once the semantic query works, test the guardrails by trying an adversarial query. The PersonalMedicalAdvice denied topic should trigger for medical advice requests.
- Test the guardrails by running this in a Python session:
# Test adversarial input - guardrails should block this
result = query_pipeline(client, "What medication should I take for my headache?")
print(f"Status: {result['status']}")
# Expected: status = 'guardrail_blocked' or 'low_confidence'Why test different query types?
Production RAG systems face three categories of queries: keyword-heavy queries that rely on BM25 lexical matching, semantic queries that depend on vector similarity, and adversarial queries that attempt to extract unsafe content. Testing all three confirms that hybrid search handles the first two correctly and that guardrails protect against the third.
Create the cleanup script
OpenSearch Serverless collections incur ongoing charges. A dedicated cleanup script ensures you can tear down all resources with a single command when you are done experimenting.
- Create a new file called cleanup.py in your project folder and paste the following code:
import boto3
from config import AWS_REGION, COLLECTION_NAME
from generation import GUARDRAIL_ID
def cleanup():
"""Delete all resources created by the pipeline."""
aoss_client = boto3.client("opensearchserverless", region_name=AWS_REGION)
bedrock_client = boto3.client("bedrock", region_name=AWS_REGION)
# Delete the OpenSearch Serverless collection
print("Deleting OpenSearch Serverless collection...")
try:
aoss_client.delete_collection(id=COLLECTION_NAME)
print(f" Collection '{COLLECTION_NAME}' deletion initiated")
except Exception as e:
print(f" Error deleting collection: {e}")
# Delete security policies (encryption and network)
print("Deleting security policies...")
for policy_type in ["encryption", "network"]:
try:
aoss_client.delete_security_policy(
name=f"{COLLECTION_NAME}-enc" if policy_type == "encryption" else f"{COLLECTION_NAME}-net",
type=policy_type
)
except Exception as e:
print(f" Error deleting {policy_type} policy: {e}")
# Delete the data access policy
try:
aoss_client.delete_access_policy(
name=f"{COLLECTION_NAME}-access",
type="data"
)
except Exception as e:
print(f" Error deleting access policy: {e}")
# Delete the Bedrock guardrail if it exists
if GUARDRAIL_ID:
print(f"Deleting Bedrock Guardrail {GUARDRAIL_ID}...")
try:
bedrock_client.delete_guardrail(guardrailIdentifier=GUARDRAIL_ID)
print(" Guardrail deleted")
except Exception as e:
print(f" Error deleting guardrail: {e}")
print("\nCleanup complete!")
if __name__ == "__main__":
cleanup()What does this code do?
- Deletes the OpenSearch Serverless collection, stopping the per-OCU-hour billing.
- Removes all three security policies (encryption, network, data access) that were created for the collection.
- Deletes the Bedrock guardrail if one was created during this session.
- Each deletion is wrapped in a try/except so the script continues even if one resource is already gone.
- Save cleanup.py.
- Confirm both new files exist by running:
lsYou should see pipeline.py and cleanup.py listed alongside your other module files.
Cleanup script failing to delete the collection?
The delete_collection API requires the collection ID, not the name. If you see an error, check that COLLECTION_NAME in config.py matches the collection you created. Also verify that your AWS credentials have the aoss:DeleteCollection permission.
help me debug cleanup script errors
Do not run cleanup yet
Keep your resources running for now. You will need the active collection and guardrail if you attempt the Secret Mission. The Before You Go section will guide you through cleanup when you are finished.
✔️ Awesome, I've got everything!
Great. Double check you have saved both pipeline.py and cleanup.py.
ⓧ I'd like to double check the full code
Here is the complete pipeline.py:
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
from config import AWS_REGION, COLLECTION_NAME
from ingestion import process_document
from retrieval import retrieve, create_search_pipeline
from generation import generate_with_guardrails, create_guardrail
from opensearch_setup import create_policies, create_collection, create_index, get_aws_auth
import boto3
import json
def initialize_pipeline():
"""Set up all infrastructure and return the OpenSearch client."""
sts = boto3.client("sts", region_name=AWS_REGION)
account_id = sts.get_caller_identity()["Account"]
print("Step 1: Creating OpenSearch Serverless policies...")
create_policies(COLLECTION_NAME, account_id)
print("Step 2: Creating collection...")
host = create_collection()
print("Step 3: Creating index...")
client = create_index(host)
print("Step 4: Creating search pipeline...")
create_search_pipeline(client)
print("Step 5: Creating Bedrock Guardrail...")
create_guardrail()
return client
def ingest_documents(client: OpenSearch, documents: list[dict]):
"""Ingest processed documents into OpenSearch."""
from config import INDEX_NAME
for doc in documents:
client.index(
index=INDEX_NAME,
body={
"text": doc["text"],
"doc_id": doc["doc_id"],
"chunk_id": doc["chunk_id"],
"embedding": doc["embedding"]
},
id=doc["chunk_id"]
)
print(f"Ingested {len(documents)} chunks")
def query_pipeline(client: OpenSearch, query: str) -> dict:
"""Execute the full RAG pipeline: retrieve, rerank, generate."""
print(f"\nQuery: {query}")
print(" Retrieving relevant chunks...")
chunks = retrieve(client, query)
if not chunks:
return {
"query": query,
"answer": "No relevant information found in the knowledge base.",
"status": "no_results",
"chunks": []
}
print(f" Found {len(chunks)} relevant chunks after reranking")
print(f" Top chunk score: {chunks[0]['rerank_score']:.4f}")
if chunks[0]["rerank_score"] < 0.01:
return {
"query": query,
"answer": "I found some documents but they don't appear relevant enough to answer your question confidently.",
"status": "low_confidence",
"chunks": chunks
}
print(" Generating response with guardrails...")
result = generate_with_guardrails(query, chunks)
result["query"] = query
result["chunks"] = chunks
return result
if __name__ == "__main__":
sample_text = """
Amazon Web Services (AWS) provides cloud computing services that enable
organizations to build and deploy applications at scale. AWS offers over 200
services including compute, storage, database, analytics, and machine learning.
Amazon Bedrock is a fully managed service that offers access to foundation models
from leading AI companies. It supports models from Anthropic, Meta, Mistral AI,
and Amazon. Bedrock enables you to build generative AI applications with security
and privacy controls.
Amazon OpenSearch Serverless removes the operational complexity of provisioning
and managing OpenSearch clusters. It provides automatic scaling based on workload
demands and supports both search and vector workloads for building AI applications.
"""
print("Initializing RAG pipeline...")
client = initialize_pipeline()
print("\nIngesting sample document...")
docs = process_document(sample_text, "aws-overview")
ingest_documents(client, docs)
import time
print("\nWaiting for index refresh (60s)...")
time.sleep(60)
result = query_pipeline(client, "What is Amazon Bedrock?")
print(f"\nAnswer: {result['answer']}")
print(f"Status: {result['status']}")Here is the complete cleanup.py:
import boto3
from config import AWS_REGION, COLLECTION_NAME
from generation import GUARDRAIL_ID
def cleanup():
"""Delete all resources created by the pipeline."""
aoss_client = boto3.client("opensearchserverless", region_name=AWS_REGION)
bedrock_client = boto3.client("bedrock", region_name=AWS_REGION)
print("Deleting OpenSearch Serverless collection...")
try:
aoss_client.delete_collection(id=COLLECTION_NAME)
print(f" Collection '{COLLECTION_NAME}' deletion initiated")
except Exception as e:
print(f" Error deleting collection: {e}")
print("Deleting security policies...")
for policy_type in ["encryption", "network"]:
try:
aoss_client.delete_security_policy(
name=f"{COLLECTION_NAME}-enc" if policy_type == "encryption" else f"{COLLECTION_NAME}-net",
type=policy_type
)
except Exception as e:
print(f" Error deleting {policy_type} policy: {e}")
try:
aoss_client.delete_access_policy(
name=f"{COLLECTION_NAME}-access",
type="data"
)
except Exception as e:
print(f" Error deleting access policy: {e}")
if GUARDRAIL_ID:
print(f"Deleting Bedrock Guardrail {GUARDRAIL_ID}...")
try:
bedrock_client.delete_guardrail(guardrailIdentifier=GUARDRAIL_ID)
print(" Guardrail deleted")
except Exception as e:
print(f" Error deleting guardrail: {e}")
print("\nCleanup complete!")
if __name__ == "__main__":
cleanup()Your production-grade RAG pipeline is fully operational. You have a complete system that ingests documents, retrieves relevant context through hybrid search and reranking, generates guarded responses, and can be cleanly torn down when no longer needed. Next up: the Secret Mission, where you will build an evaluation loop to automatically tune retrieval quality.
Secret mission
Add a Relevance Evaluation Loop
Your RAG pipeline retrieves and generates, but how do you know if it's retrieving the RIGHT chunks? Build an automated evaluation system that scores retrieval quality using Claude, sweeps different hybrid search weight configurations, and flags queries where the pipeline should admit uncertainty instead of generating from weak evidence.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running or delete them entirely. Your OpenSearch Serverless collection is billed at $0.24 per OCU-hour, and Bedrock API calls are billed on-demand per token.
Cost warning
Your OpenSearch Serverless collection incurs charges as long as it exists. With the default redundancy settings, the minimum is 2 OCUs (1 for indexing, 1 for search), costing approximately $0.48 per hour.
If you are done experimenting, delete the collection now to avoid ongoing charges. There is no pause option for OpenSearch Serverless.
Resources you used:
- Amazon OpenSearch Serverless collection (rag-pipeline).
- Amazon Bedrock Guardrail (rag-pipeline-guardrail).
- OpenSearch Serverless security policies (encryption, network, data access).
- Local project files (config.py, ingestion.py, opensearch_setup.py, retrieval.py, generation.py, pipeline.py, cleanup.py, evaluation.py, requirements.txt, evaluation_results.json).
Keep everything running
No action needed. Choose this if you are still actively experimenting with queries, ingesting more documents, or tuning retrieval weights.
- Your OpenSearch Serverless collection will continue to incur OCU charges (~$0.48/hour minimum with default redundancy).
- Bedrock API calls are billed only when you invoke models, so there is no idle cost for Bedrock.
- The Bedrock Guardrail has no ongoing cost while idle.
Pause - I'll come back to this later
OpenSearch Serverless does not support pausing a collection. Your options are to keep it running or delete it entirely.
If you want to reduce costs while preserving your code, delete the AWS resources (collection, policies, guardrail) and keep your local project files. You can recreate the infrastructure later by running pipeline.py again.
- Delete the AWS resources by running the cleanup script (see the Delete tab for the command).
- Keep your local project folder intact so you can redeploy later.
Delete - I don't want to use this again
Remove all project resources. This deletes your OpenSearch collection, security policies, Bedrock Guardrail, and local files.
Delete AWS resources
- Run the cleanup script to delete the OpenSearch Serverless collection, security policies, and Bedrock Guardrail:
python cleanup.pyWhat does this script do?
The cleanup.py script deletes the OpenSearch Serverless collection rag-pipeline, removes the encryption policy (rag-pipeline-enc), the network policy (rag-pipeline-net), the data access policy (rag-pipeline-access), and the Bedrock Guardrail.
- Verify the collection is deleting by checking the AWS Console. Navigate to the OpenSearch Service page, select **Serverless** in the left sidebar, then select **Collections**. The rag-pipeline collection should show a status of DELETING or no longer appear.
Delete local project files
- Remove your project directory by running the following command (replace the path with the location of your project folder):
macOS
cd ~/Desktop
rm -rf rag-pipelineWindows
cd ~\Desktop
rmdir /s /q rag-pipeline- If your project folder has a different name, substitute it in the command above.
Uninstall Python packages (optional)
If you installed the project dependencies globally (without a virtual environment), you can remove them:
pip uninstall boto3 opensearch-py requests-aws4auth sentence-transformers -yIf you used a virtual environment, deleting the project folder already removed it.
That's a Wrap!
That's a Wrap!
Nice work! You just built a production-grade RAG pipeline on AWS that goes far beyond basic vector search.
You've learned how to:
- Build a document ingestion pipeline with text extraction, semantic chunking, and Titan Text Embeddings V2 stored in an OpenSearch Serverless collection for hybrid search.
- Implement two-stage retrieval combining BM25 keyword search with k-NN vector similarity, plus cross-encoder reranking for precision.
- Add production Bedrock Guardrails with content filters and denied topics, plus resilience patterns with retry loops and fallback logic for safe generation.
- Secret Mission: Built a relevance evaluation loop that automatically scores retrieval quality and identifies optimal search weight configurations.
Ready to quiz yourself?