File size: 3,112 Bytes
332e48b 2693f75 eb7cc40 332e48b d48b3cc 332e48b d48b3cc 332e48b eb7cc40 ffdfd85 5c63a78 d48b3cc 5c63a78 d48b3cc 5c63a78 ffdfd85 eb7cc40 2693f75 5c63a78 ffdfd85 5c63a78 ffdfd85 2693f75 d48b3cc 5c63a78 ffdfd85 eb7cc40 2693f75 eb7cc40 d48b3cc eb7cc40 d48b3cc ffdfd85 eb7cc40 ffdfd85 d48b3cc eb7cc40 ffdfd85 332e48b d48b3cc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 |
import os
import io
import pandas as pd
import requests
from openai import OpenAI
SKIPPED_TASKS = {
# Tasks requiring video, image, or audio
"a1e91b78-d3d8-4675-bb8d-62741b4b68a6", # YouTube birds
"cca530fc-4052-43b2-b130-b30968d8aa44", # Chess image
"9d191bce-651d-4746-be2d-7ef8ecadb9c2", # Teal'c audio
"99c9cc74-fdc8-46c6-8f8d-3ce2d3bfeea3", # Strawberry pie.mp3
"1f975693-876d-457b-a649-393859e79bf3" # Homework.mp3
}
class GaiaAgent:
def __init__(self):
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.instructions = (
"You are a precise and logical assistant solving GAIA benchmark questions. "
"Use any context or data provided. Respond with only the final answer."
)
self.api_url = "https://agents-course-unit4-scoring.hf.space"
def analyze_csv(self, csv_text: str, question: str) -> str:
try:
df = pd.read_csv(io.StringIO(csv_text))
q = question.lower()
if "total" in q and "food" in q and "not including drinks" in q:
food_items = df[df["category"].str.lower() == "food"]
return f"Total food sales: ${food_items["sales"].sum():.2f}"
return f"Sample row: {df.iloc[0].to_dict()}"
except Exception as e:
return f"[CSV parse failed: {e}]"
def fetch_file_context(self, task_id: str, question: str) -> str:
try:
url = f"{self.api_url}/files/{task_id}"
response = requests.get(url, timeout=10)
response.raise_for_status()
content_type = response.headers.get("Content-Type", "")
if "csv" in content_type or url.endswith(".csv"):
return self.analyze_csv(response.text, question)
elif "json" in content_type:
return f"JSON Preview: {response.text[:1000]}"
elif "text/plain" in content_type:
return f"Text Preview: {response.text[:1000]}"
elif "pdf" in content_type:
return "[PDF detected. OCR not supported.]"
else:
return f"[Unsupported file type: {content_type}]"
except Exception as e:
return f"[File error: {e}]"
def __call__(self, question: str, task_id: str = None) -> str:
if task_id in SKIPPED_TASKS:
return "SKIPPED"
file_fact = ""
if task_id:
file_fact = self.fetch_file_context(task_id, question)
file_fact = f"FILE CONTEXT:\n{file_fact}\n"
prompt = f"{self.instructions}\n\n{file_fact}QUESTION: {question}\nANSWER:"
try:
response = self.client.chat.completions.create(
model="gpt-4-turbo",
messages=[
{"role": "system", "content": self.instructions},
{"role": "user", "content": prompt}
],
temperature=0.0,
)
return response.choices[0].message.content.strip()
except Exception as e:
return f"[Agent error: {e}]"
|