Build routed GAIA agent v1

.gitignore

@@ -6,3 +6,4 @@ __pycache__/
 .mypy_cache/
 .ruff_cache/
 *.egg-info/
+.gaia_cache/

README.md

@@ -39,3 +39,26 @@ OPENAI_API_KEY=...
 ```
 
 If you run a LiteLLM proxy, set `LITELLM_API_BASE` and `LITELLM_API_KEY`.
+
+## Agent V1
+
+The agent uses a routed LangGraph flow:
+
+```text
+ingest_task -> classify_task -> specialized solver -> verify_answer -> normalize_final_answer
+```
+
+Routes cover direct reasoning, computation/table questions, web search, YouTube
+transcripts, spreadsheets, Python files, audio files, and image files.
+
+For audio questions, configure a LiteLLM transcription-compatible provider. By
+default the code uses `AUDIO_TRANSCRIPTION_MODEL=whisper-1` and reads
+`OPENAI_API_KEY`, or you can set:
+
+```env
+AUDIO_TRANSCRIPTION_MODEL=whisper-1
+AUDIO_TRANSCRIPTION_API_KEY=...
+AUDIO_TRANSCRIPTION_API_BASE=...
+```
+
+For image questions, use a vision-capable `LITELLM_MODEL`.

app.py

@@ -74,6 +74,8 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
     for item in questions_data:
         task_id = item.get("task_id")
         question_text = item.get("question")
+        file_name = item.get("file_name") or ""
+        level = item.get("Level") or ""
         if not task_id or question_text is None:
             print(f"Skipping item with missing task_id or question: {item}")
             continue
@@ -83,12 +85,32 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
                 session_id=session_id,
                 user_id=username.strip(),
                 task_id=task_id,
+                file_name=file_name,
+                level=level,
+            )
+            answers_payload.append(
+                {"task_id": task_id, "submitted_answer": submitted_answer}
+            )
+            results_log.append(
+                {
+                    "Task ID": task_id,
+                    "Level": level,
+                    "File": file_name,
+                    "Question": question_text,
+                    "Submitted Answer": submitted_answer,
+                }
             )
-            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}"})
+            print(f"Error running agent on task {task_id}: {e}")
+            results_log.append(
+                {
+                    "Task ID": task_id,
+                    "Level": level,
+                    "File": file_name,
+                    "Question": question_text,
+                    "Submitted Answer": f"AGENT ERROR: {e}",
+                }
+            )
 
     if not answers_payload:
         print("Agent did not produce any answers to submit.")

gaia_agent/agent.py

@@ -14,6 +14,9 @@ class GaiaAgent:
         session_id: str | None = None,
         user_id: str | None = None,
         task_id: str | None = None,
+        file_name: str | None = None,
+        file_path: str | None = None,
+        level: str | None = None,
     ) -> str:
         print(f"Agent received question (first 80 chars): {question[:80]}...")
         with trace_agent_run(
@@ -23,7 +26,15 @@ class GaiaAgent:
             task_id=task_id,
         ) as trace:
             graph = build_graph(trace=trace, llm=self.llm)
-            result = graph.invoke({"question": question})
+            initial_state = {
+                "question": question,
+                "task_id": task_id or "",
+                "file_name": file_name or "",
+                "level": level or "",
+            }
+            if file_path:
+                initial_state["file_path"] = file_path
+            result = graph.invoke(initial_state)
         final_answer = result["final_answer"]
         print(f"Agent returning answer: {final_answer}")
         return final_answer

gaia_agent/answer.py

@@ -1,3 +1,30 @@
+import re
+
+
 def normalize_answer(answer: str) -> str:
     """Apply minimal GAIA answer cleanup without changing meaning."""
-    return answer.strip().removesuffix(".")
+    cleaned = str(answer).strip()
+
+    final_answer_match = re.search(
+        r"FINAL\s+ANSWER\s*:\s*(.+)\s*$",
+        cleaned,
+        flags=re.IGNORECASE | re.DOTALL,
+    )
+    if final_answer_match:
+        cleaned = final_answer_match.group(1).strip()
+
+    cleaned = cleaned.strip("` \n\t")
+    cleaned = cleaned.strip()
+    if (
+        len(cleaned) >= 2
+        and cleaned[0] == cleaned[-1]
+        and cleaned[0] in {"'", '"'}
+    ):
+        cleaned = cleaned[1:-1].strip()
+
+    cleaned = cleaned.removesuffix(".").strip()
+
+    if re.fullmatch(r"\$?-?\d[\d,]*(?:\.\d+)?", cleaned):
+        cleaned = cleaned.removeprefix("$").replace(",", "")
+
+    return cleaned

gaia_agent/graph.py

@@ -1,38 +1,454 @@
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import Any
+
+from langchain_core.messages import HumanMessage, SystemMessage
 from langgraph.graph import END, StateGraph
 
 from gaia_agent.answer import normalize_answer
+from gaia_agent.config import settings
 from gaia_agent.llms import create_chat_model
 from gaia_agent.observability import traced_step
-from gaia_agent.prompts import DUMMY_LLM_TEST_PROMPT
+from gaia_agent.prompts import (
+    GAIA_AGENT_SYSTEM_PROMPT,
+    GAIA_QUERY_PROMPT,
+    GAIA_VERIFY_PROMPT,
+)
 from gaia_agent.state import GaiaState
