File size: 3,053 Bytes
332e48b 5fffd11 1bf6d60 6a05ca9 6acc56a 08aa3fd eb7cc40 332e48b 5fffd11 8dcca97 08aa3fd 1bf6d60 08aa3fd 6acc56a 08aa3fd 6acc56a 08aa3fd 6a05ca9 0e46560 ddbce07 0e46560 ddbce07 0e46560 ddbce07 08aa3fd 1bf6d60 8dcca97 0e46560 6a05ca9 ddbce07 6a05ca9 ddbce07 08aa3fd 1bf6d60 0e46560 1bf6d60 08aa3fd 0e46560 ddbce07 0e46560 ddbce07 0e46560 1bf6d60 ddbce07 0e46560 1bf6d60 |
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 re
import requests
import base64
import io
import pandas as pd
from openai import OpenAI
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"
def clean(self, text):
return text.strip().replace("Final Answer:", "").replace("\n", "").replace(".", "").strip()
def fetch_file(self, task_id):
try:
r = requests.get(f"{self.api_url}/files/{task_id}", timeout=10)
r.raise_for_status()
return r.content, r.headers.get("Content-Type", "")
except Exception as e:
return None, f"[Fetch error: {e}]"
def ask(self, prompt: str, model="gpt-4-turbo") -> str:
res = self.client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "You are a precise assistant. Think step by step and return only the exact answer."},
{"role": "user", "content": prompt + "\n\nReturn only the final answer. Do not explain. Format it exactly as expected."}
],
temperature=0.0,
)
return self.clean(res.choices[0].message.content)
def q_excel_sales(self, file: bytes, question: str) -> str:
try:
df = pd.read_excel(io.BytesIO(file), engine="openpyxl")
food = df[df['category'].str.lower() == 'food']
total = food['sales'].sum()
return f"${total:.2f}"
except Exception as e:
return f"[Excel error: {e}]"
def q_audio_transcribe(self, file: bytes, question: str) -> str:
audio_path = "/tmp/audio.mp3"
with open(audio_path, "wb") as f:
f.write(file)
transcript = self.client.audio.transcriptions.create(
model="whisper-1",
file=open(audio_path, "rb")
)
content = transcript.text[:3000]
prompt = f"Transcript: {content}\n\nQuestion: {question}"
return self.ask(prompt)
def __call__(self, question: str, task_id: str = None) -> str:
# File-based branching
if task_id:
file, content_type = self.fetch_file(task_id)
if task_id == "7bd855d8-463d-4ed5-93ca-5fe35145f733" and isinstance(file, bytes):
return self.q_excel_sales(file, question)
if task_id in [
"99c9cc74-fdc8-46c6-8f8d-3ce2d3bfeea3",
"1f975693-876d-457b-a649-393859e79bf3"
] and isinstance(file, bytes):
return self.q_audio_transcribe(file, question)
if isinstance(file, bytes) and content_type and "text" in content_type:
try:
text = file.decode("utf-8", errors="ignore")[:3000]
prompt = f"Document:\n{text}\n\nQuestion: {question}"
return self.ask(prompt)
except:
pass
# Fallback
return self.ask(f"Question: {question}") |