ZerdaZerda

guard_and_learning module

On this page

ContextGuard

Python · 6 lines
class ContextGuard(ContextGuardStrategy):
    def __init__(
        self,
        config:   ContextEngineConfig,
        embed_fn: Optional[Callable[[str], np.ndarray]] = None,
    ): ...

validate(context, query, analysis)

Python · 6 lines
def validate(
    self,
    context:  str,
    query:    str,
    analysis: QueryAnalysis,
) -> Tuple[bool, List[str]]: ...

Algorithm (checks run in this exact order; some short-circuit, some only warn)

  1. If GuardConfig.enabled is False, returns (True, []) immediately — no other check runs at all, including the empty-context check.
  2. If check_empty and context.strip() is falsy → returns (False, ["context is empty ⚠"]) immediately, skipping every remaining check.
  3. Length check: if len(context) < min_context_length → appends a warning and sets passed = False, but does not return early — remaining checks still run.
  4. If len(context) > max_context_length → appends a warning but does not set passed = False. This is purely informational.
  5. If check_source_coverage → checks for a regex match of \[(?:|Source|source): (note: the empty alternative (?:|Source|source) makes this also match a bare [: — likely an unintentional artifact of the regex, but as written it also matches any [anything: style bracket-colon pattern, not just literal "Source:") or the literal substring ---. If neither is found, appends a warning but does not fail the check.
  6. If check_relevance is True and an embed_fn was supplied at construction time, runs _check_relevance. If check_relevance is True but no embed_fn was given, this step is silently skipped — no warning, no error.
  7. Logs all warnings together at WARNING level if any exist, otherwise logs a DEBUG confirmation.

_check_relevance(context, query)

Embeds query and the first 512 characters of context, computes cosine similarity, compares against relevance_threshold. On any exception (e.g. embed_fn raising), logs at DEBUG and returns (True, "") — i.e. embedding failures pass the relevance check by default rather than failing it.

Returns

Tuple[bool, List[str]] = (passed, warnings). Note passed=False can coexist with the context still being returned and used by ContextEngine.run() — the guard does not prevent the context from being included in ContextResult; it only sets ContextResult.guard_passed and skips caching the result (see engine.md).

Parameters

NameTypeRequired
contextstrYes
querystrYes — only used by the optional relevance check
analysisQueryAnalysisYes, but entirely unused in the current implementation — accepted for interface conformance only

Thread safety

Holds only read-only config and an optional embed_fn reference — safe for concurrent validate() calls.

AdaptiveLearningSystem

Python · 4 lines
class AdaptiveLearningSystem:
    WINDOW = 100

    def __init__(self, config: ContextEngineConfig): ...

Maintains four collections.deque(maxlen=100) rolling windows: response times, context lengths, chunk counts, and guard-failure booleans.

record(query, analysis, context_length, chunk_count, response_ms, guard_passed)

Python · 9 lines
def record(
    self,
    query:          str,
    analysis:       QueryAnalysis,
    context_length: int,
    chunk_count:    int,
    response_ms:    float,
    guard_passed:   bool,
) -> None: ...

No-op if config.enable_adaptive is False. Otherwise appends to all four deques and increments _total_queries; every 20th call triggers _auto_tune(). Note: query and analysis parameters are accepted but not stored or used anywhere inside record() — only their side effects via the other parameters matter.

suggest_budget(analysis)

Python
def suggest_budget(self, analysis: QueryAnalysis) -> int: ...

Algorithm

  1. base_budget = analysis.context_budget (whatever RuleBasedQueryAnalyzer assigned).
  2. If adaptive learning is disabled, or fewer than 10 samples exist yet, returns base_budget unchanged.
  3. Otherwise computes avg_actual (mean of historical _context_lengths). If avg_actual < base_budget * 0.5 (i.e. historically, actual composed contexts have been using less than half the assigned budget), returns max(int(avg_actual * 1.3), 200) — a 30%-margin-adjusted, floor-200 reduced budget.
  4. Otherwise returns base_budget unchanged.

This is the only mechanism in the package that actually adapts a numeric setting based on history — it only ever shrinks the budget, never grows it beyond what the analyzer already assigned.

get_performance_report()

Returns averages of all four rolling windows plus total_queries and samples (= number of entries currently in the response-time deque, which is bounded at 100 regardless of how many total queries have run).

_auto_tune()

Called automatically every 20th record(). Only logs — see the note in Architecture for the exact behavior; it does not mutate any config or internal state.

Budget allocation strategies

All three implement BudgetAllocationStrategy.

EqualBudgetAllocation

Python · 2 lines
class EqualBudgetAllocation(BudgetAllocationStrategy):
    def allocate(self, total_budget: int, n_queries: int) -> List[int]: ...

Integer-divides total_budget by n_queries; the first query in the list absorbs any remainder (budgets[0] += total_budget - sum(budgets)), so allocations are not perfectly equal when total_budget doesn't divide evenly — the first query always gets the leftover. Returns [] for n_queries <= 0.

WeightedBudgetAllocation

Python · 3 lines
class WeightedBudgetAllocation(BudgetAllocationStrategy):
    def __init__(self, weights: List[float]): ...
    def allocate(self, total_budget: int, n_queries: int) -> List[int]: ...

Raises ValueError if len(self.weights) != n_queries. Each query gets max(1, int(total_budget * weight / sum(weights))) (floor — every query gets at least 1), with rounding remainder again absorbed by the first query.

ComplexityBasedAllocation

Python · 8 lines
class ComplexityBasedAllocation(BudgetAllocationStrategy):
    _COMPLEXITY_WEIGHTS = {LOW: 1.0, MEDIUM: 1.5, HIGH: 2.5}
    def allocate(
        self,
        total_budget: int,
        n_queries:    int,
        analyses:     Optional[List[QueryAnalysis]] = None,
    ) -> List[int]: ...

If analyses is missing or doesn't match n_queries in length, falls back to EqualBudgetAllocation silently (no warning). Otherwise maps each analysis's complexity to a weight and delegates to WeightedBudgetAllocation.

ContextCache

Python · 2 lines
class ContextCache:
    def __init__(self, max_size: int = 256, ttl_seconds: int = 300): ...

A from-scratch LRU+TTL cache: a plain dict (key -> (context, timestamp)) plus a deque tracking insertion order for eviction.

get(key)

Returns None and increments _misses if the key is absent or expired (age > ttl_seconds, in which case the entry is also deleted from the store — but not removed from the _order deque). Otherwise increments _hits and returns the cached context string.

put(key, context)

Python
def put(self, key: str, context: str) -> None: ...
  • If key already exists, just updates its value/timestamp in place — and does not touch _order at all in this branch.
  • Otherwise, if the store is at max_size, evicts the oldest entry by popping the leftmost item from _order and deleting it from the store (if it's even still there).
  • Appends the new key to _order and stores it.

[!WARNING] _order can accumulate duplicate/stale entries Because put() on an existing key doesn't move or re-add it in _order, and a get()-expired key is deleted from _store but left in _order, the deque can contain key references that no longer exist in _store, or miss the "this key was just touched" recency information an LRU cache normally needs. The eviction logic in put() calls self._order.popleft() and then does self._store.pop(oldest, None) (using .pop with a default rather than del, specifically to tolerate the oldest key already being gone) — this is defensive code that strongly suggests the implementers were aware _order can go stale, but the cache as implemented is closer to FIFO-with-known-staleness than a textbook LRU.

make_key(query, filters=None)

Python
def make_key(self, query: str, filters: Optional[Dict] = None) -> str: ...

MD5 of the lower-cased, stripped query, with str(sorted(filters.items())) appended if filters are given. Note: MD5 is used here purely as a cache key digest, not for any security purpose.

stats (property)

Returns size, max_size, hits, misses, hit_rate (formatted as a percentage string, e.g. "42.0%", or the literal string "0%" if no requests have been made yet), and ttl_sec.

clear()

Empties both _store and _order. Does not reset _hits/_misses counters.

Examples

1. Basic — guard:

Python · 3 lines
from zerda.context import ContextEngineConfig, ContextGuard
guard = ContextGuard(ContextEngineConfig())
passed, warnings = guard.validate(context="some text", query="q", analysis=analysis)

2. Intermediate — guard with relevance checking:

Python · 4 lines
cfg = ContextEngineConfig()
cfg.guard.check_relevance = True
guard = ContextGuard(cfg, embed_fn=my_embed_fn)
passed, warnings = guard.validate(context, query, analysis)

3. Advanced — manual adaptive learning report:

Python · 5 lines
from zerda.context import AdaptiveLearningSystem
adaptive = AdaptiveLearningSystem(cfg)
for q, a, ctx_len, n_chunks, ms, ok in usage_log:
    adaptive.record(q, a, ctx_len, n_chunks, ms, ok)
print(adaptive.get_performance_report())

4. Error handling — mismatched weights:

Python · 5 lines
from zerda.context import WeightedBudgetAllocation
try:
    WeightedBudgetAllocation(weights=[1.0, 2.0]).allocate(total_budget=1000, n_queries=3)
except ValueError as e:
    print(e)  # "weights length must equal n_queries."

5. Production usage — complexity-aware multi-query budgeting:

Python · 3 lines
from zerda.context import ComplexityBasedAllocation
analyses = [analyzer.analyze(q) for q in queries]
budgets = ComplexityBasedAllocation().allocate(total_budget=6000, n_queries=len(queries), analyses=analyses)

Best practices

  • Don't rely on ContextGuard.validate() to block low-quality context from reaching your LLM — it only reports; ContextEngine.run() still returns the composed context regardless of guard_passed. Check result.guard_passed yourself and decide what to do.
  • If you need real LRU eviction semantics (not FIFO-with-staleness), consider wrapping ContextCache or substituting functools.lru_cache-backed logic at the call site.

Common mistakes

  • Enabling check_relevance without passing embed_fn to ContextGuard — this fails silently rather than raising, so it's easy to believe relevance checking is active when it isn't.
  • Calling ComplexityBasedAllocation.allocate() without analyses and expecting complexity-aware splitting — it silently degrades to equal allocation.

← Back to API overview

Related pages