+from gaia_agent.tools.files import (
+    download_task_file,
+    read_text_file,
+    summarize_spreadsheet,
+)
+from gaia_agent.tools.media import image_data_url, transcribe_audio_file
+from gaia_agent.tools.python_repl import run_python_file
+from gaia_agent.tools.web import (
+    extract_urls,
+    fetch_url,
+    get_youtube_transcript,
+    web_search,
+)
+
+
+MAX_EVIDENCE_CHARS = 36_000
+MAX_WEB_PAGES = 4
 
 
 def build_graph(trace=None, llm=None):
     graph = StateGraph(GaiaState)
     chat_model = llm or create_chat_model()
 
-    def draft_answer(state: GaiaState) -> GaiaState:
+    def ingest_task(state: GaiaState) -> dict[str, Any]:
+        def run() -> dict[str, Any]:
+            evidence = list(state.get("evidence", []))
+            output: dict[str, Any] = {
+                "evidence": evidence,
+                "tool_outputs": list(state.get("tool_outputs", [])),
+            }
+            if state.get("file_path") or not state.get("file_name"):
+                return output
+
+            try:
+                path = download_task_file(
+                    settings.gaia_api_url,
+                    state["task_id"],
+                    state.get("file_name"),
+                )
+                output["file_path"] = str(path)
+                evidence.append(f"Downloaded attached file to {path}.")
+            except Exception as exc:
+                evidence.append(f"Could not download attached file: {exc}")
+                output["error"] = str(exc)
+            return output
+
+        return traced_step(trace, "ingest_task", run)
+
+    def classify_task(state: GaiaState) -> dict[str, str]:
         def run() -> dict[str, str]:
