ZerdaZerda

pipeline module

On this page

Overview

This module owns everything between "raw retrieved chunks" and "final context string": filtering (ChunkFilter), ranking (CompositeRanker), and composition/templating (SmartContextComposer + ContextTemplate).

Utility functions

char_counter(text)

Python · 2 lines
def char_counter(text: str) -> int:
    return len(text) if text else 0

The default LengthCounter. Treats None/empty string as length 0 explicitly (rather than relying on len(None) raising).

make_token_counter(tokenizer)

Python
def make_token_counter(tokenizer) -> LengthCounter: ...

Returns a closure lambda text: len(tokenizer.encode(text)). Parameter contract: tokenizer must expose an .encode(text) -> Sequence. This matches tiktoken.Encoding directly; for Hugging Face tokenizers, pass tokenizer itself if it has .encode(), or wrap it.

Raises: Whatever tokenizer.encode() raises, propagated unmodified — there's no error handling in the closure.

ChunkFilter

Python · 2 lines
class ChunkFilter:
    def __init__(self, config: ContextEngineConfig): ...

Implements ChunkFilterStrategy (note: not declared with explicit inheritance in the class signature, but it satisfies the interface structurally and is used everywhere ChunkFilterStrategy is expected). Compiles FilterConfig.noise_patterns into regex objects once at construction.

filter(chunks)

Python
def filter(self, chunks: List[ScoredChunk]) -> List[ScoredChunk]: ...

Runs three sub-steps in this fixed order — always, regardless of FilterConfig.dedup_method (see the warning in Configuration Reference):

  1. _filter_noise — drops any chunk whose stripped text matches (re.match, anchored at the start) one of the compiled noise patterns. In practice all three default patterns in FilterConfig.noise_patterns already end in $, so they behave as full-string matches; a custom pattern without a trailing $ would only need to match a prefix, since re.match (unlike re.fullmatch) doesn't require consuming the whole string.
  2. _filter_length — keeps only chunks where min_chunk_length <= char_count <= max_chunk_length.
  3. _dedup_hash — removes exact duplicates by content_hash, keeping the first occurrence.

Logs a debug line with before/after counts only if something was actually removed.

Complexity

O(n) in the number of chunks (each step is a single linear pass; noise/length are independent per-chunk, hash dedup uses a set for O(1) membership checks).

filter_with_embeddings(chunks, embeddings)

Python · 5 lines
def filter_with_embeddings(
    self,
    chunks:     List[ScoredChunk],
    embeddings: List[Optional[np.ndarray]],
) -> List[ScoredChunk]: ...

Calls filter() first, then semantic dedup. Important: embeddings must be in 1:1 positional correspondence with the original chunks list passed in, but after filter() runs internally, the chunk list may have shrunk — yet _dedup_semantic is called with the post-filter chunks and the original, unfiltered embeddings list. Reading _dedup_semantic, it immediately returns the input chunks unchanged if len(chunks) != len(embeddings) — so in virtually any case where filter() actually removed something, semantic dedup silently becomes a no-op for that call, since the lengths will no longer match.

[!WARNING] filter_with_embeddings likely doesn't do what it looks like it does Call _dedup_semantic yourself directly with pre-filtered, length-matched lists if you need working semantic deduplication, rather than relying on filter_with_embeddings().

_dedup_semantic(chunks, embeddings)

O(n²) pairwise cosine-similarity comparison (only among chunks with a non-None embedding) against FilterConfig.semantic_sim_threshold. For every pair above threshold, the later chunk in the list is dropped, the earlier one kept.

CompositeRanker

Python · 2 lines
class CompositeRanker(ChunkRankingStrategy):
    def __init__(self, config: ContextEngineConfig): ...

rank(chunks, analysis, max_n=None)

Python · 6 lines
def rank(
    self,
    chunks:   List[ScoredChunk],
    analysis: QueryAnalysis,
    max_n:    Optional[int] = None,
) -> List[ScoredChunk]: ...

Algorithm

For every chunk, overwrites composite_score (discarding whatever HybridRetriever had set) with:

Text · 5 lines
composite = vector_score * weight_vector_score
          + keyword_boost * weight_keyword_score
          + (source_quality.value / 3.0) * weight_source_quality
          + recency_score * weight_recency
          + position_score * weight_position

Where:

  • keyword_boost = fraction of analysis.keywords found as a literal substring in the chunk's lowercased text, capped at 1.0.
  • position_score = max(0.0, 1.0 - retrieval_rank * 0.01) — i.e. decays to 0 once retrieval_rank >= 100.

Sorted descending, truncated to max_n (or RankingConfig.max_chunks_to_rank if max_n is None).

Parameters

NameTypeRequiredNotes
chunksList[ScoredChunk]YesReturns [] immediately if empty
analysisQueryAnalysisYesOnly .keywords is read
max_nOptional[int]NoDefaults to config value

Returns

List[ScoredChunk], length ≤ max_n, sorted by recomputed composite_score descending.

Complexity

O(n log n) — dominated by the final sort; per-chunk scoring is O(k) where k = number of query keywords (small, capped at 12 by the analyzer).

Side effects

Mutates composite_score in place on every input ScoredChunk object — callers holding references to the original (pre-rank) list will see their composite_score values change too, since ScoredChunk objects are not copied.

ContextTemplate

Python · 2 lines
class ContextTemplate:
    def __init__(self, header: str, chunk_fmt: str, chunk_fmt_score: str, footer: str): ...

