Feedback Engine

Closes the routing loop: records routing outcomes, computes a per-route exponential-moving-average-derived score bias, applies it back onto Route.score_bias, and optionally persists learned state to JSON.

Score adjustment model

Documented directly in the module docstring:

ema_n = alpha * outcome + (1 - alpha) * ema_{n-1}
bias  = clip((ema - 0.5) * 0.6, -0.3, +0.3)

• outcome is 1.0 for success, 0.0 for failure.
• alpha controls adaptation speed (constructor default 0.1 — not read from FeedbackConfig.score_decay; see the warning under FeedbackEngine.__init__ below).
• EMA starts at 0.5 (neutral) → initial bias 0.
• EMA → 1.0 (all successes) → bias → +0.3 (route preferred in future scoring).
• EMA → 0.0 (all failures) → bias → -0.3 (route penalized).
• Only routes with n_samples >= min_samples_to_adapt actually have their score_bias updated — avoids overreacting to a handful of early observations.

FeedbackRecord

@dataclass
class FeedbackRecord:
    route_name:  str
    group_name:  Optional[str]
    query_hash:  str
    success:     bool
    score:       float
    timestamp:   float = field(default_factory=time.time)

A single logged feedback event, appended to FeedbackEngine's internal history. query_hash stores only a truncated MD5 hash of the raw query (first 8 hex chars) rather than the query text itself — a lightweight privacy measure noted in record()'s docstring.

RouteFeedbackState

@dataclass
class RouteFeedbackState:
    route_name: str
    ema:        float = 0.5
    n_samples:  int   = 0
    n_success:  int   = 0
    n_failure:  int   = 0

Per-route running feedback statistics.

Properties

success_rate -> float

n_success / n_samples if n_samples > 0, else 0.0. This is the raw success rate — distinct from ema, which is exponentially weighted toward recent observations.

computed_bias -> float

max(-0.3, min(0.3, (self.ema - 0.5) * 0.6)) — the clipped bias value derived from the current EMA.

Methods

update(self, success: bool, alpha: float)

Updates ema via the EMA formula above, increments n_samples, and increments n_success or n_failure accordingly. Mutates the instance in place; no return value.

to_dict(self) -> Dict[str, Any]

Returns {"route_name", "ema" (6dp), "n_samples", "n_success", "n_failure", "success_rate" (4dp), "bias" (4dp)}.

FeedbackEngine

class FeedbackEngine:
    def __init__(self, cfg: FeedbackConfig, alpha: float = 0.1): ...

Thread-safe (via a single threading.Lock guarding all state mutation) adaptive feedback loop.

Constructor

Parameters:

Side effects: Initializes empty _states dict and _history list. If cfg.persist_path is set and the file already exists on disk, immediately calls self._load(cfg.persist_path) to restore prior learned state — meaning a fresh FeedbackEngine can start with non-neutral biases if a persistence file from a previous run is found.

[!WARNING] FeedbackConfig.score_decay has no effect on FeedbackEngine's actual smoothing The EMA smoothing factor used throughout FeedbackEngine is the constructor's alpha parameter (default 0.1), not FeedbackConfig.score_decay (default 0.05). HierarchicalRouter constructs FeedbackEngine(self._cfg.feedback) — passing only the config, not a custom alpha — so every HierarchicalRouter-created FeedbackEngine uses alpha=0.1 regardless of what score_decay is set to in your RouterConfig. If you need a different smoothing rate, you would need to construct FeedbackEngine yourself with an explicit alpha argument rather than relying on RouterConfig.feedback.score_decay.

Recording

record(self, route_name: str, group_name: Optional[str], success: bool, score: float, query: str = "")

No-op if cfg.enabled is False. Otherwise: hashes query (MD5, truncated to 8 hex chars) for the log record, gets-or-creates a RouteFeedbackState for route_name, calls state.update(success, self._alpha), and appends a FeedbackRecord to self._history. If history exceeds 10,000 records, trims to the most recent 10,000 (self._history[-10_000:]) — a hardcoded bound not configurable via FeedbackConfig.

Thread safety: The entire state-mutation block is wrapped in with self._lock:.

bulk_record(self, records: List[Dict[str, Any]])

