Complete GAIA agent implementation with local LLM

agent/local_llm.py

@@ -0,0 +1,32 @@
+from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
+import torch
+
+class LocalLLM:
+    def __init__(self):
+        self.model_name = "HuggingFaceH4/zephyr-7b-beta"
+        self.device = "cuda" if torch.cuda.is_available() else "cpu"
+        self.pipeline = self._load_model()
+    
+    def _load_model(self):
+        tokenizer = AutoTokenizer.from_pretrained(self.model_name)
+        model = AutoModelForCausalLM.from_pretrained(
+            self.model_name,
+            torch_dtype=torch.float16,
+            device_map="auto",
+            load_in_4bit=True  # Quantization to save memory
+        )
+        return pipeline(
+            "text-generation",
+            model=model,
+            tokenizer=tokenizer,
+            device=self.device
+        )
+    
+    def generate(self, prompt: str) -> str:
+        outputs = self.pipeline(
+            prompt,
+            max_new_tokens=256,
+            do_sample=True,
+            temperature=0.7
+        )
+        return outputs[0]['generated_text']

agent/tools.py

@@ -0,0 +1,17 @@
+from llama_index.core.tools import FunctionTool
+from utils.gaia_api import GaiaAPI
+
+def get_gaia_questions() -> str:
+    """Fetch all GAIA benchmark questions"""
+    questions = GaiaAPI.get_questions()
+    return "\n".join([f"{q['task_id']}: {q['question']}" for q in questions])
+
+def get_random_question() -> str:
+    """Get a single random GAIA question"""
+    question = GaiaAPI.get_random_question()
+    return f"{question['task_id']}: {question['question']}"
+
+gaia_tools = [
+    FunctionTool.from_defaults(fn=get_gaia_questions),
+    FunctionTool.from_defaults(fn=get_random_question)
+]

app.py

@@ -0,0 +1,55 @@
+import gradio as gr
+from agent.local_llm import LocalLLM
+from agent.tools import gaia_tools
+from llama_index.core.agent import ReActAgent
+from utils.gaia_api import GaiaAPI
+
+# Initialize components
+llm = LocalLLM()
+agent = ReActAgent.from_tools(gaia_tools, llm=llm.pipeline)
+
+def process_question(question_text: str) -> str:
+    """Process GAIA question through agent"""
+    try:
+        response = agent.query(question_text)
+        return str(response)
+    except Exception as e:
+        return f"Error: {str(e)}"
+
+def submit_to_gaia(username: str, code_url: str) -> str:
+    """Submit all answers to GAIA"""
+    try:
+        questions = GaiaAPI.get_questions()
+        answers = []
+        for q in questions:
+            answer = process_question(q['question'])
+            answers.append({
+                "task_id": q['task_id'],
+                "submitted_answer": answer
+            })
+        result = GaiaAPI.submit_answers(username, code_url, answers)
+        return f"Submitted! Score: {result.get('score', 'N/A')}"
+    except Exception as e:
+        return f"Submission failed: {str(e)}"
+
+with gr.Blocks() as demo:
+    gr.Markdown("# GAIA Benchmark Agent")
+    
+    with gr.Tab("Question Processing"):
+        question_input = gr.Textbox(label="Enter GAIA Question")
+        answer_output = gr.Textbox(label="Agent Answer")
+        process_btn = gr.Button("Process Question")
+        process_btn.click(process_question, inputs=question_input, outputs=answer_output)
+    
+    with gr.Tab("GAIA Submission"):
+        username_input = gr.Textbox(label="HF Username")
+        code_url_input = gr.Textbox(label="Space Code URL")
+        submit_btn = gr.Button("Submit to GAIA")
+        submission_output = gr.Textbox(label="Submission Result")
+        submit_btn.click(
+            submit_to_gaia,
+            inputs=[username_input, code_url_input],
+            outputs=submission_output
+        )
+
+demo.launch()

packages.txt

@@ -0,0 +1,2 @@
+libgl1
+libglib2.0-0

requirements.txt

@@ -0,0 +1,9 @@
+llama-index==0.10.0
+transformers==4.34.0
+torch==2.0.1
+accelerate==0.23.0
+sentence-transformers==2.2.2
+python-dotenv==1.0.0
+gradio==3.41.0
+requests==2.31.0
+bitsandbytes==0.41.1

utils/gaia_api.py

@@ -0,0 +1,31 @@
+import requests
+from typing import List, Dict, Optional
+
+class GaiaAPI:
+    BASE_URL = "https://https://agents-course-unit4-scoring.hf.space/docs"  # Actual GAIA API URL
+    
+    @classmethod
+    def get_questions(cls) -> List[Dict]:
+        """Fetch all questions from GAIA"""
+        response = requests.get(f"{cls.BASE_URL}/questions")
+        response.raise_for_status()
+        return response.json()
+    
+    @classmethod
+    def get_random_question(cls) -> Dict:
+        """Get single random question"""
+        response = requests.get(f"{cls.BASE_URL}/random-question")
+        response.raise_for_status()
+        return response.json()
+    
+    @classmethod
+    def submit_answers(cls, username: str, code_url: str, answers: List[Dict]) -> Dict:
+        """Submit answers to GAIA"""
+        payload = {
+            "username": username,
+            "agent_code": code_url,
+            "answers": answers
+        }
+        response = requests.post(f"{cls.BASE_URL}/submit", json=payload)
+        response.raise_for_status()
+        return response.json()