I tried using the vector search added to DynamoDB, vectorizing it with Streams and using it from Strands Agents

Introduction

Hello, I'm Kamino from the consulting department, a supermarket enthusiast.

On August 5, 2026, Amazon DynamoDB's native vector search feature became generally available (GA)!!

https://aws.amazon.com/blogs/aws/amazon-dynamodb-now-supports-real-time-vector-search-at-any-scale/

What naturally catches my attention is the use case for AI agent knowledge retrieval. Since Bedrock Knowledge Bases doesn't currently support DynamoDB as a data source, if you want an agent to search data stored in DynamoDB, you need to build the mechanism yourself.

So this time, I tried the following!

1. A pipeline that automatically vectorizes records upon registration using DynamoDB Streams
2. A Strands Agents tool that calls the native vector search (SearchVectors API)
3. Deploying that agent to Amazon Bedrock AgentCore Runtime

The infrastructure is built primarily with AWS CDK! The complete code for this article is available in the repository below, so please refer to it as needed.

https://github.com/yuu551/agentcore-strands-dynamo

Architecture

The architecture we are building this time is as follows.

On the write side, when an application registers a record to the DynamoDB table, DynamoDB Streams triggers a Lambda function, which vectorizes the text using Amazon Titan Text Embeddings V2. The vectorized result is saved to the same record as an attribute named embedding. The attribute name is arbitrary, and which attribute to target for vector search is specified during index creation (described later; this attribute is referred to as the embedding attribute going forward). Since the vector index is updated asynchronously, the write side doesn't need to be aware of vectorization at all.

On the search side, a Strands Agent on AgentCore Runtime vectorizes the query via a tool call, retrieves similar documents using the SearchVectors API, and generates a response.

Prerequisites

The environment and versions used are as follows.

Also, Claude Haiku 4.5 is used as the agent model.

Implementation

CDK Stack

First, let's define the entire infrastructure with CDK. The DynamoDB table, vectorization Lambda, and AgentCore Runtime are combined into a single stack.

import * as cdk from "aws-cdk-lib";
import {
  Duration,
  RemovalPolicy,
  Stack,
  StackProps,
  aws_bedrockagentcore as agentcore,
  aws_dynamodb as dynamodb,
  aws_ecr_assets as ecrAssets,
  aws_iam as iam,
  aws_lambda as lambda,
  aws_lambda_event_sources as eventSources,
} from "aws-cdk-lib";
import { Construct } from "constructs";

const EMBED_MODEL_ID = "amazon.titan-embed-text-v2:0";

class VectorSearchStack extends Stack {
  constructor(scope: Construct, id: string, props?: StackProps) {
    super(scope, id, props);

    const table = new dynamodb.Table(this, "KnowledgeTable", {
      tableName: "agent-knowledge",
      partitionKey: { name: "doc_id", type: dynamodb.AttributeType.STRING },
      billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
      stream: dynamodb.StreamViewType.NEW_AND_OLD_IMAGES,
      removalPolicy: RemovalPolicy.DESTROY,
    });

    const embedder = new lambda.Function(this, "EmbedderFunction", {
      functionName: "agent-knowledge-embedder",
      runtime: lambda.Runtime.PYTHON_3_13,
      handler: "embedder.handler",
      code: lambda.Code.fromAsset("cdk/lambda"),
      timeout: Duration.seconds(60),
      environment: {
        TABLE_NAME: table.tableName,
        EMBED_MODEL_ID: EMBED_MODEL_ID,
      },
    });

    table.grantWriteData(embedder);
    embedder.addToRolePolicy(
      new iam.PolicyStatement({
        actions: ["bedrock:InvokeModel"],
        resources: [
          `arn:aws:bedrock:${this.region}::foundation-model/${EMBED_MODEL_ID}`,
        ],
      }),
    );

    embedder.addEventSource(
      new eventSources.DynamoEventSource(table, {
        startingPosition: lambda.StartingPosition.LATEST,
        batchSize: 10,
        retryAttempts: 3,
      }),
    );

    // ---- AgentCore Runtime (Strands Agent) ----
    const runtime = new agentcore.Runtime(this, "KnowledgeAgentRuntime", {
      runtimeName: "knowledge_agent",
      agentRuntimeArtifact: agentcore.AgentRuntimeArtifact.fromAsset("agent", {
        platform: ecrAssets.Platform.LINUX_ARM64,
      }),
    });

    runtime.role.addToPrincipalPolicy(
      new iam.PolicyStatement({
        actions: ["bedrock:InvokeModel", "bedrock:InvokeModelWithResponseStream"],
        resources: [
          "arn:aws:bedrock:*::foundation-model/*",
          `arn:aws:bedrock:*:${this.account}:inference-profile/*`,
        ],
      }),
    );
    runtime.role.addToPrincipalPolicy(
      new iam.PolicyStatement({
        actions: ["dynamodb:SearchVectors"],
        resources: [table.tableArn, `${table.tableArn}/*`],
      }),
    );

    new cdk.CfnOutput(this, "TableName", { value: table.tableName });
    new cdk.CfnOutput(this, "AgentRuntimeArn", {
      value: runtime.agentRuntimeArn,
    });
  }
}