Convenience wrapper for post-hoc/batch labeling pipelines. Iterates records, calling self.record(...) once per entry, reading route_name/success as required keys (KeyError if missing) and group_name/score/query as optional (defaulting to None/0.0/"").

Example — bulk import from an offline evaluation:

engine.bulk_record([
    {"route_name": "docs_qa", "success": True, "score": 0.9, "query": "policy question"},
    {"route_name": "weather", "success": False, "score": 0.3},
])

Applying biases

apply_biases(self, route_map: Dict[str, Route])

No-op if cfg.enabled is False. Otherwise, for every tracked route with n_samples >= cfg.min_samples_to_adapt, looks it up in route_map and directly assigns route.score_bias = state.computed_bias (an outright overwrite, not an incremental adjustment via Route.apply_score_bias's clip-and-add pattern — though the end result lands in the same [-0.3, 0.3] range since computed_bias is already clipped).

Should be called periodically (per the docstring) — e.g. HierarchicalRouter calls it immediately after every record() call (both in _post_route's auto-feedback path and in the manual router.feedback(...) method), so biases take effect on the very next scoring pass in practice.

Persistence

save(self, path: Optional[str] = None)

Writes {route_name: state.to_dict()} for every tracked route as indented JSON to path (or cfg.persist_path if path is omitted). No-op if neither is set. Reads self._states under the lock, but performs the actual file write outside the lock.

_load(self, path: str) (private)

Reads a JSON file previously written by save(), reconstructing a RouteFeedbackState per entry (defaulting missing fields to their dataclass defaults). Silently ignores json.JSONDecodeError and OSError (corrupt or unreadable file) — a load failure leaves _states empty rather than raising, with no logging of the failure in the analyzed source.

Inspection

get_state(self, route_name: str) -> Optional[RouteFeedbackState]

Direct lookup, None if untracked.

summary(self) -> Dict[str, Any]

Returns {"enabled", "alpha", "min_samples", "tracked_routes" (count), "total_records" (history length), "states" (dict of every route's to_dict())}, read under the lock.

reset(self, route_name: Optional[str] = None)

If route_name given, removes just that route's tracked state (_states.pop(...)) — its history entries in _history are not removed. If route_name is omitted, clears both _states and _history entirely.

Examples

1. Basic:

from zerda.router.feedback.engine import FeedbackEngine
from zerda.router import FeedbackConfig

engine = FeedbackEngine(FeedbackConfig(min_samples_to_adapt=5))
engine.record(route_name="docs_qa", group_name="rag", success=True, score=0.9)
engine.apply_biases(route_map)  # route_map: Dict[str, Route]

2. Custom alpha for faster adaptation:

engine = FeedbackEngine(FeedbackConfig(), alpha=0.3)  # reacts faster to recent outcomes

3. Persisting and restoring learned state:

engine = FeedbackEngine(FeedbackConfig(persist_path="feedback.json"))
# ... record many outcomes ...
engine.save()  # writes feedback.json

# On next process start:
engine2 = FeedbackEngine(FeedbackConfig(persist_path="feedback.json"))
# engine2 automatically loaded prior state from feedback.json in its constructor

4. Inspecting a specific route's learned bias:

state = engine.get_state("docs_qa")
if state:
    print(f"EMA={state.ema:.3f}, bias={state.computed_bias:+.3f}, n={state.n_samples}")

5. Resetting after a route's handler implementation changed significantly:

engine.reset(route_name="docs_qa")  # forget learned bias, keep other routes' state
route.score_bias = 0.0  # also manually reset the route's live bias if needed immediately

Best practices

• Set min_samples_to_adapt proportional to your expected traffic per route — too low risks noisy early biasing from a handful of lucky/unlucky outcomes; too high delays adaptation indefinitely for low-traffic routes.
• If you rely on FeedbackConfig.score_decay to tune adaptation speed, be aware it is not actually wired to FeedbackEngine's alpha when constructed via HierarchicalRouter — construct your own FeedbackEngine(cfg, alpha=...) if you need non-default smoothing.
• Call save() explicitly (or use HierarchicalRouter as a context manager) if you want learned biases to survive process restarts — nothing persists automatically outside of an explicit save() call or the __exit__ hook.

Related APIs: Route.score_bias, HierarchicalRouter.feedback, FeedbackConfig