engine module

ContextEngine

class ContextEngine:
    def __init__(
        self,
        config:            Optional[ContextEngineConfig]      = None,
        query_analyzer:    Optional[QueryAnalyzerStrategy]    = None,
        retriever:         Optional[HybridRetriever]          = None,
        chunk_filter:      Optional[ChunkFilterStrategy]      = None,
        ranker:            Optional[ChunkRankingStrategy]     = None,
        composer:          Optional[ContextComposerStrategy]  = None,
        guard:             Optional[ContextGuardStrategy]     = None,
        budget_allocator:  Optional[BudgetAllocationStrategy] = None,
        length_counter:    Optional[LengthCounter]            = None,
    ): ...

Overview

Why it exists: to provide one call (run()) that takes a raw query and a pre-indexed corpus and returns a fully validated, budgeted context string plus complete diagnostics — without the caller needing to know about, or directly call, any of the six underlying strategy components.

Constructor — what gets built

self._counter defaults to char_counter.

Side effects: logs one INFO line on construction ("ContextEngine: Ready ✓ (template=%s)").

index(chunks, embed_fn=None, vector_store_adapter=None, vector_db=None, add_to_vector_db=True, source_quality_map=None)

def index(
    self,
    chunks:               List[Any],
    embed_fn:             Optional[Callable[[str], np.ndarray]] = None,
    vector_store_adapter: Optional[Any]                          = None,
    vector_db:            Optional[Any]                          = None,
    add_to_vector_db:     bool                                   = True,
    source_quality_map:   Optional[Dict[str, SourceQuality]]    = None,
) -> "ContextEngine": ...

Parameters

Algorithm

1. Bails out early (logs warning, returns self) if chunks is empty.
2. Stores source_quality_map (or {}).
3. Always constructs a fresh VectorRetriever, KeywordRetriever, MetadataRetriever — even if self._retriever was already injected at construction time (these freshly-built sub-retrievers are simply discarded in that case, since step 6 only assembles a HybridRetriever if self._retriever is None).
4. Vector-indexing branch, checked in this order:
   • If vector_db is given: when add_to_vector_db is True (default), calls vec_ret.index_vector_db(chunks, vector_db), which itself calls vector_db.add(chunks) — so the backend's own configured embedder (or each chunk's pre-computed .embedding) is used, not embed_fn. When add_to_vector_db=False, the retriever is simply pointed at vector_db without re-adding anything.
   • Otherwise, if embed_fn is given and vector_store_adapter is None: calls vec_ret.index_chunks(chunks, embed_fn) — this runs even if self._retriever is already set, meaning a throwaway VectorRetriever gets fully embedded for no benefit if you already injected your own retriever.