-            response = chat_model.invoke(
-                [
-                    ("system", DUMMY_LLM_TEST_PROMPT),
-                    ("user", state["question"]),
-                ]
+            question = state["question"].lower()
+            file_name = state.get("file_name", "").lower()
+
+            if file_name.endswith((".xlsx", ".xls", ".csv")):
+                task_type = "spreadsheet"
+            elif file_name.endswith(".py"):
+                task_type = "python_file"
+            elif file_name.endswith((".mp3", ".wav", ".m4a", ".ogg", ".flac")):
+                task_type = "audio"
+            elif file_name.endswith((".png", ".jpg", ".jpeg", ".webp")):
+                task_type = "image"
+            elif "youtube.com" in question or "youtu.be" in question:
+                task_type = "youtube"
+            elif _looks_like_computation(question):
+                task_type = "compute"
+            elif _looks_like_direct(question):
+                task_type = "direct"
+            else:
+                task_type = "web"
+            return {"task_type": task_type}
+
+        return traced_step(trace, "classify_task", run)
+
+    def solve_direct(state: GaiaState) -> dict[str, Any]:
+        def run() -> dict[str, Any]:
+            answer = _invoke_text(
+                chat_model,
+                GAIA_AGENT_SYSTEM_PROMPT,
+                f"Question:\n{state['question']}",
+            )
+            return {"draft_answer": answer}
+
+        return traced_step(trace, "solve_direct", run)
+
+    def solve_compute(state: GaiaState) -> dict[str, Any]:
+        def run() -> dict[str, Any]:
+            answer = _invoke_text(
+                chat_model,
+                GAIA_AGENT_SYSTEM_PROMPT,
+                (
+                    "Solve this question carefully. If it includes a table or "
+                    "formal rule, compute the requested value exactly.\n\n"
+                    f"Question:\n{state['question']}"
+                ),
+            )
+            return {"draft_answer": answer}
+
+        return traced_step(trace, "solve_compute", run)
+
+    def solve_spreadsheet(state: GaiaState) -> dict[str, Any]:
+        def run() -> dict[str, Any]:
+            evidence = list(state.get("evidence", []))
+            path = state.get("file_path")
+            if not path:
+                evidence.append("Attached spreadsheet is unavailable.")
+                answer = _invoke_text(
+                    chat_model,
+                    GAIA_AGENT_SYSTEM_PROMPT,
+                    _question_with_evidence(state["question"], evidence),
+                )
+                return {"evidence": evidence, "draft_answer": answer}
+
+            summary = summarize_spreadsheet(path)
+            evidence.append(f"Spreadsheet summary:\n{summary}")
+            answer = _invoke_text(
+                chat_model,
+                GAIA_AGENT_SYSTEM_PROMPT,
+                _question_with_evidence(state["question"], evidence),
+            )
+            return {"evidence": evidence, "draft_answer": answer}
+
+        return traced_step(trace, "solve_spreadsheet", run)
+
+    def solve_python_file(state: GaiaState) -> dict[str, Any]:
+        def run() -> dict[str, Any]:
+            evidence = list(state.get("evidence", []))
+            path = state.get("file_path")
+            if not path:
+                evidence.append("Attached Python file is unavailable.")
+                answer = _invoke_text(
+                    chat_model,
+                    GAIA_AGENT_SYSTEM_PROMPT,
+                    _question_with_evidence(state["question"], evidence),
+                )
+                return {"evidence": evidence, "draft_answer": answer}
+
+            source = read_text_file(path, max_chars=30_000)
+            result = run_python_file(path)
+            evidence.append(f"Attached Python source:\n{source}")
+            evidence.append(
+                "Python execution result:\n"
+                f"exit_code={result['exit_code']}\n"
+                f"stdout:\n{result['stdout']}\n"
+                f"stderr:\n{result['stderr']}"
+            )
+
+            stdout = str(result.get("stdout", "")).strip()
+            if stdout and not str(result.get("stderr", "")).strip():
+                draft = stdout.splitlines()[-1]
+                verified = draft
+            else:
+                draft = _invoke_text(
+                    chat_model,
+                    GAIA_AGENT_SYSTEM_PROMPT,
+                    _question_with_evidence(state["question"], evidence),
+                )
+                verified = ""
+            output = {"evidence": evidence, "draft_answer": draft}
+            if verified:
+                output["verified_answer"] = verified
+            return output
+
+        return traced_step(trace, "solve_python_file", run)
+
+    def solve_audio(state: GaiaState) -> dict[str, Any]:
+        def run() -> dict[str, Any]:
+            evidence = list(state.get("evidence", []))
+            path = state.get("file_path")
+            if not path:
+                evidence.append("Attached audio file is unavailable.")
+            else:
+                try:
+                    transcript = transcribe_audio_file(path)
+                    evidence.append(f"Audio transcript:\n{transcript}")
+                except Exception as exc:
+                    evidence.append(f"Audio transcription failed: {exc}")
+
+            answer = _invoke_text(
+                chat_model,
+                GAIA_AGENT_SYSTEM_PROMPT,
+                _question_with_evidence(state["question"], evidence),
             )
-            return {"draft_answer": str(response.content)}
+            return {"evidence": evidence, "draft_answer": answer}
+
+        return traced_step(trace, "solve_audio", run)
 
-        return traced_step(trace, "draft_answer", run)
+    def solve_image(state: GaiaState) -> dict[str, Any]:
+        def run() -> dict[str, Any]:
+            evidence = list(state.get("evidence", []))
+            path = state.get("file_path")
+            if path:
+                try:
+                    answer = _invoke_image(chat_model, state["question"], path)
+                    evidence.append(f"Image analyzed from {path}.")
+                except Exception as exc:
+                    evidence.append(f"Image analysis failed: {exc}")
+                    answer = _invoke_text(
+                        chat_model,
+                        GAIA_AGENT_SYSTEM_PROMPT,
+                        _question_with_evidence(state["question"], evidence),
+                    )
+            else:
+                evidence.append("Attached image file is unavailable.")
+                answer = _invoke_text(
+                    chat_model,
+                    GAIA_AGENT_SYSTEM_PROMPT,
+                    _question_with_evidence(state["question"], evidence),
+                )
+            return {"evidence": evidence, "draft_answer": answer}
+
+        return traced_step(trace, "solve_image", run)
+
+    def solve_youtube(state: GaiaState) -> dict[str, Any]:
+        def run() -> dict[str, Any]:
+            evidence = list(state.get("evidence", []))
+            urls = extract_urls(state["question"])
+            for url in urls:
+                if "youtube.com" not in url and "youtu.be" not in url:
+                    continue
+                try:
+                    transcript = get_youtube_transcript(url)
+                    evidence.append(f"YouTube transcript for {url}:\n{transcript}")
+                except Exception as exc:
+                    evidence.append(f"YouTube transcript failed for {url}: {exc}")
+
+            answer = _invoke_text(
+                chat_model,
+                GAIA_AGENT_SYSTEM_PROMPT,
+                _question_with_evidence(state["question"], evidence),
+            )
+            return {"evidence": evidence, "draft_answer": answer}
 
-    def normalize_final_answer(state: GaiaState) -> GaiaState:
+        return traced_step(trace, "solve_youtube", run)
+
+    def solve_web(state: GaiaState) -> dict[str, Any]:
+        def run() -> dict[str, Any]:
+            evidence = list(state.get("evidence", []))
+            queries = _build_search_queries(chat_model, state["question"])
+            seen_urls: set[str] = set()
+
+            for query in queries:
+                try:
+                    results = web_search(query, max_results=5)
+                except Exception as exc:
+                    evidence.append(f"Search failed for {query!r}: {exc}")
+                    continue
+
+                if results:
+                    evidence.append(
+                        "Search results for "
+                        f"{query!r}:\n"
+                        + "\n".join(f"- {item.title}: {item.url}" for item in results)
+                    )
+
+                for result in results:
+                    if len(seen_urls) >= MAX_WEB_PAGES:
+                        break
+                    if result.url in seen_urls:
+                        continue
+                    seen_urls.add(result.url)
+                    try:
+                        page_text = fetch_url(result.url, max_chars=12_000)
+                    except Exception as exc:
+                        evidence.append(f"Fetch failed for {result.url}: {exc}")
+                        continue
+                    evidence.append(f"Page: {result.title}\nURL: {result.url}\n{page_text}")
+
+            answer = _invoke_text(
+                chat_model,
+                GAIA_AGENT_SYSTEM_PROMPT,
+                _question_with_evidence(state["question"], evidence),
+            )
+            return {"evidence": evidence, "draft_answer": answer}
+
+        return traced_step(trace, "solve_web", run)
+
+    def verify_answer(state: GaiaState) -> dict[str, str]:
         def run() -> dict[str, str]:
-            return {"final_answer": normalize_answer(state["draft_answer"])}
+            if state.get("verified_answer"):
+                return {"verified_answer": state["verified_answer"]}
+
+            evidence = _trim_evidence(state.get("evidence", []))
+            verified = _invoke_text(
+                chat_model,
+                GAIA_VERIFY_PROMPT,
+                (
+                    f"Question:\n{state['question']}\n\n"
+                    f"Evidence:\n{evidence}\n\n"
+                    f"Draft answer:\n{state.get('draft_answer', '')}"
+                ),
+            )
+            return {"verified_answer": verified}
+
+        return traced_step(trace, "verify_answer", run)
+
+    def normalize_final_answer(state: GaiaState) -> dict[str, str]:
+        def run() -> dict[str, str]:
+            answer = state.get("verified_answer") or state.get("draft_answer", "")
+            return {"final_answer": normalize_answer(answer)}
 
         return traced_step(trace, "normalize_final_answer", run)
 
-    graph.add_node("draft_answer", draft_answer)
+    graph.add_node("ingest_task", ingest_task)
+    graph.add_node("classify_task", classify_task)
+    graph.add_node("solve_direct", solve_direct)
+    graph.add_node("solve_compute", solve_compute)
+    graph.add_node("solve_spreadsheet", solve_spreadsheet)
+    graph.add_node("solve_python_file", solve_python_file)
+    graph.add_node("solve_audio", solve_audio)
+    graph.add_node("solve_image", solve_image)
+    graph.add_node("solve_youtube", solve_youtube)
+    graph.add_node("solve_web", solve_web)
+    graph.add_node("verify_answer", verify_answer)
     graph.add_node("normalize_final_answer", normalize_final_answer)
-    graph.set_entry_point("draft_answer")
-    graph.add_edge("draft_answer", "normalize_final_answer")
+
+    graph.set_entry_point("ingest_task")
+    graph.add_edge("ingest_task", "classify_task")
+    graph.add_conditional_edges(
+        "classify_task",
+        lambda state: state.get("task_type", "web"),
+        {
+            "direct": "solve_direct",
+            "compute": "solve_compute",
+            "spreadsheet": "solve_spreadsheet",
+            "python_file": "solve_python_file",
+            "audio": "solve_audio",
+            "image": "solve_image",
+            "youtube": "solve_youtube",
+            "web": "solve_web",
+        },
+    )
+    for node in (
+        "solve_direct",
+        "solve_compute",
+        "solve_spreadsheet",
+        "solve_python_file",
+        "solve_audio",
+        "solve_image",
+        "solve_youtube",
+        "solve_web",
+    ):
+        graph.add_edge(node, "verify_answer")
+    graph.add_edge("verify_answer", "normalize_final_answer")
     graph.add_edge("normalize_final_answer", END)
 
     return graph.compile()
+
+
+def _invoke_text(chat_model, system_prompt: str, user_prompt: str) -> str:
+    response = chat_model.invoke(
+        [
+            ("system", system_prompt),
+            ("user", user_prompt),
+        ]
+    )
+    return str(response.content)
+
+
+def _invoke_image(chat_model, question: str, path: str | Path) -> str:
+    response = chat_model.invoke(
+        [
+            SystemMessage(content=GAIA_AGENT_SYSTEM_PROMPT),
+            HumanMessage(
+                content=[
+                    {"type": "text", "text": question},
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": image_data_url(path)},
+                    },
+                ]
+            ),
+        ]
+    )
+    return str(response.content)
+
+
+def _build_search_queries(chat_model, question: str) -> list[str]:
+    raw_queries = _invoke_text(
+        chat_model,
+        GAIA_QUERY_PROMPT,
+        f"Question:\n{question}",
+    )
+    queries = [
+        re.sub(r"^\s*[-*\d.)]+\s*", "", line).strip()
+        for line in raw_queries.splitlines()
+        if line.strip()
+    ]
+    queries = [query.strip("\"'") for query in queries if len(query.strip("\"'")) > 3]
+    if question not in queries:
+        queries.append(question)
+    return queries[:3]
+
+
+def _question_with_evidence(question: str, evidence: list[str]) -> str:
+    return f"Question:\n{question}\n\nEvidence:\n{_trim_evidence(evidence)}"
+
+
+def _trim_evidence(evidence: list[str]) -> str:
+    text = "\n\n---\n\n".join(evidence)
+    if len(text) <= MAX_EVIDENCE_CHARS:
+        return text
+    return f"{text[:MAX_EVIDENCE_CHARS]}\n\n[trimmed after {MAX_EVIDENCE_CHARS} chars]"
+
+
+def _looks_like_computation(question: str) -> bool:
+    markers = (
+        "given this table",
+        "provide the subset",
+        "counter-examples",
+        "not commutative",
+        "calculate",
+        "numeric output",
+    )
+    return any(marker in question for marker in markers)
+
+
+def _looks_like_direct(question: str) -> bool:
+    if question.count(" ") <= 8:
+        return True
+    if _looks_reversed(question):
+        return True
+    direct_markers = (
+        "grocery list",
+        "categorizing things",
+        "write the opposite",
+    )
+    return any(marker in question for marker in direct_markers)
+
+
+def _looks_reversed(question: str) -> bool:
+    words = re.findall(r"[a-z]{4,}", question)
+    if len(words) < 3:
+        return False
+    reversed_common = {"rewsna", "drow", "etirw", "ecnetnes", "dnatsrednu"}
+    return len(reversed_common.intersection(words)) >= 2

