File size: 4,436 Bytes
bde2f3a | 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 130 131 132 | #!/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
from app.core.logging import get_logger
logger = get_logger(__name__)
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():
logger.info("Model Evaluation Harness")
logger.info("=" * 40)
models = ["qwen2.5-coder:7b", "mistral:7b"]
results = await compare_models(models)
logger.info(json.dumps(results["leaderboard"], indent=2))
logger.info(f"\nBest model for scam detection: {results['best_model']}")
asyncio.run(main())
|