File size: 3,495 Bytes
332e48b
5fffd11
6a05ca9
6acc56a
08aa3fd
 
eb7cc40
332e48b
 
 
 
5fffd11
8dcca97
08aa3fd
ddbce07
08aa3fd
 
6acc56a
08aa3fd
 
 
6acc56a
08aa3fd
6a05ca9
ddbce07
 
 
 
 
 
 
 
08aa3fd
ddbce07
6acc56a
ddbce07
 
08aa3fd
ddbce07
08aa3fd
 
 
ddbce07
08aa3fd
 
 
 
ddbce07
 
8dcca97
ddbce07
6a05ca9
ddbce07
 
 
 
6a05ca9
ddbce07
08aa3fd
 
ddbce07
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
81
82
83
84
85
86
import os
import re
import base64
import io
import requests
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("\n", "").replace(".", "").replace("Final Answer:", "").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 factual assistant. Reason step-by-step and return only the final answer."},
                {"role": "user", "content": prompt + "\nFinal Answer:"}
            ],
            temperature=0.0,
        )
        return res.choices[0].message.content.strip()

    def q_chess_image(self, image_bytes):
        b64 = base64.b64encode(image_bytes).decode()
        messages = [
            {"role": "system", "content": "You are a chess expert."},
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": "Analyze the chessboard image. Black to move. Return only the best move in algebraic notation."},
                    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
                ]
            }
        ]
        res = self.client.chat.completions.create(model="gpt-4o", messages=messages)
        return res.choices[0].message.content.strip()

    def q_excel_total_sales(self, file):
        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 __call__(self, question: str, task_id: str = None) -> str:
        # image support
        if task_id == "cca530fc-4052-43b2-b130-b30968d8aa44":
            file, _ = self.fetch_file(task_id)
            if isinstance(file, bytes):
                return self.clean(self.q_chess_image(file))

        # excel support
        if task_id == "7bd855d8-463d-4ed5-93ca-5fe35145f733":
            file, _ = self.fetch_file(task_id)
            if isinstance(file, bytes):
                return self.clean(self.q_excel_total_sales(file))

        # text fallback
        prompt = f"Question: {question}\nIf needed, reason through data, code, or information."
        if task_id:
            file_data, content_type = self.fetch_file(task_id)
            if isinstance(file_data, bytes):
                try:
                    if content_type and "text" in content_type:
                        prompt = f"File Content:\n{file_data.decode('utf-8')[:3000]}\n\n{prompt}"
                    elif content_type and ("audio" in content_type or "mp3" in content_type):
                        prompt = f"This task involves an audio file. Transcribe it and extract only what is asked.\n\n{prompt}"
                except Exception:
                    pass

        return self.clean(self.ask(prompt))