gaia_agent/prompts.py

@@ -5,6 +5,35 @@ list of numbers and/or strings.
 """.strip()
 
 
+GAIA_AGENT_SYSTEM_PROMPT = """
+You are a GAIA benchmark assistant.
+Answer real-world questions by using the provided evidence and tool output.
+
+Rules:
+- Give only the final answer, with no explanation.
+- Keep the answer as short as possible.
+- If the answer is numeric, do not include units unless the question explicitly asks for them.
+- If the answer is a list, use a comma-separated list.
+- Do not invent facts that are not supported by evidence.
+""".strip()
+
+
+GAIA_QUERY_PROMPT = """
+Create up to three concise web search queries that would help answer the GAIA
+question. Return only the queries, one per line, without numbering.
+""".strip()
+
+
+GAIA_VERIFY_PROMPT = """
+You are checking a GAIA answer before submission.
+Use the question, evidence, and draft answer to produce the final answer.
+If the draft answer was computed directly by a tool, preserve it unless the
+evidence clearly contradicts it.
+
+Return only the final answer. Do not include reasoning or a prefix.
+""".strip()
+
+
 DUMMY_LLM_TEST_PROMPT = """
 You are testing the LLM connection for a GAIA agent.
 Answer the user question directly in a few words.

gaia_agent/state.py