const app = new cdk.App();
new VectorSearchStack(app, "AgentKnowledgeVectorSearchStack", {
  env: { region: "ap-northeast-1" },
});
app.synth();

NEW_AND_OLD_IMAGES is specified for the table stream. This is a setting that includes not only the post-change record but also the pre-change record content in the events flowing into Streams. Both are needed for the Lambda described later to compare the content before and after the change to prevent loops. Also, since vector indexes are only supported for tables with on-demand capacity mode, billingMode is set to PAY_PER_REQUEST.

Auto-Vectorization Lambda

We implement a Lambda that is triggered by DynamoDB Streams, vectorizes the text attribute of the record, and saves it to the embedding attribute of the same record.

import json
import os

import boto3

TABLE_NAME = os.environ["TABLE_NAME"]
EMBED_MODEL_ID = os.environ["EMBED_MODEL_ID"]

dynamodb = boto3.client("dynamodb")
bedrock = boto3.client("bedrock-runtime")

def embed(text: str) -> list[float]:
    response = bedrock.invoke_model(
        modelId=EMBED_MODEL_ID,
        body=json.dumps({"inputText": text, "dimensions": 1024, "normalize": True}),
    )
    return json.loads(response["body"].read())["embedding"]

def handler(event, context):
    for record in event["Records"]:
        if record["eventName"] not in ("INSERT", "MODIFY"):
            continue

        new_image = record["dynamodb"].get("NewImage", {})
        old_image = record["dynamodb"].get("OldImage", {})
        text = new_image.get("text", {}).get("S")
        if not text:
            continue

        # Skip if re-triggered by this Lambda's own embedding write (infinite loop prevention)
        if "embedding" in new_image and old_image.get("text", {}).get("S") == text:
            continue

        embedding = embed(text)
        dynamodb.update_item(
            TableName=TABLE_NAME,
            Key={"doc_id": new_image["doc_id"]},
            UpdateExpression="SET embedding = :emb",
            ExpressionAttributeValues={
                ":emb": {"L": [{"N": str(v)} for v in embedding]}
            },
        )
        print(f"embedded: {new_image['doc_id']['S']} ({len(embedding)} dims)")

    return {"statusCode": 200}

One important note: when this Lambda itself writes the embedding via update_item, that write flows back into Streams and triggers the Lambda again. To handle this, if the post-change record (NewImage) already has an embedding and the text is the same as in the pre-change record (OldImage), it is determined to be a self-write and processing moves on without doing anything. Conversely, if text has changed, re-vectorization occurs, so the implementation also handles record updates.

The vector storage format is a list of numbers (L type containing N type). Rather than a dedicated vector type being added to DynamoDB, the idea is to store vectors as regular attributes and create an index on them.

Creating the Vector Index

Since CloudFormation is not supported as mentioned above, we add the vector index using the UpdateTable API.

import time

import boto3

TABLE_NAME = "agent-knowledge"
INDEX_NAME = "embedding-index"

dynamodb = boto3.client("dynamodb", region_name="ap-northeast-1")

