跳转至

API reference

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

Other

AdaptiveRagPipeline

Python
AdaptiveRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, classifier: Callable[[RagQuery], Complexity] | None = None, max_iterations: int = 3)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/adaptive/adaptive_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    classifier: Callable[[RagQuery], Complexity] | None = None,
    max_iterations: int = 3,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "AdaptiveRagPipeline requires chunker, embedder, store and generator"
        )
    if max_iterations < 1:
        raise PipelineConfigError("max_iterations must be >= 1")
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._classifier = classifier or heuristic_complexity
    self._max_iterations = max_iterations

name class-attribute instance-attribute

Python
name = 'adaptive'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/adaptive/adaptive_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/adaptive/adaptive_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    await self._emit(
        "query.received", query_id=query.id, text=query.text, pipeline_type=self.name,
    )
    complexity, retrieved = await self._route(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.sources = retrieved
    response.query_id = query.id
    response.traces.append({"pipeline": self.name, "complexity": complexity})
    await self._emit(
        "query.generated", query_id=query.id, complexity=complexity,
    )
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/adaptive/adaptive_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    _, retrieved = await self._route(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

AdvancedRagPipeline

Python
AdvancedRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, query_rewriter: IQueryRewriter | None = None, reranker: IReranker | None = None, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, dedupe_by_parent: bool = True)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/advanced/advanced_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913 — DI surface
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    query_rewriter: IQueryRewriter | None = None,
    reranker: IReranker | None = None,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    dedupe_by_parent: bool = True,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "AdvancedRagPipeline requires chunker, embedder, vector_store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._rewriter = query_rewriter or PassThroughRewriter()
    self._reranker = reranker or NoOpReranker()
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._dedupe = dedupe_by_parent

name class-attribute instance-attribute

Python
name = 'advanced'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/advanced/advanced_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/advanced/advanced_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.query_id = query.id
    await self._emit(
        "query.generated",
        query_id=query.id,
        tokens=response.usage.get("total_tokens", 0),
        finish_reason=response.finish_reason,
    )
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/advanced/advanced_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk
    await self._emit("query.generated", query_id=query.id, streamed=True)

AgenticRagPipeline

Python
AgenticRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, max_steps: int = 5, extra_tools: dict[str, Callable[[str], str]] | None = None)

Bases: ReActRagPipeline

Source code in apogee_ai_rag/infrastructure/pipelines/agentic/agentic_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    max_steps: int = 5,
    extra_tools: dict[str, Callable[[str], str]] | None = None,
) -> None:
    super().__init__(
        chunker=chunker,
        embedder=embedder,
        vector_store=vector_store,
        generator=generator,
        observability=observability,
        generation_config=generation_config,
        max_steps=max_steps,
    )
    self._extra_tools = extra_tools or {}

name class-attribute instance-attribute

Python
name = 'agentic'

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/agentic/agentic_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    response = await super().run(query)
    if self._extra_tools:
        response.traces.append({
            "pipeline": self.name,
            "extra_tools": sorted(self._extra_tools.keys()),
        })
    else:
        response.traces.append({"pipeline": self.name, "extra_tools": []})
    return response

ApogeeProvidersGenerator

Python
ApogeeProvidersGenerator(chat_provider: IChatCompletionProvider, default_model: str = 'gpt-4o-mini', system_prompt: str | None = None)

Bridges :class:IGenerator to apogee-ai-providers chat providers.

The chat provider implements IChatCompletionProvider from apogee_ai_providers and may be backed by any of the 14 LLMs supported by that package (OpenAI, Anthropic, Gemini, Bedrock, OpenRouter, …).

Source code in apogee_ai_rag/infrastructure/generators/apogee_providers_generator.py
Python
def __init__(
    self,
    chat_provider: IChatCompletionProvider,
    default_model: str = "gpt-4o-mini",
    system_prompt: str | None = None,
) -> None:
    self._chat = chat_provider
    self._default_model = default_model
    self._system_prompt = system_prompt or _DEFAULT_SYSTEM_PROMPT

name class-attribute instance-attribute

Python
name = 'apogee_providers'

generate async

Python
generate(prompt: str, context: list[RetrievedChunk], config: GenerationConfig | None = None) -> RagResponse
Source code in apogee_ai_rag/infrastructure/generators/apogee_providers_generator.py
Python
async def generate(
    self,
    prompt: str,
    context: list[RetrievedChunk],
    config: GenerationConfig | None = None,
) -> RagResponse:
    try:
        from apogee_ai_providers import ChatRequest
    except ImportError as exc:  # pragma: no cover - covered indirectly
        raise GenerationError(
            "apogee-ai-providers is required for ApogeeProvidersGenerator"
        ) from exc
    messages, model = self._build_messages(prompt, context, config)
    request = ChatRequest(
        model=model,
        messages=messages,
        temperature=(config.temperature if config else 0.2),
        max_tokens=(config.max_tokens if config else 1024),
    )
    try:
        chat_response = await self._chat.complete(request)
    except Exception as exc:  # pragma: no cover - real provider errors
        raise GenerationError("chat provider failure", stage="generate", cause=exc) from exc

    choice = chat_response.choices[0] if chat_response.choices else None
    answer = (choice.message.content if choice and choice.message else "") or ""
    finish = choice.finish_reason.value if choice and choice.finish_reason else None
    usage_obj = getattr(chat_response, "usage", None)
    usage: dict = {}
    if usage_obj is not None:
        usage = {
            "prompt_tokens": getattr(usage_obj, "prompt_tokens", 0),
            "completion_tokens": getattr(usage_obj, "completion_tokens", 0),
            "total_tokens": getattr(usage_obj, "total_tokens", 0),
        }
    return RagResponse(
        answer=answer,
        sources=context,
        finish_reason=finish,
        usage=usage,
    )

stream async

Python
stream(prompt: str, context: list[RetrievedChunk], config: GenerationConfig | None = None) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/generators/apogee_providers_generator.py
Python
async def stream(
    self,
    prompt: str,
    context: list[RetrievedChunk],
    config: GenerationConfig | None = None,
) -> AsyncIterator[RagChunk]:
    try:
        from apogee_ai_providers import ChatRequest
    except ImportError as exc:  # pragma: no cover
        raise GenerationError(
            "apogee-ai-providers is required for ApogeeProvidersGenerator"
        ) from exc
    messages, model = self._build_messages(prompt, context, config)
    request = ChatRequest(
        model=model,
        messages=messages,
        temperature=(config.temperature if config else 0.2),
        max_tokens=(config.max_tokens if config else 1024),
        stream=True,
    )
    try:
        async for chunk in self._chat.stream(request):
            yield RagChunk(
                delta=chunk.delta or "",
                sources=[],
                finish_reason=(chunk.finish_reason.value if chunk.finish_reason else None),
            )
    except Exception as exc:  # pragma: no cover
        raise GenerationError("chat provider stream failure", stage="stream", cause=exc) from exc
    yield RagChunk(delta="", sources=context, finish_reason="stop")

BM25Okapi dataclass

Python
BM25Okapi(k1: float = 1.5, b: float = 0.75, _docs: list[list[str]] = list(), _doc_lengths: list[int] = list(), _avg_dl: float = 0.0, _df: Counter[str] = Counter(), _idf: dict[str, float] = dict())

In-memory BM25 index over a collection of token lists.

k1 class-attribute instance-attribute

Python
k1: float = 1.5

b class-attribute instance-attribute

Python
b: float = 0.75

fit

Python
fit(corpus: list[list[str]]) -> None
Source code in apogee_ai_rag/infrastructure/search/bm25.py
Python
def fit(self, corpus: list[list[str]]) -> None:
    self._docs = list(corpus)
    self._doc_lengths = [len(doc) for doc in self._docs]
    self._avg_dl = (sum(self._doc_lengths) / len(self._docs)) if self._docs else 0.0
    self._df = Counter()
    for doc in self._docs:
        for term in set(doc):
            self._df[term] += 1
    n = len(self._docs)
    self._idf = {
        term: math.log(1.0 + (n - df + 0.5) / (df + 0.5))
        for term, df in self._df.items()
    }

score

Python
score(query_tokens: list[str], doc_index: int) -> float
Source code in apogee_ai_rag/infrastructure/search/bm25.py
Python
def score(self, query_tokens: list[str], doc_index: int) -> float:
    if not self._docs or doc_index >= len(self._docs):
        return 0.0
    doc = self._docs[doc_index]
    if not doc:
        return 0.0
    tf = Counter(doc)
    dl = self._doc_lengths[doc_index]
    norm = self.k1 * (1.0 - self.b + self.b * dl / self._avg_dl) if self._avg_dl else 0.0
    score = 0.0
    for term in query_tokens:
        if term not in self._idf:
            continue
        f = tf[term]
        if f == 0:
            continue
        score += self._idf[term] * (f * (self.k1 + 1.0)) / (f + norm)
    return score

search

Python
search(query_tokens: list[str], top_k: int = 10) -> list[tuple[int, float]]
Source code in apogee_ai_rag/infrastructure/search/bm25.py
Python
def search(self, query_tokens: list[str], top_k: int = 10) -> list[tuple[int, float]]:
    scored = [(i, self.score(query_tokens, i)) for i in range(len(self._docs))]
    scored.sort(key=lambda t: t[1], reverse=True)
    return scored[:top_k]

BM25Search

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

Convenience wrapper that owns a chunk index and returns RetrievedChunk.

Source code in apogee_ai_rag/infrastructure/search/bm25.py
Python
def __init__(self, k1: float = 1.5, b: float = 0.75) -> None:
    self._index = BM25Okapi(k1=k1, b=b)
    self._chunks: list[Chunk] = []

name class-attribute instance-attribute

Python
name = 'bm25'

fit

Python
fit(chunks: list[Chunk]) -> None
Source code in apogee_ai_rag/infrastructure/search/bm25.py
Python
def fit(self, chunks: list[Chunk]) -> None:
    self._chunks = list(chunks)
    self._index.fit([tokenize(c.text) for c in self._chunks])

search

Python
search(query: str, top_k: int = 10) -> list[RetrievedChunk]
Source code in apogee_ai_rag/infrastructure/search/bm25.py
Python
def search(self, query: str, top_k: int = 10) -> list[RetrievedChunk]:
    if not self._chunks:
        return []
    ranked = self._index.search(tokenize(query), top_k=top_k)
    return [
        RetrievedChunk(chunk=self._chunks[i], score=score, retriever="bm25")
        for i, score in ranked
        if score > 0.0
    ]

CAGPipeline

Python
CAGPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, cache_store: ICacheStore | None = None, ttl_seconds: int = 3600)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/cag/cag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    cache_store: ICacheStore | None = None,
    ttl_seconds: int = 3600,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "CAGPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._cache: ICacheStore = cache_store or InMemoryCacheStore()
    self._ttl = ttl_seconds

name class-attribute instance-attribute

Python
name = 'cag'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/cag/cag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    result = await self._ingest_default(job)
    # Indexing invalidates the cache so we never serve stale answers.
    await self._cache.invalidate("")
    return result

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/cag/cag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    key = _cache_key(query.text, retrieved)
    cached = await self._cache.get(key)
    if cached is not None:
        await self._emit("cag.hit", query_id=query.id, key=key)
        return RagResponse(
            answer=cached.value.decode("utf-8"),
            sources=retrieved,
            query_id=query.id,
            traces=[{"pipeline": self.name, "cache_hit": True}],
        )

    await self._emit("cag.miss", query_id=query.id, key=key)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.sources = retrieved
    response.query_id = query.id
    response.traces.append({"pipeline": self.name, "cache_hit": False})
    await self._cache.set(
        CacheEntry(
            key=key,
            value=response.answer.encode("utf-8"),
            ttl_seconds=self._ttl,
        ),
    )
    await self._emit("query.generated", query_id=query.id)
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/cag/cag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

CRagPipeline

Python
CRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, upper_threshold: float = 0.6, lower_threshold: float = 0.2, web_search_fn: WebSearchFn | None = None, max_web_results: int = 4)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/crag/crag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    upper_threshold: float = 0.6,
    lower_threshold: float = 0.2,
    web_search_fn: WebSearchFn | None = None,
    max_web_results: int = 4,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "CRagPipeline requires chunker, embedder, store and generator"
        )
    if not 0.0 <= lower_threshold <= upper_threshold <= 1.0:
        raise PipelineConfigError(
            "thresholds must satisfy 0 <= lower <= upper <= 1"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._upper = upper_threshold
    self._lower = lower_threshold
    self._web = web_search_fn
    self._max_web = max_web_results

name class-attribute instance-attribute

Python
name = 'crag'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/crag/crag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/crag/crag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved, verdict = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.sources = retrieved
    response.query_id = query.id
    response.traces.append({"pipeline": self.name, "verdict": verdict})
    await self._emit(
        "query.generated",
        query_id=query.id,
        verdict=verdict,
        finish_reason=response.finish_reason,
    )
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/crag/crag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved, _ = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

CacheConfig dataclass

Python
CacheConfig(enabled: bool = False, backend: str = 'in_memory', ttl_seconds: int = 3600, max_size: int = 1024)

enabled class-attribute instance-attribute

Python
enabled: bool = False

backend class-attribute instance-attribute

Python
backend: str = 'in_memory'

ttl_seconds class-attribute instance-attribute

Python
ttl_seconds: int = 3600

max_size class-attribute instance-attribute

Python
max_size: int = 1024

CacheEntry dataclass

Python
CacheEntry(key: str, value: bytes, ttl_seconds: int | None = None, metadata: dict = dict())

key instance-attribute

Python
key: str

value instance-attribute

Python
value: bytes

ttl_seconds class-attribute instance-attribute

Python
ttl_seconds: int | None = None

metadata class-attribute instance-attribute

Python
metadata: dict = field(default_factory=dict)

Chunk dataclass

Python
Chunk(text: str, parent_id: str, position: int = 0, id: str = _new_chunk_id(), metadata: dict = dict(), modality: Modality = TEXT, embedding: list[float] = list())

text instance-attribute

Python
text: str

parent_id instance-attribute

Python
parent_id: str

position class-attribute instance-attribute

Python
position: int = 0

id class-attribute instance-attribute

Python
id: str = field(default_factory=_new_chunk_id)

metadata class-attribute instance-attribute

Python
metadata: dict = field(default_factory=dict)

modality class-attribute instance-attribute

Python
modality: Modality = TEXT

embedding class-attribute instance-attribute

Python
embedding: list[float] = field(default_factory=list)

ChunkStrategy dataclass

Python
ChunkStrategy(strategy: ChunkingStrategy = RECURSIVE, chunk_size: int = 400, chunk_overlap: int = 40, separators: tuple[str, ...] = ('\n\n', '\n', '. ', ' ', ''), metadata: dict = dict())

strategy class-attribute instance-attribute

Python
strategy: ChunkingStrategy = RECURSIVE

chunk_size class-attribute instance-attribute

Python
chunk_size: int = 400

chunk_overlap class-attribute instance-attribute

Python
chunk_overlap: int = 40

separators class-attribute instance-attribute

Python
separators: tuple[str, ...] = ('\n\n', '\n', '. ', ' ', '')

metadata class-attribute instance-attribute

Python
metadata: dict = field(default_factory=dict)

ChunkingStrategy

Bases: StrEnum

FIXED class-attribute instance-attribute

Python
FIXED = 'fixed'

RECURSIVE class-attribute instance-attribute

Python
RECURSIVE = 'recursive'

SEMANTIC class-attribute instance-attribute

Python
SEMANTIC = 'semantic'

AGENTIC class-attribute instance-attribute

Python
AGENTIC = 'agentic'

MARKDOWN_AWARE class-attribute instance-attribute

Python
MARKDOWN_AWARE = 'markdown_aware'

CODE_AWARE class-attribute instance-attribute

Python
CODE_AWARE = 'code_aware'

LAYOUT_AWARE class-attribute instance-attribute

Python
LAYOUT_AWARE = 'layout_aware'

ContextualRagPipeline

Python
ContextualRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, max_context_chars: int = 200)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/contextual/contextual_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    max_context_chars: int = 200,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "ContextualRagPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._max_context = max_context_chars

