File size: 4,341 Bytes
9513328 | 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 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | #!/usr/bin/env python3
"""#8 — Model Evaluation Harness. Benchmarks models on Real-CATS scam data.
Runs lm-eval locally or via Ollama. Picks the best model per task."""
import asyncio
import json
import os
import time
from pathlib import Path
from typing import Any
import httpx
OLLAMA = os.getenv("OLLAMA_HOST", "http://localhost:11434")
REAL_CATS_PATH = Path(os.getenv("REAL_CATS_PATH", str(Path.home() / "rmi/backend/data/real_cats.json")))
# Test prompts for scam classification
BENCHMARK_TASKS = {
"scam_detection": {
"prompts": [
{
"input": "Token has mint authority enabled, liquidity is 0.5 SOL unlocked, deployer created 50 tokens before. Is this a scam?",
"expected": "yes",
},
{
"input": "Token has renounced mint, liquidity locked for 1 year, verified contract, audited by CertiK. Is this a scam?",
"expected": "no",
},
{
"input": "Token has honeypot detection enabled, 99% sell tax, unverified contract, anonymous team. Is this a scam?",
"expected": "yes",
},
{
"input": "Token listed on Binance, $50M market cap, 100K holders, 2 years old. Is this a scam?",
"expected": "no",
},
],
"metric": "accuracy",
},
}
async def evaluate_model(model: str, task_name: str) -> dict[str, Any]:
"""Evaluate a model on a benchmark task."""
task = BENCHMARK_TASKS.get(task_name)
if not task:
return {"error": f"Unknown task: {task_name}"}
correct = 0
total = 0
total_time = 0.0
results = []
async with httpx.AsyncClient(timeout=60) as c:
for item in task["prompts"]:
start = time.perf_counter()
try:
r = await c.post(
f"{OLLAMA}/api/generate",
json={
"model": model,
"prompt": f"Answer only YES or NO. {item['input']}",
"stream": False,
"options": {"num_predict": 5, "temperature": 0.1},
},
)
elapsed = time.perf_counter() - start
total_time += elapsed
response = r.json().get("response", "").strip().upper()
is_correct = item["expected"].upper() in response
if is_correct:
correct += 1
total += 1
results.append(
{
"input": item["input"][:80],
"expected": item["expected"],
"got": response[:20],
"correct": is_correct,
"time_ms": round(elapsed * 1000),
}
)
except Exception as e:
results.append({"input": item["input"][:80], "error": str(e)})
total += 1
accuracy = (correct / total * 100) if total > 0 else 0
return {
"model": model,
"task": task_name,
"accuracy": round(accuracy, 1),
"correct": correct,
"total": total,
"avg_time_ms": round((total_time / total) * 1000) if total > 0 else 0,
"results": results,
}
async def compare_models(models: list[str], task: str = "scam_detection"):
"""Compare multiple models on a benchmark task."""
scores = []
for model in models:
result = await evaluate_model(model, task)
scores.append(result)
scores.sort(key=lambda s: s["accuracy"], reverse=True)
return {
"task": task,
"models_compared": len(scores),
"leaderboard": [
{"model": s["model"], "accuracy": s["accuracy"], "avg_time_ms": s["avg_time_ms"]} for s in scores
],
"best_model": scores[0]["model"] if scores else None,
}
if __name__ == "__main__":
async def main():
print("Model Evaluation Harness")
print("=" * 40)
models = ["qwen2.5-coder:7b", "mistral:7b"]
results = await compare_models(models)
print(json.dumps(results["leaderboard"], indent=2))
print(f"\nBest model for scam detection: {results['best_model']}")
asyncio.run(main())
|