File size: 2,885 Bytes
c63c390 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | import json
import sys
from pathlib import Path
from sentence_transformers import SentenceTransformer
from memory_policy import evaluate_memory_write
from memory_store import MemoryStore, load_memories_from_json
ROOT = Path(__file__).resolve().parents[1]
def main():
cases = json.loads((ROOT / "eval" / "cases.json").read_text(encoding="utf-8"))
model = SentenceTransformer("all-MiniLM-L6-v2")
store = MemoryStore(load_memories_from_json(ROOT / "sample_memories.json"), embedding_model=model)
retrieval_rows = []
retrieval_failures = []
reciprocal_ranks = []
for case in cases["retrieval_cases"]:
results = store.retrieve(case["query"], top_k=3)
ranked_ids = [item.memory.memory_id for item in results]
try:
rank = ranked_ids.index(case["expected_memory_id"]) + 1
reciprocal_rank = 1.0 / rank
except ValueError:
rank = None
reciprocal_rank = 0.0
retrieval_failures.append(case["query"])
reciprocal_ranks.append(reciprocal_rank)
retrieval_rows.append(
{
"query": case["query"],
"expected_memory": case["expected_memory_id"],
"retrieved": ranked_ids,
"rank": rank,
}
)
policy_rows = []
policy_failures = []
for case in cases["write_policy_cases"]:
decision = evaluate_memory_write(case["query"])
passed = decision.outcome == case["expected_outcome"]
policy_rows.append(
{
"query": case["query"],
"expected": case["expected_outcome"],
"actual": decision.outcome,
"passed": passed,
}
)
if not passed:
policy_failures.append(case["query"])
mrr = sum(reciprocal_ranks) / len(reciprocal_ranks)
hit_at_3 = 1.0 - (len(retrieval_failures) / len(cases["retrieval_cases"]))
print("Memory retrieval evaluation:")
for row in retrieval_rows:
print(row)
print("\nMemory write-policy evaluation:")
for row in policy_rows:
print(row)
print("\nSummary:")
print({
"retrieval_cases": len(retrieval_rows),
"hit_at_3": round(hit_at_3, 3),
"mrr": round(mrr, 3),
"write_policy_passed": len(policy_rows) - len(policy_failures),
"write_policy_cases": len(policy_rows),
})
if retrieval_failures or mrr < 0.75 or policy_failures:
if retrieval_failures:
print(f"Retrieval failures: {retrieval_failures}", file=sys.stderr)
if mrr < 0.75:
print(f"MRR below required threshold: {mrr:.3f}", file=sys.stderr)
if policy_failures:
print(f"Write-policy failures: {policy_failures}", file=sys.stderr)
raise SystemExit(1)
if __name__ == "__main__":
main()
|