Upload 10 files

README.md

@@ -0,0 +1,113 @@
+---
+title: GAIA Agent (Final Assignment of HF Agents Course)
+emoji: 🕵🏻‍♂️
+colorFrom: indigo
+colorTo: indigo
+sdk: gradio
+sdk_version: 5.33.0
+app_file: app.py
+pinned: false
+hf_oauth: true
+# optional, default duration is 8 hours/480 minutes. Max duration is 30 days/43200 minutes.
+hf_oauth_expiration_minutes: 480
+---
+
+
+# GAIA AI Agent via LangGraph
+
+This repository contains a **LangGraph‑powered** agent that scores over 30% on the GAIA Level‑1 benchmark *without any RAG leaks*.
+It routes questions, invokes the right tool, and returns an exact‑match string for the grader.
+
+## 📜 What is GAIA?
+
+**GAIA = _“General AI Assistants”_** – a multi-domain benchmark introduced in the paper   [GAIA: A Benchmark for General AI Assistants](https://arxiv.org/abs/2311.12983).
+The public leaderboard is hosted on Hugging Face:
+<https://huggingface.co/spaces/gaia-benchmark/leaderboard>
+
+---
+
+## ✨ Key features
+
+| Capability | Implementation |
+|------------|---------------|
+| Multi‑step routing | LangGraph state machine (`route_question → invoke_tools → synthesize_response → format_output`) |
+| Web & Wiki search | Tavily ➜ DuckDuckGo fallback |
+| YouTube | `youtube_transcript_api` ➜ generate captions |
+| Spreadsheets | `analyze_excel_file` (*pandas* one‑liner generator) |
+| Attached code | Safe `subprocess` sandbox via `run_py` |
+| Audio | OpenAI‑Whisper |
+| Vision | VLM (GPT-4o-mini)|
+
+---
+
+## 📂 Repository guide
+
+| File | Purpose |
+|------|---------|
+| `app.py` | Gradio UI, API submission, LangGraph workflow |
+| `tools.py` | All custom LangChain tools (search, Excel, Whisper, *etc*.) |
+| `prompts.yaml` | LLM prompts |
+| `helpers.py` | Tiny utilities (debug prints *etc*.) |
+| `debug_agent.py` | Run agent on a single GAIA question from CLI |
+| `requirements.txt` | Runtime deps |
+| `requirements-dev.txt` | Dev / lint deps |
+
+---
+
+## 🚀 Quick start
+
+    # clone repo / space
+    pip install -r requirements.txt   # Python ≥ 3.11
+    python app.py                     # launches local Gradio UI
+
+Run **one** task from CLI (handy while tuning prompts):
+
+    python debug_agent.py <GAIA_task_id>
+
+### Environment variables
+
+| Var | Used for | Example |
+|-----|----------|---------|
+| `OPENAI_API_KEY` | Router & answer LLM (OpenAI) | `sk‑…` |
+| `TAVILY_API_KEY` | Higher‑quality web search (optional) | `tvly_…` |
+
+*(Agent falls back to DuckDuckGo if `TAVILY_API_KEY` is absent.)*
+
+---
+
+##  Agent Routing & Tool-Execution Flow
+
+
+![GAIA  Agent Routing & Tool-Execution Flow](agent_routing.png)
+
+- **route_question** routes to one of eight labels.
+- **invoke_tools** invokes the matching tool and stores context.
+- **synthesize_response** calls the answer LLM unless the answer was computed.
+- **format_output** normalizes output for GAIA’s exact‑match scorer.
+
+
+## 📝 Prompt snippet
+
+All LLM prompts are available in `prompts.yaml`:
+
+## 🛠️ Dev helpers
+
+1️⃣ Create the virtual environment and activate it.
+
+```
+uv venv --python 3.11
+source ./.venv/bin/activate
+```
+
+2️⃣ Install Python dependencies:
+
+```
+uv pip install -r requirements.txt
+uv pip install -r requirements-dev.txt
+```
+
+3️⃣ [Optional] Install Git hooks for code quality checks :
+
+```
+pre-commit install
+```

agent-architecture.mermaid

@@ -0,0 +1,33 @@
+graph TD
+    Start([User Question]) --> RouteQuestion[📋 Route Question]
+    RouteQuestion --> ExecuteTools[🔧 Execute Tools]
+    ExecuteTools --> CheckAttachment{Has Attachment?}
+    CheckAttachment -->|Yes| AttachmentType{Attachment Type?}
+    CheckAttachment -->|No| CheckLabel{Label Type?}
+
+    AttachmentType -->|Python Code| RunPy[🐍 run_py]
+    AttachmentType -->|Excel/CSV| AnalyzeExcel[📊 analyze_excel_file]
+    AttachmentType -->|Audio| TranscribeAudio[🎵 transcribe_via_whisper]
+    AttachmentType -->|Image| VisionTask[👁️ vision_task]
+
+    CheckLabel -->|math| Calculator[🧮 calculator]
+    CheckLabel -->|youtube| YouTubeTranscript[📹 youtube_transcript]
+    CheckLabel -->|search| WebSearch[🔍 web_multi_search]
+    CheckLabel -->|general| NoTool[💭 No specific tool]
+
+    RunPy --> SynthesizeResponse[🧠 Synthesize Response]
+    AnalyzeExcel --> SynthesizeResponse
+    TranscribeAudio --> SynthesizeResponse
+    VisionTask --> SynthesizeResponse
+    Calculator --> SynthesizeResponse
+    YouTubeTranscript --> SynthesizeResponse
+    WebSearch --> SynthesizeResponse
+    NoTool --> SynthesizeResponse
+
+    SynthesizeResponse --> NeedsSynthesis{Needs Additional<br/>Synthesis?}
+    NeedsSynthesis -->|No: code excel<br/>image math| DirectAnswer[✅ Use tool output directly<br/>Already complete]
+    NeedsSynthesis -->|Yes: youtube audio<br/>search general| UseSynthesisLLM[🤖 Additional LLM synthesis<br/>Combine with context]
+
+    DirectAnswer --> FormatOutput[✨ Format Output]
+    UseSynthesisLLM --> FormatOutput
+    FormatOutput --> End([Final Answer])

app.py

@@ -0,0 +1,443 @@
+import os
+import re
+from typing import Literal, TypedDict, get_args
+
+import gradio as gr
+import pandas as pd
+import requests
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_openai import ChatOpenAI
+from langgraph.graph import END, StateGraph
+
+from helpers import fetch_task_attachment, get_prompt, sniff_excel_type
+from tools import (
+    analyze_excel_file,
+    calculator,
+    run_py,
+    transcribe_via_whisper,
+    vision_task,
+    web_multi_search,
+    wiki_search,
+    youtube_transcript,
+)
+
+# --------------------------------------------------------------------------- #
+#                              CONFIGURATION                                  #
+# --------------------------------------------------------------------------- #
+DEFAULT_API_URL: str = "https://agents-course-unit4-scoring.hf.space"
+MODEL_NAME: str = "o4-mini"  # "gpt-4.1-mini"
+TEMPERATURE: float = 0.1
+
+# --------------------------------------------------------------------------- #
+#                           QUESTION  CLASSIFIER                               #
+# --------------------------------------------------------------------------- #
+_LABELS = Literal[
+    "math",
+    "youtube",
+    "image",
+    "code",
+    "excel",
+    "audio",
+    "general",
+]
+
+
+# --------------------------------------------------------------------------- #
+# -------------------------------  AGENT STATE  ----------------------------- #
+# --------------------------------------------------------------------------- #
+class AgentState(TypedDict):
+    question: str
+    label: str
+    context: str
+    answer: str
+    task_id: str | None = None
+
+
+# --------------------------------------------------------------------------- #
+#                         NODES  (LangGraph  functions)                        #
+# --------------------------------------------------------------------------- #
+
+_llm_router = ChatOpenAI(model=MODEL_NAME)
+_llm_answer = ChatOpenAI(model=MODEL_NAME)
+
+
+def route_question(state: AgentState) -> AgentState:  # noqa: D401
+    """Label the task so we know which toolchain to invoke."""
+    question = state["question"]
+
+    label_values = set(get_args(_LABELS))  # -> ("math", "youtube", ...)
+    prompt = get_prompt(
+        prompt_key="router",
+        question=question,
+        labels=", ".join(repr(v) for v in label_values),
+    )
+    resp = _llm_router.invoke(prompt).content.strip().lower()
+    state["label"] = resp if resp in label_values else "general"
+    return state
+
+
+def invoke_tools_context(state: AgentState) -> AgentState:
+    question, label, task_id = state["question"], state["label"], state["task_id"]
+
+    matched_pattern = r"https?://\S+"
+    matched_obj = re.search(matched_pattern, question)
+
+    # ---- attachment detection ------------------------------------------------
+    if task_id:
+        blob, ctype = fetch_task_attachment(api_url=DEFAULT_API_URL, task_id=task_id)
+
+        if any([blob, ctype]):
+            print(f"[DEBUG] attachment type={ctype} ")
+            # ── Python code ------------------------------------------------------
+            if "python" in ctype:
+                print("[DEBUG] Working with a Python attachment file")
+                state["answer"] = run_py.invoke({"code": blob.decode("utf-8")})
+                state["label"] = "code"
+                return state
+
+            # ── Excel / CSV ------------------------------------------------------
+            # 1) Header hints
+            header_says_sheet = any(key in ctype for key in ("excel", "sheet", "csv"))
+            # 2) Magic-number sniff (works when ctype is application/octet-stream)
+            blob_says_sheet = sniff_excel_type(blob) in {"xlsx", "xls", "csv"}
+
+            if header_says_sheet or blob_says_sheet:
+                if blob_says_sheet:
+                    print(f"[DEBUG] octet-stream sniffed as {sniff_excel_type(blob)}")
+
+                print("[DEBUG] Working with a Excel/CSV attachment file")
+                state["answer"] = analyze_excel_file.invoke(
+                    {"xls_bytes": blob, "question": question}
+                )
+                state["label"] = "excel"
+                return state
+
+            # ── Audio --------------------------------------------------------
+            if "audio" in ctype:
+                print("[DEBUG] Working with an audio attachment file")
+                state["context"] = transcribe_via_whisper.invoke({"audio_bytes": blob})
+                state["label"] = "audio"
+                return state
+
+            # ── Image --------------------------------------------------------
+            if "image" in ctype:
+                print("[DEBUG] Working with an image attachment file")
+                state["answer"] = vision_task.invoke(
+                    {"img_bytes": blob, "question": question}
+                )
+                state["label"] = "image"
+                return state
+
+    if label == "math":
+        print("[TOOL] calculator")
+        expr = re.sub(r"\s+", "", question)
+        state["answer"] = calculator.invoke({"expression": expr})
+    elif label == "youtube" and matched_obj:
+        print("[TOOL] youtube_transcript")
+        if matched_obj:
+            url = matched_obj[0]
+            state["context"] = youtube_transcript.invoke({"url": url})
+    elif label == "search":
+        print("[TOOL] web search")
+        search_json = web_multi_search.invoke({"query": question})
+        wiki_text = wiki_search.invoke({"query": question})
+        state["context"] = f"{search_json}\n\n{wiki_text}"
+    else:
+        print("[TOOL] reasoning only (no search)")
+        state["context"] = ""
+    return state
+
+
+def synthesize_response(state: AgentState) -> AgentState:
+    # Skip LLM for deterministic labels or tasks that already used LLMs
+    if state["label"] in {"code", "excel", "image", "math"}:
+        print(f"[DEBUG] ANSWER ({state['label']}) >>> {state['answer']}")
+        return state
+
+    prompt = [
+        SystemMessage(content=get_prompt("final_llm_system")),
+        HumanMessage(
+            content=get_prompt(
+                prompt_key="final_llm_user",
+                question=state["question"],
+                context=state["context"],
+            )
+        ),
+    ]
+    raw = _llm_answer.invoke(prompt).content.strip()
+    state["answer"] = raw
+    return state
+
+
+def format_output(state: AgentState) -> AgentState:
+    txt = re.sub(r"^(final answer:?\s*)", "", state["answer"], flags=re.I).strip()
+
+    # If question demands a single token (first name / one word), enforce it
+    if any(kw in state["question"].lower() for kw in ["first name", "single word"]):
+        txt = txt.split(" ")[0]
+
+    state["answer"] = txt.rstrip(".")
+    return state
+
+
+# --------------------------------------------------------------------------- #
+#                              BUILD  THE  GRAPH                              #
+# --------------------------------------------------------------------------- #
+def build_graph() -> StateGraph:
+    g = StateGraph(AgentState)
+    g.set_entry_point("route_question")
+
+    g.add_node("route_question", route_question)
+    g.add_node("invoke_tools", invoke_tools_context)
+    g.add_node("synthesize_response", synthesize_response)
+    g.add_node("format_output", format_output)
+
+    g.add_edge("route_question", "invoke_tools")
+    g.add_edge("invoke_tools", "synthesize_response")
+    g.add_edge("synthesize_response", "format_output")
+    g.add_edge("format_output", END)
+
+    return g.compile()
+
+
+# --------------------------------------------------------------------------- #
+# -------------------------------  GAIA AGENT  ------------------------------ #
+# --------------------------------------------------------------------------- #
+class GAIAAgent:
+    """Callable wrapper used by run_and_submit_all."""
+
+    def __init__(self) -> None:
+        self.graph = build_graph()
+
+    def __call__(self, question: str, task_id: str | None = None) -> str:
+        state: AgentState = {
+            "question": question,
+            "label": "general",
+            "context": "",
+            "answer": "",
+            "task_id": task_id,
+        }
+        final = self.graph.invoke(state)
+
+        # ── Debug trace ───────────────────────────────────────────────
+        route = final["label"]
+        llm_used = route != "math"  # math path skips the generation LLM
+        print(f"[DEBUG] route='{route}' | LLM_used={llm_used}")
+        # ─────────────────────────────────────────────────────────────
+
+        return final["answer"]
+
+
+def run_and_submit_all(
+    profile: gr.OAuthProfile | None,
+) -> tuple[str, pd.DataFrame | None]:
+    """
+    Fetches all questions, runs the BasicAgent on them, submits all answers,
+    and displays the results.
+    """
+    # --- Determine HF Space Runtime URL and Repo URL ---
+    space_id = os.getenv("SPACE_ID")  # Get the SPACE_ID for sending link to the code
+
+    if profile:
+        username = f"{profile.username}"
+        print(f"User logged in: {username}")
+    else:
+        print("User not logged in.")
+        return "Please Login to Hugging Face with the button.", None
+
+    api_url = DEFAULT_API_URL
+    questions_url = f"{api_url}/questions"
+    submit_url = f"{api_url}/submit"
+
+    # 1. Instantiate Agent ( modify this part to create your agent)
+    try:
+        agent = GAIAAgent()
+        print("GAIA Agent initialized successfully")
+    except Exception as e:
+        print(f"Error instantiating agent: {e}")
+        return f"Error initializing agent: {e}", None
+    # In the case of an app running as a hugging Face space, this link points toward your codebase ( usefull for others so please keep it public)
+    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
+    print(agent_code)
+
+    # 2. Fetch Questions
+    print(f"Fetching questions from: {questions_url}")
+    try:
+        response = requests.get(questions_url, timeout=15)
+        response.raise_for_status()
+        questions_data = response.json()
+        if not questions_data:
+            print("Fetched questions list is empty.")
+            return "Fetched questions list is empty or invalid format.", None
+        print(f"Fetched {len(questions_data)} questions.")
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching questions: {e}")
+        return f"Error fetching questions: {e}", None
+    except requests.exceptions.JSONDecodeError as e:
+        print(f"Error decoding JSON response from questions endpoint: {e}")
+        print(f"Response text: {response.text[:500]}")
+        return f"Error decoding server response for questions: {e}", None
+    except Exception as e:
+        print(f"An unexpected error occurred fetching questions: {e}")
+        return f"An unexpected error occurred fetching questions: {e}", None
+
+    # 3. Run your Agent
+    results_log = []
+    answers_payload = []
+    print(f"Running agent on {len(questions_data)} questions...")
+    for item in questions_data:
+        task_id = item.get("task_id")
+        question_text = item.get("question")
+        if not task_id or question_text is None:
+            print(f"Skipping item with missing task_id or question: {item}")
+            continue
+        try:
+            submitted_answer = agent(question=question_text, task_id=task_id)
+            answers_payload.append(
+                {"task_id": task_id, "submitted_answer": submitted_answer}
+            )
+            results_log.append(
+                {
+                    "Task ID": task_id,
+                    "Question": question_text,
+                    "Submitted Answer": submitted_answer,
+                }
+            )
+        except Exception as e:
+            print(f"Error running agent on task {task_id}: {e}")
+            results_log.append(
+                {
+                    "Task ID": task_id,
+                    "Question": question_text,
+                    "Submitted Answer": f"AGENT ERROR: {e}",
+                }
+            )
+
+    if not answers_payload:
+        print("Agent did not produce any answers to submit.")
+        return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)
+
+    # 4. Prepare Submission
+    submission_data = {
+        "username": username.strip(),
+        "agent_code": agent_code,
+        "answers": answers_payload,
+    }
+    status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..."
+    print(status_update)
+
+    # 5. Submit
+    print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
+    try:
+        response = requests.post(submit_url, json=submission_data, timeout=60)
+        response.raise_for_status()
+        result_data = response.json()
+        final_status = (
+            f"Submission Successful!\n"
+            f"User: {result_data.get('username')}\n"
+            f"Overall Score: {result_data.get('score', 'N/A')}% "
+            f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
+            f"Message: {result_data.get('message', 'No message received.')}"
+        )
+        print("Submission successful.")
+        results_df = pd.DataFrame(results_log)
+        return final_status, results_df
+    except requests.exceptions.HTTPError as e:
+        error_detail = f"Server responded with status {e.response.status_code}."
+        try:
+            error_json = e.response.json()
+            error_detail += f" Detail: {error_json.get('detail', e.response.text)}"
+        except requests.exceptions.JSONDecodeError:
+            error_detail += f" Response: {e.response.text[:500]}"
+        status_message = f"Submission Failed: {error_detail}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+    except requests.exceptions.Timeout:
+        status_message = "Submission Failed: The request timed out."
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+    except requests.exceptions.RequestException as e:
+        status_message = f"Submission Failed: Network error - {e}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+    except Exception as e:
+        status_message = f"An unexpected error occurred during submission: {e}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+
+
+# --- Build Gradio Interface using Blocks ---
+with gr.Blocks() as demo:
+    gr.Markdown("# Basic Agent Evaluation Runner")
+    gr.Markdown(
+        """
+        **Instructions:**
+
+        1.  Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ...
+        2.  Log in to your Hugging Face account using the button below. This uses your HF username for submission.
+        3.  Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.
+
+        ---
+        **Disclaimers:**
+        Once clicking on the "submit button, it can take quite some time ( this is the time for the agent to go through all the questions).
+        This space provides a basic setup and is intentionally sub-optimal to encourage you to develop your own, more robust solution. For instance for the delay process of the submit button, a solution could be to cache the answers and submit in a seperate action or even to answer the questions in async.
+        """
+    )
+
+    gr.LoginButton()
+
+    run_button = gr.Button("Run Evaluation & Submit All Answers")
+
+    status_output = gr.Textbox(
+        label="Run Status / Submission Result", lines=5, interactive=False
+    )
+    # Removed max_rows=10 from DataFrame constructor
+    results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)
+
+    run_button.click(fn=run_and_submit_all, outputs=[status_output, results_table])
+
+
+if __name__ == "__main__":
+    print("\n" + "-" * 30 + " App Starting " + "-" * 30)
+    # Check for SPACE_HOST and SPACE_ID at startup for information
+    space_host_startup = os.getenv("SPACE_HOST")
+    space_id_startup = os.getenv("SPACE_ID")  # Get SPACE_ID at startup
+
+    if space_host_startup:
+        print(f"✅ SPACE_HOST found: {space_host_startup}")
+        print(f"   Runtime URL should be: https://{space_host_startup}.hf.space")
+    else:
+        print("ℹ️  SPACE_HOST environment variable not found (running locally?).")
+
+    if space_id_startup:  # Print repo URLs if SPACE_ID is found
+        print(f"✅ SPACE_ID found: {space_id_startup}")
+        print(f"   Repo URL: https://huggingface.co/spaces/{space_id_startup}")
+        print(
+            f"   Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main"
+        )
+    else:
+        print(
+            "ℹ️  SPACE_ID environment variable not found (running locally?). Repo URL cannot be determined."
+        )
+
+    print("-" * (60 + len(" App Starting ")) + "\n")
+
+    print("Launching Gradio Interface for Basic Agent Evaluation...")
+    demo.launch(debug=True, share=False)
+
+
+## For Local testing
+# if __name__ == "__main__":
+#     agent = GAIAAgent()
+#     while True:
+#         try:
+#             q = input("\nEnter question (or blank to quit): ")
+#         except KeyboardInterrupt:
+#             break
+#         if not q.strip():
+#             break
+#         print("Answer:", agent(q))