@@ -1,7 +1,16 @@
-from typing import TypedDict
+from typing import Any, TypedDict
 
 
 class GaiaState(TypedDict, total=False):
+    task_id: str
     question: str
+    file_name: str
+    file_path: str
+    level: str
+    task_type: str
+    evidence: list[str]
+    tool_outputs: list[dict[str, Any]]
     draft_answer: str
+    verified_answer: str
     final_answer: str
+    error: str

gaia_agent/tools/files.py

@@ -1 +1,91 @@
-"""File-reading tools will live here."""
+from __future__ import annotations
+
+import mimetypes
+from pathlib import Path
+from urllib.parse import urlparse
+
+import pandas as pd
+import requests
+
+
+DEFAULT_CACHE_DIR = Path(".gaia_cache") / "files"
+
+
+def download_task_file(
+    api_url: str,
+    task_id: str,
+    file_name: str | None = None,
+    *,
+    cache_dir: Path = DEFAULT_CACHE_DIR,
+    timeout: int = 60,
+) -> Path:
+    """Download the file associated with a GAIA task ID into a local cache."""
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    target_name = file_name or task_id
+    target_path = cache_dir / Path(target_name).name
+    if target_path.exists() and target_path.stat().st_size > 0:
+        return target_path
+
+    response = requests.get(
+        f"{api_url.rstrip('/')}/files/{task_id}",
+        timeout=timeout,
+    )
+    response.raise_for_status()
+
+    content_type = response.headers.get("content-type", "")
+    if "application/json" in content_type:
+        detail = response.json().get("detail", "unknown file download error")
+        raise FileNotFoundError(detail)
+
+    if not file_name:
+        suffix = _suffix_from_content_type(content_type)
+        target_path = cache_dir / f"{task_id}{suffix}"
+
+    target_path.write_bytes(response.content)
+    return target_path
+
+
+def read_text_file(path: str | Path, *, max_chars: int = 20_000) -> str:
+    text = Path(path).read_text(encoding="utf-8", errors="replace")
+    if len(text) <= max_chars:
+        return text
+    return f"{text[:max_chars]}\n\n[truncated after {max_chars} characters]"
+
+
+def summarize_spreadsheet(path: str | Path, *, max_rows: int = 12) -> str:
+    """Return a compact textual summary of all sheets in a spreadsheet."""
+    file_path = Path(path)
+    if file_path.suffix.lower() == ".csv":
+        workbook = {file_path.stem: pd.read_csv(file_path)}
+    else:
+        workbook = pd.read_excel(file_path, sheet_name=None)
+    sections: list[str] = []
+    for sheet_name, dataframe in workbook.items():
+        sections.append(f"Sheet: {sheet_name}")
+        sections.append(f"Shape: {dataframe.shape[0]} rows x {dataframe.shape[1]} columns")
+        sections.append(f"Columns: {', '.join(map(str, dataframe.columns))}")
+
+        numeric_sums = dataframe.select_dtypes(include="number").sum(numeric_only=True)
+        if not numeric_sums.empty:
+            sums = ", ".join(
+                f"{column}={value}" for column, value in numeric_sums.items()
+            )
+            sections.append(f"Numeric column sums: {sums}")
+
+        preview = dataframe.head(max_rows).to_csv(index=False)
+        sections.append(f"Preview CSV:\n{preview.strip()}")
+    return "\n\n".join(sections)
+
+
+def _suffix_from_content_type(content_type: str) -> str:
+    media_type = content_type.split(";", 1)[0].strip()
+    guessed = mimetypes.guess_extension(media_type)
+    if guessed:
+        return guessed
+
+    parsed = urlparse(media_type)
+    if parsed.path:
+        suffix = Path(parsed.path).suffix
+        if suffix:
+            return suffix
+    return ".bin"

gaia_agent/tools/media.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import base64
+import mimetypes
+import os
+from pathlib import Path
+
+
+def transcribe_audio_file(path: str | Path) -> str:
+    """Transcribe an audio file with LiteLLM's transcription API."""
+    model = os.getenv("AUDIO_TRANSCRIPTION_MODEL", "whisper-1")
+    api_key = os.getenv("AUDIO_TRANSCRIPTION_API_KEY") or os.getenv("OPENAI_API_KEY")
+    api_base = os.getenv("AUDIO_TRANSCRIPTION_API_BASE")
+
+    try:
+        import litellm
+    except ImportError as exc:
+        raise RuntimeError("litellm is required for audio transcription.") from exc
+
+    with Path(path).open("rb") as audio:
+        result = litellm.transcription(
+            model=model,
+            file=audio,
+            api_key=api_key,
+            api_base=api_base,
+            response_format="json",
+        )
+
+    if isinstance(result, str):
+        return result
+    if isinstance(result, dict):
+        return str(result.get("text", result))
+    return str(getattr(result, "text", result))
+
+
+def image_data_url(path: str | Path) -> str:
+    image_path = Path(path)
+    media_type = mimetypes.guess_type(image_path.name)[0] or "image/png"
+    payload = base64.b64encode(image_path.read_bytes()).decode("ascii")
+    return f"data:{media_type};base64,{payload}"

