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

Introduction

Hello, I'm Kamino from the Consulting Department, and I love supermarkets.

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 search. Since Bedrock Knowledge Bases does not 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 Lambda, which vectorizes the text using Amazon Titan Text Embeddings V2. The vectorized result is stored in the same record as an attribute named embedding. The attribute name is arbitrary, and which attribute to target for vector search is specified when creating the index described later (this attribute will be referred to as the embedding attribute hereafter). Since the vector index is updated asynchronously, the write side does not 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 an answer.

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 consolidated 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's stream. This setting includes not only the post-change record but also the pre-change record content in the events flowing through Streams. As described later in the Lambda, both are needed to compare the before and after content to prevent loops. Also, since vector indexes are only supported on on-demand capacity mode tables, billingMode is set to PAY_PER_REQUEST.

Auto-Vectorization Lambda

Let's implement a Lambda that is triggered by DynamoDB Streams, vectorizes the text attribute of a 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 triggered again by the embedding write itself (prevent infinite loop)
        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}

Note that when this Lambda itself writes the embedding via update_item, that write flows back through Streams and triggers Lambda again. Therefore, if the post-change record (NewImage) has an embedding and the text is the same as the pre-change record (OldImage), it is judged as the Lambda's own write and skipped to proceed to the next record. Conversely, if the text has been updated, it will be re-vectorized, so changes to records can also be tracked.

The vector storage format is a list of numbers (N type inside L type). A dedicated vector type hasn't been added to DynamoDB; rather, the idea is to store the vector as a regular attribute and create an index on it.

Creating the Vector Index

Since CloudFormation is not supported as mentioned earlier, the vector index is added 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 completed")

if __name__ == "__main__":
    main()

The main parameters are as follows.

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

There are two types of SearchSchema: INLINE_FILTER and HASH. The INLINE_FILTER used here is an attribute that can optionally be used as a filtering condition during search. The other, HASH, partitions the search space itself by the values of that attribute, and when searching, an equality condition for that attribute must always be included in SearchConditionExpression. A typical use case would be setting a tenant ID as HASH to ensure data from different tenants never gets mixed up in search results.

Note that Dimensions and DistanceFunction cannot be changed after index creation.
If you switch the embedding model, the index will need to be rebuilt, so it is best to finalize the model selection early.

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

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

Strands Agents Tool Implementation

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

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: The 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 about information not found in the search results; answer that you don't know."
        ),
    )

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

For the SearchVectors call, the search query is vectorized using the same embedding model and same number of dimensions, then passed to SearchVector. The number of results can be specified with TopK, returning up to 100 results. Since search results include Item (attributes specified in Projection) and Score, there is no need to perform 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