name class-attribute instance-attribute

Python
name = 'contextual'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/contextual/contextual_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    await self._emit(
        "ingestion.started",
        job_id=job.id,
        n_documents=len(job.documents),
        chunker=self._chunker.name,
        embedder=self._embedder.model,
        store=self._store.name,
    )
    chunks = await self._chunker.chunk(job.documents)
    await self._emit("ingestion.chunked", job_id=job.id, n_chunks=len(chunks))
    if not chunks:
        return IngestionResult(
            job_id=job.id,
            documents_ingested=len(job.documents),
            chunks_produced=0,
            vectors_upserted=0,
        )
    contextualised: list[Chunk] = []
    for chunk in chunks:
        contextualised.append(await self._contextualise(chunk))
    await self._emit(
        "contextual.contextualised",
        job_id=job.id,
        n_chunks=sum(1 for c in contextualised if c.metadata.get("contextualised")),
    )
    embeddings = await self._embedder.embed([c.text for c in contextualised])
    for chunk, vec in zip(contextualised, embeddings, strict=False):
        chunk.embedding = vec
    upserted = await self._store.upsert(contextualised)
    await self._emit("ingestion.upserted", job_id=job.id, n_upserted=upserted)
    return IngestionResult(
        job_id=job.id,
        documents_ingested=len(job.documents),
        chunks_produced=len(contextualised),
        vectors_upserted=upserted,
    )

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/contextual/contextual_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.sources = retrieved
    response.query_id = query.id
    await self._emit("query.generated", query_id=query.id)
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/contextual/contextual_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

ConversationalRagPipeline

Python
ConversationalRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, history_window: int = 6)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/conversational/conversational_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    history_window: int = 6,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "ConversationalRagPipeline requires chunker, embedder, store and generator"
        )
    if history_window < 1:
        raise PipelineConfigError("history_window must be >= 1")
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._history_window = history_window

name class-attribute instance-attribute

Python
name = 'conversational'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/conversational/conversational_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/conversational/conversational_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved, expanded = await self._retrieve(query)
    response = await self._generator.generate(
        expanded, retrieved, self._generation_config,
    )
    response.sources = retrieved
    response.query_id = query.id
    response.traces.append({"pipeline": self.name, "history_turns": len(query.history)})
    await self._emit("query.generated", query_id=query.id)
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/conversational/conversational_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved, expanded = await self._retrieve(query)
    async for chunk in self._generator.stream(
        expanded, retrieved, self._generation_config,
    ):
        yield chunk

DiskCacheStore

Python
DiskCacheStore(*, directory: str | Path)
Source code in apogee_ai_rag/infrastructure/cache_stores/disk_cache_store.py
Python
def __init__(self, *, directory: str | Path) -> None:
    self._dir = Path(directory)
    self._dir.mkdir(parents=True, exist_ok=True)

name class-attribute instance-attribute

Python
name = 'disk'

get async

Python
get(key: str) -> CacheEntry | None
Source code in apogee_ai_rag/infrastructure/cache_stores/disk_cache_store.py
Python
async def get(self, key: str) -> CacheEntry | None:
    path = self._path(key)
    if not path.is_file():
        return None
    meta = self._meta(key)
    ttl = None
    if meta.is_file():
        try:
            ttl = float(meta.read_text(encoding="utf-8").strip())
        except ValueError:
            ttl = None
    if ttl is not None and ttl < time.time():
        for p in (path, meta, self._key_file(key)):
            await asyncio.to_thread(p.unlink, missing_ok=True)
        return None
    value = await asyncio.to_thread(path.read_bytes)
    return CacheEntry(key=key, value=value, ttl_seconds=None)

set async

Python
set(entry: CacheEntry) -> None
Source code in apogee_ai_rag/infrastructure/cache_stores/disk_cache_store.py
Python
async def set(self, entry: CacheEntry) -> None:
    await asyncio.to_thread(self._path(entry.key).write_bytes, entry.value)
    await asyncio.to_thread(
        self._key_file(entry.key).write_text, entry.key, encoding="utf-8",
    )
    if entry.ttl_seconds is not None:
        await asyncio.to_thread(
            self._meta(entry.key).write_text,
            str(time.time() + entry.ttl_seconds),
            encoding="utf-8",
        )

invalidate async

Python
invalidate(prefix: str) -> int
Source code in apogee_ai_rag/infrastructure/cache_stores/disk_cache_store.py
Python
async def invalidate(self, prefix: str) -> int:
    removed = 0
    for kfile in list(self._dir.glob("*.key")):
        try:
            stored_key = kfile.read_text(encoding="utf-8")
        except OSError:
            continue
        if not stored_key.startswith(prefix):
            continue
        stem = kfile.stem
        for suffix in (".bin", ".meta", ".key"):
            p = self._dir / f"{stem}{suffix}"
            await asyncio.to_thread(p.unlink, missing_ok=True)
        removed += 1
    return removed

Document dataclass

Python
Document(text: str = '', metadata: dict = dict(), id: str = _new_doc_id(), modality: Modality = TEXT, image_b64: str | None = None, audio_url: str | None = None, video_url: str | None = None, embedding: list[float] = list())

text class-attribute instance-attribute

Python
text: str = ''

metadata class-attribute instance-attribute

Python
metadata: dict = field(default_factory=dict)

id class-attribute instance-attribute

Python
id: str = field(default_factory=_new_doc_id)

modality class-attribute instance-attribute

Python
modality: Modality = TEXT

image_b64 class-attribute instance-attribute

Python
image_b64: str | None = None

audio_url class-attribute instance-attribute

Python
audio_url: str | None = None

video_url class-attribute instance-attribute

Python
video_url: str | None = None

embedding class-attribute instance-attribute

Python
embedding: list[float] = field(default_factory=list)

EchoGenerator

Deterministic generator that mirrors the prompt and cites sources.

Used by the smoke E2E test (no API key needed) and as a sane default for pipelines created without an explicit IGenerator.

name class-attribute instance-attribute

Python
name = 'echo'

generate async

Python
generate(prompt: str, context: list[RetrievedChunk], config: GenerationConfig | None = None) -> RagResponse
Source code in apogee_ai_rag/infrastructure/generators/echo_generator.py
Python
async def generate(
    self,
    prompt: str,
    context: list[RetrievedChunk],
    config: GenerationConfig | None = None,
) -> RagResponse:
    bullets = "\n".join(
        f"- [{i + 1}] {c.chunk.text}" for i, c in enumerate(context)
    )
    body = (
        f"Question: {prompt}\n\n"
        f"Context:\n{bullets}" if bullets else f"Question: {prompt}\n\n(no context)"
    )
    return RagResponse(
        answer=body,
        sources=context,
        confidence=1.0 if context else 0.0,
        finish_reason="stop",
        usage={"prompt_chars": len(prompt), "context_chunks": len(context)},
    )

stream async

Python
stream(prompt: str, context: list[RetrievedChunk], config: GenerationConfig | None = None) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/generators/echo_generator.py
Python
async def stream(
    self,
    prompt: str,
    context: list[RetrievedChunk],
    config: GenerationConfig | None = None,
) -> AsyncIterator[RagChunk]:
    response = await self.generate(prompt, context, config)
    for word in response.answer.split(" "):
        yield RagChunk(delta=word + " ", sources=[])
    yield RagChunk(delta="", sources=response.sources, finish_reason="stop")

EmbedderProvider

Bases: StrEnum

HASHING class-attribute instance-attribute

Python
HASHING = 'hashing'

OPENAI class-attribute instance-attribute

Python
OPENAI = 'openai'

COHERE class-attribute instance-attribute

Python
COHERE = 'cohere'

VOYAGE class-attribute instance-attribute

Python
VOYAGE = 'voyage'

SENTENCE_TRANSFORMERS class-attribute instance-attribute

Python
SENTENCE_TRANSFORMERS = 'sentence_transformers'

HUGGINGFACE class-attribute instance-attribute

Python
HUGGINGFACE = 'huggingface'

BGE class-attribute instance-attribute

Python
BGE = 'bge'

E5 class-attribute instance-attribute

Python
E5 = 'e5'

NOMIC class-attribute instance-attribute

Python
NOMIC = 'nomic'

JINA class-attribute instance-attribute

Python
JINA = 'jina'

INSTRUCTOR class-attribute instance-attribute

Python
INSTRUCTOR = 'instructor'

GTE class-attribute instance-attribute

Python
GTE = 'gte'

EmbeddingConfig dataclass

Python
EmbeddingConfig(provider: EmbedderProvider = HASHING, model: str = 'hashing-128', dims: int = 128, batch_size: int = 32)

provider class-attribute instance-attribute

Python
provider: EmbedderProvider = HASHING

model class-attribute instance-attribute

Python
model: str = 'hashing-128'

dims class-attribute instance-attribute

Python
dims: int = 128

batch_size class-attribute instance-attribute

Python
batch_size: int = 32

EvalCase dataclass

Python
EvalCase(question: str, ground_truth: str | None = None, expected_substrings: list[str] = list(), contexts: list[str] = list(), metadata: dict = dict())

question instance-attribute

Python
question: str

ground_truth class-attribute instance-attribute

Python
ground_truth: str | None = None

expected_substrings class-attribute instance-attribute

Python
expected_substrings: list[str] = field(default_factory=list)

contexts class-attribute instance-attribute

Python
contexts: list[str] = field(default_factory=list)

metadata class-attribute instance-attribute

Python
metadata: dict = field(default_factory=dict)

EvalConfig dataclass

Python
EvalConfig(evaluator: Evaluator = RAGAS, metrics: tuple[str, ...] = ('faithfulness', 'answer_relevancy', 'context_precision', 'context_recall'), threshold: float = 0.7, options: dict = dict())

evaluator class-attribute instance-attribute

Python
evaluator: Evaluator = RAGAS

metrics class-attribute instance-attribute

Python
metrics: tuple[str, ...] = ('faithfulness', 'answer_relevancy', 'context_precision', 'context_recall')

threshold class-attribute instance-attribute

Python
threshold: float = 0.7

options class-attribute instance-attribute

Python
options: dict = field(default_factory=dict)

EvalReport dataclass

Python
EvalReport(metrics: dict[str, float] = dict(), per_case: list[EvalResult] = list(), passed: int = 0, failed: int = 0)

metrics class-attribute instance-attribute

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

per_case class-attribute instance-attribute

Python
per_case: list[EvalResult] = field(default_factory=list)

passed class-attribute instance-attribute

Python
passed: int = 0

failed class-attribute instance-attribute

Python
failed: int = 0

total property

Python
total: int

score property

Python
score: float

EvalResult dataclass

Python
EvalResult(case: EvalCase, answer: str, metrics: dict = dict(), passed: bool = False, error: str | None = None)

case instance-attribute

Python
case: EvalCase

answer instance-attribute

Python
answer: str

metrics class-attribute instance-attribute

Python
metrics: dict = field(default_factory=dict)

passed class-attribute instance-attribute

Python
passed: bool = False

error class-attribute instance-attribute

Python
error: str | None = None

Evaluator

Bases: StrEnum

RAGAS class-attribute instance-attribute

Python
RAGAS = 'ragas'

TRULENS class-attribute instance-attribute

Python
TRULENS = 'trulens'

DEEPEVAL class-attribute instance-attribute

Python
DEEPEVAL = 'deepeval'

ARES class-attribute instance-attribute

Python
ARES = 'ares'

PHOENIX class-attribute instance-attribute

Python
PHOENIX = 'phoenix'

GISKARD class-attribute instance-attribute

Python
GISKARD = 'giskard'

LANGSMITH class-attribute instance-attribute

Python
LANGSMITH = 'langsmith'

LANGFUSE class-attribute instance-attribute

Python
LANGFUSE = 'langfuse'

PROMPTFOO class-attribute instance-attribute

Python
PROMPTFOO = 'promptfoo'

CONTINUOUS_EVAL class-attribute instance-attribute

Python
CONTINUOUS_EVAL = 'continuous_eval'

FLAREPipeline

Python
FLAREPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, max_re_retrievals: int = 2, min_segment_chars: int = 32)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/flare/flare_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    max_re_retrievals: int = 2,
    min_segment_chars: int = 32,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "FLAREPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._max_re = max_re_retrievals
    self._min_segment = min_segment_chars

name class-attribute instance-attribute

Python
name = 'flare'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/flare/flare_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/flare/flare_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    await self._emit(
        "query.received", query_id=query.id, text=query.text, pipeline_type=self.name,
    )
    retrieved = await self._retrieve(query.text, query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )

    triggers = 0
    while (
        triggers < self._max_re
        and _looks_uncertain(response.answer, self._min_segment)
    ):
        triggers += 1
        await self._emit(
            "flare.uncertain", query_id=query.id, attempt=triggers,
        )
        extra = await self._retrieve(
            f"{query.text} {response.answer}", query,
        )
        seen = {r.chunk.id for r in retrieved}
        retrieved.extend(r for r in extra if r.chunk.id not in seen)
        response = await self._generator.generate(
            query.text, retrieved, self._generation_config,
        )

    response.sources = retrieved
    response.query_id = query.id
    response.traces.append({"pipeline": self.name, "re_retrievals": triggers})
    await self._emit(
        "query.generated", query_id=query.id, re_retrievals=triggers,
    )
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/flare/flare_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    response = await self.run(query)
    async for chunk in self._generator.stream(
        query.text, response.sources, self._generation_config,
    ):
        yield chunk

FeedbackInput dataclass

Python
FeedbackInput(query_id: str, score: float, comment: str | None = None, metadata: dict | None = None)

query_id instance-attribute

Python
query_id: str

score instance-attribute

Python
score: float

comment class-attribute instance-attribute

Python
comment: str | None = None

metadata class-attribute instance-attribute

Python
metadata: dict | None = None

Framework

Bases: StrEnum

NATIVE class-attribute instance-attribute

Python
NATIVE = 'native'

LANGCHAIN class-attribute instance-attribute

Python
LANGCHAIN = 'langchain'

LLAMAINDEX class-attribute instance-attribute

Python
LLAMAINDEX = 'llamaindex'

HAYSTACK class-attribute instance-attribute

Python
HAYSTACK = 'haystack'

DSPY class-attribute instance-attribute

Python
DSPY = 'dspy'

LANGGRAPH class-attribute instance-attribute

Python
LANGGRAPH = 'langgraph'

LLAMA_STACK class-attribute instance-attribute

Python
LLAMA_STACK = 'llama_stack'

SEMANTIC_KERNEL class-attribute instance-attribute

Python
SEMANTIC_KERNEL = 'semantic_kernel'

TXTAI class-attribute instance-attribute

Python
TXTAI = 'txtai'

EMBEDCHAIN class-attribute instance-attribute

Python
EMBEDCHAIN = 'embedchain'

RAGFLOW class-attribute instance-attribute

Python
RAGFLOW = 'ragflow'

VERBA class-attribute instance-attribute

Python
VERBA = 'verba'

COGNITA class-attribute instance-attribute

Python
COGNITA = 'cognita'

CANOPY class-attribute instance-attribute

Python
CANOPY = 'canopy'

FLASHRAG class-attribute instance-attribute

Python
FLASHRAG = 'flashrag'

AUTORAG class-attribute instance-attribute

Python
AUTORAG = 'autorag'

R2R class-attribute instance-attribute

Python
R2R = 'r2r'

LIGHTRAG class-attribute instance-attribute

Python
LIGHTRAG = 'lightrag'

GRAPHRAG_MS class-attribute instance-attribute

Python
GRAPHRAG_MS = 'graphrag_ms'

NEMO_RETRIEVER class-attribute instance-attribute

Python
NEMO_RETRIEVER = 'nemo_retriever'

GenerationConfig dataclass

Python
GenerationConfig(model: str = 'gpt-4o-mini', temperature: float = 0.2, max_tokens: int | None = 1024, system_prompt: str | None = None, stream: bool = False)

