from uuid import uuid4 from pathlib import Path from typing import Optional, Union, Callable, List from pydantic import BaseModel, ConfigDict, field_validator, Field from phi.agent import Agent, RunResponse from phi.utils.log import logger, set_log_level_to_debug from phi.utils.timer import Timer class AccuracyResult(BaseModel): score: int = Field(..., description="Accuracy Score between 1 and 10 assigned to the Agent's answer.") reason: str = Field(..., description="Detailed reasoning for the accuracy score.") class EvalResult(BaseModel): accuracy_score: int = Field(..., description="Accuracy Score between 1 to 10.") accuracy_reason: str = Field(..., description="Reasoning for the accuracy score.") class Eval(BaseModel): # Evaluation name name: Optional[str] = None # Evaluation UUID (autogenerated if not set) eval_id: Optional[str] = Field(None, validate_default=True) # Agent to evaluate agent: Optional[Agent] = None # Question to evaluate question: str answer: Optional[str] = None # Expected Answer for the question expected_answer: str # Result of the evaluation result: Optional[EvalResult] = None accuracy_evaluator: Optional[Agent] = None # Guidelines for the accuracy evaluator accuracy_guidelines: Optional[List[str]] = None # Additional context to the accuracy evaluator accuracy_context: Optional[str] = None accuracy_result: Optional[AccuracyResult] = None # Save the result to a file save_result_to_file: Optional[str] = None # debug_mode=True enables debug logs debug_mode: bool = False model_config = ConfigDict(arbitrary_types_allowed=True) @field_validator("eval_id", mode="before") def set_eval_id(cls, v: Optional[str] = None) -> str: return v or str(uuid4()) @field_validator("debug_mode", mode="before") def set_log_level(cls, v: bool) -> bool: if v: set_log_level_to_debug() logger.debug("Debug logs enabled") return v def get_accuracy_evaluator(self) -> Agent: if self.accuracy_evaluator is not None: return self.accuracy_evaluator try: from phi.model.openai import OpenAIChat except ImportError as e: logger.exception(e) logger.error( "phidata uses `openai` as the default model provider. Please run `pip install openai` to use the default evaluator." ) exit(1) accuracy_guidelines = "" if self.accuracy_guidelines is not None and len(self.accuracy_guidelines) > 0: accuracy_guidelines = "\n## Guidelines for the AI Agent's answer:\n" accuracy_guidelines += "\n- ".join(self.accuracy_guidelines) accuracy_guidelines += "\n" accuracy_context = "" if self.accuracy_context is not None and len(self.accuracy_context) > 0: accuracy_context = "## Additional Context:\n" accuracy_context += self.accuracy_context accuracy_context += "\n" return Agent( model=OpenAIChat(id="gpt-4o-mini"), description=f"""\ You are an expert evaluator tasked with assessing the accuracy of an AI Agent's answer compared to an expected answer for a given question. Your task is to provide a detailed analysis and assign a score on a scale of 1 to 10, where 10 indicates a perfect match to the expected answer. ## Question: {self.question} ## Expected Answer: {self.expected_answer} ## Evaluation Criteria: 1. Accuracy of information 2. Completeness of the answer 3. Relevance to the question 4. Use of key concepts and ideas 5. Overall structure and clarity of presentation {accuracy_guidelines}{accuracy_context} ## Instructions: 1. Carefully compare the AI Agent's answer to the expected answer. 2. Provide a detailed analysis, highlighting: - Specific similarities and differences - Key points included or missed - Any inaccuracies or misconceptions 3. Explicitly reference the evaluation criteria and any provided guidelines in your reasoning. 4. Assign a score from 1 to 10 (use only whole numbers) based on the following scale: 1-2: Completely incorrect or irrelevant 3-4: Major inaccuracies or missing crucial information 5-6: Partially correct, but with significant omissions or errors 7-8: Mostly accurate and complete, with minor issues 9-10: Highly accurate and complete, matching the expected answer closely Your evaluation should be objective, thorough, and well-reasoned. Provide specific examples from both answers to support your assessment.""", response_model=AccuracyResult, ) def run(self, answer: Optional[Union[str, Callable]] = None) -> Optional[EvalResult]: logger.debug(f"*********** Evaluation Start: {self.eval_id} ***********") answer_to_evaluate: Optional[RunResponse] = None if answer is None: if self.agent is not None: logger.debug("Getting answer from agent") answer_to_evaluate = self.agent.run(self.question) if self.answer is not None: answer_to_evaluate = RunResponse(content=self.answer) else: try: if callable(answer): logger.debug("Getting answer from callable") answer_to_evaluate = RunResponse(content=answer()) else: answer_to_evaluate = RunResponse(content=answer) except Exception as e: logger.error(f"Failed to get answer: {e}") raise if answer_to_evaluate is None: raise ValueError("No Answer to evaluate.") else: self.answer = answer_to_evaluate.content logger.debug("************************ Evaluating ************************") logger.debug(f"Question: {self.question}") logger.debug(f"Expected Answer: {self.expected_answer}") logger.debug(f"Answer: {answer_to_evaluate}") logger.debug("************************************************************") logger.debug("Evaluating accuracy...") accuracy_evaluator = self.get_accuracy_evaluator() try: self.accuracy_result: AccuracyResult = accuracy_evaluator.run( answer_to_evaluate.content, stream=False ).content except Exception as e: logger.error(f"Failed to evaluate accuracy: {e}") return None if self.accuracy_result is not None: self.result = EvalResult( accuracy_score=self.accuracy_result.score, accuracy_reason=self.accuracy_result.reason, ) # -*- Save result to file if save_result_to_file is set if self.save_result_to_file is not None and self.result is not None: try: fn_path = Path(self.save_result_to_file.format(name=self.name, eval_id=self.eval_id)) if not fn_path.parent.exists(): fn_path.parent.mkdir(parents=True, exist_ok=True) fn_path.write_text(self.result.model_dump_json(indent=4)) except Exception as e: logger.warning(f"Failed to save result to file: {e}") logger.debug(f"*********** Evaluation End: {self.eval_id} ***********") return self.result def print_result(self, answer: Optional[Union[str, Callable]] = None) -> Optional[EvalResult]: from phi.cli.console import console from rich.table import Table from rich.progress import Progress, SpinnerColumn, TextColumn from rich.box import ROUNDED response_timer = Timer() response_timer.start() with Progress(SpinnerColumn(spinner_name="dots"), TextColumn("{task.description}"), transient=True) as progress: progress.add_task("Working...") result: Optional[EvalResult] = self.run(answer=answer) response_timer.stop() if result is None: return None table = Table( box=ROUNDED, border_style="blue", show_header=False, title="[ Evaluation Result ]", title_style="bold sky_blue1", title_justify="center", ) table.add_row("Question", self.question) table.add_row("Answer", self.answer) table.add_row("Expected Answer", self.expected_answer) table.add_row("Accuracy Score", f"{str(result.accuracy_score)}/10") table.add_row("Accuracy Reason", result.accuracy_reason) table.add_row("Time Taken", f"{response_timer.elapsed:.1f}s") console.print(table) return result