gaia_agent/tools/python_repl.py

@@ -1 +1,47 @@
-"""Python execution tools will live here."""
+from __future__ import annotations
+
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+
+def run_python_file(path: str | Path, *, timeout: int = 20) -> dict[str, str | int]:
+    file_path = Path(path)
+    return _run_python([str(file_path)], cwd=file_path.parent, timeout=timeout)
+
+
+def run_python_code(code: str, *, timeout: int = 20) -> dict[str, str | int]:
+    with tempfile.TemporaryDirectory(prefix="gaia-python-") as tmpdir:
+        file_path = Path(tmpdir) / "snippet.py"
+        file_path.write_text(code, encoding="utf-8")
+        return _run_python([str(file_path)], cwd=Path(tmpdir), timeout=timeout)
+
+
+def _run_python(
+    args: list[str],
+    *,
+    cwd: Path,
+    timeout: int,
+) -> dict[str, str | int]:
+    try:
+        completed = subprocess.run(
+            [sys.executable, *args],
+            cwd=cwd,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+            check=False,
+        )
+    except subprocess.TimeoutExpired as exc:
+        return {
+            "exit_code": 124,
+            "stdout": exc.stdout or "",
+            "stderr": f"Python execution timed out after {timeout}s.",
+        }
+
+    return {
+        "exit_code": completed.returncode,
+        "stdout": completed.stdout,
+        "stderr": completed.stderr,
+    }

gaia_agent/tools/search.py

@@ -1 +1,4 @@
-"""Search tools will live here."""
+from gaia_agent.tools.web import SearchResult, web_search
+
+
+__all__ = ["SearchResult", "web_search"]

gaia_agent/tools/web.py

