"""
agent.py – Claude-smolagents based solution for GAIA challenge
-----------------------------------------------------------
Environment
-----------
ANTHROPIC_API_KEY – API key from Anthropic (set in Hugging Face space secrets)
GAIA_API_URL – (optional) override for the GAIA scoring endpoint
"""
from __future__ import annotations
import base64
import mimetypes
import os
import re
import tempfile
from typing import List, Dict, Any, Optional
import json
import requests
from urllib.parse import urlparse
from smolagents import (
CodeAgent,
DuckDuckGoSearchTool,
PythonInterpreterTool,
LiteLLMModel,
tool,
)
# --------------------------------------------------------------------------- #
# constants & helpers
# --------------------------------------------------------------------------- #
DEFAULT_API_URL = os.getenv(
"GAIA_API_URL", "https://agents-course-unit4-scoring.hf.space"
)
FILE_TAG = re.compile(r"<file:([^>]+)>") # <file:xyz>
def _download_file(file_id: str) -> bytes:
"""Download the attachment for a GAIA task."""
url = f"{DEFAULT_API_URL}/files/{file_id}"
resp = requests.get(url, timeout=30)
resp.raise_for_status()
return resp.content
# --------------------------------------------------------------------------- #
# custom tool: fetch GAIA attachments
# --------------------------------------------------------------------------- #
@tool
def gaia_file_reader(file_id: str) -> str:
"""
Download a GAIA attachment and return its contents.
Args:
file_id: identifier that appears inside a <file:...> placeholder.
Returns:
base64-encoded string for binary files (images, PDFs, …) or decoded
UTF-8 text for textual files.
"""
try:
raw = _download_file(file_id)
mime = mimetypes.guess_type(file_id)[0] or "application/octet-stream"
if mime.startswith("text") or mime in ("application/json",):
return raw.decode(errors="ignore")
return base64.b64encode(raw).decode()
except Exception as exc:
return f"ERROR downloading {file_id}: {exc}"
# --------------------------------------------------------------------------- #
# additional tool functions
# --------------------------------------------------------------------------- #
@tool
def save_and_read_file(content: str, filename: Optional[str] = None) -> str:
"""
Save content to a temporary file and return the path.
Useful for processing files from the GAIA API.
Args:
content: The content to save to the file
filename: Optional filename, will generate a random name if not provided
Returns:
Path to the saved file
"""
temp_dir = tempfile.gettempdir()
if filename is None:
temp_file = tempfile.NamedTemporaryFile(delete=False)
filepath = temp_file.name
else:
filepath = os.path.join(temp_dir, filename)
# Write content to the file
with open(filepath, 'w') as f:
f.write(content)
return f"File saved to {filepath}. You can read this file to process its contents."
@tool
def download_file_from_url(url: str, filename: Optional[str] = None) -> str:
"""
Download a file from a URL and save it to a temporary location.
Args:
url: The URL to download from
filename: Optional filename, will generate one based on URL if not provided
Returns:
Path to the downloaded file
"""
try:
# Parse URL to get filename if not provided
if not filename:
path = urlparse(url).path
filename = os.path.basename(path)
if not filename:
# Generate a random name if we couldn't extract one
import uuid
filename = f"downloaded_{uuid.uuid4().hex[:8]}"
# Create temporary file
temp_dir = tempfile.gettempdir()
filepath = os.path.join(temp_dir, filename)
# Download the file
response = requests.get(url, stream=True)
response.raise_for_status()
# Save the file
with open(filepath, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
return f"File downloaded to {filepath}. You can now process this file."
except Exception as e:
return f"Error downloading file: {str(e)}"
@tool
def extract_text_from_image(image_path: str) -> str:
"""
Extract text from an image using pytesseract (if available).
Args:
image_path: Path to the image file
Returns:
Extracted text or error message
"""
try:
# Try to import pytesseract
import pytesseract
from PIL import Image
# Open the image
image = Image.open(image_path)
# Extract text
text = pytesseract.image_to_string(image)
return f"Extracted text from image:\n\n{text}"
except ImportError:
return "Error: pytesseract is not installed. Please install it with 'pip install pytesseract' and ensure Tesseract OCR is installed on your system."
except Exception as e:
return f"Error extracting text from image: {str(e)}"
@tool
def analyze_csv_file(file_path: str, query: str) -> str:
"""
Analyze a CSV file using pandas and answer a question about it.
Args:
file_path: Path to the CSV file
query: Question about the data
Returns:
Analysis result or error message
"""
try:
import pandas as pd
# Read the CSV file
df = pd.read_csv(file_path)
# Run various analyses based on the query
result = f"CSV file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
result += f"Columns: {', '.join(df.columns)}\n\n"
# Add summary statistics
result += "Summary statistics:\n"
result += str(df.describe())
return result
except ImportError:
return "Error: pandas is not installed. Please install it with 'pip install pandas'."
except Exception as e:
return f"Error analyzing CSV file: {str(e)}"
@tool
def analyze_excel_file(file_path: str, query: str) -> str:
"""
Analyze an Excel file using pandas and answer a question about it.
Args:
file_path: Path to the Excel file
query: Question about the data
Returns:
Analysis result or error message
"""
try:
import pandas as pd
# Read the Excel file
df = pd.read_excel(file_path)
# Run various analyses based on the query
result = f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
result += f"Columns: {', '.join(df.columns)}\n\n"
# Add summary statistics
result += "Summary statistics:\n"
result += str(df.describe())
return result
except ImportError:
return "Error: pandas and openpyxl are not installed. Please install them with 'pip install pandas openpyxl'."
except Exception as e:
return f"Error analyzing Excel file: {str(e)}"
# --------------------------------------------------------------------------- #
# GAIAAgent class
# --------------------------------------------------------------------------- #
class GAIAAgent:
def __init__(
self,
api_key: Optional[str] = None,
temperature: float = 0.1,
verbose: bool = False,
system_prompt: Optional[str] = None
):
"""
Initialize a GAIAAgent with Claude model
Args:
api_key: Anthropic API key (fetched from environment if not provided)
temperature: Temperature for text generation
verbose: Enable verbose logging
system_prompt: Custom system prompt (optional)
"""
# Set verbosity
self.verbose = verbose
self.system_prompt = system_prompt or """You are a concise, highly accurate assistant specialized in solving challenges for the GAIA benchmark.
Unless explicitly required, reply with ONE short sentence.
Your answers should be precise, direct, and exactly match the expected format.
All answers are graded by exact string match, so format carefully!"""
# Get API key
if api_key is None:
api_key = os.getenv("ANTHROPIC_API_KEY")
if not api_key:
raise ValueError("No Anthropic token provided. Please set ANTHROPIC_API_KEY environment variable or pass api_key parameter.")
if self.verbose:
print(f"Using Anthropic token: {api_key[:5]}...")
# Initialize Claude model
self.model = LiteLLMModel(
model_id="anthropic/claude-3-5-sonnet-20240620", # Use Claude 3.5 Sonnet
api_key=api_key,
temperature=temperature
)
if self.verbose:
print(f"Initialized model: LiteLLMModel - anthropic/claude-3-5-sonnet-20240620")
# Initialize default tools
self.tools = [
DuckDuckGoSearchTool(),
PythonInterpreterTool(),
save_and_read_file,
download_file_from_url,
analyze_csv_file,
analyze_excel_file,
gaia_file_reader
]
# Add extract_text_from_image if PIL and pytesseract are available
try:
import pytesseract
from PIL import Image
self.tools.append(extract_text_from_image)
if self.verbose:
print("Added image processing tool")
except ImportError:
if self.verbose:
print("Image processing libraries not available")
if self.verbose:
print(f"Initialized with {len(self.tools)} tools")
# Setup imports allowed
self.imports = ["pandas", "numpy", "datetime", "json", "re", "math", "os", "requests", "csv", "urllib"]
# Initialize the CodeAgent
self.agent = CodeAgent(
tools=self.tools,
model=self.model,
additional_authorized_imports=self.imports,
executor_type="local",
verbosity_level=2 if self.verbose else 0
)
if self.verbose:
print("Agent initialized and ready")
def answer_question(self, question: str, task_file_path: Optional[str] = None) -> str:
"""
Process a GAIA benchmark question and return the answer
Args:
question: The question to answer
task_file_path: Optional path to a file associated with the question
Returns:
The answer to the question
"""
try:
if self.verbose:
print(f"Processing question: {question}")
if task_file_path:
print(f"With associated file: {task_file_path}")
# Create a context with file information if available
context = question
file_content = None
# If there's a file, read it and include its content in the context
if task_file_path:
try:
with open(task_file_path, 'r', errors='ignore') as f:
file_content = f.read()
# Determine file type from extension
import os
file_ext = os.path.splitext(task_file_path)[1].lower()
context = f"""
Question: {question}
This question has an associated file. Here is the file content:
```{file_ext}
{file_content}
```
Analyze the file content above to answer the question.
"""
except Exception as file_e:
try:
# Try to read in binary mode
with open(task_file_path, 'rb') as f:
binary_content = f.read()
# For image files
if file_ext.lower() in ['.jpg', '.jpeg', '.png', '.gif', '.bmp']:
context = f"""
Question: {question}
This question has an associated image file. Please use the extract_text_from_image tool to process it.
File path: {task_file_path}
"""
else:
context = f"""
Question: {question}
This question has an associated file at path: {task_file_path}
This is a binary file. Use appropriate tools to analyze it.
"""
except Exception as binary_e:
context = f"""
Question: {question}
This question has an associated file at path: {task_file_path}
However, there was an error reading the file: {file_e}
You can still try to answer the question based on the information provided.
"""
# Check for special cases that need specific formatting
# Reversed text questions
if question.startswith(".") or ".rewsna eht sa" in question:
context = f"""
This question appears to be in reversed text. Here's the reversed version:
{question[::-1]}
Now answer the question above. Remember to format your answer exactly as requested.
"""
# Add a prompt to ensure precise answers
full_prompt = f"""{context}
When answering, provide ONLY the precise answer requested.
Do not include explanations, steps, reasoning, or additional text.
Be direct and specific. GAIA benchmark requires exact matching answers.
For example, if asked "What is the capital of France?", respond simply with "Paris".
"""
# Run the agent with the question
answer = self.agent.run(full_prompt)
# Clean up the answer to ensure it's in the expected format
# Remove common prefixes that models often add
answer = self._clean_answer(answer)
if self.verbose:
print(f"Generated answer: {answer}")
return answer
except Exception as e:
error_msg = f"Error answering question: {e}"
if self.verbose:
print(error_msg)
return error_msg
def _clean_answer(self, answer: any) -> str:
"""
Clean up the answer to remove common prefixes and formatting
that models often add but that can cause exact match failures.
Args:
answer: The raw answer from the model
Returns:
The cleaned answer as a string
"""
# Convert non-string types to strings
if not isinstance(answer, str):
# Handle numeric types (float, int)
if isinstance(answer, float):
# Format floating point numbers properly
# Check if it's an integer value in float form (e.g., 12.0)
if answer.is_integer():
formatted_answer = str(int(answer))
else:
# For currency values that might need formatting
if abs(answer) >= 1000:
formatted_answer = f"${answer:,.2f}"
else:
formatted_answer = str(answer)
return formatted_answer
elif isinstance(answer, int):
return str(answer)
else:
# For any other type
return str(answer)
# Now we know answer is a string, so we can safely use string methods
# Normalize whitespace
answer = answer.strip()
# Remove common prefixes and formatting that models add
prefixes_to_remove = [
"The answer is ",
"Answer: ",
"Final answer: ",
"The result is ",
"To answer this question: ",
"Based on the information provided, ",
"According to the information: ",
]
for prefix in prefixes_to_remove:
if answer.startswith(prefix):
answer = answer[len(prefix):].strip()
# Remove quotes if they wrap the entire answer
if (answer.startswith('"') and answer.endswith('"')) or (answer.startswith("'") and answer.endswith("'")):
answer = answer[1:-1].strip()
return answer
# --------------------------------------------------------------------------- #
# GeminiAgent class - Wrapper around GAIAAgent
# --------------------------------------------------------------------------- #
class ClaudeAgent:
"""Claude-enhanced agent for GAIA challenge"""
def __init__(self):
# Try to initialize GAIAAgent with Claude
try:
# Get API key
api_key = os.getenv("ANTHROPIC_API_KEY")
if not api_key:
raise ValueError("ANTHROPIC_API_KEY environment variable not found")
print("✅ Initializing GAIAAgent with Claude")
# Create GAIAAgent instance
self.agent = GAIAAgent(
api_key=api_key,
temperature=0.1, # Use low temperature for precise answers
verbose=True, # Enable verbose logging
)
except Exception as e:
print(f"Error initializing GAIAAgent: {e}")
raise
def __call__(self, question: str) -> str:
"""
Process a GAIA question and return the answer
Args:
question: The question to answer
Returns:
The answer to the question
"""
try:
print(f"Received question: {question[:100]}..." if len(question) > 100 else f"Received question: {question}")
# Detect reversed text
if question.startswith(".") or ".rewsna eht sa" in question:
print("Detected reversed text question")
# GAIAAgent handles reversed text internally
# Detect if there's a file
file_match = re.search(r"<file:([^>]+)>", question)
if file_match:
file_id = file_match.group(1)
print(f"Detected file reference: {file_id}")
# Download the file
try:
file_content = _download_file(file_id)
# Create temporary file for the file
temp_dir = tempfile.gettempdir()
file_path = os.path.join(temp_dir, file_id)
# Save file content
with open(file_path, 'wb') as f:
f.write(file_content)
print(f"File downloaded to: {file_path}")
# Remove file tag from question
clean_question = re.sub(r"<file:[^>]+>", "", question).strip()
# Process question with file path
answer = self.agent.answer_question(clean_question, file_path)
return self._clean_answer(answer)
except Exception as e:
print(f"Error processing file: {e}")
# Fall back to processing without file
# Process standard question
answer = self.agent.answer_question(question)
return self._clean_answer(answer)
except Exception as e:
print(f"Error processing question: {e}")
error_msg = f"Unable to process question: {str(e)}"
return error_msg
def _clean_answer(self, answer: str) -> str:
"""
Final cleanup of answer to ensure correct format
Reuses GAIAAgent's cleaning method
"""
# Already cleaned in GAIAAgent, but do additional checks
if isinstance(answer, str):
# Remove any trailing periods and whitespace
answer = answer.rstrip(". \t\n\r")
# Ensure it's not too long an answer - GAIA usually needs concise responses
if len(answer) > 1000:
# Try to find the first sentence or statement of the answer
sentences = answer.split('. ')
if len(sentences) > 1:
return sentences[0].strip()
return answer