Spaces:
Running on Zero
Running on Zero
| from __future__ import annotations | |
| import csv | |
| import json | |
| import re | |
| import time | |
| from pathlib import Path | |
| from typing import Any | |
| from llm_service import GeminiService | |
| from memory_engine import ( | |
| apply_memories, | |
| build_scene, | |
| detect_generation_overrides, | |
| detect_memory_reversals, | |
| extract_edit_candidates, | |
| find_memory, | |
| incorporate_llm_proposals, | |
| learn_from_edit, | |
| memory_context, | |
| memory_summary, | |
| normalize_memory, | |
| trim_history, | |
| ) | |
| from routing_policy import ( | |
| candidate_ids as route_candidate_ids, | |
| local_memory_decision, | |
| should_escalate_to_llm, | |
| ) | |
| from semantic_memory import quick_retrieve_memories, retrieve_memories | |
| REQUIRED_COLUMNS = ["chain_id", "step", "action", "mode"] | |
| OPTIONAL_COLUMNS = { | |
| "note_title", | |
| "input", | |
| "before", | |
| "after", | |
| "expect_output", | |
| "expect_contains", | |
| "expect_not_contains", | |
| "expect_event", | |
| "expect_memory", | |
| "expect_memory_absent", | |
| "expect_status", | |
| "expect_memory_status", | |
| "expect_negative_min", | |
| "expect_positive_min", | |
| "expect_exception_min", | |
| "expect_scope", | |
| "expect_route", | |
| "comment", | |
| "reset", | |
| } | |
| DEPRECATED_COLUMNS = {"use_llm", "use_ml"} | |
| EVENT_GROUPS = { | |
| "learn": {"memory_candidate", "memory_learned"}, | |
| "ignore": { | |
| "ignored_edit", | |
| "no_change", | |
| "undo", | |
| "no_memory_reversal", | |
| "no_generation_override", | |
| "context_skip", | |
| }, | |
| "negative_evidence": { | |
| "memory_negative_evidence", | |
| "memory_override", | |
| "memory_conflict", | |
| }, | |
| "apply": { | |
| "fast_path", | |
| "context_llm", | |
| "llm_format", | |
| "fast_fallback", | |
| }, | |
| "do_not_apply": {"context_skip"}, | |
| } | |
| def _bool(value: str) -> bool: | |
| return str(value or "").strip().lower() in { | |
| "1", "true", "yes", "y" | |
| } | |
| def _split(value: str) -> list[str]: | |
| value = str(value or "").strip() | |
| if not value: | |
| return [] | |
| value = value.replace("|", ";").replace("\n", ";") | |
| return [item.strip() for item in value.split(";") if item.strip()] | |
| def _memory_pair(value: str) -> tuple[str, str] | None: | |
| value = str(value or "").strip() | |
| if "=>" not in value: | |
| return None | |
| source, target = value.split("=>", 1) | |
| source = source.strip() | |
| target = target.strip() | |
| return (source, target) if source and target else None | |
| def _memory_pairs(value: str) -> list[tuple[str, str]]: | |
| return [ | |
| pair | |
| for item in _split(value) | |
| for pair in [_memory_pair(item)] | |
| if pair is not None | |
| ] | |
| def _memory_status_map(value: str) -> dict[tuple[str, str], str]: | |
| result: dict[tuple[str, str], str] = {} | |
| for item in _split(value): | |
| if "=" not in item: | |
| continue | |
| pair_text, status = item.rsplit("=", 1) | |
| pair = _memory_pair(pair_text) | |
| status = status.strip() | |
| if pair and status: | |
| result[pair] = status | |
| return result | |
| def _memory_pair_exact_exists( | |
| memories: list[dict[str, Any]], | |
| pair: tuple[str, str], | |
| ) -> bool: | |
| source, target = pair | |
| return any( | |
| str(memory.get("source", "")) == source | |
| and str(memory.get("target", "")) == target | |
| for memory in memories | |
| ) | |
| def _normalize_route(route: str) -> str: | |
| value = str(route or "").strip().lower() | |
| if value in {"deterministic", "local", "local no-memory"}: | |
| return "local" | |
| if value in {"semantic", "ml", "minilm"}: | |
| return "ml" | |
| if value in {"llm", "gemini"}: | |
| return "llm" | |
| return value or "local" | |
| def _semantic_event(result: dict[str, Any], action: str) -> str: | |
| event = str(result.get("event", "")) | |
| if action == "edit": | |
| for group, values in EVENT_GROUPS.items(): | |
| if event in values: | |
| return group | |
| return event | |
| changes = result.get("changes") or [] | |
| if changes: | |
| return "apply" | |
| candidate_ids = result.get("candidate_memory_ids") or [] | |
| if event == "context_skip": | |
| return "do_not_apply" if candidate_ids else "format" | |
| if event in {"llm_format", "fast_path", "ml_path", "fast_fallback", "context_llm"}: | |
| return "do_not_apply" if candidate_ids else "format" | |
| return event | |
| def _memory_snapshot(memories: list[dict[str, Any]]) -> list[dict[str, Any]]: | |
| return [ | |
| { | |
| "id": memory["id"], | |
| "source": memory["source"], | |
| "target": memory["target"], | |
| "kind": memory["kind"], | |
| "status": memory["status"], | |
| "confidence": memory["confidence"], | |
| "positive": memory["positive_evidence_count"], | |
| "negative": memory["negative_evidence_count"], | |
| "explicit": memory["explicit_evidence_count"], | |
| "exceptions": len(memory.get("exceptions", []) or []), | |
| "scope": memory.get("scope", "global"), | |
| } | |
| for memory in memory_summary(memories)["memories"] | |
| ] | |
| def _assert_row(row: dict[str, str], result: dict[str, Any]) -> tuple[bool, list[str]]: | |
| failures: list[str] = [] | |
| actual_output = str(result.get("actual_output", "")) | |
| memories = result.get("memories", []) | |
| semantic_event = str(result.get("semantic_event", "")) | |
| route = _normalize_route(result.get("route", "")) | |
| expected_output = str(row.get("expect_output", "")).strip() | |
| if expected_output and actual_output != expected_output: | |
| failures.append( | |
| f"output mismatch: expected={expected_output!r} actual={actual_output!r}" | |
| ) | |
| actual_folded = actual_output.casefold() | |
| for needle in _split(row.get("expect_contains", "")): | |
| if needle.casefold() not in actual_folded: | |
| failures.append(f"output missing {needle!r}") | |
| for needle in _split(row.get("expect_not_contains", "")): | |
| if needle not in actual_output: | |
| continue | |
| failures.append( | |
| f"output unexpectedly contains {needle!r}" | |
| ) | |
| expected_event = str(row.get("expect_event", "")).strip().lower() | |
| if expected_event: | |
| allowed = EVENT_GROUPS.get( | |
| expected_event, | |
| {expected_event}, | |
| ) | |
| if semantic_event != expected_event and str(result.get("event", "")) not in allowed: | |
| failures.append( | |
| f"event mismatch: expected semantic={expected_event!r} " | |
| f"actual semantic={semantic_event!r} raw={result.get('event', '')!r}" | |
| ) | |
| expected_route = str(row.get("expect_route", "")).strip().lower() | |
| if expected_route and route != _normalize_route(expected_route): | |
| failures.append( | |
| f"route mismatch: expected={expected_route!r} actual={route!r}" | |
| ) | |
| expected_pairs = _memory_pairs(row.get("expect_memory", "")) | |
| expected_status = str( | |
| row.get("expect_status", "") | |
| ).strip() | |
| expected_memory_status = _memory_status_map( | |
| row.get("expect_memory_status", "") | |
| ) | |
| for expected_pair in expected_pairs: | |
| memory = find_memory( | |
| memories, | |
| *expected_pair, | |
| ) | |
| if memory is None: | |
| failures.append( | |
| "expected memory missing: " | |
| f"{expected_pair[0]} => {expected_pair[1]}" | |
| ) | |
| continue | |
| pair_expected_status = expected_memory_status.get( | |
| expected_pair, | |
| expected_status, | |
| ) | |
| if pair_expected_status and memory.get("status") != pair_expected_status: | |
| failures.append( | |
| "memory status mismatch for " | |
| f"{expected_pair[0]} => {expected_pair[1]}: " | |
| f"expected={pair_expected_status!r} " | |
| f"actual={memory.get('status')!r}" | |
| ) | |
| for field, label in ( | |
| ("expect_negative_min", "negative evidence"), | |
| ("expect_positive_min", "positive evidence"), | |
| ): | |
| value = str(row.get(field, "")).strip() | |
| if not value: | |
| continue | |
| minimum = int(float(value)) | |
| key = ( | |
| "negative_evidence_count" | |
| if field == "expect_negative_min" | |
| else "positive_evidence_count" | |
| ) | |
| actual = int(memory.get(key, 0) or 0) | |
| if actual < minimum: | |
| failures.append( | |
| f"{label} too low for {expected_pair[0]} => {expected_pair[1]}: " | |
| f"expected>={minimum} actual={actual}" | |
| ) | |
| exception_min = str( | |
| row.get("expect_exception_min", "") | |
| ).strip() | |
| if exception_min: | |
| minimum = int(float(exception_min)) | |
| actual = len(memory.get("exceptions", []) or []) | |
| if actual < minimum: | |
| failures.append( | |
| f"exceptions too few for {expected_pair[0]} => {expected_pair[1]}: " | |
| f"expected>={minimum} actual={actual}" | |
| ) | |
| expected_scope = str( | |
| row.get("expect_scope", "") | |
| ).strip() | |
| if expected_scope and memory.get("scope") != expected_scope: | |
| failures.append( | |
| f"scope mismatch for {expected_pair[0]} => {expected_pair[1]}: " | |
| f"expected={expected_scope!r} actual={memory.get('scope')!r}" | |
| ) | |
| for value in _split(row.get("expect_memory_absent", "")): | |
| pair = _memory_pair(value) | |
| if pair and _memory_pair_exact_exists(memories, pair): | |
| failures.append( | |
| "unexpected exact memory exists: " | |
| f"{pair[0]} => {pair[1]}" | |
| ) | |
| return not failures, failures | |
| class ChainState: | |
| def __init__(self) -> None: | |
| self.notes: dict[str, str] = {} | |
| self.memories: list[dict[str, Any]] = [] | |
| self.last_generation: dict[str, Any] = {} | |
| self.last_trace: dict[str, Any] = {} | |
| self.history: list[dict[str, Any]] = [] | |
| def note(self, title: str) -> str: | |
| return self.notes.get(title, "") | |
| def set_note(self, title: str, content: str) -> None: | |
| self.notes[title] = content | |
| def reset(self) -> None: | |
| self.notes.clear() | |
| self.memories.clear() | |
| self.last_generation = {} | |
| self.last_trace = {} | |
| self.history.clear() | |
| class EvaluationRunner: | |
| def __init__(self, gemini: GeminiService | None = None) -> None: | |
| self.gemini = gemini or GeminiService() | |
| def _retrieve( | |
| self, | |
| text: str, | |
| note_title: str, | |
| note_content: str, | |
| memories: list[dict[str, Any]], | |
| policy: str, | |
| ) -> tuple[list[dict[str, Any]], float, str]: | |
| started = time.perf_counter() | |
| allow_ml = policy in {"auto", "ml_assisted", "force_llm"} | |
| if allow_ml: | |
| results = retrieve_memories( | |
| text, | |
| memories, | |
| note_title=note_title, | |
| scene_text=note_content[-600:], | |
| top_k=5, | |
| ) | |
| backend = ( | |
| results[0].get("retrieval_backend", "local") | |
| if results | |
| else "local/no-memory" | |
| ) | |
| else: | |
| results, _ = quick_retrieve_memories( | |
| text, | |
| memories, | |
| top_k=5, | |
| note_title=note_title, | |
| scene_text=note_content[-600:], | |
| ) | |
| backend = ( | |
| results[0].get( | |
| "retrieval_backend", | |
| "local lexical + context", | |
| ) | |
| if results | |
| else "local/no-memory" | |
| ) | |
| elapsed = round( | |
| (time.perf_counter() - started) * 1000, | |
| 2, | |
| ) | |
| return results, elapsed, backend | |
| def _fast_format(self, text: str) -> str: | |
| value = re.sub( | |
| r"\s+", | |
| " ", | |
| (text or "").strip(), | |
| ) | |
| value = re.sub( | |
| r"\b(?:um+|uh+|er+|ah+|hmm+)\b[\s,]*", | |
| " ", | |
| value, | |
| flags=re.IGNORECASE, | |
| ) | |
| value = re.sub( | |
| r"\b([A-Za-z]+)(?:\s+\1\b)+", | |
| r"\1", | |
| value, | |
| flags=re.IGNORECASE, | |
| ) | |
| value = re.sub( | |
| r"\s+([,.!?;:])", | |
| r"\1", | |
| value, | |
| ) | |
| value = re.sub( | |
| r"([,.!?;:])(?=\w)", | |
| r"\1 ", | |
| value, | |
| ) | |
| value = re.sub( | |
| r"\s{2,}", | |
| " ", | |
| value, | |
| ).strip() | |
| if not value: | |
| return "" | |
| value = value[0].upper() + value[1:] | |
| if value[-1] not in ".!?": | |
| value += "." | |
| return value | |
| def _candidate_ids(self, retrieved: list[dict[str, Any]]) -> set[str]: | |
| return route_candidate_ids(retrieved) | |
| def _route_name(self, backend: str, llm_used: bool) -> str: | |
| backend_text = str(backend or "").lower() | |
| ml_used = ( | |
| "minilm" in backend_text | |
| or "tf-idf" in backend_text | |
| or backend_text.startswith("ml ") | |
| or backend_text.startswith("ml:") | |
| ) | |
| if ml_used and llm_used: | |
| return "ml+llm" | |
| if llm_used: | |
| return "llm" | |
| if ml_used: | |
| return "ml" | |
| return "local" | |
| def _generate( | |
| self, | |
| state: ChainState, | |
| row: dict[str, str], | |
| policy: str, | |
| ) -> dict[str, Any]: | |
| note_title = str(row.get("note_title", "Default") or "Default") | |
| text = str(row.get("input", "") or "").strip() | |
| mode = str(row.get("mode", "Raw ASR") or "Raw ASR") | |
| note_content = state.note(note_title) | |
| scene = build_scene(note_title, note_content, text) | |
| retrieved, retrieval_ms, backend = self._retrieve( | |
| text, | |
| note_title, | |
| note_content, | |
| state.memories, | |
| policy, | |
| ) | |
| selected_ids = self._candidate_ids(retrieved) | |
| decision, local_ids, decision_reason, decision_details = local_memory_decision(retrieved) | |
| llm_metrics: dict[str, Any] = {"model": None, "api": None, "latency_ms": 0.0} | |
| llm_used = False | |
| rewritten = self._fast_format(text) | |
| decision_type = "context_skip" | |
| reason = decision_reason | |
| if policy == "force_llm" and self.gemini.available: | |
| try: | |
| result, llm_metrics = self.gemini.format_text( | |
| text, | |
| note_title, | |
| mode, | |
| scene["text"], | |
| memory_context(state.memories, selected_ids, limit=4), | |
| ) | |
| llm_used = True | |
| rewritten = result.formatted_text.strip() | |
| proposed = {memory_id for memory_id in result.memory_ids if memory_id in selected_ids} | |
| selected_ids = proposed | |
| decision_type = "llm_format" | |
| reason = "Force-LLM ablation requested a full LLM formatting pass." | |
| except Exception as exc: | |
| llm_metrics = {"model": self.gemini.model, "api": "interactions_v2", "latency_ms": 0.0, "error": str(exc)} | |
| decision_type = "fast_fallback" | |
| reason = "Gemini failed; Lexi used the local path." | |
| elif decision == "apply": | |
| selected_ids = local_ids | |
| decision_type = "fast_path" | |
| reason = decision_reason | |
| elif decision == "ambiguous" and policy == "auto" and should_escalate_to_llm( | |
| retrieved, | |
| decision, | |
| self.gemini.available, | |
| ): | |
| try: | |
| decision_result, llm_metrics = self.gemini.decide_memory_application( | |
| text, | |
| scene["text"], | |
| note_title, | |
| memory_context(state.memories, selected_ids, limit=4), | |
| ) | |
| llm_used = True | |
| proposed_ids = {memory_id for memory_id in decision_result.memory_ids if memory_id in selected_ids} | |
| if decision_result.decision == "apply" and proposed_ids: | |
| selected_ids = proposed_ids | |
| decision_type = "context_llm" | |
| reason = f"Local and ML evidence was ambiguous; Gemini resolved only among retrieved IDs: {decision_result.rationale}" | |
| else: | |
| selected_ids = set() | |
| decision_type = "context_skip" | |
| reason = f"Ambiguity remained unresolved, so Lexi did not intervene: {decision_result.rationale}" | |
| except Exception as exc: | |
| llm_metrics = {"model": self.gemini.model, "api": "interactions_v2", "latency_ms": 0.0, "error": str(exc)} | |
| selected_ids = set() | |
| decision_type = "context_skip" | |
| reason = "Context remained ambiguous and Lexi deliberately avoided intervention." | |
| else: | |
| selected_ids = set() | |
| decision_type = "context_skip" | |
| reason = decision_reason | |
| rewritten, changes = apply_memories( | |
| rewritten, | |
| state.memories, | |
| selected_ids=selected_ids, | |
| note_title=note_title, | |
| scene_text=note_content[-600:], | |
| ) | |
| state.set_note( | |
| note_title, | |
| rewritten if not note_content.strip() else note_content + "\n\n" + rewritten, | |
| ) | |
| state.last_generation = { | |
| "note_title": note_title, | |
| "input": text, | |
| "output": rewritten, | |
| "changes": changes, | |
| } | |
| route = self._route_name(backend, llm_used) | |
| trace = { | |
| "stage": "generation", | |
| "decision": decision_type, | |
| "semantic_event": _semantic_event( | |
| {"event": decision_type, "changes": changes, "retrieved": retrieved, "candidate_memory_ids": [item["memory_id"] for item in retrieved if item.get("term_present")]}, | |
| "generate", | |
| ), | |
| "mode": mode, | |
| "input": text, | |
| "output": rewritten, | |
| "reason": reason, | |
| "route": route, | |
| "retrieved": retrieved, | |
| "changes": changes, | |
| "retrieval_backend": backend, | |
| "routing": { | |
| "local_decision": decision, | |
| "local_details": decision_details, | |
| "llm_escalated": llm_used, | |
| }, | |
| "timing": {"retrieval_ms": retrieval_ms, "llm_ms": llm_metrics.get("latency_ms", 0.0)}, | |
| "model_metrics": llm_metrics, | |
| } | |
| state.last_trace = trace | |
| state.history.append(trace) | |
| return { | |
| "actual_output": rewritten, | |
| "event": decision_type, | |
| "semantic_event": trace["semantic_event"], | |
| "route": route, | |
| "retrieval_backend": backend, | |
| "retrieval_ms": retrieval_ms, | |
| "llm_ms": llm_metrics.get("latency_ms", 0.0), | |
| "ml_used": route in {"ml", "ml+llm"}, | |
| "llm_used": llm_used, | |
| "changes": changes, | |
| "candidate_memory_ids": [item["memory_id"] for item in retrieved if item.get("term_present") or item.get("retrieval_score", 0) >= 0.56], | |
| "memories": state.memories, | |
| "trace": trace, | |
| } | |
| def _edit( | |
| self, | |
| state: ChainState, | |
| row: dict[str, str], | |
| policy: str, | |
| ) -> dict[str, Any]: | |
| note_title = str(row.get("note_title", "Default") or "Default") | |
| before = str(row.get("before", "") or "") | |
| after = str(row.get("after", "") or "") | |
| state.set_note(note_title, after) | |
| previous_edit = state.history[-1] if state.history else {} | |
| last_generation = state.last_generation or {} | |
| llm_metrics: dict[str, Any] = {} | |
| llm_used = False | |
| updated, event = detect_generation_overrides( | |
| before, | |
| after, | |
| state.memories, | |
| last_generation, | |
| note_title, | |
| ) | |
| state.last_generation = {} | |
| if event.get("type") != "memory_override": | |
| updated, event = detect_memory_reversals( | |
| before, | |
| after, | |
| state.memories, | |
| note_title, | |
| ) | |
| if event.get("type") not in {"memory_override", "memory_negative_evidence"}: | |
| if policy == "force_llm" and self.gemini.available: | |
| try: | |
| analysis, llm_metrics = self.gemini.analyze_edit( | |
| before, | |
| after, | |
| note_title, | |
| build_scene(note_title, before, after)["text"], | |
| ) | |
| llm_used = True | |
| updated, event = incorporate_llm_proposals( | |
| state.memories, | |
| [proposal.model_dump() for proposal in analysis.proposals], | |
| before, | |
| after, | |
| note_title=note_title, | |
| decision=analysis.decision, | |
| ) | |
| event["llm_decision"] = analysis.model_dump() | |
| except Exception as exc: | |
| llm_metrics = {"model": self.gemini.model, "api": "interactions_v2", "latency_ms": 0.0, "error": str(exc)} | |
| updated, event = learn_from_edit( | |
| before, after, state.memories, | |
| previous_edit=previous_edit, | |
| last_generation={}, | |
| note_title=note_title, | |
| ) | |
| else: | |
| updated, event = learn_from_edit( | |
| before, after, state.memories, | |
| previous_edit=previous_edit, | |
| last_generation={}, | |
| note_title=note_title, | |
| ) | |
| state.memories = [normalize_memory(memory) for memory in updated] | |
| semantic_event = _semantic_event({"event": event.get("type", "unknown")}, "edit") | |
| route = "llm" if llm_used else "local" | |
| trace = { | |
| "stage": "learning", | |
| "decision": event.get("type", "unknown"), | |
| "semantic_event": semantic_event, | |
| "before": before, | |
| "after": after, | |
| "note_title": note_title, | |
| "event": event, | |
| "route": route, | |
| "model_metrics": llm_metrics, | |
| } | |
| state.last_trace = trace | |
| state.history.append(trace) | |
| return { | |
| "actual_output": after, | |
| "event": event.get("type", "unknown"), | |
| "semantic_event": semantic_event, | |
| "route": route, | |
| "retrieval_backend": "n/a", | |
| "retrieval_ms": 0.0, | |
| "llm_ms": llm_metrics.get("latency_ms", 0.0), | |
| "ml_used": False, | |
| "llm_used": llm_used, | |
| "changes": [], | |
| "memories": state.memories, | |
| "trace": trace, | |
| } | |
| def run_rows( | |
| self, | |
| rows: list[dict[str, str]], | |
| policy: str = "auto", | |
| ) -> dict[str, Any]: | |
| if policy not in { | |
| "auto", | |
| "local_only", | |
| "ml_assisted", | |
| "no_ml", | |
| "force_llm", | |
| }: | |
| policy = "auto" | |
| chains: dict[str, ChainState] = {} | |
| results: list[dict[str, Any]] = [] | |
| started_total = time.perf_counter() | |
| for raw in rows: | |
| row = { | |
| str(k): str(v or "") | |
| for k, v in raw.items() | |
| } | |
| chain_id = row.get("chain_id", "default") or "default" | |
| state = chains.setdefault( | |
| chain_id, | |
| ChainState(), | |
| ) | |
| if _bool(row.get("reset", "0")): | |
| state.reset() | |
| action = ( | |
| row.get("action", "generate") | |
| or "generate" | |
| ).strip().lower() | |
| if policy == "force_llm" and not self.gemini.available: | |
| actual = { | |
| "actual_output": "", | |
| "event": "error", | |
| "semantic_event": "error", | |
| "route": "llm", | |
| "memories": state.memories, | |
| "trace": {}, | |
| } | |
| actual["error"] = "Gemini is not configured." | |
| elif action == "reset": | |
| state.reset() | |
| actual = { | |
| "actual_output": "", | |
| "event": "reset", | |
| "semantic_event": "reset", | |
| "route": "local", | |
| "memories": state.memories, | |
| "trace": {}, | |
| } | |
| elif action == "edit": | |
| actual = self._edit( | |
| state, | |
| row, | |
| policy, | |
| ) | |
| elif action == "generate": | |
| if policy == "force_llm" and self.gemini.available: | |
| effective_policy = "auto" | |
| original = row.get("mode", "Raw ASR") | |
| if original == "Formatted": | |
| actual = self._generate( | |
| state, | |
| row, | |
| effective_policy, | |
| ) | |
| else: | |
| actual = self._generate( | |
| state, | |
| row, | |
| effective_policy, | |
| ) | |
| else: | |
| actual = self._generate( | |
| state, | |
| row, | |
| policy, | |
| ) | |
| else: | |
| actual = { | |
| "actual_output": "", | |
| "event": "error", | |
| "semantic_event": "error", | |
| "route": "local", | |
| "memories": state.memories, | |
| "trace": {}, | |
| "error": f"Unknown action: {action}", | |
| } | |
| passed, failures = _assert_row( | |
| row, | |
| actual, | |
| ) | |
| results.append( | |
| { | |
| "chain_id": chain_id, | |
| "step": row.get("step", ""), | |
| "action": action, | |
| "mode": row.get("mode", ""), | |
| "note_title": row.get("note_title", ""), | |
| "status": "PASS" if passed else "FAIL", | |
| "passed": passed, | |
| "event": actual.get("event", ""), | |
| "semantic_event": actual.get("semantic_event", ""), | |
| "route": actual.get("route", ""), | |
| "ml_used": actual.get("ml_used", False), | |
| "llm_used": actual.get("llm_used", False), | |
| "actual_output": actual.get("actual_output", ""), | |
| "failures": " | ".join(failures), | |
| "retrieval_backend": actual.get("retrieval_backend", ""), | |
| "retrieval_ms": actual.get("retrieval_ms", 0.0), | |
| "llm_ms": actual.get("llm_ms", 0.0), | |
| "memory_count": len(actual.get("memories", [])), | |
| "memory_state": json.dumps( | |
| _memory_snapshot( | |
| actual.get("memories", []) | |
| ), | |
| ensure_ascii=False, | |
| ), | |
| "trace": json.dumps( | |
| actual.get("trace", {}), | |
| ensure_ascii=False, | |
| ), | |
| "comment": row.get("comment", ""), | |
| } | |
| ) | |
| passed = sum( | |
| bool(item["passed"]) | |
| for item in results | |
| ) | |
| llm_calls = sum( | |
| bool(item.get("llm_used")) | |
| for item in results | |
| ) | |
| ml_calls = sum( | |
| bool(item.get("ml_used")) | |
| for item in results | |
| ) | |
| route_counts: dict[str, int] = {} | |
| for item in results: | |
| route = _normalize_route(item.get("route", "")) | |
| route_counts[route] = route_counts.get(route, 0) + 1 | |
| return { | |
| "summary": { | |
| "policy": policy, | |
| "total_steps": len(results), | |
| "passed_steps": passed, | |
| "failed_steps": len(results) - passed, | |
| "pass_rate": round( | |
| passed / len(results), | |
| 4, | |
| ) if results else 0.0, | |
| "llm_calls": llm_calls, | |
| "ml_calls": ml_calls, | |
| "route_counts": route_counts, | |
| "total_llm_ms": round( | |
| sum( | |
| float(item.get("llm_ms", 0.0) or 0.0) | |
| for item in results | |
| ), | |
| 1, | |
| ), | |
| "total_ml_ms": round( | |
| sum( | |
| float(item.get("retrieval_ms", 0.0) or 0.0) | |
| for item in results | |
| if item.get("ml_used") | |
| ), | |
| 1, | |
| ), | |
| "total_runtime_ms": round( | |
| ( | |
| time.perf_counter() | |
| - started_total | |
| ) * 1000, | |
| 1, | |
| ), | |
| "gemini_available": self.gemini.available, | |
| }, | |
| "results": results, | |
| } | |
| def load_csv(path: str | Path) -> list[dict[str, str]]: | |
| with open( | |
| path, | |
| "r", | |
| encoding="utf-8-sig", | |
| newline="", | |
| ) as handle: | |
| reader = csv.DictReader(handle) | |
| if reader.fieldnames is None: | |
| raise ValueError("CSV has no header row.") | |
| field_map = { | |
| str(field).strip(): field | |
| for field in reader.fieldnames | |
| if field is not None | |
| } | |
| fields = set(field_map) | |
| missing = [ | |
| column | |
| for column in REQUIRED_COLUMNS | |
| if column not in fields | |
| ] | |
| if missing: | |
| raise ValueError( | |
| "CSV is missing required columns: " | |
| + ", ".join(missing) | |
| + ". Required columns are chain_id, step, action, mode." | |
| ) | |
| deprecated = sorted(fields.intersection(DEPRECATED_COLUMNS)) | |
| if deprecated: | |
| raise ValueError( | |
| "Remove " | |
| + ", ".join(deprecated) | |
| + ". Lexi now chooses ML/LLM routing automatically; " | |
| "the CSV must describe only the user journey and expected behavior." | |
| ) | |
| rows = [] | |
| for index, raw in enumerate(reader, start=2): | |
| row = { | |
| clean_key: str(raw.get(original_key, "") or "") | |
| for clean_key, original_key in field_map.items() | |
| } | |
| if not any(value.strip() for value in row.values()): | |
| continue | |
| action = row.get("action", "").strip().lower() | |
| if action == "generate" and not row.get("input", "").strip(): | |
| raise ValueError( | |
| f"Row {index}: generate rows require input." | |
| ) | |
| if action == "edit" and ( | |
| not row.get("before", "").strip() | |
| and not row.get("after", "").strip() | |
| ): | |
| raise ValueError( | |
| f"Row {index}: edit rows require before/after." | |
| ) | |
| rows.append(row) | |
| return rows | |
| def write_reports( | |
| report: dict[str, Any], | |
| output_dir: str | Path, | |
| ) -> tuple[str, str]: | |
| output = Path(output_dir) | |
| output.mkdir( | |
| parents=True, | |
| exist_ok=True, | |
| ) | |
| stamp = time.strftime( | |
| "%Y%m%d_%H%M%S" | |
| ) | |
| json_path = output / f"lexi_evaluation_{stamp}.json" | |
| csv_path = output / f"lexi_evaluation_{stamp}.csv" | |
| json_path.write_text( | |
| json.dumps( | |
| report, | |
| indent=2, | |
| ensure_ascii=False, | |
| ), | |
| encoding="utf-8", | |
| ) | |
| results = report.get("results", []) | |
| fieldnames = list(results[0].keys()) if results else [] | |
| with csv_path.open( | |
| "w", | |
| encoding="utf-8", | |
| newline="", | |
| ) as handle: | |
| writer = csv.DictWriter( | |
| handle, | |
| fieldnames=fieldnames, | |
| ) | |
| if fieldnames: | |
| writer.writeheader() | |
| writer.writerows(results) | |
| return str(json_path), str(csv_path) | |