""" core/evaluator.py - LLM-as-Judge Evaluation System (Phase 4A) ============================================================ Uses GPT-5 (or GPT-4o) as an expert evaluator to score RAG pipeline answers on multiple quality dimensions. Scoring Dimensions: - Correctness (0-10): Factual accuracy - Relevance (0-10): Addresses the question - Completeness (0-10): Sufficient detail - Clarity (0-10): Clear and understandable - Conciseness (0-10): Not overly verbose """ from typing import List, Dict, Optional, Tuple from dataclasses import dataclass import json import time from openai import AzureOpenAI import os from dotenv import load_dotenv load_dotenv() @dataclass class EvaluationScores: """Container for multi-dimensional quality scores""" correctness_score: float # 0-10 relevance_score: float # 0-10 completeness_score: float # 0-10 clarity_score: float # 0-10 conciseness_score: float # 0-10 overall_score: float # 0-10 (weighted average) confidence: float # 0-1 (judge's confidence) explanation: str # Why these scores? issues: List[str] # List of specific problems found evaluator_model: str # Model used as judge evaluation_cost_usd: float # Cost of this evaluation evaluation_time_ms: float # Latency class LLMJudge: """ GPT-5/GPT-4o as an expert judge for RAG answer quality Evaluates answers on 5 dimensions and provides detailed feedback. """ # Cost per 1K tokens for judge model (GPT-5 or GPT-4o) JUDGE_COSTS = { "gpt-5-chat": (0.005, 0.015), # $5/$15 per 1M tokens "gpt-4o": (0.005, 0.015), "gpt-4o-mini": (0.00015, 0.0006), } def __init__( self, judge_model: str = "gpt-5-chat", azure_endpoint: Optional[str] = None, azure_api_key: Optional[str] = None, azure_deployment: Optional[str] = None, temperature: float = 0.0, # Deterministic scoring verbose: bool = False ): """ Initialize LLM judge Args: judge_model: Model name (gpt-5-chat, gpt-4o, gpt-4o-mini) azure_endpoint: Azure OpenAI endpoint azure_api_key: Azure API key azure_deployment: Azure deployment name temperature: 0.0 for consistent scoring verbose: Print evaluation details """ self.judge_model = judge_model self.temperature = temperature self.verbose = verbose # Initialize Azure OpenAI client for judge self.client = AzureOpenAI( api_key=azure_api_key or os.getenv("AZURE_OPENAI_API_KEY"), api_version="2024-02-01", azure_endpoint=azure_endpoint or os.getenv("AZURE_OPENAI_ENDPOINT"), ) self.deployment = azure_deployment or os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME") or judge_model if self.verbose: print(f"[LLMJudge] Initialized with {self.judge_model}") print(f"[LLMJudge] Deployment: {self.deployment}") def evaluate( self, query: str, ground_truth_answers: List[str], generated_answer: str, retrieved_context: List[str] ) -> EvaluationScores: """ Evaluate a generated answer against ground truth Args: query: The question asked ground_truth_answers: List of acceptable answers generated_answer: Answer generated by RAG pipeline retrieved_context: Context chunks used for generation Returns: EvaluationScores with multi-dimensional ratings """ # Build evaluation prompt system_prompt = self._build_system_prompt() user_prompt = self._build_user_prompt( query=query, ground_truth_answers=ground_truth_answers, generated_answer=generated_answer, retrieved_context=retrieved_context ) # Call judge model start_time = time.time() try: response = self.client.chat.completions.create( model=self.deployment, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], temperature=self.temperature, max_tokens=1000, response_format={"type": "json_object"} # Force JSON output ) # Extract tokens and calculate cost prompt_tokens = response.usage.prompt_tokens completion_tokens = response.usage.completion_tokens input_cost, output_cost = self.JUDGE_COSTS[self.judge_model] cost = (prompt_tokens / 1000 * input_cost) + (completion_tokens / 1000 * output_cost) latency_ms = (time.time() - start_time) * 1000 # Parse JSON response eval_data = json.loads(response.choices[0].message.content) # Extract scores scores = EvaluationScores( correctness_score=float(eval_data.get("correctness", 0)), relevance_score=float(eval_data.get("relevance", 0)), completeness_score=float(eval_data.get("completeness", 0)), clarity_score=float(eval_data.get("clarity", 0)), conciseness_score=float(eval_data.get("conciseness", 0)), overall_score=float(eval_data.get("overall_score", 0)), confidence=float(eval_data.get("confidence", 0.5)), explanation=eval_data.get("explanation", ""), issues=eval_data.get("issues", []), evaluator_model=self.judge_model, evaluation_cost_usd=cost, evaluation_time_ms=latency_ms ) if self.verbose: print(f"[LLMJudge] Evaluated in {latency_ms:.0f}ms (${cost:.6f})") print(f"[LLMJudge] Overall Score: {scores.overall_score:.1f}/10") return scores except Exception as e: print(f"[LLMJudge] ERROR: {e}") # Return default (failed) scores return EvaluationScores( correctness_score=0.0, relevance_score=0.0, completeness_score=0.0, clarity_score=0.0, conciseness_score=0.0, overall_score=0.0, confidence=0.0, explanation=f"Evaluation failed: {str(e)}", issues=["evaluation_error"], evaluator_model=self.judge_model, evaluation_cost_usd=0.0, evaluation_time_ms=(time.time() - start_time) * 1000 ) def _build_system_prompt(self) -> str: """Build system prompt for the judge""" return """You are an expert evaluator for Retrieval-Augmented Generation (RAG) systems. Your task is to evaluate the quality of answers generated by RAG pipelines. **Evaluation Criteria:** 1. **Correctness (0-10)**: Is the answer factually accurate compared to ground truth? - 10: Perfect match with ground truth - 7-9: Correct with minor differences in phrasing - 4-6: Partially correct or missing key details - 1-3: Mostly incorrect - 0: Completely wrong or "I don't know" when answer exists 2. **Relevance (0-10)**: Does the answer directly address the question? - 10: Perfectly on-topic - 7-9: Addresses question with minor tangents - 4-6: Partially relevant - 1-3: Mostly off-topic - 0: Completely irrelevant 3. **Completeness (0-10)**: Is the answer sufficiently detailed? - 10: Comprehensive, includes all necessary information - 7-9: Good detail, minor gaps - 4-6: Basic answer, missing context - 1-3: Too brief, incomplete - 0: No meaningful content 4. **Clarity (0-10)**: Is the answer clear and understandable? - 10: Crystal clear, well-structured - 7-9: Clear with minor ambiguity - 4-6: Somewhat confusing - 1-3: Very unclear - 0: Incomprehensible 5. **Conciseness (0-10)**: Is the answer appropriately brief? - 10: Perfect length, no fluff - 7-9: Slightly verbose but acceptable - 4-6: Somewhat wordy - 1-3: Very verbose - 0: Extremely long-winded **Important Notes:** - Be **lenient** if the generated answer is semantically equivalent to ground truth even with different wording - If the model makes **reasonable inferences** from context, give credit - Only penalize "I don't know" responses if ground truth exists - Consider that retrieved context may limit what the model can answer **Output Format (JSON only):** { "correctness": <0-10>, "relevance": <0-10>, "completeness": <0-10>, "clarity": <0-10>, "conciseness": <0-10>, "overall_score": <0-10 weighted average>, "confidence": <0.0-1.0>, "explanation": "", "issues": ["", "", ...] } **Issue Types:** - "hallucination" - Answer contains false information - "retrieval_failure" - Context didn't contain answer - "generation_error" - Model failed to use context properly - "factual_error" - Wrong facts - "incomplete" - Missing key information - "off_topic" - Doesn't address question - "too_verbose" - Unnecessarily long""" def _build_user_prompt( self, query: str, ground_truth_answers: List[str], generated_answer: str, retrieved_context: List[str] ) -> str: """Build user prompt with evaluation data""" # Format ground truth gt_text = "\n".join([f"- {ans}" for ans in ground_truth_answers]) # Format context context_text = "\n\n".join([f"[{i+1}] {chunk[:300]}..." if len(chunk) > 300 else f"[{i+1}] {chunk}" for i, chunk in enumerate(retrieved_context[:5])]) # Limit to 5 chunks return f"""**Question:** {query} **Ground Truth Answers:** {gt_text} **Model's Generated Answer:** {generated_answer} **Retrieved Context (used by model):** {context_text} --- Evaluate the model's answer and provide scores in JSON format.""" def evaluate_batch( judge: LLMJudge, evaluation_results: List[Dict] ) -> List[Tuple[Dict, EvaluationScores]]: """ Evaluate a batch of evaluation results Args: judge: LLMJudge instance evaluation_results: List of dicts with keys: - query - ground_truth_answers (JSON list string) - generated_answer - retrieved_chunks (JSON list string) Returns: List of (original_result, scores) tuples """ evaluated = [] for result in evaluation_results: # Parse JSON strings ground_truth = json.loads(result['ground_truth_answers']) context = json.loads(result['retrieved_chunks']) # Evaluate scores = judge.evaluate( query=result['query'], ground_truth_answers=ground_truth, generated_answer=result['generated_answer'], retrieved_context=context ) evaluated.append((result, scores)) return evaluated # ============================================================================ # USAGE EXAMPLE - Test Judge on Sample Answer # ============================================================================ if __name__ == "__main__": print("šŸ§‘ā€āš–ļø LLM Judge Test - Phase 4A") print("=" * 80) # Initialize judge (using your GPT-5 deployment) judge = LLMJudge( judge_model="gpt-5-chat", azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"), azure_api_key=os.getenv("AZURE_OPENAI_API_KEY"), azure_deployment=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"), verbose=True ) # Sample evaluation case query = "What is the capital of France?" ground_truth = ["Paris"] generated_answer = "The capital of France is Paris, a major European city known for the Eiffel Tower." context = [ "Paris is the capital and most populous city of France.", "The Eiffel Tower is an iron lattice tower in Paris.", "France is a country in Western Europe." ] print(f"\nšŸ“ Evaluating:") print(f" Q: {query}") print(f" Ground Truth: {ground_truth}") print(f" Generated: {generated_answer}") print("\n" + "-" * 80) # Evaluate scores = judge.evaluate( query=query, ground_truth_answers=ground_truth, generated_answer=generated_answer, retrieved_context=context ) # Print results print(f"\nšŸ“Š Evaluation Results:") print(f" Correctness: {scores.correctness_score:.1f}/10") print(f" Relevance: {scores.relevance_score:.1f}/10") print(f" Completeness: {scores.completeness_score:.1f}/10") print(f" Clarity: {scores.clarity_score:.1f}/10") print(f" Conciseness: {scores.conciseness_score:.1f}/10") print(f" Overall: {scores.overall_score:.1f}/10") print(f"\n Confidence: {scores.confidence:.2f}") print(f" Explanation: {scores.explanation}") print(f" Issues: {scores.issues}") print(f"\n Cost: ${scores.evaluation_cost_usd:.6f}") print(f" Time: {scores.evaluation_time_ms:.0f}ms") print("\n" + "=" * 80) print("āœ… Judge test complete!") print("\nšŸš€ Next: Create database schema for storing these scores")