def main():
    dynamodb.update_table(
        TableName=TABLE_NAME,
        AttributeDefinitions=[
            {"AttributeName": "category", "AttributeType": "S"}
        ],
        VectorIndexUpdates=[
            {
                "Create": {
                    "IndexName": INDEX_NAME,
                    "VectorAttribute": {"AttributeName": "embedding"},
                    "Dimensions": 1024,
                    "DistanceFunction": "COSINE",
                    "SearchSchema": [
                        {
                            "AttributeName": "category",
                            "SearchSchemaElementType": "INLINE_FILTER",
                        }
                    ],
                    "Projection": {"ProjectionType": "ALL"},
                }
            }
        ],
    )

    while True:
        table = dynamodb.describe_table(TableName=TABLE_NAME)["Table"]
        indexes = table.get("VectorIndexes", [])
        status = indexes[0]["IndexStatus"] if indexes else "NOT_FOUND"
        print(f"IndexStatus: {status}")
        if status == "ACTIVE":
            break
        time.sleep(10)

    print("Vector index creation complete")

if __name__ == "__main__":
    main()

The main parameters are as follows.

Attributes specified in SearchSchema require definition in AttributeDefinitions, similar to GSI.

There are two types of SearchSchema: INLINE_FILTER and HASH. The INLINE_FILTER we used this time is an attribute that can optionally be used as a filtering condition during search. The other type, HASH, divides the search space itself by the value of that attribute, and when searching, an equality condition on that attribute must be included in SearchConditionExpression. It can be used for scenarios like making a tenant ID a HASH to ensure data from different tenants never gets mixed in search results.

Note that Dimensions and DistanceFunction cannot be changed after index creation.
If you replace the embedding model, you'll need to recreate the index, so it's best to finalize the model selection ahead of time.

When executed, with around 6 records, it became ACTIVE in about 1 minute.

IndexStatus: CREATING
IndexStatus: CREATING
...
IndexStatus: ACTIVE
Vector index creation complete

Strands Agents Tool Implementation

On the agent side, we define a tool using the @tool decorator that vectorizes the query and calls the SearchVectors API.

import json
import sys

import boto3
from strands import Agent, tool
from strands.models import BedrockModel

REGION = "ap-northeast-1"
TABLE_NAME = "agent-knowledge"
INDEX_NAME = "embedding-index"
EMBED_MODEL_ID = "amazon.titan-embed-text-v2:0"

bedrock_runtime = boto3.client("bedrock-runtime", region_name=REGION)
dynamodb = boto3.client("dynamodb", region_name=REGION)

@tool
def search_knowledge(query: str) -> list[dict]:
    """Searches the internal knowledge base using vector search and returns relevant documents.

    Used to answer questions about internal regulations and procedures related to expenses, attendance, and IT.

    Args:
        query: Content to search for (natural language)
    """
    response = bedrock_runtime.invoke_model(
        modelId=EMBED_MODEL_ID,
        body=json.dumps({"inputText": query, "dimensions": 1024, "normalize": True}),
    )
    embedding = json.loads(response["body"].read())["embedding"]

    results = dynamodb.search_vectors(
        TableName=TABLE_NAME,
        IndexName=INDEX_NAME,
        SearchVector=[{"N": str(v)} for v in embedding],
        TopK=3,
    )
    return [
        {
            "title": r["Item"]["title"]["S"],
            "text": r["Item"]["text"]["S"],
            "category": r["Item"]["category"]["S"],
            "score": r["Score"],
        }
        for r in results["SearchResults"]
    ]

def create_agent() -> Agent:
    return Agent(
        model=BedrockModel(
            model_id="jp.anthropic.claude-haiku-4-5-20251001-v1:0",
            region_name=REGION,
        ),
        tools=[search_knowledge],
        system_prompt=(
            "You are an internal helpdesk assistant. "
            "For questions about internal regulations, always search using the search_knowledge tool "
            "and answer based on the search results. "
            "Do not speculate on information not found in the search results; say you don't know."
        ),
    )

if __name__ == "__main__":
    agent = create_agent()
    agent(sys.argv[1] if len(sys.argv) > 1 else "What is the expense reimbursement deadline?")

For the SearchVectors call, the search query is vectorized using the same embedding model and the same number of dimensions, then passed to SearchVector. You can specify the number of results with TopK, returning up to 100 results. The search results include Item (attributes specified in Projection) and Score, so there is no need to do a separate GetItem.