boto3 1.43.66 or later is specified as the minimum version in requirements.txt. This is because if the boto3 in the container is outdated, 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 packages (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 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 automatically handled by Lambda via Streams.

uv run python scripts/seed_items.py

put: doc-001 Expense report deadline
put: doc-002 Business trip expense limit
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 company Wi-Fi

After waiting about 20 seconds, let's verify that the embedding attribute has been added to each record. This is a script that scans the table and outputs whether the embedding attribute is present and its 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!

Directly Calling the SearchVectors API

Before incorporating 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 report deadline'), (0.9086, 'Business trip expense limit')]
How much does the hotel cover? -> [(0.6383, 'Business trip expense limit'), (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 in first place in all cases!

The Score is the cosine distance when COSINE is specified, and a smaller value indicates higher similarity. Be careful not to set a threshold with the intuition of a similarity score (where larger means more similar), as it would be the 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 company Wi-Fi
0.8222 Company PC replacement cycle

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

Asking the Agent on AgentCore Runtime a Question

Let's call the deployed agent. The Runtime ARN is displayed in the CDK deploy 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 "When is the expense report 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 ride in the Shinkansen Green Car?"

Based on our internal regulations, here is the answer.

## Business Trip Expenses

**Hotel costs (accommodation)**
- For domestic business trips, the upper limit is **12,000 yen per night**.

**Shinkansen Green Car**
- The Shinkansen may be used up to **ordinary reserved seats**.
- 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 answer based on the vector search results! It accurately retrieved and answered both the hotel cost and Green Car information from a single document about business trip expense limits.

Verifying the Case of Adding Data Afterwards

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

table.put_item(Item={
    "doc_id": "doc-007",
    "category": "Benefits",
    "title": "Book purchase subsidy program",
    "text": "Work-related books can be purchased at company expense up to 5,000 yen per month. Purchase requests are made through the General Affairs Department form, and e-books are also eligible.",
})

15 seconds after inserting, asking a question with the earlier script...

uv run python scripts/invoke_runtime.py \
  arn:aws:bedrock-agentcore:ap-northeast-1:xxxxxxxxxxxx:runtime/knowledge_agent-xxxxxxxxxx \
  "Is there any program to buy technical books with company money?"

Yes, there is!

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

## Program Details
- **Upper limit**: Up to 5,000 yen per month at company expense
- **Eligible**: Work-related books
- **How to apply**: Submit a purchase request through the General Affairs Department form
- **E-books**: Included as eligible

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

It successfully answered using the knowledge that was just added! In this environment, it took about 15 seconds from insertion to being searchable, including DynamoDB Streams triggering Lambda → vectorization → index update (the time required varies as it is asynchronous processing).

Supplement (Split long documents before registering)

With Knowledge Bases, long documents are automatically split into searchable units during data ingestion, but since this configuration lacks that mechanism, splitting must be done manually.

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

1. Split before registration. The application divides the document into small 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 within Lambda. Lambda splits the text received via Streams and registers them as child records. This requires no changes on the registration side, 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 tried inserting an information security policy consisting of three sections, split by section.

document = {
    "doc_id": "doc-008",
    "category": "IT",
    "title": "Information Security Policy",
    "chunks": [
        "Passwords must be at least 12 characters long and contain at least 3 of the following: uppercase letters, lowercase letters, numbers, and symbols...",
        "When sending materials outside the company, if the confidentiality classification is 'confidential' or above, approval from a supervisor and...",
        "If a security incident is discovered, contact the Information Systems Department within 30 minutes...",
    ],
}

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 characters)
put: doc-008#chunk-1 (84 characters)
put: doc-008#chunk-2 (90 characters)

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

How many characters are required for a password? -> [(0.3302, 'doc-008#chunk-0'), (0.8678, 'doc-005')]
Who should I contact if I find a security incident? -> [(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 correctly 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 will cause the embedding itself to fail. When operating in production, it would be worth thinking about a chunking strategy as well...!
This is also an area where I'd love to see DynamoDB eventually supported as a Knowledge Bases data source.

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

In this configuration, vectorization is executed in the background at a different time from record registration. Furthermore, reflection to the vector index is also asynchronous. Therefore, in this environment, there was about a 15-second lag from registration until the data became discoverable via search. This time may vary depending on data volume and conditions.

This configuration cannot guarantee that data will be immediately reflected in search right after registration. For example, it is not suited for use cases where you search for and display data immediately after pressing a "Save" button, so please confirm in advance whether your requirements can tolerate a time lag between registration and search. If the time lag is not acceptable, one option is to handle vectorization synchronously within the registration process itself, 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 felt that the appeal of vector search with DynamoDB is the ability to perform vector search together with the storage of business data.

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

Additionally, by specifying HASH in the schema, you can separate the search space itself by the value of that attribute. For multi-tenant applications, you could use the tenant ID as the HASH to prevent data from different tenants from mixing in search results.

Honestly, I'm still thinking about what cases this would be really effective for...so I'll share more ideas if I come up with any!

Conclusion

Vector search was completed entirely within DynamoDB, and by combining it with Streams, vectorization was achieved without touching the write side!
I feel that support as a Knowledge Bases data source in the future could broaden its applicability even further!

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