A plain formatting-string container. Not exported from the package root (context/__init__.py) — accessible only via from context.pipeline import ContextTemplate or indirectly through SmartContextComposer.

Built-in templates (_TEMPLATES dict, module-private)

KeyHeaderChunk format (no score)Footer
"arabic" (default)📚 المعلومات المسترجعة:[المصدر: {source}]\n{text}─── نهاية السياق ───
"english"📚 Retrieved Information:[Source: {source}]\n{text}─── End of Context ───
"minimal"(empty){text}(empty)
"structured"=== CONTEXT START ===--- [{idx}] {source} ---\n{text}=== CONTEXT END ===

Each also has a chunk_fmt_score variant used when ComposerConfig.include_scores=True, embedding {score:.3f} in the header.

SmartContextComposer

Python · 6 lines
class SmartContextComposer(ContextComposerStrategy):
    def __init__(
        self,
        config:  ContextEngineConfig,
        counter: Optional[LengthCounter] = None,
    ): ...

counter defaults to char_counter if not provided. self._tmpl is resolved at construction time from config.composer.template — changing config.composer.template after construction has no effect on an already-constructed SmartContextComposer instance, since the template object is cached in self._tmpl, not re-read on every call.

compose(chunks, analysis, budget)

Python · 6 lines
def compose(
    self,
    chunks:   List[ScoredChunk],
    analysis: QueryAnalysis,
    budget:   int,
) -> str: ...

Algorithm

  1. If chunks is empty, returns _empty_context() (a short "no information available" message in the active template's language).
  2. If ComposerConfig.group_by_source, reorders chunks by source-group, ordering groups by their best member's score.
  3. Formats every chunk via _format_chunk (applies the template, optionally appending up to 5 metadata key/value pairs).
  4. _assemble(): greedily packs formatted chunks (in their current order) into the budget, accounting for separator and footer overhead. Guarantees at least one chunk is included — even if the very first chunk alone exceeds the available budget, it is hard-truncated and included rather than yielding an empty context. After packing, the loop breaks on the first chunk that doesn't fit (it does not skip ahead to try a smaller chunk further down the list).
  5. Safety clamp: if the assembled string still somehow exceeds budget (can happen due to header/footer/separator overhead estimation), hard-truncates the whole thing.

_hard_truncate(text, max_len)

Binary-searches for the longest prefix of text whose counted length is ≤ max_len - 3 (reserving room for a "..." suffix), then backs up to the last space if it's not too far back (space > lo * 0.7), and appends "...".

Parameters

NameTypeRequiredNotes
chunksList[ScoredChunk]YesAlready filtered + ranked, in display order
analysisQueryAnalysisYesAccepted but not read anywhere in compose() or its helpers — present for interface conformance with ContextComposerStrategy only
budgetintYesMeasured in whatever unit self._counter returns (chars by default)

Returns

str — never None, never raises for normal input; an empty chunks list produces a non-empty placeholder message instead of "".

Complexity

O(n) for formatting + O(n) for assembly (one pass each); _hard_truncate is O(log n) in text length for the binary search, called at most twice per compose() call.

Side effects

Logs a debug line with chunk count, assembled length, and budget.

get_stats(context)

Python
def get_stats(self, context: str) -> Dict[str, Any]: ...

Splits a previously-composed context by ComposerConfig.separator and reports total/avg/min/max chunk length and a length_unit field ("chars" if self._counter is char_counter, else "tokens"). Caveat: this split is purely string-based — if you pass in an arbitrary string (not one this composer produced), the stats will reflect however many times the separator happens to occur in it, which may not correspond to "chunks" at all.

Examples

1. Basic:

Python · 3 lines
from zerda.context import ContextEngineConfig, SmartContextComposer
composer = SmartContextComposer(ContextEngineConfig())
context = composer.compose(ranked_chunks, analysis, budget=1000)

2. Intermediate — structured template with scores:

Python · 5 lines
cfg = ContextEngineConfig()
cfg.composer.template = "structured"
cfg.composer.include_scores = True
composer = SmartContextComposer(cfg)
print(composer.compose(ranked_chunks, analysis, budget=1500))

3. Advanced — token-budgeted composition:

Python · 6 lines
import tiktoken
from zerda.context import make_token_counter, SmartContextComposer

tok = tiktoken.get_encoding("cl100k_base")
composer = SmartContextComposer(cfg, counter=make_token_counter(tok))
context = composer.compose(ranked_chunks, analysis, budget=512)  # 512 tokens, not chars

4. Error handling — empty input:

Python · 3 lines
composer.compose([], analysis, budget=500)
# Returns a placeholder string, e.g. "📚 المعلومات المسترجعة:\nلا تتوفر معلومات.\n─── نهاية السياق ───"
# Does not raise, does not return ""

5. Production usage — grouping by source for citation-friendly output:

Python · 4 lines
cfg.composer.group_by_source = True
cfg.composer.include_metadata = True
composer = SmartContextComposer(cfg)
context = composer.compose(ranked_chunks, analysis, budget=cfg.composer.max_context_length)

Best practices

  • Build a new SmartContextComposer (or call ContextEngine fresh) if you change template at runtime — don't mutate config.composer.template on a live instance and expect it to take effect.
  • Use get_stats() only on strings this exact composer instance produced (same separator).

Common mistakes

  • Relying on filter_with_embeddings() for semantic dedup without first verifying chunk/embedding list lengths stay aligned (see the warning under ChunkFilter above).
  • Assuming compose()'s analysis parameter affects formatting — it currently does not.

← Back to API overview

Related pages