app/training/replay_buffer.py

"""Replay buffer and failure-case mining utilities."""

from __future__ import annotations

from dataclasses import dataclass, field
import json
from pathlib import Path
from typing import Any


@dataclass
class ReplayBuffer:
    records: list[dict[str, Any]] = field(default_factory=list)

    def add(self, payload: dict[str, Any]) -> None:
        self.records.append(payload)

    def failures(self) -> list[dict[str, Any]]:
        out: list[dict[str, Any]] = []
        for row in self.records:
            reasons = row.get("failure_reasons") or []
            if reasons:
                out.append(row)
        return out

    def dump_jsonl(self, path: Path) -> Path:
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("w", encoding="utf-8") as f:
            for row in self.records:
                f.write(json.dumps(row, ensure_ascii=True) + "\n")
        return path

    def dump_failures_json(self, path: Path) -> Path:
        path.parent.mkdir(parents=True, exist_ok=True)
        failures = self.failures()
        path.write_text(json.dumps(failures, ensure_ascii=True, indent=2), encoding="utf-8")
        return path


def failure_mining_summary(rows: list[dict[str, Any]]) -> dict[str, Any]:
    reason_counts: dict[str, int] = {}
    for row in rows:
        for reason in row.get("failure_reasons") or []:
            reason_counts[reason] = reason_counts.get(reason, 0) + 1
    ranked = sorted(reason_counts.items(), key=lambda item: item[1], reverse=True)
    return {
        "total_rows": len(rows),
        "failure_rows": sum(1 for row in rows if row.get("failure_reasons")),
        "top_failure_reasons": [{"reason": k, "count": v} for k, v in ranked[:20]],
    }