model class-attribute instance-attribute

Python
model: str = 'gpt-4o-mini'

temperature class-attribute instance-attribute

Python
temperature: float = 0.2

max_tokens class-attribute instance-attribute

Python
max_tokens: int | None = 1024

system_prompt class-attribute instance-attribute

Python
system_prompt: str | None = None

stream class-attribute instance-attribute

Python
stream: bool = False

GraphConfig dataclass

Python
GraphConfig(enable_communities: bool = True, community_algorithm: str = 'leiden', max_depth: int = 2, entity_extractor: str = 'llm')

enable_communities class-attribute instance-attribute

Python
enable_communities: bool = True

community_algorithm class-attribute instance-attribute

Python
community_algorithm: str = 'leiden'

max_depth class-attribute instance-attribute

Python
max_depth: int = 2

entity_extractor class-attribute instance-attribute

Python
entity_extractor: str = 'llm'

GraphRagPipeline

Python
GraphRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, graph_store: IGraphStore | None = None, traversal_depth: int = 1)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/graph/graph_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    graph_store: IGraphStore | None = None,
    traversal_depth: int = 1,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "GraphRagPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._graph: IGraphStore = graph_store or NetworkXGraphStore()
    self._depth = traversal_depth
    self._community_summaries: list[tuple[set[str], str]] = []

name class-attribute instance-attribute

Python
name = 'graph'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/graph/graph_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    result = await self._ingest_default(job)
    chunks = list(getattr(self._store, "_chunks", {}).values())  # type: ignore[attr-defined]

    # Build the entity graph.
    for chunk in chunks:
        entities = _extract_entities(chunk.text)
        for entity in entities:
            await self._graph.add_node(
                KnowledgeNode(label=entity, properties={"chunk_id": chunk.id}),
            )
        for i, src in enumerate(entities):
            for dst in entities[i + 1 :]:
                await self._graph.add_edge(
                    KnowledgeEdge(source_id=src, target_id=dst, label="co_occurs"),
                )
    # Cache one short summary per community.
    communities: list[list[str]] = getattr(self._graph, "communities", lambda: [])()
    self._community_summaries = []
    for component in communities:
        tokens = Counter()
        for entity in component:
            for chunk in chunks:
                if entity in chunk.text:
                    tokens.update(chunk.text.lower().split())
        top = " ".join(t for t, _ in tokens.most_common(10))
        self._community_summaries.append((set(component), top))
    all_nodes_fn = getattr(self._graph, "all_nodes", None)
    n_entities = 0
    if callable(all_nodes_fn):
        try:
            nodes_value = all_nodes_fn()
            n_entities = sum(1 for _ in nodes_value)  # type: ignore[unused-ignore]
        except Exception:
            n_entities = 0
    await self._emit(
        "graph.indexed",
        job_id=job.id,
        communities=len(self._community_summaries),
        entities=n_entities,
    )
    return result

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/graph/graph_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.sources = retrieved
    response.query_id = query.id
    response.traces.append({"pipeline": self.name, "n_sources": len(retrieved)})
    await self._emit("query.generated", query_id=query.id)
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/graph/graph_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

HashingEmbedder

Python
HashingEmbedder(dims: int = 128, model: str | None = None)

Deterministic, dependency-free token-hashing embedder.

Useful for tests, offline development and smoke pipelines without API keys. Output vectors are L2-normalised so cosine similarity matches dot product.

Source code in apogee_ai_rag/infrastructure/embeddings/hashing_embedder.py
Python
def __init__(self, dims: int = 128, model: str | None = None) -> None:
    if dims <= 0:
        raise ValueError("dims must be > 0")
    self.dims = dims
    self.model = model or f"hashing-{dims}"

dims instance-attribute

Python
dims = dims

model instance-attribute

Python
model = model or f'hashing-{dims}'

embed async

Python
embed(texts: list[str], modality: Modality = TEXT) -> list[list[float]]
Source code in apogee_ai_rag/infrastructure/embeddings/hashing_embedder.py
Python
async def embed(
    self, texts: list[str], modality: Modality = Modality.TEXT
) -> list[list[float]]:
    if modality is not Modality.TEXT:
        raise UnsupportedModalityError(
            f"HashingEmbedder only supports TEXT, got {modality.value}"
        )
    return [self._embed_one(t) for t in texts]

HeuristicQueryRewriter

Python
HeuristicQueryRewriter(max_variants: int = 3)

Generates a small set of paraphrases without calling any LLM.

Strategies: - the original query; - keyword-only form (drop interrogative prefixes and stopwords); - a contextual phrasing prepended with about.

Good enough as a baseline before plugging :class:LLMQueryRewriter.

Source code in apogee_ai_rag/infrastructure/query_rewriters/heuristic_query_rewriter.py
Python
def __init__(self, max_variants: int = 3) -> None:
    if max_variants <= 0:
        raise ValueError("max_variants must be positive")
    self._max = max_variants

name class-attribute instance-attribute

Python
name = 'heuristic'

rewrite async

Python
rewrite(query: RagQuery) -> list[str]
Source code in apogee_ai_rag/infrastructure/query_rewriters/heuristic_query_rewriter.py
Python
async def rewrite(self, query: RagQuery) -> list[str]:
    text = query.text.strip()
    if not text:
        return [""]
    variants: list[str] = [text]

    lowered = text.lower()
    for prefix in _QUESTION_PREFIXES:
        if lowered.startswith(prefix):
            variants.append(text[len(prefix) :].strip(" ?."))
            break

    keywords = [
        tok for tok in tokenize(text)
        if tok not in self._STOPWORDS and len(tok) > 2
    ]
    if keywords:
        variants.append(" ".join(keywords))
        variants.append("about " + " ".join(keywords[:5]))

    seen: set[str] = set()
    deduped: list[str] = []
    for variant in variants:
        if variant and variant not in seen:
            seen.add(variant)
            deduped.append(variant)
    return deduped[: self._max]

HierarchicalRagPipeline

Python
HierarchicalRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/hierarchical/hierarchical_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "HierarchicalRagPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config

name class-attribute instance-attribute

Python
name = 'hierarchical'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/hierarchical/hierarchical_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/hierarchical/hierarchical_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.sources = retrieved
    response.query_id = query.id
    await self._emit("query.generated", query_id=query.id)
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/hierarchical/hierarchical_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

HyDERagPipeline

Python
HyDERagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, hypothesis_max_chars: int = 600)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/hyde/hyde_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    hypothesis_max_chars: int = 600,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "HyDERagPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._hypothesis_cap = hypothesis_max_chars

name class-attribute instance-attribute

Python
name = 'hyde'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/hyde/hyde_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/hyde/hyde_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.query_id = query.id
    await self._emit(
        "query.generated", query_id=query.id, finish_reason=response.finish_reason,
    )
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/hyde/hyde_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

HybridRagPipeline

Python
HybridRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, rrf_k: int = 60)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/hybrid/hybrid_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    rrf_k: int = 60,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "HybridRagPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._bm25 = BM25Search()
    self._chunks: list[Chunk] = []
    self._rrf_k = rrf_k

name class-attribute instance-attribute

Python
name = 'hybrid'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/hybrid/hybrid_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    result = await self._ingest_default(job)
    # Rebuild BM25 from the in-memory store when available; otherwise
    # accumulate the freshly chunked input.
    chunks = list(getattr(self._store, "_chunks", {}).values())  # type: ignore[attr-defined]
    if not chunks:
        chunks = self._chunks + await self._chunker.chunk(job.documents)
    self._chunks = chunks
    self._bm25.fit(self._chunks)
    return result

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/hybrid/hybrid_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.query_id = query.id
    await self._emit(
        "query.generated", query_id=query.id, finish_reason=response.finish_reason,
    )
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/hybrid/hybrid_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

InMemoryCacheStore

Python
InMemoryCacheStore(max_size: int = 1024)
Source code in apogee_ai_rag/infrastructure/cache_stores/in_memory_cache_store.py
Python
def __init__(self, max_size: int = 1024) -> None:
    if max_size <= 0:
        raise ValueError("max_size must be > 0")
    self._max = max_size
    self._store: dict[str, tuple[CacheEntry, float | None]] = {}

name class-attribute instance-attribute

Python
name = 'in_memory'

get async

Python
get(key: str) -> CacheEntry | None
Source code in apogee_ai_rag/infrastructure/cache_stores/in_memory_cache_store.py
Python
async def get(self, key: str) -> CacheEntry | None:
    self._evict_expired()
    item = self._store.get(key)
    if item is None:
        return None
    return item[0]

set async

Python
set(entry: CacheEntry) -> None
Source code in apogee_ai_rag/infrastructure/cache_stores/in_memory_cache_store.py
Python
async def set(self, entry: CacheEntry) -> None:
    if len(self._store) >= self._max:
        # Drop the oldest entry to keep the cache bounded.
        oldest_key = next(iter(self._store))
        self._store.pop(oldest_key, None)
    expiry = (
        time.time() + entry.ttl_seconds if entry.ttl_seconds is not None else None
    )
    self._store[entry.key] = (entry, expiry)

invalidate async

Python
invalidate(prefix: str) -> int
Source code in apogee_ai_rag/infrastructure/cache_stores/in_memory_cache_store.py
Python
async def invalidate(self, prefix: str) -> int:
    keys = [k for k in self._store if k.startswith(prefix)]
    for k in keys:
        self._store.pop(k, None)
    return len(keys)

size async

Python
size() -> int
Source code in apogee_ai_rag/infrastructure/cache_stores/in_memory_cache_store.py
Python
async def size(self) -> int:
    self._evict_expired()
    return len(self._store)

InMemoryObservabilityEmitter

Python
InMemoryObservabilityEmitter()

Captures emitted events in a list for tests and assertions.

Source code in apogee_ai_rag/infrastructure/observability/in_memory_observability_emitter.py
Python
def __init__(self) -> None:
    self.events: list[RagEvent] = []

name class-attribute instance-attribute

Python
name = 'in_memory'

events instance-attribute

Python
events: list[RagEvent] = []

emit async

Python
emit(event: RagEvent) -> None
Source code in apogee_ai_rag/infrastructure/observability/in_memory_observability_emitter.py
Python
async def emit(self, event: RagEvent) -> None:
    self.events.append(event)

by_name

Python
by_name(name: str) -> list[RagEvent]
Source code in apogee_ai_rag/infrastructure/observability/in_memory_observability_emitter.py
Python
def by_name(self, name: str) -> list[RagEvent]:
    return [e for e in self.events if e.name == name]

clear

Python
clear() -> None
Source code in apogee_ai_rag/infrastructure/observability/in_memory_observability_emitter.py
Python
def clear(self) -> None:
    self.events.clear()

InMemoryVectorStore

Python
InMemoryVectorStore()

Cosine-similarity store for the naive RAG MVP.

Holds chunks in a dict keyed by chunk.id. Supports filters via exact match on chunk.metadata and a lightweight hybrid mode that mixes cosine similarity with a normalised token-overlap score (sparse stand-in).

Source code in apogee_ai_rag/infrastructure/vector_stores/in_memory_vector_store.py
Python
def __init__(self) -> None:
    self._chunks: dict[str, Chunk] = {}

name class-attribute instance-attribute

Python
name = 'in_memory'

upsert async

Python
upsert(chunks: list[Chunk]) -> int
Source code in apogee_ai_rag/infrastructure/vector_stores/in_memory_vector_store.py
Python
async def upsert(self, chunks: list[Chunk]) -> int:
    for chunk in chunks:
        self._chunks[chunk.id] = chunk
    return len(chunks)

search async

Python
search(query_embedding: list[float], top_k: int = 5, threshold: float = 0.0, filters: dict | None = None) -> list[RetrievedChunk]
Source code in apogee_ai_rag/infrastructure/vector_stores/in_memory_vector_store.py
Python
async def search(
    self,
    query_embedding: list[float],
    top_k: int = 5,
    threshold: float = 0.0,
    filters: dict | None = None,
) -> list[RetrievedChunk]:
    scored: list[RetrievedChunk] = []
    for chunk in self._chunks.values():
        if not chunk.embedding:
            continue
        if not _matches_filters(chunk.metadata, filters):
            continue
        score = _cosine(query_embedding, chunk.embedding)
        scored.append(RetrievedChunk(chunk=chunk, score=score, retriever="vector"))
    scored.sort(key=lambda r: r.score, reverse=True)
    return [r for r in scored[:top_k] if r.score >= threshold]
Python
hybrid_search(query_text: str, query_embedding: list[float], top_k: int = 5, threshold: float = 0.0, filters: dict | None = None, alpha: float = 0.5) -> list[RetrievedChunk]
Source code in apogee_ai_rag/infrastructure/vector_stores/in_memory_vector_store.py
Python
async def hybrid_search(
    self,
    query_text: str,
    query_embedding: list[float],
    top_k: int = 5,
    threshold: float = 0.0,
    filters: dict | None = None,
    alpha: float = 0.5,
) -> list[RetrievedChunk]:
    query_tokens = set(_TOKEN_RE.findall(query_text.lower()))
    scored: list[RetrievedChunk] = []
    for chunk in self._chunks.values():
        if not _matches_filters(chunk.metadata, filters):
            continue
        dense = _cosine(query_embedding, chunk.embedding) if chunk.embedding else 0.0
        sparse = _bm25ish_score(query_tokens, set(_TOKEN_RE.findall(chunk.text.lower())))
        blended = alpha * dense + (1.0 - alpha) * sparse
        scored.append(
            RetrievedChunk(chunk=chunk, score=blended, retriever="hybrid"),
        )
    scored.sort(key=lambda r: r.score, reverse=True)
    return [r for r in scored[:top_k] if r.score >= threshold]

delete async

Python
delete(chunk_ids: list[str]) -> int
Source code in apogee_ai_rag/infrastructure/vector_stores/in_memory_vector_store.py
Python
async def delete(self, chunk_ids: list[str]) -> int:
    removed = 0
    for cid in chunk_ids:
        if self._chunks.pop(cid, None) is not None:
            removed += 1
    return removed

count async

Python
count() -> int
Source code in apogee_ai_rag/infrastructure/vector_stores/in_memory_vector_store.py
Python
async def count(self) -> int:
    return len(self._chunks)

IngestionJob dataclass

Python
IngestionJob(documents: list[Document], batch_size: int = 64, idempotency_key: str | None = None, id: str = (lambda: f'job_{hex[:10]}')(), metadata: dict = dict())

documents instance-attribute

Python
documents: list[Document]

batch_size class-attribute instance-attribute

Python
batch_size: int = 64

idempotency_key class-attribute instance-attribute

Python
idempotency_key: str | None = None

id class-attribute instance-attribute

Python
id: str = field(default_factory=lambda: f'job_{hex[:10]}')

metadata class-attribute instance-attribute

Python
metadata: dict = field(default_factory=dict)

IngestionResult dataclass

Python
IngestionResult(job_id: str, documents_ingested: int, chunks_produced: int, vectors_upserted: int, duration_ms: float = 0.0, errors: list[str] = list())

job_id instance-attribute

Python
job_id: str

documents_ingested instance-attribute

Python
documents_ingested: int

chunks_produced instance-attribute

Python
chunks_produced: int

vectors_upserted instance-attribute

Python
vectors_upserted: int

duration_ms class-attribute instance-attribute

Python
duration_ms: float = 0.0

errors class-attribute instance-attribute

Python
errors: list[str] = field(default_factory=list)

IterativeRagPipeline

Python
IterativeRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, n_rounds: int = 2)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/iterative/iterative_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    n_rounds: int = 2,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "IterativeRagPipeline requires chunker, embedder, store and generator"
        )
    if n_rounds < 1:
        raise PipelineConfigError("n_rounds must be >= 1")
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._n_rounds = n_rounds

name class-attribute instance-attribute