The entry point for AgentCore Runtime simply wraps this agent.

from agent import create_agent
from bedrock_agentcore.runtime import BedrockAgentCoreApp

app = BedrockAgentCoreApp()

@app.entrypoint
def invoke(payload, context):
    agent = create_agent()
    result = agent(payload.get("prompt", ""))
    return {"result": str(result)}

if __name__ == "__main__":
    app.run()

Here is the Dockerfile for the container image as well.

FROM public.ecr.aws/docker/library/python:3.13-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY agent.py main.py ./

EXPOSE 8080
CMD ["python", "main.py"]

strands-agents
bedrock-agentcore
boto3>=1.43.66

The requirements.txt specifies boto3 1.43.66 or later as a minimum. This is because if the boto3 in the container is old, search_vectors cannot be called.

Verification

Setup and Deployment

First, clone the repository and install the dependencies. CDK-related packages (aws-cdk-lib / tsx, etc.) are managed with pnpm, and Python script dependencies (boto3 / strands-agents) are managed with uv.

git clone https://github.com/yuu551/agentcore-strands-dynamo.git
cd agentcore-strands-dynamo
pnpm install
uv sync

Next, deploy with CDK.

npx cdk deploy

Once the deployment is complete, create the vector index.

uv run python scripts/create_vector_index.py

Next, seed the sample internal knowledge. Once seeded, the vectorization is handled automatically by Lambda via Streams.

uv run python scripts/seed_items.py

put: doc-001 Expense reimbursement deadline
put: doc-002 Business travel expense limits
put: doc-003 Remote work policy
put: doc-004 How to apply for paid leave
put: doc-005 Company PC replacement cycle
put: doc-006 How to connect to the company Wi-Fi

After waiting about 20 seconds, let's verify that the embedding has been added to each record. This is a script that scans the table and outputs whether the embedding attribute exists and the number of dimensions.

import boto3

TABLE_NAME = "agent-knowledge"

dynamodb = boto3.client("dynamodb", region_name="ap-northeast-1")

def main():
    paginator = dynamodb.get_paginator("scan")
    for page in paginator.paginate(
        TableName=TABLE_NAME, ProjectionExpression="doc_id, embedding"
    ):
        for item in sorted(page["Items"], key=lambda i: i["doc_id"]["S"]):
            emb = item.get("embedding", {}).get("L")
            status = f"present ({len(emb)} dimensions)" if emb else "absent"
            print(f"{item['doc_id']['S']}: embedding={status}")

if __name__ == "__main__":
    main()

uv run python scripts/check_embeddings.py

doc-001: embedding=present (1024 dimensions)
doc-002: embedding=present (1024 dimensions)
doc-003: embedding=present (1024 dimensions)
doc-004: embedding=present (1024 dimensions)
doc-005: embedding=present (1024 dimensions)
doc-006: embedding=present (1024 dimensions)

Just by inserting data into DynamoDB, it has been automatically vectorized!

Calling the SearchVectors API Directly

Before integrating it into the agent, let's check the search result tendencies. This is a script that vectorizes a query with Titan Text Embeddings V2, passes it to the SearchVectors API, and displays the top 2 results.

def embed(text: str) -> list[float]:
    response = bedrock_runtime.invoke_model(
        modelId=EMBED_MODEL_ID,
        body=json.dumps({"inputText": text, "dimensions": 1024, "normalize": True}),
    )
    return json.loads(response["body"].read())["embedding"]

for query in queries:
    vector = [{"N": str(v)} for v in embed(query)]
    results = dynamodb.search_vectors(
        TableName=TABLE_NAME,
        IndexName=INDEX_NAME,
        SearchVector=vector,
        TopK=2,
    )
    hits = [
        (round(r["Score"], 4), r["Item"]["title"]["S"])
        for r in results["SearchResults"]
    ]
    print(f"{query} -> {hits}")

uv run python scripts/search_test.py

When is the expense deadline? -> [(0.4065, 'Expense reimbursement deadline'), (0.9086, 'Business travel expense limits')]
How much is covered for hotel stays? -> [(0.6383, 'Business travel expense limits'), (0.913, 'Remote work policy')]
What should I do if my computer breaks? -> [(0.6795, 'Company PC replacement cycle'), (0.8679, 'Remote work policy')]

