| import os | |
| import re | |
| import io | |
| import base64 | |
| import requests | |
| import pandas as pd | |
| from word2number import w2n | |
| from openai import OpenAI | |
| from langchain_community.tools import DuckDuckGoSearchRun | |
| class GaiaAgent: | |
| def __init__(self): | |
| self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) | |
| self.api_url = "https://agents-course-unit4-scoring.hf.space" | |
| self.search_tool = DuckDuckGoSearchRun() | |
| def fetch_file(self, task_id): | |
| try: | |
| url = f"{self.api_url}/files/{task_id}" | |
| r = requests.get(url, timeout=10) | |
| r.raise_for_status() | |
| return r.content, r.headers.get("Content-Type", "") | |
| except: | |
| return None, None | |
| def ask(self, prompt): | |
| try: | |
| r = self.client.chat.completions.create( | |
| model="gpt-4-turbo", | |
| messages=[{"role": "user", "content": prompt}], | |
| temperature=0, | |
| timeout=30 | |
| ) | |
| return r.choices[0].message.content.strip() | |
| except: | |
| return "[ERROR: ask failed]" | |
| def search_context(self, query): | |
| try: | |
| result = self.search_tool.run(query) | |
| return result[:2000] if result else "[NO WEB RESULT]" | |
| except: | |
| return "[WEB ERROR]" | |
| def handle_file(self, content, ctype, question): | |
| try: | |
| if "image" in ctype: | |
| b64 = base64.b64encode(content).decode("utf-8") | |
| result = self.client.chat.completions.create( | |
| model="gpt-4o", | |
| messages=[ | |
| {"role": "system", "content": "You're a chess assistant. Reply only with the best move in algebraic notation (e.g., Qd1#)."}, | |
| {"role": "user", "content": [ | |
| {"type": "text", "text": question}, | |
| {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}} | |
| ]} | |
| ] | |
| ) | |
| return result.choices[0].message.content.strip() | |
| if "audio" in ctype: | |
| with open("/tmp/audio.mp3", "wb") as f: | |
| f.write(content) | |
| result = self.client.audio.transcriptions.create(model="whisper-1", file=open("/tmp/audio.mp3", "rb")) | |
| return result.text | |
| if "excel" in ctype: | |
| df = pd.read_excel(io.BytesIO(content), engine="openpyxl") | |
| df.columns = [c.lower().strip() for c in df.columns] | |
| if 'category' in df.columns and 'sales' in df.columns: | |
| df = df.dropna(subset=['category', 'sales']) | |
| df = df[df['category'].str.lower() == 'food'] | |
| df['sales'] = pd.to_numeric(df['sales'], errors='coerce') | |
| return f"${df['sales'].sum():.2f}" | |
| return "[NO FOOD SALES DATA]" | |
| return content.decode("utf-8", errors="ignore")[:3000] | |
| except Exception as e: | |