Python
name = 'iterative'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/iterative/iterative_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/iterative/iterative_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    await self._emit(
        "query.received", query_id=query.id, text=query.text, pipeline_type=self.name,
    )
    seen: set[str] = set()
    retrieved: list[RetrievedChunk] = []
    last_response: RagResponse | None = None
    current_text = query.text

    for round_idx in range(1, self._n_rounds + 1):
        [embedding] = await self._embedder.embed([current_text])
        results = await self._store.search(
            embedding,
            top_k=query.top_k,
            threshold=query.threshold,
            filters=query.filters or None,
        )
        new_items = [r for r in results if r.chunk.id not in seen]
        for r in new_items:
            seen.add(r.chunk.id)
        retrieved.extend(new_items)
        last_response = await self._generator.generate(
            query.text, retrieved, self._generation_config,
        )
        await self._emit(
            "iterative.round",
            query_id=query.id,
            round=round_idx,
            added=len(new_items),
        )
        if not new_items:
            break
        current_text = f"{query.text} {last_response.answer}"
        retrieved.append(
            RetrievedChunk(
                chunk=Chunk(
                    text=last_response.answer[:400],
                    parent_id=f"iterative_thought_{round_idx}",
                ),
                score=0.0,
                retriever="iterative_thought",
            ),
        )

    assert last_response is not None
    last_response.sources = [
        r for r in retrieved if r.retriever != "iterative_thought"
    ]
    last_response.query_id = query.id
    await self._emit("query.generated", query_id=query.id)
    return last_response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/iterative/iterative_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    # Streaming variant runs once with the merged context for simplicity.
    response = await self.run(query)
    async for chunk in self._generator.stream(
        query.text, response.sources, self._generation_config,
    ):
        yield chunk

KAGPipeline

Python
KAGPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, graph_store: IGraphStore | None = None)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/kag/kag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    graph_store: IGraphStore | None = None,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "KAGPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._graph: IGraphStore = graph_store or NetworkXGraphStore()
    self._entity_index: dict[str, list[str]] = {}

name class-attribute instance-attribute

Python
name = 'kag'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/kag/kag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    result = await self._ingest_default(job)
    chunks = list(getattr(self._store, "_chunks", {}).values())  # type: ignore[attr-defined]
    self._entity_index = {}
    for chunk in chunks:
        for entity in _extract_entities(chunk.text):
            await self._graph.add_node(
                KnowledgeNode(
                    label=entity, properties={"chunk_id": chunk.id, "kind": "entity"},
                ),
            )
            self._entity_index.setdefault(entity, []).append(chunk.id)
    await self._emit(
        "kag.indexed", job_id=job.id, entities=len(self._entity_index),
    )
    return result

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/kag/kag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.sources = retrieved
    response.query_id = query.id
    response.traces.append(
        {"pipeline": self.name, "entities_in_query": _extract_entities(query.text)},
    )
    await self._emit("query.generated", query_id=query.id)
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/kag/kag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

KnowledgeEdge dataclass

Python
KnowledgeEdge(source_id: str, target_id: str, label: str, weight: float = 1.0, properties: dict = dict(), id: str = (lambda: f'e_{hex[:10]}')())

source_id instance-attribute

Python
source_id: str

target_id instance-attribute

Python
target_id: str

label instance-attribute

Python
label: str

weight class-attribute instance-attribute

Python
weight: float = 1.0

properties class-attribute instance-attribute

Python
properties: dict = field(default_factory=dict)

id class-attribute instance-attribute

Python
id: str = field(default_factory=lambda: f'e_{hex[:10]}')

KnowledgeNode dataclass

Python
KnowledgeNode(label: str, properties: dict = dict(), id: str = (lambda: f'n_{hex[:10]}')())

label instance-attribute

Python
label: str

properties class-attribute instance-attribute

Python
properties: dict = field(default_factory=dict)

id class-attribute instance-attribute

Python
id: str = field(default_factory=lambda: f'n_{hex[:10]}')

LLMQueryRewriter

Python
LLMQueryRewriter(generator: IGenerator, n: int = 3, config: GenerationConfig | None = None)

Generates rewrites via an :class:IGenerator (typically an LLM).

The default prompt asks the model for n numbered alternatives. This works with any provider supported by apogee-ai-providers once you inject :class:ApogeeProvidersGenerator. With EchoGenerator it falls back to returning the original query (the echo just mirrors back).

Source code in apogee_ai_rag/infrastructure/query_rewriters/llm_query_rewriter.py
Python
def __init__(
    self,
    generator: IGenerator,
    n: int = 3,
    config: GenerationConfig | None = None,
) -> None:
    if n <= 0:
        raise ValueError("n must be positive")
    self._generator = generator
    self._n = n
    self._config = config or GenerationConfig(temperature=0.7, max_tokens=256)

name class-attribute instance-attribute

Python
name = 'llm'

rewrite async

Python
rewrite(query: RagQuery) -> list[str]
Source code in apogee_ai_rag/infrastructure/query_rewriters/llm_query_rewriter.py
Python
async def rewrite(self, query: RagQuery) -> list[str]:
    prompt = (
        f"Generate {self._n} alternative phrasings of the following user "
        f"question that would help retrieve relevant context. Output one "
        f"alternative per line, no numbering or extra commentary.\n\n"
        f"Question: {query.text}"
    )
    response = await self._generator.generate(prompt, [], self._config)
    candidates = [
        line.strip().lstrip("-•0123456789.) ").strip()
        for line in (response.answer or "").splitlines()
    ]
    candidates = [c for c in candidates if c]
    rewrites = [query.text] + candidates
    seen: set[str] = set()
    deduped: list[str] = []
    for r in rewrites:
        if r not in seen:
            seen.add(r)
            deduped.append(r)
    return deduped[: self._n + 1]

LongRagPipeline

Python
LongRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, long_chunk_size: int = 4000, long_chunk_overlap: int = 200, long_top_k: int = 2)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/long/long_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    long_chunk_size: int = 4000,
    long_chunk_overlap: int = 200,
    long_top_k: int = 2,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "LongRagPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._strategy = ChunkStrategy(
        chunk_size=long_chunk_size, chunk_overlap=long_chunk_overlap,
    )
    self._long_top_k = long_top_k

name class-attribute instance-attribute

Python
name = 'long'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/long/long_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    await self._emit(
        "ingestion.started",
        job_id=job.id,
        n_documents=len(job.documents),
        chunker=self._chunker.name,
        embedder=self._embedder.model,
        store=self._store.name,
        chunk_size=self._strategy.chunk_size,
    )
    chunks = await self._chunker.chunk(job.documents, self._strategy)
    await self._emit("ingestion.chunked", job_id=job.id, n_chunks=len(chunks))
    if not chunks:
        return IngestionResult(
            job_id=job.id,
            documents_ingested=len(job.documents),
            chunks_produced=0,
            vectors_upserted=0,
        )
    embeddings = await self._embedder.embed([c.text for c in chunks])
    for chunk, vec in zip(chunks, embeddings, strict=False):
        chunk.embedding = vec
    upserted = await self._store.upsert(chunks)
    await self._emit("ingestion.upserted", job_id=job.id, n_upserted=upserted)
    return IngestionResult(
        job_id=job.id,
        documents_ingested=len(job.documents),
        chunks_produced=len(chunks),
        vectors_upserted=upserted,
    )

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/long/long_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.sources = retrieved
    response.query_id = query.id
    await self._emit("query.generated", query_id=query.id, n_sources=len(retrieved))
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/long/long_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

Modality

Bases: StrEnum

TEXT class-attribute instance-attribute

Python
TEXT = 'text'

IMAGE class-attribute instance-attribute

Python
IMAGE = 'image'

AUDIO class-attribute instance-attribute

Python
AUDIO = 'audio'

VIDEO class-attribute instance-attribute

Python
VIDEO = 'video'

ModularRagPipeline

Python
ModularRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, stages: list[Stage] | None = None, reranker: IReranker | None = None)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/modular/modular_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    stages: list[Stage] | None = None,
    reranker: IReranker | None = None,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "ModularRagPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._reranker = reranker or NoOpReranker()
    self._stages = stages or self.default_stages()

name class-attribute instance-attribute

Python
name = 'modular'

default_stages

Python
default_stages() -> list[Stage]
Source code in apogee_ai_rag/infrastructure/pipelines/modular/modular_rag_pipeline.py
Python
def default_stages(self) -> list[Stage]:
    async def embed_query(state: ModularState) -> ModularState:
        [vec] = await self._embedder.embed([state.query.text])
        state.embedding = vec
        return state

    async def retrieve(state: ModularState) -> ModularState:
        state.retrieved = await self._store.search(
            state.embedding,
            top_k=state.query.top_k,
            threshold=state.query.threshold,
            filters=state.query.filters or None,
        )
        return state

    async def rerank(state: ModularState) -> ModularState:
        state.retrieved = await self._reranker.rerank(
            state.query.text, state.retrieved, top_n=state.query.top_k,
        )
        return state

    async def generate(state: ModularState) -> ModularState:
        state.response = await self._generator.generate(
            state.query.text, state.retrieved, self._generation_config,
        )
        return state

    return [embed_query, retrieve, rerank, generate]

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/modular/modular_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/modular/modular_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    await self._emit(
        "query.received", query_id=query.id, text=query.text, pipeline_type=self.name,
    )
    state = ModularState(query=query)
    for stage in self._stages:
        state = await stage(state)
        await self._emit(
            "modular.stage",
            query_id=query.id,
            stage=getattr(stage, "__name__", repr(stage)),
        )
    if state.response is None:
        raise PipelineConfigError("Modular pipeline finished without producing a response")
    state.response.sources = state.retrieved
    state.response.query_id = query.id
    await self._emit("query.generated", query_id=query.id)
    return state.response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/modular/modular_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    response = await self.run(query)
    async for chunk in self._generator.stream(
        query.text, response.sources, self._generation_config,
    ):
        yield chunk

ModularState dataclass

Python
ModularState(query: RagQuery, embedding: list[float] = list(), retrieved: list[RetrievedChunk] = list(), response: RagResponse | None = None, bag: dict[str, Any] = dict())

query instance-attribute

Python
query: RagQuery

embedding class-attribute instance-attribute

Python
embedding: list[float] = field(default_factory=list)

retrieved class-attribute instance-attribute

Python
retrieved: list[RetrievedChunk] = field(default_factory=list)

response class-attribute instance-attribute

Python
response: RagResponse | None = None

bag class-attribute instance-attribute

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

MultiHopRagPipeline

Python
MultiHopRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, max_hops: int = 3)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/multi_hop/multi_hop_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    max_hops: int = 3,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "MultiHopRagPipeline requires chunker, embedder, store and generator"
        )
    if max_hops < 1:
        raise PipelineConfigError("max_hops must be >= 1")
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._max_hops = max_hops

name class-attribute instance-attribute

Python
name = 'multi_hop'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/multi_hop/multi_hop_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/multi_hop/multi_hop_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.sources = retrieved
    response.query_id = query.id
    await self._emit("query.generated", query_id=query.id)
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/multi_hop/multi_hop_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

MultiModalRagPipeline

Python
MultiModalRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, image_caption_max_chars: int = 256)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/multi_modal/multi_modal_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    image_caption_max_chars: int = 256,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "MultiModalRagPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._image_caption_max = image_caption_max_chars

name class-attribute instance-attribute

Python
name = 'multi_modal'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/multi_modal/multi_modal_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/multi_modal/multi_modal_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.sources = retrieved
    response.query_id = query.id
    await self._emit("query.generated", query_id=query.id)
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/multi_modal/multi_modal_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

NaiveRagPipeline

Python
NaiveRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, reranker: IReranker | None = None, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None)

The simplest RAG pipeline — embed, search, optionally rerank, generate.