debug_agent.py

@@ -0,0 +1,57 @@
+import argparse
+import textwrap
+from typing import Any
+
+import requests
+
+from app import DEFAULT_API_URL, GAIAAgent
+
+
+def fetch_question_row(task_id: str, api: str = DEFAULT_API_URL) -> dict[str, Any]:
+    """Return the question dict associated with *task_id* (raises if not found)."""
+    resp = requests.get(f"{api}/questions", timeout=15)
+    resp.raise_for_status()
+    for row in resp.json():
+        if row["task_id"] == task_id:
+            return row
+    raise ValueError(f"task_id '{task_id}' not present in /questions.")
+
+
+def run_one(task_id: str | None, question: str | None) -> None:
+    agent = GAIAAgent()
+
+    if task_id:
+        row = fetch_question_row(task_id)
+        question = row["question"]
+        print(f"\n{row}\n")  # show full row incl. metadata
+
+    # --- show pretty question
+    print("=" * 90)
+    print(f"QUESTION ({task_id or 'adhoc'})")
+    print(textwrap.fill(question or "", width=90))
+    print("=" * 90)
+
+    assert question is not None, "Internal error: question was None"
+    answer = agent(question, task_id=task_id)
+    print(f"\nFINAL ANSWER --> {answer}")
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Run one GAIAAgent query locally.")
+    parser.add_argument("--task_id", help="GAIA task_id to fetch & run")
+    parser.add_argument("question", nargs="?", help="Ad-hoc question text (positional)")
+
+    ns = parser.parse_args()
+
+    # mutual-exclusion checks
+    if ns.task_id and ns.question:
+        parser.error("Provide either --task_id OR a question, not both.")
+    if ns.task_id is None and ns.question is None:
+        parser.error("You must supply a GAIA --task_id or a question.")
+
+    return ns
+
+
+if __name__ == "__main__":
+    args = parse_args()
+    run_one(task_id=args.task_id, question=args.question)

