Amazon Bedrock Knowledge Bases: Building a Security-First, Production-Ready AI Agent using Lambda, OpenSearch, DynamoDB, CloudWatch and S3
This guide shows how to build a production-ready RAG system on AWS using Amazon Bedrock Knowledge Bases and Agents, with S3 for documents, DynamoDB for session state, and CloudWatch for observability. It focuses not just on setup, but on the key engineering decisions around vector stores, chunking, state management, and monitoring.
Most RAG stacks on AWS end up as the same five pieces wired together by hand: an S3 bucket for source documents, some chunking and embedding code, a vector store, a place to keep conversation state, and a pile of logging you bolt on after the first production incident. Amazon Bedrock's Knowledge Bases and Agents collapse a good chunk of that into managed services, but "managed" doesn't mean "no decisions to make." You still choose your vector store, your chunking strategy, your session storage, and how much you actually observe once it's live.
This is a hands-on guide to building the whole thing: an S3-backed Knowledge Base, a Bedrock Agent with a tool of its own, DynamoDB for session state, and CloudWatch wired in from the start instead of bolted on later. By the end you'll have a working pipeline and a clear sense of where the real engineering decisions are, not just which API calls to make.
The mental model first
Four pieces do most of the work:
- S3 bucket, the source of truth for your documents. Bedrock never mutates it. It reads, chunks, and embeds, but your original files stay put.
- Knowledge Base, the managed retrieval layer. It owns the ingestion pipeline (chunk, embed, index) and the query-time retrieval (vector search, optionally reranked). You point it at a vector store, either one it provisions for you or one you already run.
- Agent, the orchestration layer. It decides when to call the Knowledge Base, when to call a tool (an "action group" backed by a Lambda function), and how to compose a final answer. This is where multi-step reasoning happens, not inside the Knowledge Base.
- DynamoDB, doing a job Bedrock doesn't do for you: holding conversation state across turns. Bedrock Agents manage a session during a single interaction, but if you want durable, queryable conversation history, or per-user memory that survives a session timeout, that's on you to build, and DynamoDB is the natural fit given its latency profile and pay-per-request pricing.
Worth naming up front because it shapes a few decisions below: Bedrock recently added a fully managed Knowledge Base mode with native connectors (S3, SharePoint, Confluence, web crawler, Google Drive, OneDrive) and an agentic retriever for multi-step queries, on top of the original self-managed mode where you choose your own vector store. The self-managed path (which this guide uses, since it's what most production setups still run) gives you a choice between Aurora PostgreSQL Serverless, OpenSearch Serverless, Neptune Analytics, or the newer S3 Vectors option for the vector store. Each has a different cost and operational profile, and that choice matters more than most of the other configuration you'll do here.
Prerequisites
You'll need:
- An AWS account with Bedrock model access enabled for at least one embedding model (Titan Text Embeddings, for example) and one foundation model for generation.
- An IAM role that Bedrock will assume to read from S3 and write to your vector store. Don't reuse this role for anything else. Scope it tight, more on that later.
- boto3 installed and configured with credentials that can create Bedrock, S3, DynamoDB, and IAM resources.
1
pip install boto3
1
2
3
4
5
6
7
import boto3
region = "us-west-2"
s3 = boto3.client("s3", region_name=region)
bedrock_agent = boto3.client("bedrock-agent", region_name=region)
bedrock_runtime = boto3.client("bedrock-agent-runtime", region_name=region)
dynamodb = boto3.resource("dynamodb", region_name=region)
Step 1: get your documents into S3
Nothing exotic here, but two things matter for what comes later. Use a prefix structure you can filter on (
documents/product-docs/, documents/support-articles/), because inclusion prefixes are how you'll scope a data source later without standing up a separate bucket. And if this bucket will ever hold anything sensitive, turn on default encryption now, not after the Knowledge Base is live.1
2
3
4
5
6
7
8
9
10
11
12
bucket_name = "acme-kb-source-docs"
s3.create_bucket(
Bucket=bucket_name,
CreateBucketConfiguration={"LocationConstraint": region},
)
s3.put_bucket_encryption(
Bucket=bucket_name,
ServerSideEncryptionConfiguration={
"Rules": [{"ApplyServerSideEncryptionByDefault": {"SSEAlgorithm": "aws:kms"}}]
},
)
s3.upload_file("product-docs/rate-limits.pdf", bucket_name, "documents/product-docs/rate-limits.pdf")
Step 2: create the Knowledge Base
This is where the vector store choice lands. OpenSearch Serverless is the common default for teams that want a managed, scalable vector store without running Aurora themselves, so that's what's used here.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
kb_role_arn = "arn:aws:iam::123456789012:role/BedrockKnowledgeBaseRole"
kb_response = bedrock_agent.create_knowledge_base(
name="product-docs-kb",
description="Product documentation and support articles.",
roleArn=kb_role_arn,
knowledgeBaseConfiguration={
"type": "VECTOR",
"vectorKnowledgeBaseConfiguration": {
"embeddingModelArn": f"arn:aws:bedrock:{region}::foundation-model/amazon.titan-embed-text-v2:0",
},
},
storageConfiguration={
"type": "OPENSEARCH_SERVERLESS",
"opensearchServerlessConfiguration": {
"collectionArn": "arn:aws:aoss:us-west-2:123456789012:collection/abc123xyz",
"vectorIndexName": "product-docs-index",
"fieldMapping": {
"vectorField": "embedding",
"textField": "text",
"metadataField": "metadata",
},
},
},
)
kb_id = kb_response["knowledgeBase"]["knowledgeBaseId"]
Then wire in the S3 data source, scoped to the prefix from Step 1:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
ds_response = bedrock_agent.create_data_source(
knowledgeBaseId=kb_id,
name="product-docs-s3",
dataSourceConfiguration={
"type": "S3",
"s3Configuration": {
"bucketArn": f"arn:aws:s3:::{bucket_name}",
"inclusionPrefixes": ["documents/product-docs/"],
},
},
vectorIngestionConfiguration={
"chunkingConfiguration": {
"chunkingStrategy": "HIERARCHICAL",
"hierarchicalChunkingConfiguration": {
"levelConfigurations": [{"maxTokens": 1500}, {"maxTokens": 300}],
"overlapTokens": 60,
},
}
},
)
data_source_id = ds_response["dataSource"]["dataSourceId"]
Hierarchical chunking is worth defaulting to over a flat fixed-size split for most document sets. It keeps a large-chunk pass for coarse retrieval and a small-chunk pass for precision, and Bedrock handles reconciling the two at query time. If your content is already short and self-contained (FAQ entries, for instance), fixed-size or even
NONE (one chunk per file) will do less work for the same result.Step 3: run ingestion and wait for it to finish
Creating the data source doesn't ingest anything by itself. You have to kick off a job, and it's asynchronous.
1
2
3
4
5
6
7
8
9
10
11
12
13
import time
job = bedrock_agent.start_ingestion_job(knowledgeBaseId=kb_id, dataSourceId=data_source_id)
job_id = job["ingestionJob"]["ingestionJobId"]
while True:
status = bedrock_agent.get_ingestion_job(
knowledgeBaseId=kb_id, dataSourceId=data_source_id, ingestionJobId=job_id
)["ingestionJob"]["status"]
if status in ("COMPLETE", "FAILED"):
print(f"Ingestion finished: {status}")
break
time.sleep(15)
Query it directly before building anything else on top, so you know retrieval is actually working:
1
2
3
4
5
6
7
result = bedrock_runtime.retrieve(
knowledgeBaseId=kb_id,
retrievalQuery={"text": "What is the current rate limit for the export API?"},
retrievalConfiguration={"vectorSearchConfiguration": {"numberOfResults": 5}},
)
for r in result["retrievalResults"]:
print(r["score"], r["content"]["text"][:120])
Step 4: give the agent a tool and somewhere to keep state
A Knowledge Base alone gets you retrieval. An Agent gets you retrieval plus the ability to call your own code mid-conversation, an action group backed by Lambda. This is also where DynamoDB earns its place, since Bedrock's own session handling doesn't give you durable, queryable history once a session ends.
Create the table first:
1
2
3
4
5
6
7
8
9
10
11
12
13
table = dynamodb.create_table(
TableName="agent-sessions",
KeySchema=[
{"AttributeName": "session_id", "KeyType": "HASH"},
{"AttributeName": "turn_ts", "KeyType": "RANGE"},
],
AttributeDefinitions=[
{"AttributeName": "session_id", "AttributeType": "S"},
{"AttributeName": "turn_ts", "AttributeType": "N"},
],
BillingMode="PAY_PER_REQUEST",
)
table.wait_until_exists()
PAY_PER_REQUEST is the right default here. Conversation traffic is bursty and hard to forecast, and on-demand billing means you're not paying for provisioned capacity during the long stretches when nobody's chatting with the agent.Now create the agent itself and point it at the Knowledge Base:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
agent_role_arn = "arn:aws:iam::123456789012:role/BedrockAgentExecutionRole"
agent_response = bedrock_agent.create_agent(
agentName="support-agent",
agentResourceRoleArn=agent_role_arn,
foundationModel="anthropic.claude-sonnet-4-20250514-v1:0",
instruction=(
"You are a support assistant. Use the knowledge base to answer questions "
"about product documentation. Cite sources. If a question needs account-specific "
"data, call the lookup_account tool."
),
)
agent_id = agent_response["agent"]["agentId"]
bedrock_agent.associate_agent_knowledge_base(
agentId=agent_id,
agentVersion="DRAFT",
knowledgeBaseId=kb_id,
knowledgeBaseState="ENABLED",
description="Product documentation and support articles.",
)
The action group is where a Lambda function becomes a callable tool. Define its interface with a small OpenAPI schema and register it:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
bedrock_agent.create_agent_action_group(
agentId=agent_id,
agentVersion="DRAFT",
actionGroupName="account-lookup",
actionGroupExecutor={"lambda": "arn:aws:lambda:us-west-2:123456789012:function:lookup-account"},
apiSchema={
"payload": (
'{"openapi":"3.0.0","info":{"title":"Account Lookup","version":"1.0"},'
'"paths":{"/lookup_account":{"post":{"description":"Look up account status by ID",'
'"operationId":"lookupAccount","requestBody":{"required":true,"content":{"application/json":'
'{"schema":{"type":"object","properties":{"account_id":{"type":"string"}},"required":["account_id"]}}}}'
',"responses":{"200":{"description":"Account status"}}}}}}'
)
},
)
bedrock_agent.prepare_agent(agentId=agent_id)
Inside the Lambda, write each turn to DynamoDB yourself. Bedrock doesn't do this part for you:
1
2
3
4
5
6
7
8
9
10
import time, boto3
def handler(event, context):
table = boto3.resource("dynamodb").Table("agent-sessions")
table.put_item(Item={
"session_id": event["sessionId"],
"turn_ts": int(time.time() * 1000),
"account_id": event["requestBody"]["content"]["application/json"]["properties"][0]["value"],
})
# ... look up the account, return the result in the shape Bedrock expects
Step 5: call it
Once the agent is prepared, invoking it is a single streaming call:
1
2
3
4
5
6
7
8
9
10
response = bedrock_runtime.invoke_agent(
agentId=agent_id,
agentAliasId="TSTALIASID",
sessionId="session-4471",
inputText="What's the rate limit on the export API, and is account A-9931 in good standing?",
)
for event in response["completion"]:
if "chunk" in event:
print(event["chunk"]["bytes"].decode("utf-8"), end="")
The response streams back interleaved with trace events if you ask for them (
enableTrace=True on the call). Turn that on in anything past a local test. It's the equivalent of the planner activity trace you'd want from any retrieval system, showing which tool got called, what the Knowledge Base returned, and how the model reasoned across both before answering.Step 6: wire in CloudWatch from day one
This is the step that's easiest to skip and most expensive to skip. Two things to turn on before you call this done:
Model invocation logging, which captures every prompt and completion:
1
2
3
4
5
6
7
8
9
10
bedrock_client = boto3.client("bedrock", region_name=region)
bedrock_client.put_model_invocation_logging_configuration(
loggingConfig={
"cloudWatchConfig": {
"logGroupName": "/bedrock/model-invocations",
"roleArn": "arn:aws:iam::123456789012:role/BedrockLoggingRole",
},
"textDataDeliveryEnabled": True,
}
)
A CloudWatch alarm on throttling and error rates, so a quota problem shows up as a page instead of a support ticket three days later:
1
2
3
4
5
6
7
8
9
10
11
12
13
cloudwatch = boto3.client("cloudwatch", region_name=region)
cloudwatch.put_metric_alarm(
AlarmName="bedrock-agent-throttles",
Namespace="AWS/Bedrock",
MetricName="InvocationThrottles",
Dimensions=[{"Name": "ModelId", "Value": "anthropic.claude-sonnet-4-20250514-v1:0"}],
Statistic="Sum",
Period=300,
EvaluationPeriods=1,
Threshold=5,
ComparisonOperator="GreaterThanThreshold",
AlarmActions=["arn:aws:sns:us-west-2:123456789012:bedrock-alerts"],
)
Once logs are flowing, CloudWatch Logs Insights is where you'll actually debug a bad answer. A query like this surfaces every turn where the agent's confidence, however you're capturing it, dropped below a threshold, letting you pull the exact retrieval results that led to a weak answer instead of guessing:
1
2
3
4
5
fields @timestamp, input.text, output.text
| filter output.text like /I don't have enough information/
| sort @timestamp desc
| limit 50
Four auth surfaces on this stack
Same underlying problem as any managed retrieval system: getting the pipeline working and getting it safe to point at real data are two different milestones. On AWS the four surfaces look like this.
Control plane. Who can create, modify, or delete Knowledge Bases, agents, and data sources. This is ordinary IAM: scope a policy to
bedrock:CreateKnowledgeBase, bedrock:CreateAgent, and friends, attached to the people or CI roles that actually provision infrastructure, not to every engineer with account access.Source auth. How the Knowledge Base's execution role reaches into S3. Keep this role narrow:
s3:GetObject and s3:ListBucket on exactly the bucket and prefix the data source needs, nothing broader. If the bucket holds anything sensitive, add a KMS key and grant the role kms:Decrypt explicitly rather than relying on a bucket-wide default.Connection auth. How a caller reaches the agent at all. If you're fronting the agent with AgentCore Gateway, this is Inbound Auth, either IAM SigV4 for service-to-service calls or an OAuth/JWT bearer token (commonly via Cognito) for end-user callers. Pick one per Gateway. A runtime can't support both at once, so decide early whether this agent is meant to be called by other services or by people, since that shapes the whole auth setup downstream.
Content-level permissions, the one that actually protects your data. By default, nothing about a Knowledge Base query knows who's asking. If the S3 documents behind it have different sensitivity levels, your Knowledge Base will happily retrieve across all of them for anyone who can invoke the agent. AWS gives you two different mechanisms depending on where the data lives:
For DynamoDB, use IAM policy conditions with
dynamodb:LeadingKeys to restrict a caller to items whose partition key matches their own identity. This is the same row-level pattern AWS has documented for years for mobile apps with per-user data, and it applies just as well to per-user session history:1
2
3
4
5
6
7
8
9
10
{
"Effect": "Allow",
"Action": ["dynamodb:GetItem", "dynamodb:Query"],
"Resource": "arn:aws:dynamodb:us-west-2:123456789012:table/agent-sessions",
"Condition": {
"ForAllValues:StringEquals": {
"dynamodb:LeadingKeys": ["${aws:userid}"]
}
}
}
For the Gateway itself, this is what Outbound Auth's "caller IAM credentials" option is for. Instead of the Gateway always acting as its own service role, it can assume a role on behalf of the calling user through AWS's Federated Access Service and sign the outbound request with that identity. That's what lets a downstream S3 bucket policy or DynamoDB condition actually see who's asking, rather than seeing an undifferentiated service principal for every request.
A test worth running before this goes anywhere near production data: create two IAM principals with different scopes, run the same query through the agent as each one, and diff what comes back. If the more restricted principal ever gets a result the less restricted one's data alone should have produced, you've got a gap in either the Knowledge Base's data source scoping or the Gateway's outbound auth configuration, and it's worth making that diff a permanent test rather than a one-time check.
Production considerations before you commit
- Vector store choice is a cost decision, not just a technical one. OpenSearch Serverless bills for the OCUs it provisions even at low query volume. Aurora Serverless v2 scales to near-zero when idle but adds cold-start latency. S3 Vectors is the newest and cheapest option for workloads that can tolerate its current limits. Model your actual query volume before picking one, not just your data volume.
- DynamoDB billing mode should match your traffic shape.
PAY_PER_REQUESTfor anything bursty or new. Move to provisioned capacity with auto-scaling only once you have enough traffic history to set sensible targets, since guessing wrong on provisioned capacity costs you either money (over-provisioned) or throttled requests (under-provisioned). - IAM roles for Bedrock resources should be single-purpose. One role for the Knowledge Base's S3 and OpenSearch access, a separate role for the Agent's model invocation and action group execution, a separate role again for CloudWatch logging. It's tempting to reuse one broad role across all three. Resist it. When one of these roles eventually needs rotating or scoping down, single-purpose roles are the difference between a quick change and an incident.
- Turn on
enableTraceeverywhere except the lightest smoke tests. The trace is your only window into why the agent chose to call a tool, skip retrieval, or answer from parametric knowledge instead of grounding in the Knowledge Base. Without it, a wrong answer is unfalsifiable. You can't tell whether retrieval failed or the model just ignored good context. - Ingestion jobs fail silently if you don't check them.
start_ingestion_jobreturns immediately, and a malformed document or a permissions gap on one file will fail that file without failing the whole job. Checkget_ingestion_jobforfailureReasons, not just the top-level status, or you'll end up with a Knowledge Base that's quietly missing content nobody noticed was missing.
Where this leaves you
The pieces here aren't novel individually. S3 for storage, a managed vector index, an orchestration layer that can call tools, DynamoDB for state, CloudWatch for observability, this is a shape most AWS-native teams already know. What Bedrock changes is how much of the retrieval plumbing you no longer have to write yourself: chunking, embedding, and vector search collapse into a handful of API calls instead of a service you maintain. The tradeoff is that the parts Bedrock doesn't manage for you, session durability and content-level access control specifically, are exactly the parts that are easy to forget because nothing forces you to think about them until something goes wrong. Build DynamoDB and the four auth surfaces in from the start, not as a hardening pass after the demo works.
References
- Amazon Web Services. "Create a knowledge base by connecting to a data source in Amazon Bedrock Knowledge Bases." docs.aws.amazon.com/bedrock/latest/userguide/knowledge-base-create.html
- Amazon Web Services. "Introducing Amazon Bedrock Managed Knowledge Base for faster, more accurate enterprise AI applications." AWS Machine Learning Blog. aws.amazon.com/blogs/machine-learning/build-enterprise-search-for-agents-with-amazon-bedrock-managed-knowledge-base
- Amazon Web Services. "Authenticate and authorize with Inbound Auth and Outbound Auth." Amazon Bedrock AgentCore Developer Guide. docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-oauth.html
- Amazon Web Services. "Using IAM policy conditions for fine-grained access control." Amazon DynamoDB Developer Guide. docs.aws.amazon.com/amazondynamodb/latest/developerguide/specifying-conditions.html
- Amazon Web Services. "Fine-Grained Access Control for Amazon DynamoDB." AWS Blog. aws.amazon.com/blogs/aws/fine-grained-access-control-for-amazon-dynamodb
Enjoyed reading this content? Let the author know!
Your likes, comments, shares, and saves help creators reach more builders.
Loading recommendations
Loading article