跳转至

API reference

Generated from the apogee-ai-search source with mkdocstrings. Every symbol below is exported from apogee_ai_search, so it is part of the supported public surface.

Application · DTOs

BenchDTO dataclass

Python
BenchDTO(documents: int = 1000, queries: int = 100)

documents class-attribute instance-attribute

Python
documents: int = 1000

queries class-attribute instance-attribute

Python
queries: int = 100

IndexDTO dataclass

Python
IndexDTO(documents: tuple[dict, ...])

documents instance-attribute

Python
documents: tuple[dict, ...]

QueryDTO dataclass

Python
QueryDTO(text: str, k: int = 10, searcher: str = 'bm25')

text instance-attribute

Python
text: str

k class-attribute instance-attribute

Python
k: int = 10

searcher class-attribute instance-attribute

Python
searcher: str = 'bm25'

Application · Use cases

BenchSearchUseCase

Synthetic corpus + queries: measure indexing + search throughput.

execute async

Python
execute(documents: int, queries: int) -> dict[str, float]
Source code in apogee_ai_search/application/use_cases/bench_search_use_case.py
Python
async def execute(self, documents: int, queries: int) -> dict[str, float]:
    if documents <= 0 or queries <= 0:
        raise ValueError("documents and queries must be positive")
    rng = random.Random(42)

    index = InMemoryInvertedIndex()
    docs = []
    for i in range(documents):
        topic = _TOPICS[i % len(_TOPICS)]
        docs.append(SearchDocument(
            id=f"d-{i:05d}",
            text=f"Documento {i}: {topic}. " + ("blah " * 30),
        ))
    idx_start = time.perf_counter()
    await index.add(docs)
    idx_elapsed = (time.perf_counter() - idx_start) * 1000.0

    searcher = BM25Searcher(index)
    hit_count = 0
    q_start = time.perf_counter()
    for q in range(queries):
        topic = _TOPICS[q % len(_TOPICS)]
        words = topic.split()
        rng.shuffle(words)
        result = await searcher.search(SearchQuery(
            text=" ".join(words[: max(2, len(words) // 2)]),
            k=5,
        ))
        hit_count += len(result.hits)
    q_elapsed = (time.perf_counter() - q_start) * 1000.0

    return {
        "documents": float(documents),
        "queries": float(queries),
        "index_ms": idx_elapsed,
        "queries_ms": q_elapsed,
        "queries_per_second": (queries / q_elapsed * 1000.0) if q_elapsed > 0 else 0.0,
        "avg_hits": hit_count / queries,
    }

BulkIndexUseCase

Python
BulkIndexUseCase(index, batch_size: int = 500)

Indexes documents in batches.

Source code in apogee_ai_search/application/use_cases/bulk_index_use_case.py
Python
def __init__(self, index, batch_size: int = 500) -> None:
    if batch_size <= 0:
        raise ValueError("batch_size must be positive")
    self._index = index
    self._batch_size = batch_size

execute async

Python
execute(documents: Iterable[SearchDocument]) -> int
Source code in apogee_ai_search/application/use_cases/bulk_index_use_case.py
Python
async def execute(self, documents: Iterable[SearchDocument]) -> int:
    items = list(documents)
    for i in range(0, len(items), self._batch_size):
        batch = items[i : i + self._batch_size]
        await self._index.add(batch)
    return await self._index.count()

IndexDocumentsUseCase

Python
IndexDocumentsUseCase(index)
Source code in apogee_ai_search/application/use_cases/index_documents_use_case.py
Python
def __init__(self, index) -> None:
    self._index = index

execute async

Python
execute(documents: Iterable[SearchDocument]) -> int
Source code in apogee_ai_search/application/use_cases/index_documents_use_case.py
Python
async def execute(self, documents: Iterable[SearchDocument]) -> int:
    items = list(documents)
    await self._index.add(items)
    return await self._index.count()

SearchUseCase

Python
SearchUseCase(searcher)
Source code in apogee_ai_search/application/use_cases/search_use_case.py
Python
def __init__(self, searcher) -> None:
    self._searcher = searcher

execute async

Python
execute(query: SearchQuery) -> SearchResult
Source code in apogee_ai_search/application/use_cases/search_use_case.py
Python
async def execute(self, query: SearchQuery) -> SearchResult:
    return await self._searcher.search(query)

Domain

SearchDocument dataclass

Python
SearchDocument(id: str, text: str = '', vector: tuple[float, ...] = (), metadata: dict[str, str] = dict())

id instance-attribute

Python
id: str

text class-attribute instance-attribute

Python
text: str = ''

vector class-attribute instance-attribute

Python
vector: tuple[float, ...] = ()

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

SearchHit dataclass

Python
SearchHit(document: SearchDocument, score: float, metadata: dict[str, str] = dict())

document instance-attribute

Python
document: SearchDocument

score instance-attribute

Python
score: float

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

SearchQuery dataclass

Python
SearchQuery(text: str, k: int = 10, filters: dict[str, str] = dict(), vector: tuple[float, ...] = ())

text instance-attribute

Python
text: str

k class-attribute instance-attribute

Python
k: int = 10

filters class-attribute instance-attribute

Python
filters: dict[str, str] = field(default_factory=dict)

vector class-attribute instance-attribute

Python
vector: tuple[float, ...] = ()

SearchResult dataclass

Python
SearchResult(query: str, hits: tuple[SearchHit, ...] = (), searcher: str = 'bm25', latency_ms: float = 0.0, metadata: dict[str, str] = dict())

query instance-attribute

Python
query: str

hits class-attribute instance-attribute

Python
hits: tuple[SearchHit, ...] = ()

searcher class-attribute instance-attribute

Python
searcher: str = 'bm25'

latency_ms class-attribute instance-attribute

Python
latency_ms: float = 0.0

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

Domain · Enums

IndexKind

Bases: str, Enum

INVERTED class-attribute instance-attribute

Python
INVERTED = 'inverted'

VECTOR class-attribute instance-attribute

Python
VECTOR = 'vector'

ELASTICSEARCH class-attribute instance-attribute

Python
ELASTICSEARCH = 'elasticsearch'

OPENSEARCH class-attribute instance-attribute

Python
OPENSEARCH = 'opensearch'

SearcherKind

Bases: str, Enum

BM25 class-attribute instance-attribute

Python
BM25 = 'bm25'

VECTOR class-attribute instance-attribute

Python
VECTOR = 'vector'

HYBRID class-attribute instance-attribute

Python
HYBRID = 'hybrid'

Domain · Exceptions

IndexNotReadyError

Bases: SearchError

SearchError

Bases: Exception

Base for apogee-ai-search errors.

Domain · Protocols (ports)

IIndex

Bases: Protocol

name instance-attribute

Python
name: str

add async

Python
add(documents: Iterable[SearchDocument]) -> None
Source code in apogee_ai_search/domain/services/i_index.py
Python
async def add(self, documents: Iterable[SearchDocument]) -> None: ...

remove async

Python
remove(doc_id: str) -> None
Source code in apogee_ai_search/domain/services/i_index.py
Python
async def remove(self, doc_id: str) -> None: ...

clear async

Python
clear() -> None
Source code in apogee_ai_search/domain/services/i_index.py
Python
async def clear(self) -> None: ...

count async

Python
count() -> int
Source code in apogee_ai_search/domain/services/i_index.py
Python
async def count(self) -> int: ...

ISearcher

Bases: Protocol

name instance-attribute

Python
name: str

search async

Python
search(query: SearchQuery) -> SearchResult
Source code in apogee_ai_search/domain/services/i_searcher.py
Python
async def search(self, query: SearchQuery) -> SearchResult: ...

Infrastructure

BM25Searcher

Python
BM25Searcher(index: InMemoryInvertedIndex)
Source code in apogee_ai_search/infrastructure/searchers/bm25_searcher.py
Python
def __init__(self, index: InMemoryInvertedIndex) -> None:
    self._index = index

name class-attribute instance-attribute

Python
name = 'bm25'

search async

Python
search(query: SearchQuery) -> SearchResult
Source code in apogee_ai_search/infrastructure/searchers/bm25_searcher.py
Python
async def search(self, query: SearchQuery) -> SearchResult:
    start = time.perf_counter()
    q_tokens = tokenize(query.text)
    scored: list[SearchHit] = []
    for doc in self._index.documents():
        if query.filters and not all(
            doc.metadata.get(k) == v for k, v in query.filters.items()
        ):
            continue
        score = self._index.score_bm25(q_tokens, doc.id)
        if score > 0:
            scored.append(SearchHit(document=doc, score=score))
    scored.sort(key=lambda h: h.score, reverse=True)
    elapsed = (time.perf_counter() - start) * 1000.0
    return SearchResult(
        query=query.text,
        hits=tuple(scored[: query.k]),
        searcher=self.name,
        latency_ms=elapsed,
    )

HybridSearcher

Python
HybridSearcher(searchers: list, rrf_k: int = 60)

Reciprocal Rank Fusion across multiple searchers.

Source code in apogee_ai_search/infrastructure/searchers/hybrid_searcher.py
Python
def __init__(self, searchers: list, rrf_k: int = 60) -> None:
    if not searchers:
        raise ValueError("at least one searcher is required")
    self._searchers = searchers
    self._k = rrf_k

name class-attribute instance-attribute

Python
name = 'hybrid'

search async

Python
search(query: SearchQuery) -> SearchResult
Source code in apogee_ai_search/infrastructure/searchers/hybrid_searcher.py
Python
async def search(self, query: SearchQuery) -> SearchResult:
    start = time.perf_counter()
    scores: dict[str, float] = {}
    docs: dict[str, object] = {}
    contributions: dict[str, int] = {}

    for searcher in self._searchers:
        try:
            result = await searcher.search(query)
        except Exception:  # noqa: BLE001
            continue
        for rank, hit in enumerate(result.hits, 1):
            doc_id = hit.document.id
            scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (self._k + rank)
            docs.setdefault(doc_id, hit.document)
            contributions[doc_id] = contributions.get(doc_id, 0) + 1

    merged = sorted(
        (
            SearchHit(
                document=docs[doc_id],
                score=score,
                metadata={"contributors": str(contributions[doc_id])},
            )
            for doc_id, score in scores.items()
        ),
        key=lambda h: h.score,
        reverse=True,
    )
    elapsed = (time.perf_counter() - start) * 1000.0
    return SearchResult(
        query=query.text,
        hits=tuple(merged[: query.k]),
        searcher=self.name,
        latency_ms=elapsed,
        metadata={"searchers": str(len(self._searchers))},
    )

InMemoryInvertedIndex

Python
InMemoryInvertedIndex(k1: float = 1.5, b: float = 0.75)

In-memory inverted index supporting BM25 retrieval over text field.

Source code in apogee_ai_search/infrastructure/indexes/inverted_index.py
Python
def __init__(self, k1: float = 1.5, b: float = 0.75) -> None:
    self.k1 = k1
    self.b = b
    self._docs: dict[str, SearchDocument] = {}
    self._tokens: dict[str, list[str]] = {}
    self._doc_freq: Counter[str] = Counter()
    self._avg_dl: float = 0.0

name class-attribute instance-attribute

Python
name = 'inverted'

k1 instance-attribute

Python
k1 = k1

b instance-attribute

Python
b = b

add async

Python
add(documents: Iterable[SearchDocument]) -> None
Source code in apogee_ai_search/infrastructure/indexes/inverted_index.py
Python
async def add(self, documents: Iterable[SearchDocument]) -> None:
    for doc in documents:
        self._docs[doc.id] = doc
        tokens = tokenize(doc.text)
        self._tokens[doc.id] = tokens
    self._rebuild_stats()

remove async

Python
remove(doc_id: str) -> None
Source code in apogee_ai_search/infrastructure/indexes/inverted_index.py
Python
async def remove(self, doc_id: str) -> None:
    self._docs.pop(doc_id, None)
    self._tokens.pop(doc_id, None)
    self._rebuild_stats()

clear async

Python
clear() -> None
Source code in apogee_ai_search/infrastructure/indexes/inverted_index.py
Python
async def clear(self) -> None:
    self._docs.clear()
    self._tokens.clear()
    self._doc_freq.clear()
    self._avg_dl = 0.0

count async

Python
count() -> int
Source code in apogee_ai_search/infrastructure/indexes/inverted_index.py
Python
async def count(self) -> int:
    return len(self._docs)

score_bm25

Python
score_bm25(query_tokens: list[str], doc_id: str) -> float
Source code in apogee_ai_search/infrastructure/indexes/inverted_index.py
Python
def score_bm25(self, query_tokens: list[str], doc_id: str) -> float:
    tokens = self._tokens.get(doc_id, [])
    if not tokens:
        return 0.0
    tf = Counter(tokens)
    dl = len(tokens) or 1
    score = 0.0
    for term in query_tokens:
        if term not in tf:
            continue
        f = tf[term]
        idf = self._idf(term)
        num = f * (self.k1 + 1)
        den = f + self.k1 * (1 - self.b + self.b * dl / (self._avg_dl or 1))
        score += idf * num / den
    return score

documents

Python
documents() -> list[SearchDocument]
Source code in apogee_ai_search/infrastructure/indexes/inverted_index.py
Python
def documents(self) -> list[SearchDocument]:
    return list(self._docs.values())

get

Python
get(doc_id: str) -> SearchDocument | None
Source code in apogee_ai_search/infrastructure/indexes/inverted_index.py
Python
def get(self, doc_id: str) -> SearchDocument | None:
    return self._docs.get(doc_id)

InMemoryVectorIndex

Python
InMemoryVectorIndex()

Cosine search over an in-memory dict of documents with vector.

Source code in apogee_ai_search/infrastructure/indexes/vector_index.py
Python
def __init__(self) -> None:
    self._docs: dict[str, SearchDocument] = {}

name class-attribute instance-attribute

Python
name = 'vector'

add async

Python
add(documents: Iterable[SearchDocument]) -> None
Source code in apogee_ai_search/infrastructure/indexes/vector_index.py
Python
async def add(self, documents: Iterable[SearchDocument]) -> None:
    for doc in documents:
        if not doc.vector:
            raise ValueError(f"document {doc.id!r} requires a vector")
        self._docs[doc.id] = doc

remove async

Python
remove(doc_id: str) -> None
Source code in apogee_ai_search/infrastructure/indexes/vector_index.py
Python
async def remove(self, doc_id: str) -> None:
    self._docs.pop(doc_id, None)

clear async

Python
clear() -> None
Source code in apogee_ai_search/infrastructure/indexes/vector_index.py
Python
async def clear(self) -> None:
    self._docs.clear()

count async

Python
count() -> int
Source code in apogee_ai_search/infrastructure/indexes/vector_index.py
Python
async def count(self) -> int:
    return len(self._docs)

score

Python
score(query_vec, doc_id: str) -> float
Source code in apogee_ai_search/infrastructure/indexes/vector_index.py
Python
def score(self, query_vec, doc_id: str) -> float:
    doc = self._docs.get(doc_id)
    if doc is None:
        return 0.0
    return cosine(query_vec, doc.vector)

documents

Python
documents() -> list[SearchDocument]
Source code in apogee_ai_search/infrastructure/indexes/vector_index.py
Python
def documents(self) -> list[SearchDocument]:
    return list(self._docs.values())

get

Python
get(doc_id: str) -> SearchDocument | None
Source code in apogee_ai_search/infrastructure/indexes/vector_index.py
Python
def get(self, doc_id: str) -> SearchDocument | None:
    return self._docs.get(doc_id)

VectorSearcher

Python
VectorSearcher(index: InMemoryVectorIndex)
Source code in apogee_ai_search/infrastructure/searchers/vector_searcher.py
Python
def __init__(self, index: InMemoryVectorIndex) -> None:
    self._index = index

name class-attribute instance-attribute

Python
name = 'vector'

search async

Python
search(query: SearchQuery) -> SearchResult
Source code in apogee_ai_search/infrastructure/searchers/vector_searcher.py
Python
async def search(self, query: SearchQuery) -> SearchResult:
    if not query.vector:
        raise SearchError("vector searcher needs query.vector")
    start = time.perf_counter()
    scored: list[SearchHit] = []
    for doc in self._index.documents():
        if query.filters and not all(
            doc.metadata.get(k) == v for k, v in query.filters.items()
        ):
            continue
        score = self._index.score(query.vector, doc.id)
        scored.append(SearchHit(document=doc, score=score))
    scored.sort(key=lambda h: h.score, reverse=True)
    elapsed = (time.perf_counter() - start) * 1000.0
    return SearchResult(
        query=query.text,
        hits=tuple(scored[: query.k]),
        searcher=self.name,
        latency_ms=elapsed,
    )

cosine

Python
cosine(a, b) -> float
Source code in apogee_ai_search/infrastructure/indexes/vector_index.py
Python
def cosine(a, b) -> float:
    if len(a) != len(b):
        return 0.0
    num = sum(x * y for x, y in zip(a, b))
    da = math.sqrt(sum(x * x for x in a))
    db = math.sqrt(sum(y * y for y in b))
    if da == 0.0 or db == 0.0:
        return 0.0
    return num / (da * db)

tokenize

Python
tokenize(text: str) -> list[str]
Source code in apogee_ai_search/infrastructure/indexes/inverted_index.py
Python
def tokenize(text: str) -> list[str]:
    return [m.group(0).lower() for m in _TOKEN.finditer(text)]