File size: 3,486 Bytes
8a1fc38 | 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 | """
Evaluation Service — Offline LLM-as-judge pipeline for answering quality.
"""
import logging
from typing import Dict, List, Any
from backend.src.config import settings
logger = logging.getLogger("math_assistant.eval")
class EvaluationService:
"""Uses LLM-as-judge to evaluate generated math answers against reference solutions."""
def __init__(self, judge_model: str = "gemini-2.5-flash"):
self.judge_model = judge_model
def evaluate_answer(self, question: str, generated_answer: str, reference_answer: str) -> Dict[str, Any]:
"""Ask the judge model to score the generated answer from 1 to 5."""
from google import genai
from google.genai import types
api_key = settings.gemini_api_key
if not api_key:
return {"score": 0, "reasoning": "Missing API Key for evaluation"}
client = genai.Client(api_key=api_key)
prompt = f"""You are an expert mathematics teacher grading a student's answer.
Question: {question}
Reference/Correct Answer:
{reference_answer}
Student's Answer to evaluate:
{generated_answer}
Grade the student's answer from 1 to 5 based on:
1. Accuracy: Is the final answer correct?
2. Steps: Are the steps logical and mathematically sound?
Format your response EXACTLY as follows:
SCORE: [1-5]
REASONING: [1-2 sentences explaining why]
"""
config = types.GenerateContentConfig(
temperature=0.1,
max_output_tokens=150,
)
try:
response = client.models.generate_content(
model=self.judge_model,
contents=prompt,
config=config
)
text = response.text.strip()
score = 0
reasoning = text
for line in text.split('\n'):
if line.startswith("SCORE:"):
try:
score = int(line.replace("SCORE:", "").strip())
except:
pass
elif line.startswith("REASONING:"):
reasoning = line.replace("REASONING:", "").strip()
return {"score": score, "reasoning": reasoning}
except Exception as e:
logger.error(f"Evaluation failed: {e}")
return {"score": 0, "reasoning": f"Evaluation error: {e}"}
def run_eval_pipeline(self, test_cases: List[Dict[str, str]]) -> Dict[str, Any]:
"""Run evaluation on a batch of test cases."""
results = []
total_score = 0
for case in test_cases:
res = self.evaluate_answer(
question=case['question'],
generated_answer=case['generated_answer'],
reference_answer=case['reference_answer']
)
total_score += res['score']
results.append({
"question": case['question'],
"score": res['score'],
"reasoning": res['reasoning']
})
avg_score = total_score / len(test_cases) if test_cases else 0
return {
"average_score": avg_score,
"total_evaluated": len(test_cases),
"results": results
}
|