Source code in apogee_ai_rag/infrastructure/pipelines/naive/naive_rag_pipeline.py
Python
def __init__(
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    reranker: IReranker | None = None,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
) -> None:
    if chunker is None or embedder is None or vector_store is None or generator is None:
        raise PipelineConfigError(
            "NaiveRagPipeline requires chunker, embedder, vector_store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._reranker = reranker
    self._emitter = observability or NoOpObservabilityEmitter()
    self._generation_config = generation_config

name class-attribute instance-attribute

Python
name = 'naive'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/naive/naive_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    started = time.perf_counter()
    await self._emit(
        "ingestion.started",
        job_id=job.id,
        n_documents=len(job.documents),
        chunker=self._chunker.name,
        embedder=self._embedder.model,
        store=self._store.name,
    )

    chunks = await self._chunker.chunk(job.documents)
    await self._emit("ingestion.chunked", job_id=job.id, n_chunks=len(chunks))

    if not chunks:
        return IngestionResult(
            job_id=job.id,
            documents_ingested=len(job.documents),
            chunks_produced=0,
            vectors_upserted=0,
            duration_ms=(time.perf_counter() - started) * 1000.0,
        )

    embeddings = await self._embedder.embed([c.text for c in chunks])
    embedded: list[Chunk] = []
    for c, e in zip(chunks, embeddings, strict=False):
        c.embedding = e
        embedded.append(c)
    await self._emit(
        "ingestion.embedded",
        job_id=job.id,
        n_vectors=len(embedded),
        dims=self._embedder.dims,
    )

    upserted = 0
    batch = max(job.batch_size, 1)
    for start in range(0, len(embedded), batch):
        upserted += await self._store.upsert(embedded[start : start + batch])
    await self._emit("ingestion.upserted", job_id=job.id, n_upserted=upserted)

    return IngestionResult(
        job_id=job.id,
        documents_ingested=len(job.documents),
        chunks_produced=len(chunks),
        vectors_upserted=upserted,
        duration_ms=(time.perf_counter() - started) * 1000.0,
    )

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/naive/naive_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(query.text, retrieved, self._generation_config)
    response.query_id = query.id
    await self._emit(
        "query.generated",
        query_id=query.id,
        tokens=response.usage.get("total_tokens", 0),
        finish_reason=response.finish_reason,
    )
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/naive/naive_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(query.text, retrieved, self._generation_config):
        yield chunk
    await self._emit("query.generated", query_id=query.id, streamed=True)

Neo4jGraphStore

Python
Neo4jGraphStore(*, uri: str, auth: tuple[str, str] | None = None)
Source code in apogee_ai_rag/infrastructure/graph_stores/neo4j_graph_store.py
Python
def __init__(self, *, uri: str, auth: tuple[str, str] | None = None) -> None:
    try:
        import neo4j  # type: ignore[import-untyped]  # noqa: F401
    except ImportError as exc:
        raise ProviderNotInstalledError("neo4j", "neo4j") from exc
    # pragma: no cover — full driver wiring is intentionally deferred.
    self._uri = uri
    self._auth = auth

name class-attribute instance-attribute

Python
name = 'neo4j'

add_node async

Python
add_node(node: KnowledgeNode) -> str
Source code in apogee_ai_rag/infrastructure/graph_stores/neo4j_graph_store.py
Python
async def add_node(self, node: KnowledgeNode) -> str:  # pragma: no cover
    raise NotImplementedError("Neo4jGraphStore.add_node is on the F7.x roadmap")

add_edge async

Python
add_edge(edge: KnowledgeEdge) -> str
Source code in apogee_ai_rag/infrastructure/graph_stores/neo4j_graph_store.py
Python
async def add_edge(self, edge: KnowledgeEdge) -> str:  # pragma: no cover
    raise NotImplementedError("Neo4jGraphStore.add_edge is on the F7.x roadmap")

traverse async

Python
traverse(start_id: str, depth: int = 2) -> tuple[list[KnowledgeNode], list[KnowledgeEdge]]
Source code in apogee_ai_rag/infrastructure/graph_stores/neo4j_graph_store.py
Python
async def traverse(  # pragma: no cover
    self, start_id: str, depth: int = 2,
) -> tuple[list[KnowledgeNode], list[KnowledgeEdge]]:
    raise NotImplementedError("Neo4jGraphStore.traverse is on the F7.x roadmap")

NetworkXGraphStore

Python
NetworkXGraphStore()
Source code in apogee_ai_rag/infrastructure/graph_stores/networkx_graph_store.py
Python
def __init__(self) -> None:
    self._nodes: dict[str, KnowledgeNode] = {}
    self._adj: dict[str, list[KnowledgeEdge]] = {}

name class-attribute instance-attribute

Python
name = 'networkx'

add_node async

Python
add_node(node: KnowledgeNode) -> str
Source code in apogee_ai_rag/infrastructure/graph_stores/networkx_graph_store.py
Python
async def add_node(self, node: KnowledgeNode) -> str:
    self._nodes[node.id] = node
    self._adj.setdefault(node.id, [])
    return node.id

add_edge async

Python
add_edge(edge: KnowledgeEdge) -> str
Source code in apogee_ai_rag/infrastructure/graph_stores/networkx_graph_store.py
Python
async def add_edge(self, edge: KnowledgeEdge) -> str:
    self._adj.setdefault(edge.source_id, []).append(edge)
    self._adj.setdefault(edge.target_id, []).append(edge)
    return edge.id

traverse async

Python
traverse(start_id: str, depth: int = 2) -> tuple[list[KnowledgeNode], list[KnowledgeEdge]]
Source code in apogee_ai_rag/infrastructure/graph_stores/networkx_graph_store.py
Python
async def traverse(
    self, start_id: str, depth: int = 2,
) -> tuple[list[KnowledgeNode], list[KnowledgeEdge]]:
    if start_id not in self._nodes:
        return [], []
    visited_nodes: set[str] = {start_id}
    visited_edges: list[KnowledgeEdge] = []
    queue: deque[tuple[str, int]] = deque([(start_id, 0)])
    while queue:
        current, level = queue.popleft()
        if level >= depth:
            continue
        for edge in self._adj.get(current, []):
            if edge.id in {e.id for e in visited_edges}:
                continue
            visited_edges.append(edge)
            neighbour = (
                edge.target_id if edge.source_id == current else edge.source_id
            )
            if neighbour not in visited_nodes:
                visited_nodes.add(neighbour)
                queue.append((neighbour, level + 1))
    nodes = [self._nodes[nid] for nid in visited_nodes if nid in self._nodes]
    return nodes, visited_edges

all_nodes

Python
all_nodes() -> list[KnowledgeNode]
Source code in apogee_ai_rag/infrastructure/graph_stores/networkx_graph_store.py
Python
def all_nodes(self) -> list[KnowledgeNode]:
    return list(self._nodes.values())

all_edges

Python
all_edges() -> list[KnowledgeEdge]
Source code in apogee_ai_rag/infrastructure/graph_stores/networkx_graph_store.py
Python
def all_edges(self) -> list[KnowledgeEdge]:
    seen: set[str] = set()
    edges: list[KnowledgeEdge] = []
    for adj_list in self._adj.values():
        for edge in adj_list:
            if edge.id in seen:
                continue
            seen.add(edge.id)
            edges.append(edge)
    return edges

communities

Python
communities() -> list[list[str]]

Returns weakly-connected components as a community partition.

Source code in apogee_ai_rag/infrastructure/graph_stores/networkx_graph_store.py
Python
def communities(self) -> list[list[str]]:
    """Returns weakly-connected components as a community partition."""
    visited: set[str] = set()
    communities: list[list[str]] = []
    for node_id in self._nodes:
        if node_id in visited:
            continue
        stack = [node_id]
        component: list[str] = []
        while stack:
            current = stack.pop()
            if current in visited:
                continue
            visited.add(current)
            component.append(current)
            for edge in self._adj.get(current, []):
                neighbour = (
                    edge.target_id if edge.source_id == current else edge.source_id
                )
                if neighbour not in visited:
                    stack.append(neighbour)
        communities.append(component)
    return communities

NoOpObservabilityEmitter

name class-attribute instance-attribute

Python
name = 'noop'

emit async

Python
emit(event: RagEvent) -> None
Source code in apogee_ai_rag/infrastructure/observability/noop_observability_emitter.py
Python
async def emit(self, event: RagEvent) -> None:
    return None

NoOpReranker

Default reranker that preserves the upstream order, only applying top_n.

name class-attribute instance-attribute

Python
name = 'no_op'

rerank async

Python
rerank(query: str, chunks: list[RetrievedChunk], top_n: int = 5) -> list[RetrievedChunk]
Source code in apogee_ai_rag/infrastructure/rerankers/no_op_reranker.py
Python
async def rerank(
    self, query: str, chunks: list[RetrievedChunk], top_n: int = 5
) -> list[RetrievedChunk]:
    del query
    if top_n <= 0:
        return list(chunks)
    return list(chunks[:top_n])

ObservabilityProvider

Bases: StrEnum

NOOP class-attribute instance-attribute

Python
NOOP = 'noop'

LANGSMITH class-attribute instance-attribute

Python
LANGSMITH = 'langsmith'

LANGFUSE class-attribute instance-attribute

Python
LANGFUSE = 'langfuse'

HELICONE class-attribute instance-attribute

Python
HELICONE = 'helicone'

PHOENIX class-attribute instance-attribute

Python
PHOENIX = 'phoenix'

WEAVE class-attribute instance-attribute

Python
WEAVE = 'weave'

MLFLOW class-attribute instance-attribute

Python
MLFLOW = 'mlflow'

OPENLLMETRY class-attribute instance-attribute

Python
OPENLLMETRY = 'openllmetry'

OTEL class-attribute instance-attribute

Python
OTEL = 'otel'

PassThroughRewriter

Returns the original query verbatim — used as the default.

name class-attribute instance-attribute

Python
name = 'pass_through'

rewrite async

Python
rewrite(query: RagQuery) -> list[str]
Source code in apogee_ai_rag/infrastructure/query_rewriters/pass_through_rewriter.py
Python
async def rewrite(self, query: RagQuery) -> list[str]:
    return [query.text]

PipelineRuntimeConfig dataclass

Python
PipelineRuntimeConfig(max_iterations: int = 5, timeout_seconds: float = 60.0, enable_streaming: bool = False, enable_caching: bool = False, enable_observability: bool = True)

max_iterations class-attribute instance-attribute

Python
max_iterations: int = 5

timeout_seconds class-attribute instance-attribute

Python
timeout_seconds: float = 60.0

enable_streaming class-attribute instance-attribute

Python
enable_streaming: bool = False

enable_caching class-attribute instance-attribute

Python
enable_caching: bool = False

enable_observability class-attribute instance-attribute

Python
enable_observability: bool = True

PipelineSpec dataclass

Python
PipelineSpec(rag_type: RagType = NAIVE, chunker: Any | None = None, embedder: Any | None = None, vector_store: Any | None = None, retriever: Any | None = None, generator: Any | None = None, reranker: Any | None = None, query_rewriter: Any | None = None, evaluator: Any | None = None, observability: Any | None = None, graph_store: Any | None = None, cache_store: Any | None = None, options: dict = dict())

Aggregate of components that compose a RAG pipeline.

Each field is a Protocol implementation (chunker, embedder, vector store, generator, etc.). Components left as None are either optional or filled in by the factory based on rag_type.

rag_type class-attribute instance-attribute

Python
rag_type: RagType = NAIVE

chunker class-attribute instance-attribute

Python
chunker: Any | None = None

embedder class-attribute instance-attribute

Python
embedder: Any | None = None

vector_store class-attribute instance-attribute

Python
vector_store: Any | None = None

retriever class-attribute instance-attribute

Python
retriever: Any | None = None

generator class-attribute instance-attribute

Python
generator: Any | None = None

reranker class-attribute instance-attribute

Python
reranker: Any | None = None

query_rewriter class-attribute instance-attribute

Python
query_rewriter: Any | None = None

evaluator class-attribute instance-attribute

Python
evaluator: Any | None = None

observability class-attribute instance-attribute

Python
observability: Any | None = None

graph_store class-attribute instance-attribute

Python
graph_store: Any | None = None

cache_store class-attribute instance-attribute

Python
cache_store: Any | None = None

options class-attribute instance-attribute

Python
options: dict = field(default_factory=dict)

QueryRewritingRagPipeline

Python
QueryRewritingRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, query_rewriter: IQueryRewriter | None = None, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/query_rewriting/query_rewriting_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    query_rewriter: IQueryRewriter | None = None,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "QueryRewritingRagPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._rewriter = query_rewriter or HeuristicQueryRewriter()
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config

name class-attribute instance-attribute

Python
name = 'query_rewriting'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/query_rewriting/query_rewriting_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/query_rewriting/query_rewriting_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.query_id = query.id
    await self._emit(
        "query.generated", query_id=query.id, finish_reason=response.finish_reason,
    )
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/query_rewriting/query_rewriting_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

RAPTORPipeline

Python
RAPTORPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, max_levels: int = 2, cluster_similarity: float = 0.4, min_cluster_size: int = 2)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/raptor/raptor_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    max_levels: int = 2,
    cluster_similarity: float = 0.4,
    min_cluster_size: int = 2,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "RAPTORPipeline requires chunker, embedder, store and generator"
        )
    if max_levels < 1:
        raise PipelineConfigError("max_levels must be >= 1")
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._max_levels = max_levels
    self._cluster_threshold = cluster_similarity
    self._min_cluster_size = min_cluster_size

name class-attribute instance-attribute

Python
name = 'raptor'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/raptor/raptor_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    result = await self._ingest_default(job)
    # Build summary levels on top of the freshly upserted chunks.
    current_chunks = list(getattr(self._store, "_chunks", {}).values())  # type: ignore[attr-defined]
    for level in range(1, self._max_levels + 1):
        clusters = _greedy_cluster(current_chunks, self._cluster_threshold)
        qualifying = [c for c in clusters if len(c.members) >= self._min_cluster_size]
        await self._emit(
            "raptor.level",
            job_id=job.id,
            level=level,
            clusters=len(qualifying),
        )
        if not qualifying:
            break

        summaries: list[Chunk] = []
        for idx, cluster in enumerate(qualifying):
            merged_text = "\n".join(m.text for m in cluster.members)
            response = await self._generator.generate(
                f"Summarise the following passages in 2-3 sentences:\n\n{merged_text}",
                [], self._generation_config,
            )
            summary_text = (response.answer or merged_text[:400]).strip()
            summaries.append(
                Chunk(
                    text=summary_text,
                    parent_id=f"raptor_l{level}_c{idx}",
                    metadata={"raptor_level": level},
                ),
            )
        embeddings = await self._embedder.embed([s.text for s in summaries])
        for summary, vec in zip(summaries, embeddings, strict=False):
            summary.embedding = vec
        await self._store.upsert(summaries)
        result.chunks_produced += len(summaries)
        result.vectors_upserted += len(summaries)
        current_chunks = summaries
    return result

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/raptor/raptor_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.sources = retrieved
    response.query_id = query.id
    await self._emit("query.generated", query_id=query.id)
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/raptor/raptor_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

RagChunk dataclass

Python
RagChunk(delta: str, sources: list[RetrievedChunk] = list(), finish_reason: str | None = None)

Streaming delta returned by IRagPipeline.stream.

delta instance-attribute

Python
delta: str

sources class-attribute instance-attribute

Python
sources: list[RetrievedChunk] = field(default_factory=list)

finish_reason class-attribute instance-attribute

Python
finish_reason: str | None = None

RagEvent dataclass

Python
RagEvent(name: str, attributes: dict = dict(), timestamp: datetime = (lambda: now(UTC))())

name instance-attribute

Python
name: str

attributes class-attribute instance-attribute

Python
attributes: dict = field(default_factory=dict)

timestamp class-attribute instance-attribute

Python
timestamp: datetime = field(default_factory=lambda: now(UTC))

RagFactory

Selects and instantiates a concrete pipeline implementing :class:IRagPipeline.

build staticmethod

Python
build(rag_type: RagType, spec: PipelineSpec) -> IRagPipeline
Source code in apogee_ai_rag/infrastructure/factory/rag_factory.py
Python
@staticmethod
def build(rag_type: RagType, spec: PipelineSpec) -> IRagPipeline:
    builder = _BUILDERS.get(rag_type)
    if builder is None:
        return _build_not_implemented(rag_type)
    spec.rag_type = rag_type
    return builder(spec)

supported staticmethod

Python
supported() -> list[RagType]
Source code in apogee_ai_rag/infrastructure/factory/rag_factory.py
Python
@staticmethod
def supported() -> list[RagType]:
    return list(_BUILDERS.keys())

RagFusionPipeline

Python
RagFusionPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, query_rewriter: IQueryRewriter | None = None, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, rrf_k: int = 60)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/rag_fusion/rag_fusion_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    query_rewriter: IQueryRewriter | None = None,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    rrf_k: int = 60,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "RagFusionPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._rewriter = query_rewriter or HeuristicQueryRewriter(max_variants=4)
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._rrf_k = rrf_k

name class-attribute instance-attribute

Python
name = 'rag_fusion'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/rag_fusion/rag_fusion_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/rag_fusion/rag_fusion_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.query_id = query.id
    await self._emit(
        "query.generated", query_id=query.id, finish_reason=response.finish_reason,
    )
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/rag_fusion/rag_fusion_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

RagProviderCredentials dataclass

Python
RagProviderCredentials(api_key: str | None = None, endpoint: str | None = None, region: str | None = None, project: str | None = None, namespace: str | None = None, extra: dict = dict())

api_key class-attribute instance-attribute

Python
api_key: str | None = None

endpoint class-attribute instance-attribute

Python
endpoint: str | None = None

region class-attribute instance-attribute

Python
region: str | None = None

project class-attribute instance-attribute

Python
project: str | None = None

namespace class-attribute instance-attribute

Python
namespace: str | None = None

extra class-attribute instance-attribute

Python
extra: dict = field(default_factory=dict)

RagQuery dataclass

Python
RagQuery(text: str, filters: dict = dict(), top_k: int = 5, threshold: float = 0.0, modality: Modality = TEXT, history: list[dict] = list(), conversation_id: str | None = None, locale: str | None = None, image_b64: str | None = None, id: str = (lambda: f'q_{hex[:10]}')())

text instance-attribute

Python
text: str

filters class-attribute instance-attribute

Python
filters: dict = field(default_factory=dict)

top_k class-attribute instance-attribute

Python
top_k: int = 5

threshold class-attribute instance-attribute

Python
threshold: float = 0.0

modality class-attribute instance-attribute

Python
modality: Modality = TEXT

history class-attribute instance-attribute

Python
history: list[dict] = field(default_factory=list)

conversation_id class-attribute instance-attribute

Python
conversation_id: str | None = None

locale class-attribute instance-attribute

Python
locale: str | None = None

image_b64 class-attribute instance-attribute

Python
image_b64: str | None = None

id class-attribute instance-attribute

Python
id: str = field(default_factory=lambda: f'q_{hex[:10]}')

RagResponse dataclass

Python
RagResponse(answer: str, sources: list[RetrievedChunk] = list(), confidence: float | None = None, finish_reason: str | None = None, usage: dict = dict(), traces: list[dict] = list(), query_id: str | None = None)

answer instance-attribute

Python
answer: str

sources class-attribute instance-attribute

Python
sources: list[RetrievedChunk] = field(default_factory=list)

confidence class-attribute instance-attribute

Python
confidence: float | None = None

finish_reason class-attribute instance-attribute

Python
finish_reason: str | None = None

usage class-attribute instance-attribute

Python
usage: dict = field(default_factory=dict)

traces class-attribute instance-attribute

Python
traces: list[dict] = field(default_factory=list)

query_id class-attribute instance-attribute

Python
query_id: str | None = None

ReActRagPipeline

Python
ReActRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, max_steps: int = 4)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/react/react_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    max_steps: int = 4,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "ReActRagPipeline requires chunker, embedder, store and generator"
        )
    if max_steps < 1:
        raise PipelineConfigError("max_steps must be >= 1")
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._max_steps = max_steps

name class-attribute instance-attribute