@@ -1 +1,224 @@
-"""Web browsing tools will live here."""
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from html.parser import HTMLParser
+from typing import Iterable
+from urllib.parse import parse_qs, unquote, urlparse
+
+import requests
+
+
+USER_AGENT = (
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
+    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124 Safari/537.36"
+)
+
+
+@dataclass(frozen=True)
+class SearchResult:
+    title: str
+    url: str
+    snippet: str = ""
+
+
+def extract_urls(text: str) -> list[str]:
+    return re.findall(r"https?://[^\s)>\]]+", text)
+
+
+def fetch_url(url: str, *, timeout: int = 20, max_chars: int = 30_000) -> str:
+    response = requests.get(
+        url,
+        headers={"User-Agent": USER_AGENT},
+        timeout=timeout,
+    )
+    response.raise_for_status()
+    content_type = response.headers.get("content-type", "")
+    raw_text = response.text
+    if "html" in content_type:
+        raw_text = html_to_text(raw_text)
+
+    raw_text = normalize_whitespace(raw_text)
+    if len(raw_text) <= max_chars:
+        return raw_text
+    return f"{raw_text[:max_chars]}\n\n[truncated after {max_chars} characters]"
+
+
+def web_search(query: str, *, max_results: int = 5, timeout: int = 20) -> list[SearchResult]:
+    results = _duckduckgo_search(query, max_results=max_results, timeout=timeout)
+    if results:
+        return results[:max_results]
+    return _wikipedia_search(query, max_results=max_results, timeout=timeout)
+
+
+def get_youtube_transcript(url_or_id: str) -> str:
+    video_id = extract_youtube_id(url_or_id)
+    if not video_id:
+        raise ValueError(f"Could not extract a YouTube video id from {url_or_id!r}.")
+
+    try:
+        from youtube_transcript_api import YouTubeTranscriptApi
+    except ImportError as exc:
+        raise RuntimeError(
+            "youtube-transcript-api is not installed, so YouTube transcripts "
+            "cannot be fetched."
+        ) from exc
+
+    try:
+        transcript = YouTubeTranscriptApi.get_transcript(video_id)
+    except AttributeError:
+        transcript = YouTubeTranscriptApi().fetch(video_id).to_raw_data()
+
+    return "\n".join(
+        f"[{entry.get('start', 0):.1f}] {entry.get('text', '')}"
+        for entry in transcript
+    )
+
+
+def extract_youtube_id(url_or_id: str) -> str | None:
+    if re.fullmatch(r"[\w-]{11}", url_or_id):
+        return url_or_id
+
+    parsed = urlparse(url_or_id)
+    if parsed.hostname in {"youtu.be", "www.youtu.be"}:
+        return parsed.path.lstrip("/")[:11]
+    if parsed.hostname and "youtube.com" in parsed.hostname:
+        query_id = parse_qs(parsed.query).get("v", [None])[0]
+        if query_id:
+            return query_id[:11]
+        match = re.search(r"/(?:shorts|embed)/([\w-]{11})", parsed.path)
+        if match:
+            return match.group(1)
+    return None
+
+
+def html_to_text(html: str) -> str:
+    parser = _TextExtractor()
+    parser.feed(html)
+    return parser.text()
+
+
+def normalize_whitespace(text: str) -> str:
+    return re.sub(r"\s+", " ", text).strip()
+
+
+def _duckduckgo_search(
+    query: str,
+    *,
+    max_results: int,
+    timeout: int,
+) -> list[SearchResult]:
+    response = requests.get(
+        "https://duckduckgo.com/html/",
+        params={"q": query},
+        headers={"User-Agent": USER_AGENT},
+        timeout=timeout,
+    )
+    response.raise_for_status()
+    parser = _DuckDuckGoParser()
+    parser.feed(response.text)
+    return parser.results[:max_results]
+
+
+def _wikipedia_search(
+    query: str,
+    *,
+    max_results: int,
+    timeout: int,
+) -> list[SearchResult]:
+    response = requests.get(
+        "https://en.wikipedia.org/w/api.php",
+        params={
+            "action": "query",
+            "list": "search",
+            "srsearch": query,
+            "format": "json",
+            "srlimit": max_results,
+        },
+        headers={"User-Agent": USER_AGENT},
+        timeout=timeout,
+    )
+    response.raise_for_status()
+    payload = response.json()
+    results = []
+    for item in payload.get("query", {}).get("search", []):
+        title = item.get("title", "")
+        url_title = title.replace(" ", "_")
+        results.append(
+            SearchResult(
+                title=title,
+                url=f"https://en.wikipedia.org/wiki/{url_title}",
+                snippet=html_to_text(item.get("snippet", "")),
+            )
+        )
+    return results
+
+
+class _TextExtractor(HTMLParser):
+    def __init__(self) -> None:
+        super().__init__()
+        self._chunks: list[str] = []
+        self._skip_depth = 0
+
+    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        if tag in {"script", "style", "noscript", "svg"}:
+            self._skip_depth += 1
+        if tag in {"p", "br", "li", "tr", "h1", "h2", "h3", "h4"}:
+            self._chunks.append("\n")
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag in {"script", "style", "noscript", "svg"} and self._skip_depth:
+            self._skip_depth -= 1
+        if tag in {"p", "li", "tr"}:
+            self._chunks.append("\n")
+
+    def handle_data(self, data: str) -> None:
+        if not self._skip_depth:
+            self._chunks.append(data)
+
+    def text(self) -> str:
+        return "\n".join(
+            chunk.strip() for chunk in self._chunks if chunk and chunk.strip()
+        )
+
+
+class _DuckDuckGoParser(HTMLParser):
+    def __init__(self) -> None:
+        super().__init__()
+        self.results: list[SearchResult] = []
+        self._active_href: str | None = None
+        self._active_chunks: list[str] = []
+
+    def handle_starttag(self, tag: str, attrs: Iterable[tuple[str, str | None]]) -> None:
+        if tag != "a":
+            return
+        attr_map = {key: value or "" for key, value in attrs}
+        css_class = attr_map.get("class", "")
+        href = attr_map.get("href", "")
+        if "result__a" in css_class and href:
+            self._active_href = _unwrap_duckduckgo_url(href)
+            self._active_chunks = []
+
+    def handle_data(self, data: str) -> None:
+        if self._active_href:
+            self._active_chunks.append(data)
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag != "a" or not self._active_href:
+            return
+        title = normalize_whitespace(" ".join(self._active_chunks))
+        if title and self._active_href.startswith("http"):
+            self.results.append(SearchResult(title=title, url=self._active_href))
+        self._active_href = None
+        self._active_chunks = []
+
+
+def _unwrap_duckduckgo_url(url: str) -> str:
+    if url.startswith("//"):
+        url = f"https:{url}"
+    parsed = urlparse(url)
+    if "duckduckgo.com" in parsed.netloc:
+        uddg = parse_qs(parsed.query).get("uddg", [None])[0]
+        if uddg:
+            return unquote(uddg)
+    return url

pyproject.toml

@@ -8,11 +8,13 @@ dependencies = [
     "gradio[oauth]==5.25.2",
     "requests>=2.32.0",
     "pandas>=2.2.0",
+    "openpyxl>=3.1.0",
     "python-dotenv>=1.0.1",
     "langchain>=0.3.0",
     "langgraph>=0.2.60",
     "langfuse>=2.57.0",
     "langchain-litellm>=0.6.4",
+    "youtube-transcript-api>=0.6.2",
 ]
 
 [build-system]

requirements.txt

@@ -1,8 +1,11 @@
 gradio[oauth]==5.25.2
 requests>=2.32.0
 pandas>=2.2.0
+openpyxl>=3.1.0
 python-dotenv>=1.0.1
 langchain>=0.3.0
 langchain-openai>=0.3.0
 langgraph>=0.2.60
 langfuse>=2.57.0
+langchain-litellm>=0.6.4
+youtube-transcript-api>=0.6.2

uv.lock

@@ -506,6 +506,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/20/2a/1b016902351a523aa2bd446b50a5bc1175d7a7d1cf90fe2ef904f9b84ebc/cryptography-46.0.7-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:258514877e15963bd43b558917bc9f54cf7cf866c38aa576ebf47a77ddbc43a4", size = 3412829, upload-time = "2026-04-08T01:57:48.874Z" },
 ]
 
