Cookbook: Custom Strategy Implementations

from zerda.context import QueryAnalyzerStrategy, QueryAnalysis, QueryIntent, QueryComplexity

class LLMQueryAnalyzer(QueryAnalyzerStrategy):
    def __init__(self, llm_client, fallback: QueryAnalyzerStrategy):
        self.llm = llm_client
        self.fallback = fallback  # use RuleBasedQueryAnalyzer as a safety net

    def analyze(self, query: str) -> QueryAnalysis:
        try:
            structured = self.llm.classify(query)  # your own LLM call
            return QueryAnalysis(
                original_query=query,
                normalized=query.strip(),
                intent=QueryIntent(structured["intent"]),
                complexity=QueryComplexity(structured["complexity"]),
                keywords=structured.get("keywords", []),
                is_arabic=structured.get("is_arabic", False),
                language=structured.get("language", "en"),
                context_budget=structured.get("budget", 2000),
            )
        except Exception:
            return self.fallback.analyze(query)  # never let analysis itself crash the pipeline

from zerda.context import ContextEngine, ContextEngineConfig, build_query_analyzer

cfg = ContextEngineConfig()
analyzer = LLMQueryAnalyzer(llm_client=my_llm, fallback=build_query_analyzer(cfg))
engine = ContextEngine(config=cfg, query_analyzer=analyzer)

from zerda.context import ChunkFilterStrategy, ScoredChunk, ChunkFilter, ContextEngineConfig
from typing import List

class NoBoilerplateFilter(ChunkFilterStrategy):
    """Wraps the default ChunkFilter and additionally drops legal boilerplate."""
    def __init__(self, config: ContextEngineConfig):
        self._base = ChunkFilter(config)

    def filter(self, chunks: List[ScoredChunk]) -> List[ScoredChunk]:
        chunks = self._base.filter(chunks)
        return [c for c in chunks if "all rights reserved" not in c.text.lower()]

engine = ContextEngine(chunk_filter=NoBoilerplateFilter(cfg))

from zerda.context import ChunkRankingStrategy, ScoredChunk, QueryAnalysis, QueryIntent
from typing import List, Optional

class CodeIntentBooster(ChunkRankingStrategy):
    """If the query is code-intent, boost chunks that look like code blocks."""
    def __init__(self, base_ranker: ChunkRankingStrategy):
        self._base = base_ranker

    def rank(self, chunks: List[ScoredChunk], analysis: QueryAnalysis, max_n: Optional[int] = None):
        ranked = self._base.rank(chunks, analysis, max_n)
        if analysis.intent == QueryIntent.CODE:
            ranked.sort(key=lambda c: ("```" in c.text or "def " in c.text, c.composite_score), reverse=True)
        return ranked

from zerda.context import CompositeRanker
engine = ContextEngine(ranker=CodeIntentBooster(CompositeRanker(cfg)))

from zerda.context import ContextGuardStrategy, QueryAnalysis
from typing import Tuple, List

class LLMFactualityGuard(ContextGuardStrategy):
    def __init__(self, llm_client, base_guard: ContextGuardStrategy):
        self.llm = llm_client
        self.base = base_guard

    def validate(self, context: str, query: str, analysis: QueryAnalysis) -> Tuple[bool, List[str]]:
        passed, warnings = self.base.validate(context, query, analysis)
        if not passed:
            return passed, warnings
        # Add an LLM-based sufficiency check on top of the structural checks
        verdict = self.llm.check_sufficiency(context=context, query=query)
        if not verdict["sufficient"]:
            warnings.append(f"LLM guard: {verdict['reason']}")
            return False, warnings
        return True, warnings

from zerda.context import ContextGuard
engine = ContextEngine(guard=LLMFactualityGuard(my_llm, ContextGuard(cfg)))

from zerda.context import BudgetAllocationStrategy
from typing import List, Optional

class FirstQueryPriorityAllocation(BudgetAllocationStrategy):
    """Give the first query in a batch most of the budget; split the rest equally."""
    def allocate(self, total_budget: int, n_queries: int) -> List[int]:
        if n_queries <= 0:
            return []
        if n_queries == 1:
            return [total_budget]
        first = int(total_budget * 0.5)
        remainder = total_budget - first
        per = remainder // (n_queries - 1)
        budgets = [first] + [per] * (n_queries - 1)
        budgets[-1] += total_budget - sum(budgets)  # absorb rounding into the last one
        return budgets

engine = ContextEngine(budget_allocator=FirstQueryPriorityAllocation())

try:
    instance = MyCustomStrategy()
except TypeError as e:
    print(e)  # "Can't instantiate abstract class MyCustomStrategy with abstract method ..."