Python
name = 'react'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/react/react_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/react/react_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    await self._emit(
        "query.received", query_id=query.id, text=query.text, pipeline_type=self.name,
    )
    seen_ids: set[str] = set()
    evidence: list[RetrievedChunk] = []
    scratchpad = f"Question: {query.text}\n"
    last_answer = ""

    for step in range(1, self._max_steps + 1):
        thought = await self._generator.generate(
            f"{scratchpad}\nThought:", [], self._generation_config,
        )
        action_query = (thought.answer or query.text).split("\n")[0][:200]
        results = await self._search(action_query, query)
        new_items = [r for r in results if r.chunk.id not in seen_ids]
        if not new_items:
            break
        for r in new_items:
            seen_ids.add(r.chunk.id)
        evidence.extend(new_items)

        observation = "\n".join(f"- {r.chunk.text}" for r in new_items[:2])
        scratchpad += (
            f"\nThought: {thought.answer.strip()[:200]}\n"
            f"Action: search[{action_query}]\n"
            f"Observation:\n{observation}\n"
        )
        await self._emit(
            "react.step",
            query_id=query.id,
            step=step,
            action=action_query,
            added=len(new_items),
        )

        response = await self._generator.generate(
            f"{scratchpad}\nAnswer:", evidence, self._generation_config,
        )
        last_answer = response.answer or ""
        looks_complete = (
            "Answer:" in last_answer or last_answer.strip().endswith(".")
        )
        if looks_complete and (step >= 2 or len(last_answer) > 60):
            # Heuristic stop: the model produced a complete sentence.
            break

    if not last_answer:
        response = await self._generator.generate(
            query.text, evidence, self._generation_config,
        )
        last_answer = response.answer or ""

    traces = [
        {"pipeline": self.name, "scratchpad": scratchpad, "evidence": len(evidence)},
    ]
    # Surface scratchpad as a synthetic source so callers can inspect it.
    evidence.append(
        RetrievedChunk(
            chunk=Chunk(text=scratchpad[-1000:], parent_id="react_scratchpad"),
            score=0.0,
            retriever="react_scratchpad",
        ),
    )
    await self._emit("query.generated", query_id=query.id, evidence=len(evidence))
    return RagResponse(
        answer=last_answer,
        sources=[r for r in evidence if r.retriever != "react_scratchpad"],
        traces=traces,
        query_id=query.id,
    )

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/react/react_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    response = await self.run(query)
    async for chunk in self._generator.stream(
        query.text, response.sources, self._generation_config,
    ):
        yield chunk

RecursiveRagPipeline

Python
RecursiveRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, max_depth: int = 3)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/recursive/recursive_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    max_depth: int = 3,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "RecursiveRagPipeline requires chunker, embedder, store and generator"
        )
    if max_depth < 1:
        raise PipelineConfigError("max_depth must be >= 1")
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._max_depth = max_depth

name class-attribute instance-attribute

Python
name = 'recursive'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/recursive/recursive_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/recursive/recursive_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.sources = retrieved
    response.query_id = query.id
    await self._emit("query.generated", query_id=query.id)
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/recursive/recursive_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

RecursiveTextChunker

Python
RecursiveTextChunker(chunk_size: int = 400, chunk_overlap: int = 40, separators: tuple[str, ...] | None = None)

Splits documents recursively along configurable separators.

Mirrors the behaviour of LangChain's RecursiveCharacterTextSplitter while staying dependency-free. Adds a per-chunk overlap by prepending the tail of the previous chunk.

Source code in apogee_ai_rag/infrastructure/chunkers/recursive_text_chunker.py
Python
def __init__(
    self,
    chunk_size: int = 400,
    chunk_overlap: int = 40,
    separators: tuple[str, ...] | None = None,
) -> None:
    if chunk_size <= 0:
        raise ValueError("chunk_size must be > 0")
    if chunk_overlap < 0 or chunk_overlap >= chunk_size:
        raise ValueError("chunk_overlap must satisfy 0 <= overlap < chunk_size")
    self._chunk_size = chunk_size
    self._chunk_overlap = chunk_overlap
    self._separators = separators or _DEFAULT_STRATEGY.separators

name class-attribute instance-attribute

Python
name = 'recursive_text'

chunk async

Python
chunk(documents: list[Document], strategy: ChunkStrategy | None = None) -> list[Chunk]
Source code in apogee_ai_rag/infrastructure/chunkers/recursive_text_chunker.py
Python
async def chunk(
    self, documents: list[Document], strategy: ChunkStrategy | None = None
) -> list[Chunk]:
    size = strategy.chunk_size if strategy else self._chunk_size
    overlap = strategy.chunk_overlap if strategy else self._chunk_overlap
    seps = strategy.separators if strategy else self._separators

    chunks: list[Chunk] = []
    for doc in documents:
        if not doc.text:
            continue
        parts = _recursive_split(doc.text, seps, size)
        parts = _apply_overlap(parts, overlap)
        for position, text in enumerate(parts):
            chunks.append(
                Chunk(
                    text=text,
                    parent_id=doc.id,
                    position=position,
                    metadata=dict(doc.metadata),
                    modality=doc.modality,
                )
            )
    return chunks

RedisCacheStore

Python
RedisCacheStore(*, url: str, namespace: str = 'rag')
Source code in apogee_ai_rag/infrastructure/cache_stores/redis_cache_store.py
Python
def __init__(self, *, url: str, namespace: str = "rag") -> None:
    try:
        import redis.asyncio as redis_asyncio  # type: ignore[import-untyped]
    except ImportError as exc:
        raise ProviderNotInstalledError("redis", "redis") from exc
    self._client = redis_asyncio.from_url(url)
    self._ns = namespace

name class-attribute instance-attribute

Python
name = 'redis'

get async

Python
get(key: str) -> CacheEntry | None
Source code in apogee_ai_rag/infrastructure/cache_stores/redis_cache_store.py
Python
async def get(self, key: str) -> CacheEntry | None:  # pragma: no cover - I/O
    raw = await self._client.get(self._k(key))
    if raw is None:
        return None
    return CacheEntry(key=key, value=raw)

set async

Python
set(entry: CacheEntry) -> None
Source code in apogee_ai_rag/infrastructure/cache_stores/redis_cache_store.py
Python
async def set(self, entry: CacheEntry) -> None:  # pragma: no cover - I/O
    await self._client.set(
        self._k(entry.key), entry.value, ex=entry.ttl_seconds,
    )

invalidate async

Python
invalidate(prefix: str) -> int
Source code in apogee_ai_rag/infrastructure/cache_stores/redis_cache_store.py
Python
async def invalidate(self, prefix: str) -> int:  # pragma: no cover - I/O
    cursor = 0
    deleted = 0
    pattern = self._k(prefix) + "*"
    while True:
        cursor, keys = await self._client.scan(cursor, match=pattern, count=128)
        if keys:
            deleted += await self._client.delete(*keys)
        if cursor == 0:
            break
    return deleted

RerankConfig dataclass

Python
RerankConfig(provider: RerankerProvider = NONE, model: str | None = None, top_n: int = 5)

provider class-attribute instance-attribute

Python
provider: RerankerProvider = NONE

model class-attribute instance-attribute

Python
model: str | None = None

top_n class-attribute instance-attribute

Python
top_n: int = 5

RerankerProvider

Bases: StrEnum

NONE class-attribute instance-attribute

Python
NONE = 'none'

COHERE class-attribute instance-attribute

Python
COHERE = 'cohere'

BGE class-attribute instance-attribute

Python
BGE = 'bge'

CROSS_ENCODER class-attribute instance-attribute

Python
CROSS_ENCODER = 'cross_encoder'

COLBERT class-attribute instance-attribute

Python
COLBERT = 'colbert'

JINA class-attribute instance-attribute

Python
JINA = 'jina'

VOYAGE class-attribute instance-attribute

Python
VOYAGE = 'voyage'

RANKGPT class-attribute instance-attribute

Python
RANKGPT = 'rankgpt'

MONOT5 class-attribute instance-attribute

Python
MONOT5 = 'monot5'

RerankingRagPipeline

Python
RerankingRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, reranker: IReranker, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, over_fetch: int = 4)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/reranking/reranking_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    reranker: IReranker,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    over_fetch: int = 4,
) -> None:
    if reranker is None:
        raise PipelineConfigError("RerankingRagPipeline requires a reranker")
    if over_fetch < 1:
        raise PipelineConfigError("over_fetch must be >= 1")
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._reranker = reranker
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._over_fetch = over_fetch

name class-attribute instance-attribute

Python
name = 'reranking'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/reranking/reranking_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/reranking/reranking_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    response = await self._generator.generate(
        query.text, retrieved, self._generation_config,
    )
    response.query_id = query.id
    await self._emit(
        "query.generated", query_id=query.id, finish_reason=response.finish_reason,
    )
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/reranking/reranking_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    retrieved = await self._retrieve(query)
    async for chunk in self._generator.stream(
        query.text, retrieved, self._generation_config,
    ):
        yield chunk

RetrievalConfig dataclass

Python
RetrievalConfig(top_k: int = 5, threshold: float = 0.0, algorithm: SearchAlgorithm = COSINE, filters: dict = dict(), hybrid_alpha: float = 0.5)

top_k class-attribute instance-attribute

Python
top_k: int = 5

threshold class-attribute instance-attribute

Python
threshold: float = 0.0

algorithm class-attribute instance-attribute

Python
algorithm: SearchAlgorithm = COSINE

filters class-attribute instance-attribute

Python
filters: dict = field(default_factory=dict)

hybrid_alpha class-attribute instance-attribute

Python
hybrid_alpha: float = 0.5

RetrievedChunk dataclass

Python
RetrievedChunk(chunk: Chunk, score: float, retriever: str = 'vector')

chunk instance-attribute

Python
chunk: Chunk

score instance-attribute

Python
score: float

retriever class-attribute instance-attribute

Python
retriever: str = 'vector'

ScoreThresholdReranker

Python
ScoreThresholdReranker(min_score: float = 0.0)

Drops every chunk whose score falls below min_score and truncates to top_n.

Source code in apogee_ai_rag/infrastructure/rerankers/score_threshold_reranker.py
Python
def __init__(self, min_score: float = 0.0) -> None:
    self._min_score = min_score

name class-attribute instance-attribute

Python
name = 'score_threshold'

rerank async

Python
rerank(query: str, chunks: list[RetrievedChunk], top_n: int = 5) -> list[RetrievedChunk]
Source code in apogee_ai_rag/infrastructure/rerankers/score_threshold_reranker.py
Python
async def rerank(
    self, query: str, chunks: list[RetrievedChunk], top_n: int = 5
) -> list[RetrievedChunk]:
    del query
    kept = [c for c in chunks if c.score >= self._min_score]
    kept.sort(key=lambda r: r.score, reverse=True)
    return kept[:top_n]

SearchAlgorithm

Bases: StrEnum

COSINE class-attribute instance-attribute

Python
COSINE = 'cosine'

DOT class-attribute instance-attribute

Python
DOT = 'dot'

EUCLIDEAN class-attribute instance-attribute

Python
EUCLIDEAN = 'euclidean'

BM25 class-attribute instance-attribute

Python
BM25 = 'bm25'

TFIDF class-attribute instance-attribute

Python
TFIDF = 'tfidf'

HNSW class-attribute instance-attribute

Python
HNSW = 'hnsw'

IVF class-attribute instance-attribute

Python
IVF = 'ivf'

IVF_PQ class-attribute instance-attribute

Python
IVF_PQ = 'ivf_pq'

DISKANN class-attribute instance-attribute

Python
DISKANN = 'diskann'

SCANN class-attribute instance-attribute

Python
SCANN = 'scann'

DPR class-attribute instance-attribute

Python
DPR = 'dpr'

SPLADE class-attribute instance-attribute

Python
SPLADE = 'splade'

COLBERT class-attribute instance-attribute

Python
COLBERT = 'colbert'

HYBRID class-attribute instance-attribute

Python
HYBRID = 'hybrid'

RRF class-attribute instance-attribute

Python
RRF = 'rrf'

SelfRagPipeline

Python
SelfRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, retrieve_min_words: int = 4, relevance_threshold: float = 0.15)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/self_rag/self_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    retrieve_min_words: int = 4,
    relevance_threshold: float = 0.15,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "SelfRagPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._generator = generator
    self._emitter = default_emitter(observability)
    self._generation_config = generation_config
    self._retrieve_min_words = retrieve_min_words
    self._relevance_threshold = relevance_threshold

name class-attribute instance-attribute

Python
name = 'self_rag'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/self_rag/self_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/self_rag/self_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    await self._emit(
        "query.received", query_id=query.id, text=query.text, pipeline_type=self.name,
    )
    sources, verdicts = await self._retrieve_if_needed(query)
    response = await self._generator.generate(
        query.text, sources, self._generation_config,
    )
    supported = self._is_supported(response.answer, sources)
    response.sources = sources
    response.query_id = query.id
    response.confidence = (
        0.95 if supported and verdicts["relevant"]
        else 0.5 if verdicts["retrieve"]
        else 0.3
    )
    response.traces.append({
        "pipeline": self.name,
        "verdicts": verdicts,
        "supported": supported,
    })
    await self._emit(
        "query.generated",
        query_id=query.id,
        confidence=response.confidence,
        supported=supported,
    )
    return response

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/self_rag/self_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    sources, _ = await self._retrieve_if_needed(query)
    async for chunk in self._generator.stream(
        query.text, sources, self._generation_config,
    ):
        yield chunk

SpeculativeRagPipeline

Python
SpeculativeRagPipeline(chunker: IChunker, embedder: IEmbedder, vector_store: IVectorStore, generator: IGenerator, observability: IRagObservabilityEmitter | None = None, generation_config: GenerationConfig | None = None, drafter: IGenerator | None = None, drafter_config: GenerationConfig | None = None)

Bases: IngestionMixin

Source code in apogee_ai_rag/infrastructure/pipelines/speculative/speculative_rag_pipeline.py
Python
def __init__(  # noqa: PLR0913
    self,
    chunker: IChunker,
    embedder: IEmbedder,
    vector_store: IVectorStore,
    generator: IGenerator,
    observability: IRagObservabilityEmitter | None = None,
    generation_config: GenerationConfig | None = None,
    drafter: IGenerator | None = None,
    drafter_config: GenerationConfig | None = None,
) -> None:
    if not all([chunker, embedder, vector_store, generator]):
        raise PipelineConfigError(
            "SpeculativeRagPipeline requires chunker, embedder, store and generator"
        )
    self._chunker = chunker
    self._embedder = embedder
    self._store = vector_store
    self._verifier = generator
    self._drafter = drafter or generator
    self._emitter = default_emitter(observability)
    self._verifier_config = generation_config
    self._drafter_config = drafter_config or GenerationConfig(
        temperature=0.0, max_tokens=192,
    )

name class-attribute instance-attribute

Python
name = 'speculative'

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/infrastructure/pipelines/speculative/speculative_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult:
    return await self._ingest_default(job)

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/infrastructure/pipelines/speculative/speculative_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse:
    retrieved = await self._retrieve(query)
    draft_context = retrieved[:1]
    draft = await self._drafter.generate(
        query.text, draft_context, self._drafter_config,
    )
    await self._emit(
        "speculative.drafted",
        query_id=query.id,
        draft_chars=len(draft.answer or ""),
    )

    verifier_prompt = (
        f"Question: {query.text}\n\n"
        f"Draft answer (from a small model): {draft.answer}\n\n"
        "Using the supplied context, ratify or rewrite the draft. "
        "Quote the brackets when referencing sources."
    )
    verified = await self._verifier.generate(
        verifier_prompt, retrieved, self._verifier_config,
    )
    verified.sources = retrieved
    verified.query_id = query.id
    verified.traces.append({
        "pipeline": self.name,
        "draft_answer": draft.answer,
        "draft_sources": len(draft_context),
    })
    # Track the draft as an auxiliary chunk for downstream UIs.
    verified.sources = list(verified.sources) + [
        RetrievedChunk(
            chunk=Chunk(text=draft.answer or "", parent_id="speculative_draft"),
            score=0.0,
            retriever="speculative_draft",
        ),
    ]
    await self._emit("query.generated", query_id=query.id)
    return verified