+[[package]]
+name = "defusedxml"
+version = "0.7.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/0f/d5/c66da9b79e5bdb124974bfe172b4daf3c984ebd9c2a06e2b8a4dc7331c72/defusedxml-0.7.1.tar.gz", hash = "sha256:1bb3032db185915b62d7c6209c5a8792be6a32ab2fedacc84e01b52c51aa3e69", size = 75520, upload-time = "2021-03-08T10:59:26.269Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/07/6c/aa3f2f849e01cb6a001cd8554a88d4c77c5c1a31c95bdf1cf9301e6d9ef4/defusedxml-0.7.1-py2.py3-none-any.whl", hash = "sha256:a352e7e428770286cc899e2542b6cdaedb2b4953ff269a210103ec58f6198a61", size = 25604, upload-time = "2021-03-08T10:59:24.45Z" },
+]
+
 [[package]]
 name = "distro"
 version = "1.9.0"
@@ -515,6 +524,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/12/b3/231ffd4ab1fc9d679809f356cebee130ac7daa00d6d6f3206dd4fd137e9e/distro-1.9.0-py3-none-any.whl", hash = "sha256:7bffd925d65168f85027d8da9af6bddab658135b840670a223589bc0c8ef02b2", size = 20277, upload-time = "2023-12-24T09:54:30.421Z" },
 ]
 
+[[package]]
+name = "et-xmlfile"
+version = "2.0.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/d3/38/af70d7ab1ae9d4da450eeec1fa3918940a5fafb9055e934af8d6eb0c2313/et_xmlfile-2.0.0.tar.gz", hash = "sha256:dab3f4764309081ce75662649be815c4c9081e88f0837825f90fd28317d4da54", size = 17234, upload-time = "2024-10-25T17:25:40.039Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/c1/8b/5fe2cc11fee489817272089c4203e679c63b570a5aaeb18d852ae3cbba6a/et_xmlfile-2.0.0-py3-none-any.whl", hash = "sha256:7a91720bc756843502c3b7504c77b8fe44217c85c537d85037f0f536151b2caa", size = 18059, upload-time = "2024-10-25T17:25:39.051Z" },
+]
+
 [[package]]
 name = "fastapi"
 version = "0.136.1"
@@ -725,9 +743,11 @@ dependencies = [
     { name = "langchain-litellm" },
     { name = "langfuse" },
     { name = "langgraph" },
+    { name = "openpyxl" },
     { name = "pandas" },
     { name = "python-dotenv" },
     { name = "requests" },
+    { name = "youtube-transcript-api" },
 ]
 
 [package.dev-dependencies]
@@ -742,9 +762,11 @@ requires-dist = [
     { name = "langchain-litellm", specifier = ">=0.6.4" },
     { name = "langfuse", specifier = ">=2.57.0" },
     { name = "langgraph", specifier = ">=0.2.60" },
+    { name = "openpyxl", specifier = ">=3.1.0" },
     { name = "pandas", specifier = ">=2.2.0" },
     { name = "python-dotenv", specifier = ">=1.0.1" },
     { name = "requests", specifier = ">=2.32.0" },
+    { name = "youtube-transcript-api", specifier = ">=0.6.2" },
 ]
 
 [package.metadata.requires-dev]
@@ -1613,6 +1635,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/f2/40/f090499f10514515081d09cb9da09f25b821eb20497e9423afe4f07b4ecf/openai-2.34.0-py3-none-any.whl", hash = "sha256:c996a71b1a210f3569844572ad4c609307e978515fb76877cf449b72596e549e", size = 1316535, upload-time = "2026-05-04T17:34:06.773Z" },
 ]
 
+[[package]]
+name = "openpyxl"
+version = "3.1.5"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "et-xmlfile" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/3d/f9/88d94a75de065ea32619465d2f77b29a0469500e99012523b91cc4141cd1/openpyxl-3.1.5.tar.gz", hash = "sha256:cf0e3cf56142039133628b5acffe8ef0c12bc902d2aadd3e0fe5878dc08d1050", size = 186464, upload-time = "2024-06-28T14:03:44.161Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/c0/da/977ded879c29cbd04de313843e76868e6e13408a94ed6b987245dc7c8506/openpyxl-3.1.5-py2.py3-none-any.whl", hash = "sha256:5282c12b107bffeef825f4617dc029afaf41d0ea60823bbb665ef3079dc79de2", size = 250910, upload-time = "2024-06-28T14:03:41.161Z" },
+]
+
 [[package]]
 name = "opentelemetry-api"
 version = "1.41.1"
@@ -3218,6 +3252,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/69/68/c8739671f5699c7dc470580a4f821ef37c32c4cb0b047ce223a7f115757f/yarl-1.23.0-py3-none-any.whl", hash = "sha256:a2df6afe50dea8ae15fa34c9f824a3ee958d785fd5d089063d960bae1daa0a3f", size = 48288, upload-time = "2026-03-01T22:07:51.388Z" },
 ]
 
+[[package]]
+name = "youtube-transcript-api"
+version = "1.2.4"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "defusedxml" },
+    { name = "requests" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/60/43/4104185a2eaa839daa693b30e15c37e7e58795e8e09ec414f22b3db54bec/youtube_transcript_api-1.2.4.tar.gz", hash = "sha256:b72d0e96a335df599d67cee51d49e143cff4f45b84bcafc202ff51291603ddcd", size = 469839, upload-time = "2026-01-29T09:09:17.088Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/be/95/129ea37efd6cd6ed00f62baae6543345c677810b8a3bf0026756e1d3cf3c/youtube_transcript_api-1.2.4-py3-none-any.whl", hash = "sha256:03878759356da5caf5edac77431780b91448fb3d8c21d4496015bdc8a7bc43ff", size = 485227, upload-time = "2026-01-29T09:09:15.427Z" },
+]
+
 [[package]]
 name = "zipp"
 version = "3.23.1"