#!/usr/bin/env python3 """ Final Answer Tool for GAIA Agent System Extracts precise, EXACT MATCH compliant answers from agent results """ import re import logging from typing import Dict, Any, Optional from models.qwen_client import QwenClient, ModelTier logger = logging.getLogger(__name__) class FinalAnswerTool: """ Tool for extracting precise, GAIA-compliant final answers Ensures EXACT MATCH compatibility for Unit 4 API submission """ def __init__(self, llm_client: QwenClient): self.llm_client = llm_client def extract_final_answer(self, question: str, agent_results: str, question_type: str = "") -> Dict[str, Any]: """ Extract the precise final answer in GAIA-compliant format Args: question: The original GAIA question agent_results: Combined results from multiple agents question_type: Type of question (for specialized extraction) Returns: Dict with extracted answer, confidence, and reasoning """ try: logger.info("🎯 Extracting GAIA-compliant final answer") # Create specialized extraction prompt extraction_prompt = self._create_extraction_prompt(question, agent_results, question_type) # Use 72B model for precise extraction result = self.llm_client.generate( extraction_prompt, tier=ModelTier.COMPLEX, # 72B model max_tokens=50 # Force concise answers ) if not result.success: logger.error("Final answer extraction failed") return { "answer": "Processing failed", "confidence": 0.0, "reasoning": f"Extraction failed: {result.response}" } # Parse and clean the extracted answer extracted_answer = self._clean_answer(result.response, question, question_type) # Validate answer format validation_result = self._validate_answer(extracted_answer, question_type) logger.info(f"✅ Final answer extracted: '{extracted_answer}'") return { "answer": extracted_answer, "confidence": validation_result["confidence"], "reasoning": f"Extracted using 72B model. Validation: {validation_result['status']}" } except Exception as e: error_msg = f"Final answer extraction error: {str(e)}" logger.error(error_msg) return { "answer": "Extraction error", "confidence": 0.0, "reasoning": error_msg } def _create_extraction_prompt(self, question: str, agent_results: str, question_type: str) -> str: """Create specialized extraction prompt based on question type""" base_prompt = f""" CRITICAL: This is for GAIA benchmark evaluation using EXACT MATCH comparison. Your response must be ONLY the precise answer - no explanations, no "FINAL ANSWER:", no extra text. Question: {question} Agent Analysis Results: {agent_results} EXTRACTION RULES: """ # Add type-specific rules if "mathematical" in question_type.lower() or any(word in question.lower() for word in ["how many", "count", "number", "albums"]): base_prompt += """ - If asking for a count/number: respond with ONLY the number (e.g., "5", "23", "0") - If asking for calculation: respond with ONLY the result (e.g., "42", "3.14", "100") - No units unless specifically requested in the question """ elif "text_manipulation" in question_type.lower() or "reverse" in question.lower(): base_prompt += """ - If text is reversed: provide the corrected text - If asking for opposite: provide ONLY the opposite word (e.g., "right" for opposite of "left") - If asking to decode: provide ONLY the decoded answer """ elif "yes" in question.lower() or "true" in question.lower() or "false" in question.lower(): base_prompt += """ - If yes/no question: respond with ONLY "yes" or "no" (lowercase) - If true/false question: respond with ONLY "true" or "false" (lowercase) """ elif any(word in question.lower() for word in ["name", "who", "which person"]): base_prompt += """ - If asking for a name: provide ONLY the name (e.g., "John Smith", "Einstein") - If asking for first name only: provide ONLY first name (e.g., "John") - If asking for last name only: provide ONLY last name (e.g., "Smith") """ elif any(word in question.lower() for word in ["where", "location", "city", "country"]): base_prompt += """ - If asking for location: provide ONLY the location name (e.g., "Paris", "USA", "New York") - No additional descriptors unless specifically requested """ else: base_prompt += """ - Provide ONLY the direct answer to the question - No explanations, context, or additional information - Be as concise as possible while being accurate """ base_prompt += """ EXAMPLES OF CORRECT FORMAT: - Question: "How many albums?" → Answer: "5" - Question: "What is the opposite of left?" → Answer: "right" - Question: "True or false?" → Answer: "true" - Question: "Who discovered X?" → Answer: "Einstein" - Question: "Which city?" → Answer: "London" Extract the precise answer NOW:""" return base_prompt def _clean_answer(self, raw_answer: str, question: str, question_type: str) -> str: """Clean and format the extracted answer""" # Remove common unwanted prefixes/suffixes answer = raw_answer.strip() # Remove common prefixes prefixes_to_remove = [ "the answer is", "answer:", "final answer:", "result:", "response:", "conclusion:", "based on", "according to", "from the", ] for prefix in prefixes_to_remove: if answer.lower().startswith(prefix): answer = answer[len(prefix):].strip() # Remove quotes if they wrap the entire answer if answer.startswith('"') and answer.endswith('"'): answer = answer[1:-1] if answer.startswith("'") and answer.endswith("'"): answer = answer[1:-1] # AGGRESSIVE LENGTH ENFORCEMENT FOR GAIA # If answer is too long, extract the core information if len(answer) > 50: # For different question types, extract differently if "mathematical" in question_type.lower() or any(word in question.lower() for word in ["how many", "count", "number", "albums"]): # Extract just the number for mathematical questions number_match = re.search(r'-?\d+(?:\.\d+)?', answer) if number_match: answer = number_match.group() elif "name" in question_type.lower() or any(word in question.lower() for word in ["who", "name"]): # Extract just the name (first few words) words = answer.split() if len(words) > 3: answer = ' '.join(words[:3]) # Keep only first 3 words for names elif "location" in question_type.lower() or any(word in question.lower() for word in ["where", "city", "country"]): # Extract just the location name words = answer.split() if len(words) > 2: answer = ' '.join(words[:2]) # Keep only first 2 words for locations elif "yes_no" in question_type.lower() or any(word in answer.lower() for word in ["yes", "no", "true", "false"]): # Extract yes/no/true/false if any(word in answer.lower() for word in ["yes", "no", "true", "false"]): for word in answer.lower().split(): if word in ["yes", "no", "true", "false"]: answer = word break else: # For other types, take first sentence or clause sentences = re.split(r'[.!?]', answer) if sentences: answer = sentences[0].strip() # If still too long, take first clause if len(answer) > 30: clauses = re.split(r'[,;:]', answer) if clauses: answer = clauses[0].strip() # Handle specific formatting based on question type if "text_manipulation" in question_type.lower(): # For reversed text questions, ensure clean output if len(answer.split()) == 1: # Single word answer answer = answer.lower() # Final aggressive truncation if still too long if len(answer) > 40: # Split into words and take as many as fit words = answer.split() truncated_words = [] current_length = 0 for word in words: if current_length + len(word) + 1 <= 40: truncated_words.append(word) current_length += len(word) + 1 else: break if truncated_words: answer = ' '.join(truncated_words) else: # Last resort - take first 40 characters answer = answer[:40].strip() # Remove any trailing punctuation that's not part of the answer answer = answer.rstrip('.,!?;:') return answer.strip() def _validate_answer(self, answer: str, question_type: str) -> Dict[str, Any]: """Validate the extracted answer format""" if not answer: return {"status": "empty_answer", "confidence": 0.0} # Check length - GAIA answers should be concise if len(answer) > 100: return {"status": "too_long", "confidence": 0.3} # Type-specific validation if "mathematical" in question_type.lower(): if re.match(r'^-?\d+(?:\.\d+)?$', answer): return {"status": "valid_number", "confidence": 0.9} else: return {"status": "invalid_number_format", "confidence": 0.5} elif "yes_no" in question_type.lower(): if answer.lower() in ["yes", "no", "true", "false"]: return {"status": "valid_boolean", "confidence": 0.9} else: return {"status": "invalid_boolean_format", "confidence": 0.4} # General validation - prefer short, direct answers if len(answer) <= 20: return {"status": "concise_answer", "confidence": 0.8} elif len(answer) <= 50: return {"status": "moderate_length", "confidence": 0.6} else: return {"status": "long_answer", "confidence": 0.4}