stream async

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/infrastructure/pipelines/speculative/speculative_rag_pipeline.py
Python
async def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]:
    response = await self.run(query)
    async for chunk in self._verifier.stream(
        query.text,
        [r for r in response.sources if r.retriever != "speculative_draft"],
        self._verifier_config,
    ):
        yield chunk

TokenOverlapReranker

Python
TokenOverlapReranker(alpha: float = 0.5)

Cheap dense+sparse re-ranker — blends dense score with token overlap.

Useful as a baseline when no Cohere/BGE/ColBERT key is available. The final score is alpha * dense + (1 - alpha) * overlap, both normalised to [0, 1]. Mirrors how :class:InMemoryVectorStore.hybrid_search blends signals, but is applied on top of an already-retrieved list.

Source code in apogee_ai_rag/infrastructure/rerankers/token_overlap_reranker.py
Python
def __init__(self, alpha: float = 0.5) -> None:
    if not 0.0 <= alpha <= 1.0:
        raise ValueError("alpha must be in [0, 1]")
    self._alpha = alpha

name class-attribute instance-attribute

Python
name = 'token_overlap'

rerank async

Python
rerank(query: str, chunks: list[RetrievedChunk], top_n: int = 5) -> list[RetrievedChunk]
Source code in apogee_ai_rag/infrastructure/rerankers/token_overlap_reranker.py
Python
async def rerank(
    self, query: str, chunks: list[RetrievedChunk], top_n: int = 5
) -> list[RetrievedChunk]:
    if not chunks:
        return []
    query_tokens = set(tokenize(query))
    max_dense = max((c.score for c in chunks), default=1.0) or 1.0

    rescored: list[RetrievedChunk] = []
    for retrieved in chunks:
        doc_tokens = set(tokenize(retrieved.chunk.text))
        overlap = (
            len(query_tokens & doc_tokens) / max(len(query_tokens), 1)
            if query_tokens
            else 0.0
        )
        dense_norm = retrieved.score / max_dense if max_dense else 0.0
        blended = self._alpha * dense_norm + (1.0 - self._alpha) * overlap
        rescored.append(
            RetrievedChunk(
                chunk=retrieved.chunk,
                score=blended,
                retriever=f"{retrieved.retriever}+overlap",
            ),
        )
    rescored.sort(key=lambda r: r.score, reverse=True)
    return rescored[:top_n]

VectorStoreProvider

Bases: StrEnum

IN_MEMORY class-attribute instance-attribute

Python
IN_MEMORY = 'in_memory'

PGVECTOR class-attribute instance-attribute

Python
PGVECTOR = 'pgvector'

QDRANT class-attribute instance-attribute

Python
QDRANT = 'qdrant'

WEAVIATE class-attribute instance-attribute

Python
WEAVIATE = 'weaviate'

PINECONE class-attribute instance-attribute

Python
PINECONE = 'pinecone'

MILVUS class-attribute instance-attribute

Python
MILVUS = 'milvus'

CHROMA class-attribute instance-attribute

Python
CHROMA = 'chroma'

FAISS class-attribute instance-attribute

Python
FAISS = 'faiss'

LANCEDB class-attribute instance-attribute

Python
LANCEDB = 'lancedb'

DEEP_LAKE class-attribute instance-attribute

Python
DEEP_LAKE = 'deep_lake'

ELASTICSEARCH class-attribute instance-attribute

Python
ELASTICSEARCH = 'elasticsearch'

OPENSEARCH class-attribute instance-attribute

Python
OPENSEARCH = 'opensearch'

REDIS class-attribute instance-attribute

Python
REDIS = 'redis'

MONGO_ATLAS class-attribute instance-attribute

Python
MONGO_ATLAS = 'mongo_atlas'

SINGLESTORE class-attribute instance-attribute

Python
SINGLESTORE = 'singlestore'

TYPESENSE class-attribute instance-attribute

Python
TYPESENSE = 'typesense'

VESPA class-attribute instance-attribute

Python
VESPA = 'vespa'

VALD class-attribute instance-attribute

Python
VALD = 'vald'

MARQO class-attribute instance-attribute

Python
MARQO = 'marqo'

ANNOY class-attribute instance-attribute

Python
ANNOY = 'annoy'

SCANN class-attribute instance-attribute

Python
SCANN = 'scann'

ZILLIZ class-attribute instance-attribute

Python
ZILLIZ = 'zilliz'

APERTUREDB class-attribute instance-attribute

Python
APERTUREDB = 'aperturedb'

reciprocal_rank_fusion

Python
reciprocal_rank_fusion(rankings: Iterable[list[RetrievedChunk]], *, k: int = 60, top_k: int | None = None, retriever_label: str = 'rrf') -> list[RetrievedChunk]

Fuses several ranked lists using reciprocal rank fusion.

Score for a chunk = Σ 1/(k + rank_i) across every list it appears in. Returns the merged list sorted by descending RRF score, optionally truncated to top_k items.

Source code in apogee_ai_rag/infrastructure/search/rrf.py
Python
def reciprocal_rank_fusion(
    rankings: Iterable[list[RetrievedChunk]],
    *,
    k: int = 60,
    top_k: int | None = None,
    retriever_label: str = "rrf",
) -> list[RetrievedChunk]:
    """Fuses several ranked lists using reciprocal rank fusion.

    Score for a chunk = Σ 1/(k + rank_i) across every list it appears in.
    Returns the merged list sorted by descending RRF score, optionally
    truncated to ``top_k`` items.
    """
    scores: dict[str, float] = {}
    repr_chunk: dict[str, RetrievedChunk] = {}

    for ranking in rankings:
        for rank, retrieved in enumerate(ranking, start=1):
            cid = retrieved.chunk.id
            scores[cid] = scores.get(cid, 0.0) + 1.0 / (k + rank)
            repr_chunk.setdefault(cid, retrieved)

    fused = [
        RetrievedChunk(
            chunk=repr_chunk[cid].chunk,
            score=score,
            retriever=retriever_label,
        )
        for cid, score in scores.items()
    ]
    fused.sort(key=lambda r: r.score, reverse=True)
    if top_k is not None:
        fused = fused[:top_k]
    return fused

tokenize

Python
tokenize(text: str, lowercase: bool = True) -> list[str]

Splits text into word tokens.

Locale-agnostic, dependency-free. lowercase=True matches the behaviour expected by the BM25 implementation and by most embedders.

Source code in apogee_ai_rag/infrastructure/search/tokenizer.py
Python
def tokenize(text: str, lowercase: bool = True) -> list[str]:
    """Splits text into word tokens.

    Locale-agnostic, dependency-free. ``lowercase=True`` matches the
    behaviour expected by the BM25 implementation and by most embedders.
    """
    text = text or ""
    tokens = _TOKEN_RE.findall(text)
    if lowercase:
        tokens = [t.lower() for t in tokens]
    return tokens

Other · DTOs

ChunkDTO

Bases: BaseModel

id class-attribute instance-attribute

Python
id: str | None = None

parent_id instance-attribute

Python
parent_id: str

position class-attribute instance-attribute

Python
position: int = 0

text instance-attribute

Python
text: str

metadata class-attribute instance-attribute

Python
metadata: dict = Field(default_factory=dict)

modality class-attribute instance-attribute

Python
modality: Modality = TEXT

DocumentDTO

Bases: BaseModel

id class-attribute instance-attribute

Python
id: str | None = None

text class-attribute instance-attribute

Python
text: str = ''

metadata class-attribute instance-attribute

Python
metadata: dict = Field(default_factory=dict)

modality class-attribute instance-attribute

Python
modality: Modality = TEXT

image_b64 class-attribute instance-attribute

Python
image_b64: str | None = None

audio_url class-attribute instance-attribute

Python
audio_url: str | None = None

video_url class-attribute instance-attribute

Python
video_url: str | None = None

EvalCaseDTO

Bases: BaseModel

question instance-attribute

Python
question: str

ground_truth class-attribute instance-attribute

Python
ground_truth: str | None = None

expected_substrings class-attribute instance-attribute

Python
expected_substrings: list[str] = Field(default_factory=list)

contexts class-attribute instance-attribute

Python
contexts: list[str] = Field(default_factory=list)

metadata class-attribute instance-attribute

Python
metadata: dict = Field(default_factory=dict)

EvalReportDTO

Bases: BaseModel

metrics class-attribute instance-attribute

Python
metrics: dict[str, float] = Field(default_factory=dict)

per_case class-attribute instance-attribute

Python
per_case: list[EvalResultDTO] = Field(default_factory=list)

passed class-attribute instance-attribute

Python
passed: int = 0

failed class-attribute instance-attribute

Python
failed: int = 0

score class-attribute instance-attribute

Python
score: float = 0.0

EvalResultDTO

Bases: BaseModel

case instance-attribute

Python
case: EvalCaseDTO

answer instance-attribute

Python
answer: str

metrics class-attribute instance-attribute

Python
metrics: dict[str, float] = Field(default_factory=dict)

passed class-attribute instance-attribute

Python
passed: bool = False

error class-attribute instance-attribute

Python
error: str | None = None

IngestionJobDTO

Bases: BaseModel

documents instance-attribute

Python
documents: list[DocumentDTO]

batch_size class-attribute instance-attribute

Python
batch_size: int = 64

idempotency_key class-attribute instance-attribute

Python
idempotency_key: str | None = None

metadata class-attribute instance-attribute

Python
metadata: dict = Field(default_factory=dict)

IngestionResultDTO

Bases: BaseModel

job_id instance-attribute

Python
job_id: str

documents_ingested instance-attribute

Python
documents_ingested: int

chunks_produced instance-attribute

Python
chunks_produced: int

vectors_upserted instance-attribute

Python
vectors_upserted: int

duration_ms class-attribute instance-attribute

Python
duration_ms: float = 0.0

errors class-attribute instance-attribute

Python
errors: list[str] = Field(default_factory=list)

RagChunkDTO

Bases: BaseModel

delta instance-attribute

Python
delta: str

sources class-attribute instance-attribute

Python
sources: list[RetrievedChunkDTO] = Field(default_factory=list)

finish_reason class-attribute instance-attribute

Python
finish_reason: str | None = None

RagQueryDTO

Bases: BaseModel

text instance-attribute

Python
text: str

filters class-attribute instance-attribute

Python
filters: dict = Field(default_factory=dict)

top_k class-attribute instance-attribute

Python
top_k: int = 5

threshold class-attribute instance-attribute

Python
threshold: float = 0.0

modality class-attribute instance-attribute

Python
modality: Modality = TEXT

history class-attribute instance-attribute

Python
history: list[dict] = Field(default_factory=list)

conversation_id class-attribute instance-attribute

Python
conversation_id: str | None = None

locale class-attribute instance-attribute

Python
locale: str | None = None

image_b64 class-attribute instance-attribute

Python
image_b64: str | None = None

RagResponseDTO

Bases: BaseModel

answer instance-attribute

Python
answer: str

sources class-attribute instance-attribute

Python
sources: list[RetrievedChunkDTO] = Field(default_factory=list)

confidence class-attribute instance-attribute

Python
confidence: float | None = None

finish_reason class-attribute instance-attribute

Python
finish_reason: str | None = None

usage class-attribute instance-attribute

Python
usage: dict = Field(default_factory=dict)

query_id class-attribute instance-attribute

Python
query_id: str | None = None

RetrievedChunkDTO

Bases: BaseModel

chunk instance-attribute

Python
chunk: ChunkDTO

score instance-attribute

Python
score: float

retriever class-attribute instance-attribute

Python
retriever: str = 'vector'

Other · Enums

ChunkerKind

Bases: StrEnum

RECURSIVE_TEXT class-attribute instance-attribute

Python
RECURSIVE_TEXT = 'recursive_text'

MARKDOWN class-attribute instance-attribute

Python
MARKDOWN = 'markdown'

CODE class-attribute instance-attribute

Python
CODE = 'code'

SEMANTIC class-attribute instance-attribute

Python
SEMANTIC = 'semantic'

UNSTRUCTURED class-attribute instance-attribute

Python
UNSTRUCTURED = 'unstructured'

LLAMAPARSE class-attribute instance-attribute

Python
LLAMAPARSE = 'llamaparse'

DOCLING class-attribute instance-attribute

Python
DOCLING = 'docling'

PYMUPDF class-attribute instance-attribute

Python
PYMUPDF = 'pymupdf'

PDFPLUMBER class-attribute instance-attribute

Python
PDFPLUMBER = 'pdfplumber'

MARKER class-attribute instance-attribute

Python
MARKER = 'marker'

NOUGAT class-attribute instance-attribute

Python
NOUGAT = 'nougat'

TIKA class-attribute instance-attribute

Python
TIKA = 'tika'

RagType

Bases: StrEnum

NAIVE class-attribute instance-attribute

Python
NAIVE = 'naive'

ADVANCED class-attribute instance-attribute

Python
ADVANCED = 'advanced'

MODULAR class-attribute instance-attribute

Python
MODULAR = 'modular'

SELF_RAG class-attribute instance-attribute

Python
SELF_RAG = 'self_rag'

CRAG class-attribute instance-attribute

Python
CRAG = 'crag'

SPECULATIVE class-attribute instance-attribute

Python
SPECULATIVE = 'speculative'

ADAPTIVE class-attribute instance-attribute

Python
ADAPTIVE = 'adaptive'

AGENTIC class-attribute instance-attribute

Python
AGENTIC = 'agentic'

GRAPH class-attribute instance-attribute

Python
GRAPH = 'graph'

HYDE class-attribute instance-attribute

Python
HYDE = 'hyde'

MULTI_HOP class-attribute instance-attribute

Python
MULTI_HOP = 'multi_hop'

HIERARCHICAL class-attribute instance-attribute

Python
HIERARCHICAL = 'hierarchical'

RECURSIVE class-attribute instance-attribute

Python
RECURSIVE = 'recursive'

HYBRID class-attribute instance-attribute

Python
HYBRID = 'hybrid'

RAG_FUSION class-attribute instance-attribute

Python
RAG_FUSION = 'rag_fusion'

FLARE class-attribute instance-attribute

Python
FLARE = 'flare'

RAPTOR class-attribute instance-attribute

Python
RAPTOR = 'raptor'

REACT class-attribute instance-attribute

Python
REACT = 'react'

ITERATIVE class-attribute instance-attribute

Python
ITERATIVE = 'iterative'

CONVERSATIONAL class-attribute instance-attribute

Python
CONVERSATIONAL = 'conversational'

MULTI_MODAL class-attribute instance-attribute

Python
MULTI_MODAL = 'multi_modal'

LONG class-attribute instance-attribute

Python
LONG = 'long'

QUERY_REWRITING class-attribute instance-attribute

Python
QUERY_REWRITING = 'query_rewriting'

RERANKING class-attribute instance-attribute

Python
RERANKING = 'reranking'

CONTEXTUAL class-attribute instance-attribute

Python
CONTEXTUAL = 'contextual'

KAG class-attribute instance-attribute

Python
KAG = 'kag'

CAG class-attribute instance-attribute

Python
CAG = 'cag'

Other · Exceptions

EmbeddingError

Python
EmbeddingError(message: str, *, stage: str | None = None, cause: Exception | None = None)

Bases: RagError

Source code in apogee_ai_rag/domain/exceptions/rag_error.py
Python
def __init__(self, message: str, *, stage: str | None = None, cause: Exception | None = None) -> None:
    super().__init__(message)
    self.stage = stage
    self.cause = cause

EvaluationError

Python
EvaluationError(message: str, *, stage: str | None = None, cause: Exception | None = None)

Bases: RagError

Source code in apogee_ai_rag/domain/exceptions/rag_error.py
Python
def __init__(self, message: str, *, stage: str | None = None, cause: Exception | None = None) -> None:
    super().__init__(message)
    self.stage = stage
    self.cause = cause

GenerationError

Python
GenerationError(message: str, *, stage: str | None = None, cause: Exception | None = None)

Bases: RagError