helpers.py

@@ -0,0 +1,77 @@
+import csv
+from io import BytesIO
+from pathlib import Path
+from sys import stderr
+from traceback import print_exception
+from zipfile import BadZipFile, ZipFile
+
+import requests
+from yaml import safe_load
+
+CURRENT_DIR = Path(__file__).parent
+
+_PROMPTS = safe_load(CURRENT_DIR.joinpath("prompts.yaml").read_text())
+
+
+def fetch_task_attachment(api_url: str, task_id: str) -> tuple[bytes, str]:
+    """
+    Returns (file_bytes, content_type) or (b'', '') if no attachment found.
+    Follows any redirect the endpoint issues.
+    """
+    url = f"{api_url}/files/{task_id}"
+    try:
+        r = requests.get(url, timeout=15, allow_redirects=True)
+    except requests.RequestException as e:
+        print(f"[DEBUG] GET {url} failed → {e}")
+        return b"", ""
+    if r.status_code != 200:
+        print(f"[DEBUG] GET {url} → {r.status_code}")
+        return b"", ""
+    return r.content, r.headers.get("content-type", "").lower()
+
+
+def sniff_excel_type(blob: bytes) -> str:
+    """
+    Return one of 'xlsx', 'xls', 'csv', or '' (unknown) given raw bytes.
+    """
+    # 1️⃣ XLSX / XLSM / ODS  (ZIP container)
+    if blob[:4] == b"PK\x03\x04":
+        try:
+            with ZipFile(BytesIO(blob)) as zf:
+                names = set(zf.namelist())
+                if {"xl/workbook.xml", "[Content_Types].xml"} & names:
+                    return "xlsx"
+        except BadZipFile:
+            pass  # fall through
+
+    # 2️⃣ Legacy XLS (OLE Compound File)
+    if blob[:8] == b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1":
+        return "xls"
+
+    # 3️⃣ Text-like -> CSV/TSV
+    try:
+        sample = blob[:1024].decode("utf-8", "ignore")
+        first_line = sample.splitlines()[0]
+        if any(sep in first_line for sep in (",", ";", "\t")):
+            # Confirm via csv.Sniffer to avoid random text
+            csv.Sniffer().sniff(sample)
+            return "csv"
+    except (UnicodeDecodeError, csv.Error):
+        pass
+
+    return ""
+
+
+def get_prompt(prompt_key: str, **kwargs: str) -> str:
+    """Get a prompt by key and fill in placeholders via `.format(**kwargs)`"""
+    return _PROMPTS[prompt_key].format(**kwargs)
+
+
+def print_debug_trace(err: Exception, label: str = "") -> None:
+    """
+    Print the full stack trace of `err` to STDERR so it shows up in HF logs.
+    """
+    banner = f"[TRACE {label}]" if label else "[TRACE]"
+    print(banner, file=stderr)
+    print_exception(type(err), err, err.__traceback__, file=stderr)
+    print("-" * 60, file=stderr)

