|
|
""" |
|
|
Example GAIA Agent Implementation |
|
|
|
|
|
This file shows how to implement a basic agent with external API integration. |
|
|
You can use this as a starting point and customize it for your needs. |
|
|
""" |
|
|
|
|
|
import requests |
|
|
import json |
|
|
import re |
|
|
from typing import Dict, List, Any |
|
|
from agent_implementation import BaseGAIAAgent |
|
|
|
|
|
class ExampleGAIAAgent(BaseGAIAAgent): |
|
|
""" |
|
|
Example agent implementation with basic reasoning and external API support. |
|
|
|
|
|
This agent demonstrates: |
|
|
1. Basic question analysis |
|
|
2. Mathematical calculations |
|
|
3. Simple reasoning patterns |
|
|
4. Error handling |
|
|
""" |
|
|
|
|
|
def __init__(self): |
|
|
super().__init__() |
|
|
self.calculation_patterns = [ |
|
|
r'(\d+)\s*\+\s*(\d+)', |
|
|
r'(\d+)\s*-\s*(\d+)', |
|
|
r'(\d+)\s*\*\s*(\d+)', |
|
|
r'(\d+)\s*/\s*(\d+)', |
|
|
r'(\d+)\s*%\s*(\d+)' |
|
|
] |
|
|
|
|
|
def generate_answer(self, question: Dict) -> str: |
|
|
""" |
|
|
Generate an answer for a given question. |
|
|
|
|
|
This implementation includes: |
|
|
- Question type detection |
|
|
- Mathematical calculations |
|
|
- Basic reasoning |
|
|
- File content processing |
|
|
""" |
|
|
task_id = question.get("task_id", "") |
|
|
question_text = question.get("question", "") |
|
|
|
|
|
|
|
|
file_content = self.download_file(task_id) |
|
|
|
|
|
|
|
|
question_type = self.analyze_question_type(question_text) |
|
|
|
|
|
|
|
|
if question_type == "calculation": |
|
|
answer = self.handle_calculation(question_text) |
|
|
elif question_type == "temporal": |
|
|
answer = self.handle_temporal(question_text) |
|
|
elif question_type == "factual": |
|
|
answer = self.handle_factual(question_text, file_content) |
|
|
else: |
|
|
answer = self.handle_general(question_text, file_content) |
|
|
|
|
|
return answer |
|
|
|
|
|
def analyze_question_type(self, question: str) -> str: |
|
|
"""Determine the type of question based on keywords and patterns.""" |
|
|
question_lower = question.lower() |
|
|
|
|
|
|
|
|
if any(op in question_lower for op in ["calculate", "compute", "sum", "total", "percentage", "percent"]): |
|
|
return "calculation" |
|
|
|
|
|
|
|
|
if any(word in question_lower for word in ["when", "date", "time", "year", "month", "day"]): |
|
|
return "temporal" |
|
|
|
|
|
|
|
|
if any(word in question_lower for word in ["what", "which", "who", "where"]): |
|
|
return "factual" |
|
|
|
|
|
return "general" |
|
|
|
|
|
def handle_calculation(self, question: str) -> str: |
|
|
"""Handle mathematical calculation questions.""" |
|
|
|
|
|
for pattern in self.calculation_patterns: |
|
|
match = re.search(pattern, question) |
|
|
if match: |
|
|
num1 = float(match.group(1)) |
|
|
num2 = float(match.group(2)) |
|
|
|
|
|
if '+' in pattern: |
|
|
result = num1 + num2 |
|
|
operation = "addition" |
|
|
elif '-' in pattern: |
|
|
result = num1 - num2 |
|
|
operation = "subtraction" |
|
|
elif '*' in pattern: |
|
|
result = num1 * num2 |
|
|
operation = "multiplication" |
|
|
elif '/' in pattern: |
|
|
if num2 == 0: |
|
|
return "Error: Division by zero" |
|
|
result = num1 / num2 |
|
|
operation = "division" |
|
|
elif '%' in pattern: |
|
|
result = num1 % num2 |
|
|
operation = "modulo" |
|
|
|
|
|
return f"The result of {operation} {num1} and {num2} is {result}" |
|
|
|
|
|
|
|
|
percentage_match = re.search(r'(\d+)%\s*of\s*(\d+)', question) |
|
|
if percentage_match: |
|
|
percentage = float(percentage_match.group(1)) |
|
|
value = float(percentage_match.group(2)) |
|
|
result = (percentage / 100) * value |
|
|
return f"{percentage}% of {value} is {result}" |
|
|
|
|
|
return "I cannot perform the requested calculation with the given information." |
|
|
|
|
|
def handle_temporal(self, question: str) -> str: |
|
|
"""Handle temporal questions.""" |
|
|
|
|
|
date_patterns = [ |
|
|
r'(\d{4})-(\d{2})-(\d{2})', |
|
|
r'(\d{1,2})/(\d{1,2})/(\d{4})', |
|
|
r'(\w+)\s+(\d{1,2}),?\s+(\d{4})' |
|
|
] |
|
|
|
|
|
for pattern in date_patterns: |
|
|
match = re.search(pattern, question) |
|
|
if match: |
|
|
if pattern == r'(\d{4})-(\d{2})-(\d{2})': |
|
|
year, month, day = match.groups() |
|
|
return f"The date is {year}-{month}-{day}" |
|
|
elif pattern == r'(\d{1,2})/(\d{1,2})/(\d{4})': |
|
|
month, day, year = match.groups() |
|
|
return f"The date is {month}/{day}/{year}" |
|
|
elif pattern == r'(\w+)\s+(\d{1,2}),?\s+(\d{4})': |
|
|
month, day, year = match.groups() |
|
|
return f"The date is {month} {day}, {year}" |
|
|
|
|
|
return "I cannot determine the specific date or time from the question." |
|
|
|
|
|
def handle_factual(self, question: str, file_content: str) -> str: |
|
|
"""Handle factual questions.""" |
|
|
|
|
|
if file_content: |
|
|
|
|
|
question_keywords = self.extract_keywords(question) |
|
|
file_lines = file_content.split('\n') |
|
|
|
|
|
for line in file_lines: |
|
|
if any(keyword.lower() in line.lower() for keyword in question_keywords): |
|
|
return f"Based on the file content: {line.strip()}" |
|
|
|
|
|
|
|
|
return self.generate_structured_answer(question) |
|
|
|
|
|
def handle_general(self, question: str, file_content: str) -> str: |
|
|
"""Handle general questions.""" |
|
|
|
|
|
if file_content: |
|
|
return f"Based on the provided information: {file_content[:200]}..." |
|
|
|
|
|
return self.generate_structured_answer(question) |
|
|
|
|
|
def extract_keywords(self, text: str) -> List[str]: |
|
|
"""Extract important keywords from text.""" |
|
|
|
|
|
stop_words = {"the", "a", "an", "and", "or", "but", "in", "on", "at", "to", "for", "of", "with", "by", "is", "are", "was", "were"} |
|
|
words = re.findall(r'\b\w+\b', text.lower()) |
|
|
keywords = [word for word in words if word not in stop_words and len(word) > 2] |
|
|
return keywords |
|
|
|
|
|
def generate_structured_answer(self, question: str) -> str: |
|
|
"""Generate a structured answer for general questions.""" |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
answer_parts = [ |
|
|
"Based on my analysis of the question:", |
|
|
f"- Question type: {self.analyze_question_type(question)}", |
|
|
"- Key information needed: [identify what's needed]", |
|
|
"- Reasoning approach: [describe your method]", |
|
|
"- Answer: [provide your answer]" |
|
|
] |
|
|
|
|
|
return "\n".join(answer_parts) |
|
|
|
|
|
def enhance_with_external_api(self, question: str) -> str: |
|
|
""" |
|
|
Example of how to integrate with external APIs. |
|
|
Uncomment and customize this method if you want to use external services. |
|
|
""" |
|
|
|
|
|
""" |
|
|
import openai |
|
|
|
|
|
try: |
|
|
response = openai.ChatCompletion.create( |
|
|
model="gpt-4", |
|
|
messages=[ |
|
|
{"role": "system", "content": "You are a helpful assistant that answers questions accurately and concisely."}, |
|
|
{"role": "user", "content": question} |
|
|
], |
|
|
temperature=0.1 |
|
|
) |
|
|
return response.choices[0].message.content |
|
|
except Exception as e: |
|
|
return f"Error calling external API: {e}" |
|
|
""" |
|
|
|
|
|
|
|
|
return "External API integration not configured. Please implement your own API calls." |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
|
|
|
agent = ExampleGAIAAgent() |
|
|
|
|
|
|
|
|
test_questions = [ |
|
|
{ |
|
|
"task_id": "test_001", |
|
|
"question": "What is 15 + 27?" |
|
|
}, |
|
|
{ |
|
|
"task_id": "test_002", |
|
|
"question": "What is 25% of 80?" |
|
|
}, |
|
|
{ |
|
|
"task_id": "test_003", |
|
|
"question": "What is the date 2024-03-15?" |
|
|
}, |
|
|
{ |
|
|
"task_id": "test_004", |
|
|
"question": "What is the capital of France?" |
|
|
} |
|
|
] |
|
|
|
|
|
|
|
|
print("Testing Example GAIA Agent:") |
|
|
print("=" * 50) |
|
|
|
|
|
for i, question in enumerate(test_questions, 1): |
|
|
print(f"\nTest {i}:") |
|
|
print(f"Question: {question['question']}") |
|
|
answer = agent.generate_answer(question) |
|
|
print(f"Answer: {answer}") |
|
|
print("-" * 30) |