5. Builds KeywordRetriever/MetadataRetriever from chunks regardless of which branch above ran (or didn't).
6. Only if self._retriever is None, wraps the three sub-retrievers into a new HybridRetriever and assigns it.
7. Logs an INFO line with the chunk count.

[!WARNING] Calling index() after injecting a custom retriever If you passed retriever= to __init__, calling index() will still build (and potentially embed, or write into vector_db) a redundant VectorRetriever/KeywordRetriever/MetadataRetriever set that's silently discarded — wasting embedding calls or vector-db writes if embed_fn/vector_db is set. If you're using a fully custom retriever, you likely don't need to call index() at all.

Returns

self — supports method chaining: engine.index(chunks=...).run(...).

Raises

Propagates whatever embed_fn raises during the eager embedding pass (see retriever.VectorRetriever.index_chunks), or whatever vector_db.add() raises — e.g. a missing/misconfigured embedder on the backend, an embedding-dimension mismatch, or a connection error from the underlying Chroma/FAISS/Pinecone client.

index_documents(texts, chunker, doc_ids=None, sources=None, vector_db=None, source_quality_map=None)

def index_documents(
    self,
    texts:              List[str],
    chunker:            Any,
    doc_ids:            Optional[List[str]]                  = None,
    sources:            Optional[List[str]]                  = None,
    vector_db:          Optional[Any]                         = None,
    source_quality_map: Optional[Dict[str, SourceQuality]]   = None,
) -> "ContextEngine": ...

A native, end-to-end convenience wrapper around index(): takes raw text instead of pre-built chunks, splits it with any zerda.chunks chunker, and indexes the resulting DocumentChunks — into a native zerda.vector_database backend if vector_db is given, or into keyword/metadata retrieval only otherwise. This is the recommended entry point when you want the context engine wired directly to the rest of zerda (chunking + vector storage) without hand-building DocumentChunk lists yourself.

Parameters

Algorithm

1. Bails out early (logs warning, returns self) if texts is empty.
2. Fills in default doc_ids ("doc_{i}") / sources (same as doc_id) for any not supplied.
3. For each (text, doc_id, source) triple, calls chunker.chunk(text, doc_id=doc_id, source=source) and extends a flat all_chunks list. If the chunker raises TypeError on the source keyword — the legacy ChunkingStrategy interface only accepts chunk(text, doc_id) — retries without it.
4. Logs an INFO line with the text-count → chunk-count and the chunker's class name.
5. Delegates to self.index(chunks=all_chunks, vector_db=vector_db, source_quality_map=source_quality_map) — everything documented under index() above (including the vector_db-before-embed_fn precedence and the "already-injected retriever" warning) applies here too.

Returns

self — supports method chaining.

Raises

Propagates whatever the chunker or vector_db.add() raises — same failure modes as index().

[!NOTE] Importing zerda.context stays lightweight context.retriever only imports zerda.vector_database/zerda.chunks under TYPE_CHECKING, so import zerda.context never forces chromadb/faiss/pinecone/sentence-transformers to be installed. Actually calling index_documents() with a real chunker/vector_db does require whichever of those packages your chosen chunker/backend needs (e.g. sentence-transformers for ArabicTextChunker/ContextAwareChunker; chromadb, faiss, or pinecone for the matching vector_db backend).

run(query, metadata_filters=None, max_chunks=None, override_budget=None, use_history=True, record_turn=True)

def run(
    self,
    query:            str,
    metadata_filters: Optional[Dict[str, Any]] = None,
    max_chunks:       Optional[int]             = None,
    override_budget:  Optional[int]             = None,
    use_history:      bool                      = True,
    record_turn:      bool                      = True,
) -> ContextResult: ...

This is the primary entry point. See Architecture → The run() pipeline, step by step for the full sequence diagram; this section covers the API contract.

Parameters

Returns

A ContextResult. On a cache hit, returns early with chunks_used=[] and metadata={"from_cache": True} — note that on a cache hit, total_retrieved/total_after_filter/total_after_rank are all left at their dataclass default of 0, even though real chunks were used to build the originally-cached context.

Raises

No explicit exception handling wraps the pipeline stages in run() — an exception raised by any injected custom strategy (analyzer, retriever, filter, ranker, composer, guard) propagates directly to the caller.

Important implementation detail: effective_query vs. query

The query that gets analyzed and retrieved against (effective_query) may differ from the query that gets recorded into history (query, the original parameter):

effective_query = query
if use_history and self._history and self.config.multi_turn:
    effective_query = self._history.get_enriched_query(query, n=2)

analysis = self._analyzer.analyze(effective_query)
...
if record_turn and self._history:
    self._history.add_turn(query=query, context=context, analysis=analysis)

So result.query_analysis.original_query reflects effective_query (which may include the "[Previous conversation: ...]" wrapper), not the bare query argument you passed to run(). If you need the literal original query text, keep your own reference to it rather than reading it back off result.query_analysis.original_query.

Side effects

• Reads from and writes to the cache (if enabled).
• Mutates self._history (if record_turn).
• Mutates self._adaptive's internal rolling windows (always, if enable_adaptive).
• Mutates source_quality on the retrieved ScoredChunk objects in place, based on self._source_quality_map.
• Logs at INFO level with a full summary of retrieved/filtered/ranked counts, context length, and elapsed time.

Thread safety

Not thread-safe for concurrent calls on the same ContextEngine instance: ContextCache, ConversationHistory, and AdaptiveLearningSystem are all mutated without any locking.

Complexity

Dominated by the retrieval stage — see retriever.HybridRetriever.

run_multi(queries, global_budget=None, use_weighted=True, **kwargs)

def run_multi(
    self,
    queries:       List[str],
    global_budget: Optional[int] = None,
    use_weighted:  bool          = True,
    **kwargs,
) -> str: ...

Runs run() once per query (with record_turn=False forced for each sub-call — multi-query runs are never individually recorded into conversation history), allocates a sub-budget per query (via ComplexityBasedAllocation if use_weighted, else EqualBudgetAllocation), and joins the results with a per-query header line of the form "--- {label}: {query text} ---", where label is f"Query: {i+1}" if that query was detected as Arabic, or f"Query {i+1}" otherwise — note this produces a double colon ("Query: 1: <text>") for Arabic-detected queries specifically, since the label itself already contains a colon before the outer ": {q}" is appended. A mixed-language batch will therefore have inconsistently punctuated headers between queries.

Returns: a single combined str, safety-clamped to global_budget (default: config.composer.max_context_length) via the engine's private _hard_truncate. Returns "" immediately if queries is empty.

**kwargs are forwarded to every individual run() call (e.g. you could pass metadata_filters=... to apply the same filter to every query in the batch).

Legacy methods (tuple-based API)

These exist for compatibility with an older interface that worked with List[Tuple[chunk, score]] instead of List[ScoredChunk].

build(query, chunks, max_chunks=None, max_length=None)

def build(
    self,
    query:      str,
    chunks:     List[Tuple],
    max_chunks: Optional[int] = None,
    max_length: Optional[int] = None,
) -> str: ...

Analyzes the query, wraps each (chunk, score) tuple into a ScoredChunk with vector_score = composite_score = score, then runs it through self._filter and self._ranker (note: skips retrieval and the guard entirely — this method assumes you already retrieved the chunks yourself), then composes. Returns the composed str directly (not a ContextResult).

build_multi_query_context(queries, all_chunks, global_budget=None)

Calls build() once per (query, chunks) pair and joins with "--- {query} ---" labels. Note: global_budget is accepted as a parameter but never used inside this method — each build() call uses its own per-query default budget (the query's analyzed budget), and the combined result is not clamped to global_budget at all, unlike run_multi() which does clamp.

compress_context(context, target_length)

Delegates directly to self._composer._hard_truncate(context, target_length) — reaches into a private method of the composer. Only works correctly if self._composer is a SmartContextComposer (or any object exposing a compatible _hard_truncate); if you've injected a custom composer without that method, this raises AttributeError.

get_context_stats(context)

Returns self._composer.get_stats(context) if self._composer is a SmartContextComposer, else falls back to {"total_length": len(context)}.

Diagnostics

get_performance_report()

Returns self._adaptive.get_performance_report(), with a "cache" key added (from self._cache.stats) if caching is enabled.

set_conversation(history) / get_history()

Simple setter/getter for self._history, letting you resume a previously saved ConversationHistory across process restarts (you are responsible for persisting/restoring it yourself — there's no built-in serialization).

clear_cache()

Clears self._cache if it exists; logs an INFO line. No-op (no error) if caching is disabled.

ContextManager

class ContextManager(ContextEngine):
    def __init__(
        self,
        config:               Optional[ContextEngineConfig] = None,
        *,
        length_counter:       Optional[LengthCounter]       = None,
        dedup_strategy:       Optional[Any]                 = None,
        ranking_strategy:     Optional[Any]                 = None,
        compression_strategy: Optional[Any]                 = None,
        budget_strategy:      Optional[BudgetAllocationStrategy] = None,
        **kwargs,
    ): ...

A subclass kept for backward compatibility with a still-older constructor signature. Important: dedup_strategy, ranking_strategy, and compression_strategy are accepted parameters, but reading the constructor body, they are never actually wired into the pipeline — if any of them is truthy, the constructor only logs a DEBUG line acknowledging it was provided ("... مُوفَّر (legacy mode)"). Passing a legacy dedup_strategy etc. to ContextManager today has no functional effect on retrieval, ranking, or composition; only config, length_counter, and budget_strategy (passed through as budget_allocator) actually influence behavior, via the parent ContextEngine.__init__.

ContextManager adds no new public methods of its own — every method it exposes (run, index, build, etc.) is inherited unchanged from ContextEngine.

Examples

1. Basic — end-to-end:

from zerda.context import ContextEngine
engine = ContextEngine()
engine.index(chunks=my_chunks, embed_fn=my_embed_fn)
result = engine.run("What is RAG?")
print(result.context)

2. Intermediate — custom budget and chunk limit:

result = engine.run(
    "Explain RAG in detail",
    max_chunks=5,
    override_budget=3000,
)

3. Advanced — fully custom pipeline:

from zerda.context import ContextEngine, ContextEngineConfig
from my_app.strategies import LLMQueryAnalyzer, RerankerRanker

engine = ContextEngine(
    config=ContextEngineConfig(),
    query_analyzer=LLMQueryAnalyzer(),
    ranker=RerankerRanker(),
)
engine.index(chunks=my_chunks, vector_store_adapter=my_chroma_store)
result = engine.run("query", metadata_filters={"lang": "ar"})

3b. Advanced — native zerda.chunks + zerda.vector_database, no external adapter:

from zerda.chunks import StructureAwareChunker, ChunkConfig
from zerda.vector_database import ChromaVectorDatabase
from zerda.context import ContextEngine
from zerda.embeddings import HugginFaceEmbedder

my_embedder = HugginFaceEmbedder()
chunks = StructureAwareChunker(ChunkConfig)
engine = ContextEngine()
engine.index_documents(
    texts     = my_raw_texts,
    chunker   = chunks,
    vector_db = ChromaVectorDatabase(embedder=my_embedder),
)
result = engine.run("ما هو التعلم الآلي؟")

Equivalent, if you already have DocumentChunks (e.g. produced separately, or shared with another consumer of the same vector store):

engine.index(chunks=my_document_chunks, vector_db=ChromaVectorDatabase(embedder=my_embedder))

4. Error handling — empty chunk index:

engine = ContextEngine()
engine.index(chunks=[])  # logs a warning, does nothing else; self._retriever stays None
result = engine.run("anything")
print(result.guard_passed, result.guard_warnings)
# False, ['context is empty ⚠'] — because there's no retriever, nothing was retrieved

5. Production usage — multi-query batch with shared filters:

combined_context = engine.run_multi(
    queries=["What is RAG?", "How does reranking work?"],
    global_budget=4000,
    metadata_filters={"source_type": "docs"},
)

Best practices

• Call index() (or index_documents()) exactly once per corpus snapshot; re-indexing with a different chunks/texts list does not clear previously indexed data from sub-retrievers that were already injected — it constructs new ones and only swaps in a new HybridRetriever if none existed yet (see the warning above).
• Prefer vector_db (a native zerda.vector_database backend) over the legacy vector_store_adapter when your vector store is one of ChromaVectorDatabase/FAISSVectorDatabase/PineconeVectorDatabase — it uses each backend's real .add()/.search() methods instead of a LangChain-shaped duck-typed interface, and correctly maps metadata filters onto whichever kwarg name that backend expects (filters/doc_id_filter/filter_dict — see retriever.md).
• If you're re-using a vector_db that was already populated elsewhere (a separate ingestion pipeline, a shared collection), pass add_to_vector_db=False to index() to avoid re-embedding/re-upserting the same chunks.
• Use get_performance_report() periodically in production to watch guard_failure_rate and avg_response_ms.

Common mistakes

• Expecting record_turn=False sub-calls inside run_multi() to be excludable — they're hardcoded to False, not configurable per call.
• Reading result.query_analysis.original_query expecting the literal text you passed to run(), when multi-turn enrichment is active (see the warning above).
• Passing legacy dedup_strategy/ranking_strategy/compression_strategy to ContextManager and assuming they take effect.
• Passing both embed_fn and vector_db to index() expecting embed_fn to be used as a fallback — it's simply ignored whenever vector_db is set; vector_db.add() relies entirely on the backend's own embedder (or pre-computed chunk.embedding values).
• Calling index_documents() with a chunker that requires heavy optional dependencies (ArabicTextChunker/ContextAwareChunker need torch+transformers) without having them installed — the ImportError surfaces from inside the chunker, not from context itself.

Related

• Architecture — full pipeline diagram
• models.ContextResult
• retriever.VectorRetriever — the three retrieval modes index()/index_documents() wire up
• legacy_compat — the tuple-based strategies ContextManager's constructor accepts but does not use

Version history

No formal changelog or version metadata is present in the source. One functional change is directly observable against the previously-documented behavior: index() gained vector_db/add_to_vector_db parameters, and a new index_documents() method was added, to natively wire ContextEngine into zerda.chunks (chunkers) and zerda.vector_database (ChromaVectorDatabase/FAISSVectorDatabase/PineconeVectorDatabase) without requiring a LangChain-style vector_store_adapter.

← Back to API overview