The intended document comes up first in each case!

When COSINE is specified, Score represents the cosine distance, where a smaller value indicates greater similarity. Please be careful, as setting a threshold with the intuition of a similarity score (where larger means more similar) would be the exact opposite.

Let's also try filtering by the category attribute specified in INLINE_FILTER. Just pass the condition to SearchConditionExpression.

vector = [{"N": str(v)} for v in embed("I want to connect to the company network")]
results = dynamodb.search_vectors(
    TableName=TABLE_NAME,
    IndexName=INDEX_NAME,
    SearchVector=vector,
    TopK=3,
    SearchConditionExpression="category = :c",
    ExpressionAttributeValues={":c": {"S": "IT"}},
)

--- category=IT filter ---
0.5044 How to connect to the company Wi-Fi
0.8222 Company PC replacement cycle

Vector search narrowed down to only IT category documents worked without issue.
It's convenient to be able to filter by attributes registered in records like this.

Asking Questions to the Agent on AgentCore Runtime

Let's invoke the deployed agent. The Runtime ARN is displayed in the CDK deployment output (AgentRuntimeArn), so it is passed as an argument.

import json
import sys

import boto3

client = boto3.client("bedrock-agentcore", region_name="ap-northeast-1")

def main():
    runtime_arn = sys.argv[1]
    prompt = sys.argv[2] if len(sys.argv) > 2 else "What is the expense reimbursement deadline?"

    response = client.invoke_agent_runtime(
        agentRuntimeArn=runtime_arn,
        qualifier="DEFAULT",
        payload=json.dumps({"prompt": prompt}),
    )
    print(json.loads(response["response"].read())["result"])

if __name__ == "__main__":
    main()

uv run python scripts/invoke_runtime.py \
  arn:aws:bedrock-agentcore:ap-northeast-1:xxxxxxxxxxxx:runtime/knowledge_agent-xxxxxxxxxx \
  "How much can I expense for a hotel on a business trip? Is it okay to take the Green Car on the Shinkansen?"

Based on the internal regulations, here is the answer.

## Business Travel Expenses

**Hotel Costs (Accommodation)**
- For domestic business trips, the upper limit is **¥12,000 per night**.

**Shinkansen Green Car**
- The Shinkansen is available up to **reserved ordinary car seating**.
- Use of the Green Car is not permitted.

Please note that for overseas business trips, prior approval from your department head is required.

It was able to respond based on the vector search results! It accurately retrieved both the hotel cost and Green Car information contained in a single document about business travel expense limits and answered correctly.

Also Verifying the Case of Adding Data Afterwards

Let's test whether the agent can answer using new knowledge immediately after it is inserted.

table.put_item(Item={
    "doc_id": "doc-007",
    "category": "Benefits",
    "title": "Book Purchase Assistance Program",
    "text": "Books related to work can be purchased at company expense up to ¥5,000 per month. Purchase requests are submitted through the General Affairs department form, and e-books are also eligible.",
})

15 seconds after insertion, asking with the earlier script...

uv run python scripts/invoke_runtime.py \
  arn:aws:bedrock-agentcore:ap-northeast-1:xxxxxxxxxxxx:runtime/knowledge_agent-xxxxxxxxxx \
  "I want to buy a technical book with company money, is there any program for that?"

Yes, there is!

There is a **Book Purchase Assistance Program**.

## Program Details
- **Upper limit**: Up to ¥5,000 per month at company expense
- **Eligible items**: Books related to work
- **How to apply**: Submit a request through the General Affairs department form
- **E-books**: Also eligible

Technical books qualify as work-related books, so please submit a purchase request through the General Affairs department form.

We successfully got a response using the freshly added knowledge! In this environment, it took about 15 seconds from insertion until it became searchable, including the DynamoDB Streams Lambda trigger → vectorization → index update (processing time varies as it is asynchronous).

Supplement (Split long documents before registering)

With Knowledge Bases, long documents are automatically split into searchable units during data ingestion, but since this architecture lacks that mechanism, you need to handle splitting yourself.

