strategy module
On this page
Overview
Split into two groups:
- Legacy strategies — kept for compatibility with an older tuple-based (
(chunk, score)) API. - New advanced strategies — the contracts actually used by
ContextEngineand its default components today.
All ten are plain ABC subclasses with one or two @abstractmethods each. None have default implementations, attributes, or shared base logic — there's no common parent class tying them together.
Legacy strategies
DeduplicationStrategy
class DeduplicationStrategy(ABC):
@abstractmethod
def deduplicate(self, chunks: List[Tuple]) -> List[Tuple]: ...
Implemented by legacy_compat.PrefixDeduplication and legacy_compat.HashDeduplication.
RankingStrategy
class RankingStrategy(ABC):
@abstractmethod
def rank(self, chunks: List[Tuple], max_chunks: Optional[int]) -> List[Tuple]: ...
Implemented by legacy_compat.ScoreRanking.
CompressionStrategy
class CompressionStrategy(ABC):
@abstractmethod
def compress(self, context: str, target: int, counter: Any, separator: str) -> str: ...
Implemented by legacy_compat.GreedyCompression.
BudgetAllocationStrategy
class BudgetAllocationStrategy(ABC):
@abstractmethod
def allocate(self, total_budget: int, n_queries: int) -> List[int]: ...
Note this is also used by the "new" budget allocators (EqualBudgetAllocation, WeightedBudgetAllocation, ComplexityBasedAllocation) — it sits in the "legacy" section of the source file but is in active use, including being a constructor parameter of ContextEngine.
[!NOTE]
ComplexityBasedAllocation.allocatehas an extra parameter The abstract signature only declares(total_budget, n_queries), butComplexityBasedAllocation.allocateadds an optional third parameter,analyses. Python'sABCdoes not enforce signature compatibility — only that the method name exists — so this is legal, but it meansComplexityBasedAllocationcannot be called polymorphically through theBudgetAllocationStrategyinterface without also knowing aboutanalyses.
New advanced strategy interfaces
QueryAnalyzerStrategy
class QueryAnalyzerStrategy(ABC):
@abstractmethod
def analyze(self, query: str) -> QueryAnalysis: ...
Implemented by query_analyzer.RuleBasedQueryAnalyzer. To plug in your own (e.g. an LLM-based analyzer), implement analyze() and pass it to ContextEngine(query_analyzer=...).
ChunkFilterStrategy
class ChunkFilterStrategy(ABC):
@abstractmethod
def filter(self, chunks: List[ScoredChunk]) -> List[ScoredChunk]: ...
Implemented by pipeline.ChunkFilter.
ChunkRankingStrategy
class ChunkRankingStrategy(ABC):
@abstractmethod
def rank(
self,
chunks: List[ScoredChunk],
analysis: QueryAnalysis,
max_n: Optional[int] = None,
) -> List[ScoredChunk]: ...
Implemented by pipeline.CompositeRanker. Unlike the legacy RankingStrategy, this version takes the query's QueryAnalysis so ranking can be intent/keyword-aware.
ContextComposerStrategy
class ContextComposerStrategy(ABC):
@abstractmethod
def compose(
self,
chunks: List[ScoredChunk],
analysis: QueryAnalysis,
budget: int,
) -> str: ...
Implemented by pipeline.SmartContextComposer.
ContextGuardStrategy
class ContextGuardStrategy(ABC):
@abstractmethod
def validate(
self,
context: str,
query: str,
analysis: QueryAnalysis,
) -> Tuple[bool, List[str]]: ...
Returns (passed, warnings). Implemented by guard_and_learning.ContextGuard. The docstring explicitly notes this interface could be replaced with an LLM-based hallucination checker.
HybridRetrieverStrategy
class HybridRetrieverStrategy(ABC):
@abstractmethod
def retrieve(
self,
query: str,
top_k: int,
filters: Optional[Dict[str, Any]] = None,
) -> List[ScoredChunk]: ...
@property
@abstractmethod
def method(self) -> str:
...
Implemented by retriever.VectorRetriever, retriever.KeywordRetriever, and retriever.MetadataRetriever — each individually. Note retriever.HybridRetriever (the orchestrator that combines all three) does not itself implement this interface — it's a plain class with its own retrieve() method, not an ABC subclass.
Implementing a custom strategy
Example: a custom ChunkRankingStrategy that purely favors recency:
from zerda.context import ChunkRankingStrategy, ScoredChunk, QueryAnalysis
from typing import List, Optional
class RecencyFirstRanker(ChunkRankingStrategy):
def rank(
self,
chunks: List[ScoredChunk],
analysis: QueryAnalysis,
max_n: Optional[int] = None,
) -> List[ScoredChunk]:
ranked = sorted(chunks, key=lambda c: c.recency_score, reverse=True)
return ranked[:max_n] if max_n else ranked
engine = ContextEngine(ranker=RecencyFirstRanker())
If you forget to implement an abstract method, Python raises at instantiation time, not at call time:
class BrokenRanker(ChunkRankingStrategy):
pass
BrokenRanker()
# TypeError: Can't instantiate abstract class BrokenRanker with abstract method rank
Related
- Architecture → Component substitution points
engine— where these strategies are wired together