prompts.yaml

@@ -0,0 +1,84 @@
+router: |
+  You are a *routing* assistant.
+  Your ONLY job is to print **one** of the allowed labels - nothing else.
+
+  Allowed labels
+  ==============
+  {labels}
+
+  Guidelines
+  ----------
+  - **math**: the question is a pure arithmetic/numeric expression.
+  - **youtube**: the question contains a YouTube URL and asks about its content.
+  - **code**: the task references attached Python code; caller wants its output.
+  - **excel**: the task references an attached .xlsx/.xls/.csv and asks for a sum, average, etc.
+  - **audio**: the task references an attached audio file and asks for its transcript or facts in it.
+  - **image**: the task could be either generic like "what is in the picture (e.g. Which animal is shown?) or could be a puzzle like asking for a *move, count, coordinate,* or other board-game tactic that needs an exact piece layout (e.g. "What is Black's winning move?").
+  - **search** : needs external factual information from the web
+  - **reason** : answer can be produced by analyzing the question text alone
+
+  Examples
+  ----------
+  (search) What is the last name of the person who founded Mercedes Benz company?
+  (reasoning) what is the third item of following list that is a fruit after sorting it alphabetically: ['parsley', 'orange', 'apple', 'coriander', 'lettuce', 'kiwi', 'apricot']" Answer is 'kiwi'
+
+  ~~~
+  User question:
+  {question}
+  ~~~
+
+  IMPORTANT: Respond with **one label exactly**, no punctuation, no explanation.
+
+final_llm_system: |
+  You are a precise research assistant.
+  Return ONLY the literal answer - no preamble.
+
+  Formatting rules
+  1. If the question asks for a *first name*, output the first given name only.
+  2. If the answer is purely numeric, output digits only (no commas, units, words) as a string.
+  3. Otherwise capitalize the first character of your answer **unless** doing so would change the original spelling of text you are quoting verbatim
+
+  Examples
+  Q: Which planet is fourth from the Sun?
+  A: Mars <-- capitalized
+
+  Q: What Unix command lists files?
+  A: ls <-- lower-case preserved
+
+final_llm_user: |
+  Question: {question}
+
+  Context: {context}
+
+  Answer:
+
+vision_system: |
+  You are a terse assistant. Respond with ONLY the answer to the user's question—no explanations, no punctuation except what the answer itself requires.
+  If the answer is a chess move, output it in algebraic notation.
+  IMPORTANT: Only respond with the final answer with no extra text.
+
+excel_system: |
+  You are a **pandas one-liner generator**.
+
+  Context
+  -------
+  - A full DataFrame named `df` is already loaded.
+  - Only the preview below is shown for reference.
+  - IMPORTANT: use column names from the preview to determine which columns are needed.
+
+  Preview
+  -------
+  {preview}
+
+  Formatting rules
+  ----------------
+  1. Result must be a plain Python scalar (use .item(), float(), int() …).
+  2. If the question asks for currency / 2 decimals --> wrap in an f-string.
+  3. If the question asks for a count --> wrap in int().
+  4. **Return exactly one line.**
+  5. DO NOT include any unit or currency in the output.
+  6. **Do **NOT** wrap the expression in ``` or other markdown fences.**
+
+  Question
+  --------
+  {question}

pyproject.toml

@@ -0,0 +1,50 @@
+# pyproject.toml (trimmed to just tooling – no build backend)
+
+[tool.ruff]
+line-length = 88
+target-version = "py311"
+lint.select = [
+    "E",  # pycodestyle errors
+    "W",  # pycodestyle warnings
+    "F",  # pyflakes
+    "I",  # isort
+    "C",  # flake8-comprehensions
+    "B",  # flake8-bugbear
+    "UP", # pyupgrade
+]
+lint.ignore = [
+    "E501",  # line too long, handled by black
+    "B008",  # do not perform function calls in argument defaults
+    "C901",  # too complex
+]
+fix = true
+
+[tool.mypy]
+python_version = "3.11"
+disallow_any_generics = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_defs = true
+disallow_incomplete_defs = true
+check_untyped_defs = true
+disallow_untyped_decorators = true
+no_implicit_optional = true
+warn_redundant_casts = true
+warn_unused_ignores = true
+warn_return_any = true
+implicit_reexport = false
+strict_equality = true
+disable_error_code = [
+  "misc",          # untyped decorator
+  "no-any-return", # allow Any returns temporarily
+  "operator",      # calls on unknown operator types
+]
+plugins = ["pydantic.mypy"]
+
+follow_imports = "silent"
+no_implicit_reexport = true
+
+[tool.pydantic-mypy]
+init_forbid_extra = true
+init_typed = true
+warn_required_dynamic_aliases = true

requirements-dev.txt

@@ -0,0 +1,5 @@
+pre-commit
+ruff
+mypy
+detect-secrets
+gradio[oauth]

requirements.txt

@@ -0,0 +1,26 @@
+# ── core UI / infra
+gradio
+requests
+pandas==2.2.3
+
+# ── LangGraph + LangChain stack
+langgraph==0.4.7
+langchain_openai==0.3.18
+langchain_core==0.3.61
+langchain==0.3.25
+langchain_community==0.3.24
+
+# ── Retrieval helpers
+duckduckgo-search==8.0.2          # for DuckDuckGo wrapper
+tavily-python==0.3.3              # TavilySearchResults tool
+wikipedia==1.4.0                  # WikipediaLoader
+
+# ── Media utilities
+youtube-transcript-api==1.0.3  # YouTube transcripts
+openpyxl==3.1.5                # Excel parsing when GAIA attaches .xlsx
+Pillow>=10.2.0                 # image handling for transformers
+openai-whisper==20240930
+
+# ── Lightweight vision model
+transformers>=4.41.2
+torch>=2.3.0                   # auto-installs CPU wheels on HF Spaces

tools.py

@@ -0,0 +1,289 @@
+import ast
+import json
+import operator
+import re
+import subprocess
+from base64 import b64encode
+from functools import lru_cache
+from io import BytesIO
+from tempfile import NamedTemporaryFile
+
+import numpy as np
+import pandas as pd
+from langchain_community.document_loaders import WikipediaLoader
+from langchain_community.tools.tavily_search import TavilySearchResults
+from langchain_community.utilities import DuckDuckGoSearchAPIWrapper
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_core.tools import tool
+from langchain_openai import ChatOpenAI
+from youtube_transcript_api import YouTubeTranscriptApi
+
+from helpers import get_prompt, print_debug_trace
+
+# --------------------------------------------------------------------------- #
+#                       ARITHMETIC (SAFE CALCULATOR)                         #
+# --------------------------------------------------------------------------- #
+_ALLOWED_AST_OPS = {
+    ast.Add: operator.add,
+    ast.Sub: operator.sub,
+    ast.Mult: operator.mul,
+    ast.Div: operator.truediv,
+    ast.Pow: operator.pow,
+    ast.USub: operator.neg,
+}
+
+
+def _safe_eval(node: ast.AST) -> float | int | complex:
+    """Recursively evaluate a *restricted* AST expression tree."""
+    if isinstance(node, ast.Constant):
+        return node.n
+    if isinstance(node, ast.UnaryOp) and type(node.op) in _ALLOWED_AST_OPS:
+        return _ALLOWED_AST_OPS[type(node.op)](_safe_eval(node.operand))
+    if isinstance(node, ast.BinOp) and type(node.op) in _ALLOWED_AST_OPS:
+        return _ALLOWED_AST_OPS[type(node.op)](
+            _safe_eval(node.left), _safe_eval(node.right)
+        )
+    raise ValueError("Unsafe or unsupported expression")
+
+
+@tool
+def calculator(expression: str) -> str:
+    """Safely evaluate basic arithmetic expressions (no variables, functions)."""
+    try:
+        tree = ast.parse(expression, mode="eval")
+        value = _safe_eval(tree.body)
+        return str(value)
+    except Exception as exc:
+        print_debug_trace(exc, "Calculator")
+        return f"calc_error:{exc}"
+
+
+# --------------------------------------------------------------------------- #
+#                             WEB  &  WIKI  SEARCH                           #
+# --------------------------------------------------------------------------- #
+@lru_cache(maxsize=256)
+def _ddg_search(query: str, k: int = 6) -> list[dict[str, str]]:
+    """Cached DuckDuckGo JSON search."""
+    wrapper = DuckDuckGoSearchAPIWrapper(max_results=k)
+    hits = wrapper.results(query)
+    return [
+        {
+            "title": hit.get("title", "")[:500],
+            "snippet": hit.get("snippet", "")[:750],
+            "link": hit.get("link", "")[:300],
+        }
+        for hit in hits[:k]
+    ]
+
+
+@tool
+def web_multi_search(query: str, k: int = 6) -> str:
+    """Run DuckDuckGo → Tavily fallback search. Returns JSON list[dict]."""
+    try:
+        hits = _ddg_search(query, k)
+        if hits:
+            return json.dumps(hits, ensure_ascii=False)
+    except Exception:  # fall through to Tavily
+        pass
+
+    try:
+        tavily_results = TavilySearchResults(
+            max_results=5,
+            # include_answer=True,
+            # search_depth="advanced",
+        )
+        search_result = tavily_results.invoke({"query": query})
+        print(
+            f"[TOOL] TAVILY search is triggered with following response: {search_result}"
+        )
+        formatted = [
+            {
+                "title": d.get("title", "")[:500],
+                "snippet": d.get("content", "")[:750],
+                "link": d.get("url", "")[:300],
+            }
+            for d in search_result
+        ]
+        return json.dumps(formatted, ensure_ascii=False)
+    except Exception as exc:
+        print_debug_trace(exc, "Multi Search")
+        return f"search_error:{exc}"
+
+
+@tool
+def wiki_search(query: str, max_pages: int = 2) -> str:
+    """Lightweight wrapper on WikipediaLoader; returns concatenated page texts."""
+    print(f"[TOOL] wiki_search called with query: {query}")
+    docs = WikipediaLoader(query=query, load_max_docs=max_pages).load()
+    joined = "\n\n---\n\n".join(d.page_content for d in docs)
+    return joined[:8_000]  # simple guardrail – stay within context window
+
+
+# --------------------------------------------------------------------------- #
+#                               YOUTUBE  TRANSCRIPT                          #
+# --------------------------------------------------------------------------- #
+@tool
+def youtube_transcript(url: str, chars: int = 10_000) -> str:
+    """Fetch full YouTube transcript (first *chars* characters)."""
+    video_id_match = re.search(r"[?&]v=([A-Za-z0-9_\-]{11})", url)
+    if not video_id_match:
+        return "yt_error:id_not_found"
+    try:
+        transcript = YouTubeTranscriptApi.get_transcript(video_id_match.group(1))
+        text = " ".join(piece["text"] for piece in transcript)
+        return text[:chars]
+    except Exception as exc:
+        print_debug_trace(exc, "YouTube")
+        return f"yt_error:{exc}"
+
+
+# --------------------------------------------------------------------------- #
+#                                IMAGE  DESCRIPTION                           #
+# --------------------------------------------------------------------------- #
+
+# Instantiate a lightweight CLIP‑based zero‑shot image classifier (runs on CPU)
+### The model 'openai/clip-vit-base-patch32' is a vision transformer (ViT) model trained as part of OpenAI’s CLIP project.
+### It performs zero-shot image classification by mapping images and labels into the same embedding space.
+# _image_pipe = pipeline(
+#     "image-classification", model="openai/clip-vit-base-patch32", device="cpu"
+# )
+
+# @tool
+# def image_describe(img_bytes: bytes, top_k: int = 3) -> str:
+#     """Return the top-k CLIP labels for an image supplied as raw bytes.
+
+#     typical result for a random cat photo can be:
+#     [
+#         {'label': 'tabby, tabby cat', 'score': 0.41},
+#         {'label': 'tiger cat', 'score': 0.24},
+#         {'label': 'Egyptian cat', 'score': 0.22}
+#     ]
+#     """
+
+#     try:
+#         labels = _image_pipe(BytesIO(img_bytes))[:top_k]
+#         return ", ".join(f"{d['label']} (score={d['score']:.2f})" for d in labels)
+#     except Exception as exc:
+#         return f"img_error:{exc}"
+
+
+@tool
+def vision_task(img_bytes: bytes, question: str) -> str:
+    """
+    Pass the user's question AND the referenced image to a multimodal LLM and
+    return its first line of text as the answer.  No domain assumptions made.
+    """
+    vision_llm = ChatOpenAI(
+        model="gpt-4o-mini",  # set OPENAI_API_KEY in env
+        temperature=0,
+        max_tokens=64,
+    )
+    try:
+        b64 = b64encode(img_bytes).decode()
+        messages = [
+            SystemMessage(content=get_prompt(prompt_key="vision_system")),
+            HumanMessage(
+                content=[
+                    {"type": "text", "text": question.strip()},
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": f"data:image/png;base64,{b64}"},
+                    },
+                ]
+            ),
+        ]
+        reply = vision_llm.invoke(messages).content.strip()
+        return reply
+    except Exception as exc:
+        print_debug_trace(exc, "vision")
+        return f"img_error:{exc}"
+
+
+# --------------------------------------------------------------------------- #
+#                                 FILE  UTILS                                 #
+# --------------------------------------------------------------------------- #
+@tool
+def run_py(code: str) -> str:
+    """Execute Python code in a sandboxed subprocess and return last stdout line."""
+    try:
+        with NamedTemporaryFile(delete=False, suffix=".py", mode="w") as f:
+            f.write(code)
+            path = f.name
+        proc = subprocess.run(
+            ["python", path], capture_output=True, text=True, timeout=45
+        )
+        out = proc.stdout.strip().splitlines()
+        return out[-1] if out else ""
+    except Exception as exc:
+        print_debug_trace(exc, "run_py")
+        return f"py_error:{exc}"
+
+
+@tool
+def transcribe_via_whisper(audio_bytes: bytes) -> str:
+    """Transcribe audio with Whisper (CPU)."""
+    with NamedTemporaryFile(suffix=".mp3", delete=False) as f:
+        f.write(audio_bytes)
+        path = f.name
+    try:
+        import whisper  # openai-whisper
+
+        model = whisper.load_model("base")
+        output = model.transcribe(path)["text"].strip()
+        print(f"[DEBUG] Whisper transcript (first 200 chars): {output[:200]}")
+        return output
+    except Exception as exc:
+        print_debug_trace(exc, "Whisper")
+        return f"asr_error:{exc}"
+
+
+@tool
+def analyze_excel_file(xls_bytes: bytes, question: str) -> str:
+    "Analyze Excel or CSV file by passing the data preview to LLM and getting the Python Pandas operation to run"
+    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, max_tokens=64)
+
+    try:
+        df = pd.read_excel(BytesIO(xls_bytes))
+    except Exception:
+        df = pd.read_csv(BytesIO(xls_bytes))
+
+    for col in df.select_dtypes(include="number").columns:
+        df[col] = df[col].astype(float)
+
+    # Ask the LLM for a single expression
+    prompt = get_prompt(
+        prompt_key="excel_system",
+        question=question,
+        preview=df.head(5).to_dict(orient="list"),
+    )
+    expr = llm.invoke(prompt).content.strip()
+
+    # Run generated Pandas' one-line expression
+    try:
+        result = eval(expr, {"df": df, "pd": pd, "__builtins__": {}})
+        # Normalize scalars to string
+        if isinstance(result, np.generic):
+            result = float(result)  # → plain Python float
+            return f"{result:.2f}"  # or str(result) if no decimals needed
+
+        # DataFrame / Series → single-line string
+        return (
+            result.to_string(index=False)
+            if hasattr(result, "to_string")
+            else str(result)
+        )
+    except Exception as e:
+        print_debug_trace(e, "Excel")
+        return f"eval_error:{e}"
+
+
+__all__ = [
+    "calculator",
+    "web_multi_search",
+    "wiki_search",
+    "youtube_transcript",
+    "vision_task",
+    "run_py",
+    "transcribe_via_whisper",
+    "analyze_excel_file",
+]