Skip to content

Search Strategies

Storage backends expose a search()search() method that delegates to a pluggable search strategy. A strategy implements the

SearchStrategySearchStrategy

protocol/interface: given a storage backend and a query string, it returns a ranked list of matching keys with relevance scores. Strategies that maintain an index use index()index() to update incrementally on each write; strategies that search on the fly implement it as a no-opstrategies that search on the fly omit it.

All shipped backends default to KeywordSearchStrategy. Pass a different strategy via the search_strategysearchStrategy parameter on any built-in backend to override it.

KeywordSearchStrategy

Section titled “KeywordSearchStrategy”

The default strategy shipped with both SDKs. Tokenizes the query into lowercased words, reads every key in storage, and scores each entry by the number of distinct query tokens that appear in the key and content. No index, embedding model, or external dependency required.

from strands.storage import LocalFileStorage
from strands.storage.search import KeywordSearchStrategy


strategy = KeywordSearchStrategy()
storage = LocalFileStorage("./my-data/")
results = await strategy.search(
    storage, "dark mode toggle"
)

Because it reads every entry on each search, keyword search is best suited for small-to-medium stores (up to a few thousand keys). For larger stores, use the BM25 strategy.

Section titled “BM25 search”

BM25 full-text search backed by a SQLite inverted index. Scores results by term frequency, inverse document frequency, and document length normalization, producing significantly better relevance ranking than keyword overlap for large stores. It works only with LocalFileStorage, because it keeps its index file on the host filesystem next to the storage directory. For the same reason it can’t be used with a sandboxed LocalFileStorage; use keyword search there, which works with every backend.

Bm25SearchStrategy uses the built-in sqlite3 FTS5 module, with no external dependencies. It indexes entries at write time and never re-scans storage, so only content written after the strategy is attached is searchable (there is no backfill).

from strands.storage import LocalFileStorage
from strands.storage.search import Bm25SearchStrategy


strategy = Bm25SearchStrategy()
storage = LocalFileStorage(
    "./memory/", search_strategy=strategy
)


await storage.write(
    "auth.md",
    b"OAuth2 authentication flow",
)
results = await storage.search("authentication")


await strategy.close()

The strategy persists a SQLite database alongside the storage directory. Call close()close() when the strategy is no longer needed to release the database connection.

Passing a strategy to a storage backend

Section titled “Passing a strategy to a storage backend”

Every built-in backend (InMemoryStorage, LocalFileStorage, S3Storage) accepts a search_strategysearchStrategy parameter. When set, writes automatically call the strategy’s index()index() method, and search()search() delegates to the strategy instead of the default keyword scan.

from strands.storage import LocalFileStorage
from strands.storage.search import Bm25SearchStrategy


storage = LocalFileStorage(
    "./memory/",
    search_strategy=Bm25SearchStrategy(),
)

Writing a custom strategy

Section titled “Writing a custom strategy”

To rank results your own way, implement

SearchStrategySearchStrategy

yourself. Its search()search() method receives the storage backend and the query, and returns

StorageSearchResultStorageSearchResult

entries (key, score, and optionally the stored bytes), best match first. Read entries through the storage passed in, using list()list() and read()read(), so the strategy works with every backend. When the search already read an entry’s bytes, include them in the result as data so callers don’t have to read the entry again.

If your strategy keeps its own index, such as a vector store or a full-text index, update it in index()index(). The backend calls it on every write with the key and the stored bytes.

The example below ranks entries by how often the exact query phrase appears.

SearchStrategy is a protocol, so any class with async index() and search() methods works without subclassing. index() is required; a strategy that searches on the fly implements it as a no-op.

from typing import Any


from strands.storage import InMemoryStorage, Storage
from strands.storage.search import StorageSearchResult




class PhraseSearchStrategy:
    """Rank entries by how often the exact query phrase appears."""


    async def index(
        self, storage: Storage, key: str, data: bytes, **kwargs: Any
    ) -> None:
        pass  # Searches on the fly, so there is nothing to index.


    async def search(
        self, storage: Storage, query: str, **kwargs: Any
    ) -> list[StorageSearchResult]:
        phrase = query.strip().lower()
        if not phrase:
            return []
        results = []
        for key in await storage.list(""):
            data = await storage.read(key)
            if data is None:
                continue
            text = data.decode("utf-8", errors="replace")
            count = text.lower().count(phrase)
            if count:
                results.append(
                    StorageSearchResult(key=key, score=count, data=data)
                )
        results.sort(key=lambda result: result.score, reverse=True)
        return results




storage = InMemoryStorage(search_strategy=PhraseSearchStrategy())

To pass a strategy to a sandboxed LocalFileStorage, declare requires_host_fs: Literal[False] = False on the class, as KeywordSearchStrategy does. The type checker rejects a sandboxed LocalFileStorage with a strategy that doesn’t declare it. If your strategy reads or writes the host filesystem directly, declare requires_host_fs: Literal[True] = True instead, as Bm25SearchStrategy does, and LocalFileStorage raises ValueError when a sandbox is set.