Source code in apogee_ai_rag/domain/exceptions/rag_error.py
Python
def __init__(self, message: str, *, stage: str | None = None, cause: Exception | None = None) -> None:
    super().__init__(message)
    self.stage = stage
    self.cause = cause

IngestionError

Python
IngestionError(message: str, *, stage: str | None = None, cause: Exception | None = None)

Bases: RagError

Source code in apogee_ai_rag/domain/exceptions/rag_error.py
Python
def __init__(self, message: str, *, stage: str | None = None, cause: Exception | None = None) -> None:
    super().__init__(message)
    self.stage = stage
    self.cause = cause

PipelineConfigError

Python
PipelineConfigError(message: str, *, stage: str | None = None, cause: Exception | None = None)

Bases: RagError

Source code in apogee_ai_rag/domain/exceptions/rag_error.py
Python
def __init__(self, message: str, *, stage: str | None = None, cause: Exception | None = None) -> None:
    super().__init__(message)
    self.stage = stage
    self.cause = cause

ProviderNotInstalledError

Python
ProviderNotInstalledError(provider: str, extra: str)

Bases: RagError

Raised when an optional provider is selected but its extra is missing.

Carries the suggested pip install apogee-ai-rag[<extra>] command so the caller can recover automatically (e.g. CLI surfaces it as the only message).

Source code in apogee_ai_rag/domain/exceptions/provider_not_installed_error.py
Python
def __init__(self, provider: str, extra: str) -> None:
    message = (
        f"Provider '{provider}' is not installed. "
        f"Run: pip install 'apogee-ai-rag[{extra}]'"
    )
    super().__init__(message)
    self.provider = provider
    self.extra = extra

provider instance-attribute

Python
provider = provider

extra instance-attribute

Python
extra = extra

RagError

Python
RagError(message: str, *, stage: str | None = None, cause: Exception | None = None)

Bases: Exception

Source code in apogee_ai_rag/domain/exceptions/rag_error.py
Python
def __init__(self, message: str, *, stage: str | None = None, cause: Exception | None = None) -> None:
    super().__init__(message)
    self.stage = stage
    self.cause = cause

stage instance-attribute

Python
stage = stage

cause instance-attribute

Python
cause = cause

RerankError

Python
RerankError(message: str, *, stage: str | None = None, cause: Exception | None = None)

Bases: RagError

Source code in apogee_ai_rag/domain/exceptions/rag_error.py
Python
def __init__(self, message: str, *, stage: str | None = None, cause: Exception | None = None) -> None:
    super().__init__(message)
    self.stage = stage
    self.cause = cause

RetrievalError

Python
RetrievalError(message: str, *, stage: str | None = None, cause: Exception | None = None)

Bases: RagError

Source code in apogee_ai_rag/domain/exceptions/rag_error.py
Python
def __init__(self, message: str, *, stage: str | None = None, cause: Exception | None = None) -> None:
    super().__init__(message)
    self.stage = stage
    self.cause = cause

UnsupportedModalityError

Python
UnsupportedModalityError(message: str, *, stage: str | None = None, cause: Exception | None = None)

Bases: RagError

Source code in apogee_ai_rag/domain/exceptions/rag_error.py
Python
def __init__(self, message: str, *, stage: str | None = None, cause: Exception | None = None) -> None:
    super().__init__(message)
    self.stage = stage
    self.cause = cause

VectorStoreError

Python
VectorStoreError(message: str, *, stage: str | None = None, cause: Exception | None = None)

Bases: RagError

Source code in apogee_ai_rag/domain/exceptions/rag_error.py
Python
def __init__(self, message: str, *, stage: str | None = None, cause: Exception | None = None) -> None:
    super().__init__(message)
    self.stage = stage
    self.cause = cause

Other · Protocols (ports)

ICacheStore

Bases: Protocol

name instance-attribute

Python
name: str

get async

Python
get(key: str) -> CacheEntry | None
Source code in apogee_ai_rag/domain/services/i_cache_store.py
Python
async def get(self, key: str) -> CacheEntry | None: ...

set async

Python
set(entry: CacheEntry) -> None
Source code in apogee_ai_rag/domain/services/i_cache_store.py
Python
async def set(self, entry: CacheEntry) -> None: ...

invalidate async

Python
invalidate(prefix: str) -> int
Source code in apogee_ai_rag/domain/services/i_cache_store.py
Python
async def invalidate(self, prefix: str) -> int: ...

IChunker

Bases: Protocol

name instance-attribute

Python
name: str

chunk async

Python
chunk(documents: list[Document], strategy: ChunkStrategy | None = None) -> list[Chunk]
Source code in apogee_ai_rag/domain/services/i_chunker.py
Python
async def chunk(
    self, documents: list[Document], strategy: ChunkStrategy | None = None
) -> list[Chunk]: ...

IEmbedder

Bases: Protocol

model instance-attribute

Python
model: str

dims instance-attribute

Python
dims: int

embed async

Python
embed(texts: list[str], modality: Modality = TEXT) -> list[list[float]]
Source code in apogee_ai_rag/domain/services/i_embedder.py
Python
async def embed(
    self, texts: list[str], modality: Modality = Modality.TEXT
) -> list[list[float]]: ...

IGenerator

Bases: Protocol

name instance-attribute

Python
name: str

generate async

Python
generate(prompt: str, context: list[RetrievedChunk], config: GenerationConfig | None = None) -> RagResponse
Source code in apogee_ai_rag/domain/services/i_generator.py
Python
async def generate(
    self,
    prompt: str,
    context: list[RetrievedChunk],
    config: GenerationConfig | None = None,
) -> RagResponse: ...

stream

Python
stream(prompt: str, context: list[RetrievedChunk], config: GenerationConfig | None = None) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/domain/services/i_generator.py
Python
def stream(
    self,
    prompt: str,
    context: list[RetrievedChunk],
    config: GenerationConfig | None = None,
) -> AsyncIterator[RagChunk]: ...

IGraphStore

Bases: Protocol

name instance-attribute

Python
name: str

add_node async

Python
add_node(node: KnowledgeNode) -> str
Source code in apogee_ai_rag/domain/services/i_graph_store.py
Python
async def add_node(self, node: KnowledgeNode) -> str: ...

add_edge async

Python
add_edge(edge: KnowledgeEdge) -> str
Source code in apogee_ai_rag/domain/services/i_graph_store.py
Python
async def add_edge(self, edge: KnowledgeEdge) -> str: ...

traverse async

Python
traverse(start_id: str, depth: int = 2) -> tuple[list[KnowledgeNode], list[KnowledgeEdge]]
Source code in apogee_ai_rag/domain/services/i_graph_store.py
Python
async def traverse(
    self, start_id: str, depth: int = 2
) -> tuple[list[KnowledgeNode], list[KnowledgeEdge]]: ...

IIngestor

Bases: Protocol

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/domain/services/i_ingestor.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult: ...

IQueryRewriter

Bases: Protocol

rewrite async

Python
rewrite(query: RagQuery) -> list[str]
Source code in apogee_ai_rag/domain/services/i_query_rewriter.py
Python
async def rewrite(self, query: RagQuery) -> list[str]: ...

IRagEvaluator

Bases: Protocol

name instance-attribute

Python
name: str

evaluate async

Python
evaluate(suite: list[EvalCase], pipeline: IRagPipeline) -> EvalReport
Source code in apogee_ai_rag/domain/services/i_rag_evaluator.py
Python
async def evaluate(
    self, suite: list[EvalCase], pipeline: IRagPipeline
) -> EvalReport: ...

IRagObservabilityEmitter

Bases: Protocol

name instance-attribute

Python
name: str

emit async

Python
emit(event: RagEvent) -> None
Source code in apogee_ai_rag/domain/services/i_rag_observability_emitter.py
Python
async def emit(self, event: RagEvent) -> None: ...

IRagPipeline

Bases: Protocol

name instance-attribute

Python
name: str

ingest async

Python
ingest(job: IngestionJob) -> IngestionResult
Source code in apogee_ai_rag/domain/services/i_rag_pipeline.py
Python
async def ingest(self, job: IngestionJob) -> IngestionResult: ...

run async

Python
run(query: RagQuery) -> RagResponse
Source code in apogee_ai_rag/domain/services/i_rag_pipeline.py
Python
async def run(self, query: RagQuery) -> RagResponse: ...

stream

Python
stream(query: RagQuery) -> AsyncIterator[RagChunk]
Source code in apogee_ai_rag/domain/services/i_rag_pipeline.py
Python
def stream(self, query: RagQuery) -> AsyncIterator[RagChunk]: ...

IReranker

Bases: Protocol

name instance-attribute

Python
name: str

rerank async

Python
rerank(query: str, chunks: list[RetrievedChunk], top_n: int = 5) -> list[RetrievedChunk]
Source code in apogee_ai_rag/domain/services/i_reranker.py
Python
async def rerank(
    self, query: str, chunks: list[RetrievedChunk], top_n: int = 5
) -> list[RetrievedChunk]: ...

IRetriever

Bases: Protocol

retrieve async

Python
retrieve(query: RagQuery) -> list[RetrievedChunk]
Source code in apogee_ai_rag/domain/services/i_retriever.py
Python
async def retrieve(self, query: RagQuery) -> list[RetrievedChunk]: ...

IVectorStore

Bases: Protocol

name instance-attribute

Python
name: str

upsert async

Python
upsert(chunks: list[Chunk]) -> int
Source code in apogee_ai_rag/domain/services/i_vector_store.py
Python
async def upsert(self, chunks: list[Chunk]) -> int: ...

search async

Python
search(query_embedding: list[float], top_k: int = 5, threshold: float = 0.0, filters: dict | None = None) -> list[RetrievedChunk]
Source code in apogee_ai_rag/domain/services/i_vector_store.py
Python
async def search(
    self,
    query_embedding: list[float],
    top_k: int = 5,
    threshold: float = 0.0,
    filters: dict | None = None,
) -> list[RetrievedChunk]: ...
Python
hybrid_search(query_text: str, query_embedding: list[float], top_k: int = 5, threshold: float = 0.0, filters: dict | None = None, alpha: float = 0.5) -> list[RetrievedChunk]
Source code in apogee_ai_rag/domain/services/i_vector_store.py
Python
async def hybrid_search(
    self,
    query_text: str,
    query_embedding: list[float],
    top_k: int = 5,
    threshold: float = 0.0,
    filters: dict | None = None,
    alpha: float = 0.5,
) -> list[RetrievedChunk]: ...

delete async

Python
delete(chunk_ids: list[str]) -> int
Source code in apogee_ai_rag/domain/services/i_vector_store.py
Python
async def delete(self, chunk_ids: list[str]) -> int: ...

count async

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

Other · Use cases

DeleteDocumentsUseCase

Python
DeleteDocumentsUseCase(store: IVectorStore)
Source code in apogee_ai_rag/application/use_cases/delete_documents_use_case.py
Python
def __init__(self, store: IVectorStore) -> None:
    self._store = store

execute async

Python
execute(chunk_ids: list[str]) -> int
Source code in apogee_ai_rag/application/use_cases/delete_documents_use_case.py
Python
async def execute(self, chunk_ids: list[str]) -> int:
    return await self._store.delete(chunk_ids)

EvaluateRagUseCase

Python
EvaluateRagUseCase(evaluator: IRagEvaluator, pipeline: IRagPipeline)
Source code in apogee_ai_rag/application/use_cases/evaluate_rag_use_case.py
Python
def __init__(self, evaluator: IRagEvaluator, pipeline: IRagPipeline) -> None:
    self._evaluator = evaluator
    self._pipeline = pipeline

execute async

Python
execute(suite: list[EvalCaseDTO]) -> EvalReportDTO
Source code in apogee_ai_rag/application/use_cases/evaluate_rag_use_case.py
Python
async def execute(self, suite: list[EvalCaseDTO]) -> EvalReportDTO:
    cases = [eval_case_from_dto(c) for c in suite]
    report = await self._evaluator.evaluate(cases, self._pipeline)
    return eval_report_to_dto(report)

FeedbackUseCase

Python
FeedbackUseCase(emitter: IRagObservabilityEmitter)
Source code in apogee_ai_rag/application/use_cases/feedback_use_case.py
Python
def __init__(self, emitter: IRagObservabilityEmitter) -> None:
    self._emitter = emitter

execute async

Python
execute(feedback: FeedbackInput) -> None
Source code in apogee_ai_rag/application/use_cases/feedback_use_case.py
Python
async def execute(self, feedback: FeedbackInput) -> None:
    attributes = {
        "query_id": feedback.query_id,
        "score": feedback.score,
    }
    if feedback.comment is not None:
        attributes["comment"] = feedback.comment
    if feedback.metadata:
        attributes.update(feedback.metadata)
    await self._emitter.emit(RagEvent(name="feedback.submitted", attributes=attributes))

IngestDocumentsUseCase

Python
IngestDocumentsUseCase(pipeline: IRagPipeline)
Source code in apogee_ai_rag/application/use_cases/ingest_documents_use_case.py
Python
def __init__(self, pipeline: IRagPipeline) -> None:
    self._pipeline = pipeline

execute async

Python
execute(dto: IngestionJobDTO) -> IngestionResultDTO
Source code in apogee_ai_rag/application/use_cases/ingest_documents_use_case.py
Python
async def execute(self, dto: IngestionJobDTO) -> IngestionResultDTO:
    job = ingestion_job_from_dto(dto)
    result = await self._pipeline.ingest(job)
    return ingestion_result_to_dto(result)

QueryRagUseCase

Python
QueryRagUseCase(pipeline: IRagPipeline)
Source code in apogee_ai_rag/application/use_cases/query_rag_use_case.py
Python
def __init__(self, pipeline: IRagPipeline) -> None:
    self._pipeline = pipeline

execute async

Python
execute(dto: RagQueryDTO) -> RagResponseDTO
Source code in apogee_ai_rag/application/use_cases/query_rag_use_case.py
Python
async def execute(self, dto: RagQueryDTO) -> RagResponseDTO:
    query = query_from_dto(dto)
    response = await self._pipeline.run(query)
    return response_to_dto(response)

RefreshIndexUseCase

Python
RefreshIndexUseCase(pipeline: IRagPipeline, store: IVectorStore)

Re-ingests a batch of documents after deleting their previous chunks.

The caller supplies an IVectorStore used to find and remove the chunks that share a parent_id with the incoming documents. Implementation is pluggable via the store's delete and count operations.

Source code in apogee_ai_rag/application/use_cases/refresh_index_use_case.py
Python
def __init__(self, pipeline: IRagPipeline, store: IVectorStore) -> None:
    self._ingest = IngestDocumentsUseCase(pipeline)
    self._store = store

execute async

Python
execute(dto: IngestionJobDTO) -> IngestionResultDTO
Source code in apogee_ai_rag/application/use_cases/refresh_index_use_case.py
Python
async def execute(self, dto: IngestionJobDTO) -> IngestionResultDTO:
    chunk_ids: list[str] = []
    for doc in dto.documents:
        if doc.id:
            chunk_ids.append(doc.id)
    if chunk_ids:
        await self._store.delete(chunk_ids)
    return await self._ingest.execute(dto)

StreamRagUseCase

Python
StreamRagUseCase(pipeline: IRagPipeline)
Source code in apogee_ai_rag/application/use_cases/stream_rag_use_case.py
Python
def __init__(self, pipeline: IRagPipeline) -> None:
    self._pipeline = pipeline

execute async

Python
execute(dto: RagQueryDTO) -> AsyncIterator[RagChunkDTO]
Source code in apogee_ai_rag/application/use_cases/stream_rag_use_case.py
Python
async def execute(self, dto: RagQueryDTO) -> AsyncIterator[RagChunkDTO]:
    query = query_from_dto(dto)
    async for chunk in self._pipeline.stream(query):
        yield RagChunkDTO(
            delta=chunk.delta,
            sources=[retrieved_chunk_to_dto(s) for s in chunk.sources],
            finish_reason=chunk.finish_reason,
        )