In this implementation, Lambda converts the entire text attribute into a single vector. This is sufficient for short texts like internal FAQs, but for longer documents, one of the following two approaches is needed.

1. Split before registration. The application side divides the document into smaller units and inserts each as a separate record (splitting doc_id like doc-001#chunk-0). Lambda can remain as-is, and search results are returned in those split units.
2. Split inside Lambda. Lambda receives text via Streams, splits it, and registers child records. The registration side requires no changes, but managing parent-child relationships becomes necessary, such as deleting old child records when the original document is updated.

Let's actually verify the former approach. I took an information security policy consisting of three sections and inserted it split by section.

document = {
    "doc_id": "doc-008",
    "category": "IT",
    "title": "情報セキュリティ規程",
    "chunks": [
        "パスワードは12文字以上で、英大文字・英小文字・数字・記号のうち3種類以上を...",
        "社外への資料送付は、機密区分がconfidential以上の場合、上長の承認と...",
        "セキュリティインシデントを発見した場合は、30分以内に情報システム部の...",
    ],
}

for i, chunk in enumerate(document["chunks"]):
    table.put_item(
        Item={
            "doc_id": f"{document['doc_id']}#chunk-{i}",
            "category": document["category"],
            "title": document["title"],
            "text": chunk,
        }
    )

uv run python scripts/seed_chunked_doc.py

put: doc-008#chunk-0 (97文字)
put: doc-008#chunk-1 (84文字)
put: doc-008#chunk-2 (90文字)

After waiting about 20 seconds, all three records had embeddings attached. Searching in this state, the record for the section corresponding to each question ranked first.

パスワードは何文字必要？ -> [(0.3302, 'doc-008#chunk-0'), (0.8678, 'doc-005')]
セキュリティ事故を見つけたらどこに連絡する？ -> [(0.5501, 'doc-008#chunk-2'), (0.7785, 'doc-006')]

Even when asking the agent a question spanning two sections — "Tell me the password rules. Also, who should I contact when I find a security incident?" — it successfully retrieved the content of both chunk-0 and chunk-2 and answered correctly!

Note that the input limit for Titan Text Embeddings V2 is 8,192 tokens, and exceeding this causes the embedding itself to fail. In actual operation, it would be worth thinking about chunking strategies as well...!
This is also an area where I'd hope DynamoDB will eventually support Knowledge Bases as a data source.

Supplement (With this approach, there is a time lag in vectorization)

In this architecture, vectorization is executed in the background at a different timing from record registration. Furthermore, reflection to the vector index is also asynchronous. As a result, in this environment, there was a time lag of about 15 seconds from registration until the record became discoverable via search. This duration varies depending on data volume and conditions.

This architecture cannot guarantee that data will be reflected in search immediately after registration. For example, it is not suitable for use cases like displaying data in search right after pressing a "Save" button. Please confirm in advance whether your requirements can tolerate the time lag between registration and search. If the time lag is unacceptable, one option is to perform vectorization synchronously within the registration process rather than leaving it to Streams!

Comparison with other vector search services

Finally, let me summarize how this compares to S3 Vectors and OpenSearch.

I feel that the appeal of vector search with DynamoDB is the ability to perform vector search together with where your business data already lives.

When considering cases where you want to vectorize DynamoDB data using OpenSearch or S3 Vectors, you would need to think about a mechanism to replicate DynamoDB data, as well as the metadata design and synchronization operations for the replication destination.
With this architecture, by adding an index to the table, there is no replication or synchronization operation to manage. Filtering during search can also use operational data attributes like category directly.

Additionally, specifying HASH in the schema allows you to separate the search space itself by the value of that attribute. This could be useful in multi-tenant applications, such as using a tenant ID as the HASH to prevent data from mixing across tenants in search results.

Compared to S3 Vectors, there may be cases where the lower latency is advantageous.

Honestly, I'm still thinking about exactly what cases this is particularly effective for...so I'll share again if I come up with a good idea!

Conclusion

Vector search was completed entirely within DynamoDB, and by combining it with Streams, vectorization was achieved without modifying the write side!
I felt that if it supports Knowledge Bases as a data source in the future, the range of use cases will expand further!

I hope this article has been helpful in some way. Thank you for reading to the end!