GAIA LangGraph agent: Groq + Tavily + LLM-as-judge

Replace BasicAgent with a LangGraph multi-node graph (planner, research+tools, solver, LLM-as-judge eval loop, formatter). Pydantic node-to-node contracts.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.env.example

@@ -0,0 +1,21 @@
+# --- Required secrets (fill these in) ---
+GROQ_API_KEY=
+TAVILY_API_KEY=
+
+# --- Optional overrides (sane defaults applied if omitted) ---
+# Groq text/reasoning model used by the agent + judge nodes
+GROQ_TEXT_MODEL=llama-3.3-70b-versatile
+# Groq multimodal model used by the describe_image tool
+GROQ_VISION_MODEL=meta-llama/llama-4-scout-17b-16e-instruct
+# Groq speech-to-text model used by the transcribe_audio tool
+GROQ_WHISPER_MODEL=whisper-large-v3
+
+# GAIA scoring API base URL
+GAIA_API_URL=https://agents-course-unit4-scoring.hf.space
+
+# How many times the judge may bounce a wrong answer back to research
+MAX_JUDGE_RETRIES=2
+# Hard cap on the LangGraph step budget per question
+RECURSION_LIMIT=40
+# Per-question wall-clock timeout (seconds); agent returns best-effort on expiry
+QUESTION_TIMEOUT=180

.gitignore

@@ -0,0 +1,163 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/latest/usage/project/#working-with-version-control
+.pdm.toml
+.pdm-python
+.pdm-build/
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+.langgraph_api/

README.md

@@ -12,4 +12,45 @@ hf_oauth: true
 hf_oauth_expiration_minutes: 480
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+
+---
+
+## GAIA Agent (LangGraph + Groq)
+
+This Space answers GAIA Level-1 questions with a LangGraph multi-node agent:
+
+```
+planner → [ingest_file] → research ⇄ tools → evidence → solver → judge ⇄ (loop) → formatter
+```
+
+- **LLM:** Groq (`langchain-groq`) — text/reasoning + Llama-4 vision + Whisper audio.
+- **Search:** Tavily (`langchain-tavily`) with a Wikipedia fallback.
+- **Eval loop:** an LLM-as-judge node loops wrong/ill-formatted answers back to research
+  (up to `MAX_JUDGE_RETRIES`).
+- **Structured handoffs:** every node-to-node transfer is a Pydantic model
+  (`gaia_agent/schemas.py`), so context stays precise.
+
+### Secrets (required)
+
+Set these as **Space secrets** (Settings → Variables and secrets) — locally use a `.env`
+(copy `.env.example`):
+
+| Name | Purpose |
+|------|---------|
+| `GROQ_API_KEY` | Groq LLM access |
+| `TAVILY_API_KEY` | Tavily web search |
+
+Optional overrides: `GROQ_TEXT_MODEL`, `GROQ_VISION_MODEL`, `GROQ_WHISPER_MODEL`,
+`MAX_JUDGE_RETRIES`, `RECURSION_LIMIT`. See `.env.example`.
+
+### Local development
+
+```bash
+uv venv --python 3.11
+uv pip install -e ".[dev]"
+cp .env.example .env          # then fill in your keys
+uv run pytest -q              # unit + graph/judge-loop tests
+uv run langgraph dev          # LangGraph Studio: visualise the graph
+python scripts/dry_run.py 3   # run the agent on 3 questions (no submit)
+```

app.py

@@ -3,21 +3,18 @@ import gradio as gr
 import requests
 import inspect
 import pandas as pd
+from dotenv import load_dotenv
+
+# Load secrets from a local .env when running outside an HF Space.
+load_dotenv()
 
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
-# --- Basic Agent Definition ---
-# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
-class BasicAgent:
-    def __init__(self):
-        print("BasicAgent initialized.")
-    def __call__(self, question: str) -> str:
-        print(f"Agent received question (first 50 chars): {question[:50]}...")
-        fixed_answer = "This is a default answer."
-        print(f"Agent returning fixed answer: {fixed_answer}")
-        return fixed_answer
+# --- Agent Definition ---
+# The real agent is a LangGraph multi-node graph (Groq + Tavily + LLM-as-judge).
+from gaia_agent import GaiaAgent
 
 def run_and_submit_all( profile: gr.OAuthProfile | None):
     """
@@ -40,7 +37,7 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
 
     # 1. Instantiate Agent ( modify this part to create your agent)
     try:
-        agent = BasicAgent()
+        agent = GaiaAgent()
     except Exception as e:
         print(f"Error instantiating agent: {e}")
         return f"Error initializing agent: {e}", None
@@ -80,7 +77,7 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
             print(f"Skipping item with missing task_id or question: {item}")
             continue
         try:
-            submitted_answer = agent(question_text)
+            submitted_answer = agent(question_text, 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:

gaia_agent/__init__.py

@@ -0,0 +1,5 @@
+"""LangGraph multi-agent for the HF Agents Course GAIA final assignment."""
+
+from gaia_agent.agent import GaiaAgent
+
+__all__ = ["GaiaAgent"]

gaia_agent/agent.py

@@ -0,0 +1,32 @@
+"""Public agent wrapper used by app.py."""
+
+from __future__ import annotations
+
+from gaia_agent.config import get_settings
+from gaia_agent.graph import graph
+
+
+class GaiaAgent:
+    """Callable wrapper around the compiled LangGraph graph.
+
+    ``app.py`` invokes ``agent(question, task_id)`` per question and submits the
+    returned string for exact-match scoring.
+    """
+
+    def __init__(self):
+        self._graph = graph
+        self._settings = get_settings()
+        print("GaiaAgent initialised (LangGraph + Groq).")
+
+    def __call__(self, question: str, task_id: str = "") -> str:
+        """Run the graph on one question and return the bare answer string."""
+        try:
+            result = self._graph.invoke(
+                {"question": question, "task_id": task_id},
+                config={"recursion_limit": self._settings.recursion_limit},
+            )
+            answer = (result.get("final_answer") or "").strip()
+            return answer or "Unable to determine an answer."
+        except Exception as exc:  # noqa: BLE001
+            print(f"GaiaAgent error on task {task_id}: {exc}")
+            return "Unable to determine an answer."

gaia_agent/config.py

@@ -0,0 +1,40 @@
+"""Centralised configuration loaded from environment / .env.
+
+All tunables live here so swapping a (renamed) Groq model or the scoring URL is a
+config change, not a code change.
+"""
+
+from __future__ import annotations
+
+from functools import lru_cache
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    """Runtime settings sourced from environment variables / `.env`."""
+
+    model_config = SettingsConfigDict(
+        env_file=".env", env_file_encoding="utf-8", extra="ignore"
+    )
+
+    # --- Secrets ---
+    groq_api_key: str = ""
+    tavily_api_key: str = ""
+
+    # --- Models (override via env) ---
+    groq_text_model: str = "llama-3.3-70b-versatile"
+    groq_vision_model: str = "meta-llama/llama-4-scout-17b-16e-instruct"
+    groq_whisper_model: str = "whisper-large-v3"
+
+    # --- API + control knobs ---
+    gaia_api_url: str = "https://agents-course-unit4-scoring.hf.space"
+    max_judge_retries: int = 2
+    recursion_limit: int = 40
+    question_timeout: int = 180
+
+
+@lru_cache(maxsize=1)
+def get_settings() -> Settings:
+    """Return a cached Settings instance (read once per process)."""
+    return Settings()

gaia_agent/graph.py

@@ -0,0 +1,61 @@
+"""Assemble and compile the GAIA agent StateGraph.
+
+The module-level ``graph`` is what ``langgraph.json`` points at, so ``langgraph dev``
+can render and step through it in LangGraph Studio.
+"""
+
+from __future__ import annotations
+
+from langgraph.graph import END, START, StateGraph
+from langgraph.prebuilt import ToolNode
+
+from gaia_agent.nodes import (
+    evidence,
+    formatter,
+    ingest_file,
+    judge,
+    planner,
+    research,
+    route_after_judge,
+    route_after_planner,
+    route_after_research,
+    solver,
+)
+from gaia_agent.state import GraphState
+from gaia_agent.tools import RESEARCH_TOOLS
+
+
+def build_graph():
+    """Build and compile the GAIA agent graph."""
+    g = StateGraph(GraphState)
+
+    g.add_node("planner", planner)
+    g.add_node("ingest_file", ingest_file)
+    g.add_node("research", research)
+    g.add_node("tools", ToolNode(RESEARCH_TOOLS))
+    g.add_node("evidence", evidence)
+    g.add_node("solver", solver)
+    g.add_node("judge", judge)
+    g.add_node("formatter", formatter)
+
+    g.add_edge(START, "planner")
+    g.add_conditional_edges(
+        "planner", route_after_planner, {"ingest_file": "ingest_file", "research": "research"}
+    )
+    g.add_edge("ingest_file", "research")
+    g.add_conditional_edges(
+        "research", route_after_research, {"tools": "tools", "evidence": "evidence"}
+    )
+    g.add_edge("tools", "research")
+    g.add_edge("evidence", "solver")
+    g.add_edge("solver", "judge")
+    g.add_conditional_edges(
+        "judge", route_after_judge, {"research": "research", "formatter": "formatter"}
+    )
+    g.add_edge("formatter", END)
+
+    return g.compile(name="GAIA Agent")
+
+
+# Exposed for langgraph.json and direct imports.
+graph = build_graph()

gaia_agent/llm.py

@@ -0,0 +1,33 @@
+"""Groq chat-model factories."""
+
+from __future__ import annotations
+
+from functools import lru_cache
+
+from langchain_groq import ChatGroq
+
+from gaia_agent.config import get_settings
+
+
+@lru_cache(maxsize=4)
+def get_text_llm(temperature: float = 0.0) -> ChatGroq:
+    """Return the Groq text/reasoning model used by most nodes."""
+    s = get_settings()
+    return ChatGroq(
+        model=s.groq_text_model,
+        api_key=s.groq_api_key,
+        temperature=temperature,
+        max_retries=2,
+    )
+
+
+@lru_cache(maxsize=1)
+def get_vision_llm() -> ChatGroq:
+    """Return the Groq multimodal model used for image understanding."""
+    s = get_settings()
+    return ChatGroq(
+        model=s.groq_vision_model,
+        api_key=s.groq_api_key,
+        temperature=0.0,
+        max_retries=2,
+    )

gaia_agent/nodes.py

@@ -0,0 +1,221 @@
+"""Graph nodes and routing functions for the GAIA agent.
+
+Flow: planner -> [ingest_file] -> research <-> tools -> evidence -> solver
+      -> judge -> (research on REVISE within budget | formatter) -> END
+"""
+
+from __future__ import annotations
+
+from langchain_core.messages import HumanMessage, SystemMessage
+
+from gaia_agent.config import get_settings
+from gaia_agent.llm import get_text_llm
+from gaia_agent.prompts import (
+    EVIDENCE_PROMPT,
+    FORMATTER_PROMPT,
+    GAIA_RULES,
+    JUDGE_PROMPT,
+    PLANNER_PROMPT,
+    RESEARCH_PROMPT,
+    SOLVER_PROMPT,
+)
+from gaia_agent.schemas import (
+    Candidate,
+    Evidence,
+    FileExtract,
+    FinalAnswer,
+    JudgeVerdict,
+    Plan,
+)
+from gaia_agent.state import GraphState
+from gaia_agent.tools import (
+    RESEARCH_TOOLS,
+    classify_file,
+    describe_image,
+    fetch_task_file,
+    read_spreadsheet,
+    read_text_file,
+    transcribe_audio,
+)
+
+# --------------------------------------------------------------------------- #
+# Planner
+# --------------------------------------------------------------------------- #
+
+
+def planner(state: GraphState) -> dict:
+    """Produce a structured Plan and initialise loop bookkeeping."""
+    try:
+        llm = get_text_llm().with_structured_output(Plan)
+        plan = llm.invoke(
+            PLANNER_PROMPT.format(question=state["question"], task_id=state.get("task_id", ""))
+        )
+    except Exception:  # noqa: BLE001
+        plan = Plan(needs_file=False, reasoning="planner-fallback")
+    return {"plan": plan, "attempts": 0, "context_notes": ""}
+
+
+def route_after_planner(state: GraphState) -> str:
+    """Go fetch the file only if the plan says one is needed."""
+    plan = state.get("plan")
+    return "ingest_file" if (plan and plan.needs_file) else "research"
+
+
+# --------------------------------------------------------------------------- #
+# File ingestion
+# --------------------------------------------------------------------------- #
+
+
+def ingest_file(state: GraphState) -> dict:
+    """Download and extract the task's attached file into context_notes."""
+    task_id = state.get("task_id", "")
+    path = fetch_task_file(task_id)
+    if not path:
+        return {"file_extract": None}
+
+    kind = classify_file(path)
+    question = state["question"]
+    try:
+        if kind == "spreadsheet":
+            extracted = read_spreadsheet.invoke({"path": path})
+        elif kind == "audio":
+            extracted = transcribe_audio.invoke({"path": path})
+        elif kind == "image":
+            extracted = describe_image.invoke({"path": path, "question": question})
+        elif kind in ("text", "code"):
+            extracted = read_text_file.invoke({"path": path})
+        else:
+            extracted = f"File downloaded to {path} (kind={kind}); no extractor available."
+    except Exception as exc:  # noqa: BLE001
+        extracted = f"File extraction failed: {exc}"
+
+    fe = FileExtract(kind=kind, summary=f"Attached {kind} file.", extracted=str(extracted)[:8000])
+    notes = (state.get("context_notes", "") + f"\n[FILE:{kind}]\n{fe.extracted}").strip()
+    return {"file_extract": fe, "context_notes": notes}
+
+
+# --------------------------------------------------------------------------- #
+# Research (ReAct tool-calling loop)
+# --------------------------------------------------------------------------- #
+
+
+def research(state: GraphState) -> dict:
+    """Run one step of the research tool-calling loop."""
+    llm = get_text_llm().bind_tools(RESEARCH_TOOLS)
+    if not state.get("messages"):
+        system = RESEARCH_PROMPT.format(
+            rules=GAIA_RULES,
+            question=state["question"],
+            context_notes=state.get("context_notes", "") or "(none)",
+        )
+        seed = [
+            SystemMessage(system),
+            HumanMessage(
+                f"Task id: {state.get('task_id', '')}. Research and gather the facts "
+                "needed to answer precisely. Call tools as needed; stop when confident."
+            ),
+        ]
+        ai = llm.invoke(seed)
+        return {"messages": seed + [ai]}
+    ai = llm.invoke(state["messages"])
+    return {"messages": [ai]}
+
+
+def route_after_research(state: GraphState) -> str:
+    """Route to tools if the model requested any, else summarise evidence."""
+    last = state["messages"][-1]
+    if getattr(last, "tool_calls", None):
+        return "tools"
+    return "evidence"
+
+
+# --------------------------------------------------------------------------- #
+# Evidence / Solver / Judge / Formatter
+# --------------------------------------------------------------------------- #
+
+
+def evidence(state: GraphState) -> dict:
+    """Distil the research conversation into structured Evidence."""
+    try:
+        llm = get_text_llm().with_structured_output(Evidence)
+        ev = llm.invoke([SystemMessage(EVIDENCE_PROMPT), *state.get("messages", [])])
+    except Exception:  # noqa: BLE001
+        ev = Evidence(findings=[], sources=[], confidence=0.0)
+    return {"evidence": ev}
+
+
+def solver(state: GraphState) -> dict:
+    """Synthesize evidence + context into a single Candidate answer."""
+    ev = state.get("evidence")
+    try:
+        llm = get_text_llm().with_structured_output(Candidate)
+        cand = llm.invoke(
+            SOLVER_PROMPT.format(
+                rules=GAIA_RULES,
+                question=state["question"],
+                evidence=ev.model_dump() if ev else "{}",
+                context_notes=state.get("context_notes", "") or "(none)",
+            )
+        )
+    except Exception:  # noqa: BLE001
+        cand = Candidate(answer="", justification="solver-fallback")
+    return {"candidate": cand}
+
+
+def judge(state: GraphState) -> dict:
+    """LLM-as-judge: PASS, or REVISE with feedback fed back into research."""
+    cand = state.get("candidate")
+    ev = state.get("evidence")
+    try:
+        llm = get_text_llm().with_structured_output(JudgeVerdict)
+        verdict = llm.invoke(
+            JUDGE_PROMPT.format(
+                rules=GAIA_RULES,
+                question=state["question"],
+                evidence=ev.model_dump() if ev else "{}",
+                candidate=cand.answer if cand else "",
+            )
+        )
+    except Exception:  # noqa: BLE001
+        verdict = JudgeVerdict(verdict="PASS", feedback="judge-fallback")
+
+    updates: dict = {"verdict": verdict}
+    if verdict.verdict == "REVISE":
+        attempts = state.get("attempts", 0) + 1
+        updates["attempts"] = attempts
+        feedback = (
+            f"Judge requested a revision. Feedback: {verdict.feedback}. "
+            f"Gaps to close: {verdict.missing}. Re-research as needed and improve the answer."
+        )
+        updates["messages"] = [HumanMessage(feedback)]
+        updates["context_notes"] = (
+            state.get("context_notes", "") + f"\n[JUDGE] {verdict.feedback}"
+        ).strip()
+    return updates
+
+
+def route_after_judge(state: GraphState) -> str:
+    """Loop back to research on REVISE while within the retry budget."""
+    verdict = state.get("verdict")
+    if verdict and verdict.verdict == "PASS":
+        return "formatter"
+    if state.get("attempts", 0) >= get_settings().max_judge_retries:
+        return "formatter"
+    return "research"
+
+
+def formatter(state: GraphState) -> dict:
+    """Render the candidate into the bare exact-match answer string."""
+    cand = state.get("candidate")
+    candidate_text = cand.answer if cand else ""
+    try:
+        llm = get_text_llm().with_structured_output(FinalAnswer)
+        final = llm.invoke(
+            FORMATTER_PROMPT.format(
+                rules=GAIA_RULES, question=state["question"], candidate=candidate_text
+            )
+        )
+        answer = final.answer
+    except Exception:  # noqa: BLE001
+        answer = candidate_text
+    return {"final_answer": (answer or candidate_text or "").strip()}

gaia_agent/prompts.py

@@ -0,0 +1,99 @@
+"""System prompts. GAIA exact-match rules are centralised here."""
+
+# Shared rules every node should respect.
+GAIA_RULES = """\
+You are a careful research agent solving GAIA benchmark questions. Answers are scored
+by EXACT STRING MATCH, so precision of the final token matters more than explanation.
+
+Final-answer formatting rules:
+- Output ONLY the answer. Never write the words "FINAL ANSWER".
+- A number: no thousands separators, no units (unless the question asks for units),
+  no '$' or '%' unless explicitly requested. Write digits, e.g. 1024 not "1,024".
+- A string: as few words as possible; do not use articles or abbreviations unless the
+  question demands them; spell out digits in words only if asked.
+- A comma-separated list: apply the number/string rules to each element, one space
+  after each comma.
+- Do not add trailing punctuation, quotes, or commentary.
+"""
+
+PLANNER_PROMPT = """\
+You are the PLANNER. Read the question and decide how to solve it.
+Determine whether an attached file is needed, guess its kind, list concrete subtasks,
+and name the tools likely required (tavily_search, wikipedia_search, read_spreadsheet,
+transcribe_audio, describe_image, read_text_file, python_repl).
+Be concise. Do not answer the question yet.
+
+Question:
+{question}
+
+A file MAY be attached to this task (task_id={task_id}). If the question references a
+file, attachment, image, audio, spreadsheet, table, or document, set needs_file=true.
+"""
+
+RESEARCH_PROMPT = """\
+You are the RESEARCHER. Gather the facts needed to answer the question precisely.
+Use the available tools. Prefer authoritative sources. Do exact arithmetic / string
+work with python_repl rather than guessing. Stop calling tools once you have enough.
+
+{rules}
+
+Question:
+{question}
+
+Known context (from file extraction and prior judge feedback):
+{context_notes}
+"""
+
+EVIDENCE_PROMPT = """\
+Summarise what the research established into structured evidence: the concrete findings,
+their sources, and your confidence (0-1) that they pin down the answer.
+"""
+
+SOLVER_PROMPT = """\
+You are the SOLVER. Using the evidence and context below, produce ONE candidate answer.
+Apply the GAIA formatting rules to your candidate, but you may keep a short justification.
+
+{rules}
+
+Question:
+{question}
+
+Evidence:
+{evidence}
+
+Context notes:
+{context_notes}
+"""
+
+JUDGE_PROMPT = """\
+You are the JUDGE (LLM-as-a-judge). Decide whether the candidate answer correctly and
+completely answers the question, and whether it obeys the GAIA exact-match formatting
+rules. Be strict: a near-miss in format is a REVISE.
+
+{rules}
+
+Question:
+{question}
+
+Evidence:
+{evidence}
+
+Candidate answer:
+{candidate}
+
+If correct AND well-formatted -> verdict PASS.
+Otherwise -> verdict REVISE, with specific feedback and the gaps to close.
+"""
+
+FORMATTER_PROMPT = """\
+You are the FORMATTER. Convert the candidate answer into the FINAL exact-match string.
+Output strictly the answer per the rules below — nothing else.
+
+{rules}
+
+Question:
+{question}
+
+Candidate answer:
+{candidate}
+"""

gaia_agent/schemas.py

@@ -0,0 +1,85 @@
+"""Pydantic contracts for node-to-node handoffs.
+
+Every agent->agent transfer is a typed model produced via
+``llm.with_structured_output(Model)``, so the context flowing between graph nodes
+stays precise instead of being re-parsed from free text.
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+FileKind = Literal["spreadsheet", "audio", "image", "text", "code", "pdf", "other"]
+
+
+class Plan(BaseModel):
+    """Planner output: how to approach a single GAIA question."""
+
+    needs_file: bool = Field(
+        description="True if answering requires the attached task file."
+    )
+    file_expected_kind: FileKind | None = Field(
+        default=None, description="Best guess of the attached file kind, if any."
+    )
+    subtasks: list[str] = Field(
+        default_factory=list, description="Ordered steps to solve the question."
+    )
+    tools_hint: list[str] = Field(
+        default_factory=list,
+        description="Tool names likely needed, e.g. tavily_search, python_repl.",
+    )
+    reasoning: str = Field(default="", description="Short rationale for the plan.")
+
+
+class FileExtract(BaseModel):
+    """Structured result of ingesting an attached file."""
+
+    kind: FileKind
+    summary: str = Field(description="One-paragraph summary of the file contents.")
+    extracted: str = Field(
+        description="Raw-ish extracted content (transcript, table, OCR/description)."
+    )
+
+
+class Evidence(BaseModel):
+    """Research output: facts gathered to answer the question."""
+
+    findings: list[str] = Field(
+        default_factory=list, description="Concrete facts relevant to the answer."
+    )
+    sources: list[str] = Field(
+        default_factory=list, description="URLs or tool names backing the findings."
+    )
+    confidence: float = Field(
+        default=0.0, ge=0.0, le=1.0, description="0-1 confidence in the evidence."
+    )
+
+
+class Candidate(BaseModel):
+    """Solver output: a single candidate answer with rationale."""
+
+    answer: str = Field(description="Candidate answer (pre-formatting).")
+    justification: str = Field(default="", description="Why this answer follows.")
+    assumptions: list[str] = Field(default_factory=list)
+
+
+class JudgeVerdict(BaseModel):
+    """LLM-as-judge verdict that drives the evaluation loop."""
+
+    verdict: Literal["PASS", "REVISE"]
+    feedback: str = Field(
+        default="", description="What is wrong / what to fix on REVISE."
+    )
+    missing: list[str] = Field(
+        default_factory=list, description="Specific gaps to close on REVISE."
+    )
+
+
+class FinalAnswer(BaseModel):
+    """Formatter output: the exact-match string submitted to the API."""
+
+    answer: str = Field(
+        description="Bare answer for exact-match scoring; no prose, no 'FINAL ANSWER'."
+    )

gaia_agent/state.py

@@ -0,0 +1,47 @@
+"""Graph state shared across nodes."""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+from langgraph.graph.message import add_messages
+from typing_extensions import TypedDict
+
+from gaia_agent.schemas import (
+    Candidate,
+    Evidence,
+    FileExtract,
+    JudgeVerdict,
+    Plan,
+)
+
+
+class GraphState(TypedDict, total=False):
+    """State threaded through the GAIA agent graph.
+
+    Structured node outputs (Plan/Evidence/Candidate/JudgeVerdict) live here so each
+    node reads the prior node's typed object rather than re-parsing raw messages.
+    """
+
+    # Inputs
+    task_id: str
+    question: str
+
+    # ReAct scratchpad for the research tool-calling loop
+    messages: Annotated[list, add_messages]
+
+    # Structured handoffs
+    plan: Plan
+    file_extract: FileExtract | None
+    evidence: Evidence
+    candidate: Candidate
+    verdict: JudgeVerdict
+
+    # Free-form accumulated context (file extracts + judge feedback)
+    context_notes: str
+
+    # Eval-loop bookkeeping
+    attempts: int
+
+    # Output
+    final_answer: str

gaia_agent/tools/__init__.py

@@ -0,0 +1,38 @@
+"""Tool registry. ``RESEARCH_TOOLS`` is the list bound to the research agent."""
+
+from gaia_agent.tools.files import (
+    classify_file,
+    download_task_file,
+    fetch_task_file,
+    read_spreadsheet,
+    read_text_file,
+)
+from gaia_agent.tools.media import describe_image, transcribe_audio
+from gaia_agent.tools.python_tool import python_repl
+from gaia_agent.tools.search import tavily_search, wikipedia_search
+
+# Tools the research ReAct loop may call.
+RESEARCH_TOOLS = [
+    tavily_search,
+    wikipedia_search,
+    read_spreadsheet,
+    read_text_file,
+    transcribe_audio,
+    describe_image,
+    python_repl,
+    download_task_file,
+]
+
+__all__ = [
+    "RESEARCH_TOOLS",
+    "classify_file",
+    "fetch_task_file",
+    "download_task_file",
+    "read_spreadsheet",
+    "read_text_file",
+    "transcribe_audio",
+    "describe_image",
+    "python_repl",
+    "tavily_search",
+    "wikipedia_search",
+]

gaia_agent/tools/files.py

@@ -0,0 +1,127 @@
+"""File tools: download a task's attachment and read spreadsheets / text."""
+
+from __future__ import annotations
+
+import mimetypes
+import os
+import tempfile
+
+import requests
+from langchain_core.tools import tool
+
+from gaia_agent.config import get_settings
+from gaia_agent.schemas import FileKind
+
+_DOWNLOAD_DIR = os.path.join(tempfile.gettempdir(), "gaia_files")
+os.makedirs(_DOWNLOAD_DIR, exist_ok=True)
+
+# Extension -> logical kind used by the router / readers.
+_EXT_KIND: dict[str, FileKind] = {
+    ".xlsx": "spreadsheet",
+    ".xls": "spreadsheet",
+    ".csv": "spreadsheet",
+    ".mp3": "audio",
+    ".wav": "audio",
+    ".m4a": "audio",
+    ".flac": "audio",
+    ".ogg": "audio",
+    ".png": "image",
+    ".jpg": "image",
+    ".jpeg": "image",
+    ".webp": "image",
+    ".gif": "image",
+    ".pdf": "pdf",
+    ".py": "code",
+    ".json": "code",
+    ".txt": "text",
+    ".md": "text",
+}
+
+
+def classify_file(path: str) -> FileKind:
+    """Map a local file path to a logical FileKind via its extension."""
+    _, ext = os.path.splitext(path.lower())
+    return _EXT_KIND.get(ext, "other")
+
+
+def fetch_task_file(task_id: str) -> str | None:
+    """Download ``/files/{task_id}`` to a temp path. Return path or None if absent.
+
+    Used directly by the ingest_file node (not only as an LLM tool).
+    """
+    s = get_settings()
+    url = f"{s.gaia_api_url}/files/{task_id}"
+    try:
+        resp = requests.get(url, timeout=30)
+    except requests.RequestException:
+        return None
+    if resp.status_code != 200 or not resp.content:
+        return None
+
+    # Prefer a filename from Content-Disposition; fall back to mimetype/extension.
+    filename = None
+    cd = resp.headers.get("content-disposition", "")
+    if "filename=" in cd:
+        filename = cd.split("filename=")[-1].strip().strip('"')
+    if not filename:
+        ext = mimetypes.guess_extension(resp.headers.get("content-type", "").split(";")[0]) or ".bin"
+        filename = f"{task_id}{ext}"
+
+    path = os.path.join(_DOWNLOAD_DIR, filename)
+    with open(path, "wb") as fh:
+        fh.write(resp.content)
+    return path
+
+
+@tool
+def download_task_file(task_id: str) -> str:
+    """Download the file attached to a GAIA task and return its local path and kind.
+
+    Args:
+        task_id: The task identifier whose attachment should be fetched.
+    """
+    path = fetch_task_file(task_id)
+    if not path:
+        return "No file is attached to this task (or it could not be downloaded)."
+    return f"Downloaded to {path} (kind={classify_file(path)})."
+
+
+@tool
+def read_spreadsheet(path: str) -> str:
+    """Read an Excel/CSV file and return a markdown table plus basic stats.
+
+    Args:
+        path: Local path to a .xlsx/.xls/.csv file.
+    """
+    import pandas as pd
+
+    try:
+        if path.lower().endswith(".csv"):
+            df = pd.read_csv(path)
+        else:
+            df = pd.read_excel(path)
+    except Exception as exc:  # noqa: BLE001
+        return f"Could not read spreadsheet: {exc}"
+
+    preview = df.head(50).to_markdown(index=False)
+    return (
+        f"Shape: {df.shape[0]} rows x {df.shape[1]} cols.\n"
+        f"Columns: {list(df.columns)}\n\n{preview}"
+    )
+
+
+@tool
+def read_text_file(path: str) -> str:
+    """Read a plain-text, code, JSON, or markdown file and return its contents.
+
+    Args:
+        path: Local path to a text-like file.
+    """
+    try:
+        with open(path, encoding="utf-8", errors="replace") as fh:
+            content = fh.read()
+    except Exception as exc:  # noqa: BLE001
+        return f"Could not read file: {exc}"
+    if len(content) > 20000:
+        content = content[:20000] + "\n... [truncated]"
+    return content

gaia_agent/tools/media.py

@@ -0,0 +1,69 @@
+"""Multimodal tools: audio transcription (Whisper) and image understanding (vision)."""
+
+from __future__ import annotations
+
+import base64
+import os
+
+from langchain_core.messages import HumanMessage
+from langchain_core.tools import tool
+
+from gaia_agent.config import get_settings
+
+
+@tool
+def transcribe_audio(path: str) -> str:
+    """Transcribe an audio file to text using Groq Whisper.
+
+    Args:
+        path: Local path to an audio file (mp3/wav/m4a/flac/ogg).
+    """
+    from groq import Groq
+
+    s = get_settings()
+    try:
+        client = Groq(api_key=s.groq_api_key)
+        with open(path, "rb") as fh:
+            resp = client.audio.transcriptions.create(
+                file=(os.path.basename(path), fh.read()),
+                model=s.groq_whisper_model,
+            )
+        return resp.text
+    except Exception as exc:  # noqa: BLE001
+        return f"Audio transcription failed: {exc}"
+
+
+def _mime_for(path: str) -> str:
+    ext = os.path.splitext(path.lower())[1]
+    return {
+        ".png": "image/png",
+        ".jpg": "image/jpeg",
+        ".jpeg": "image/jpeg",
+        ".webp": "image/webp",
+        ".gif": "image/gif",
+    }.get(ext, "image/png")
+
+
+@tool
+def describe_image(path: str, question: str) -> str:
+    """Answer a question about an image using the Groq vision model.
+
+    Args:
+        path: Local path to an image file.
+        question: What to determine from the image.
+    """
+    from gaia_agent.llm import get_vision_llm
+
+    try:
+        with open(path, "rb") as fh:
+            b64 = base64.b64encode(fh.read()).decode()
+        data_url = f"data:{_mime_for(path)};base64,{b64}"
+        msg = HumanMessage(
+            content=[
+                {"type": "text", "text": question},
+                {"type": "image_url", "image_url": {"url": data_url}},
+            ]
+        )
+        return get_vision_llm().invoke([msg]).content
+    except Exception as exc:  # noqa: BLE001
+        return f"Image understanding failed: {exc}"

gaia_agent/tools/python_tool.py

@@ -0,0 +1,58 @@
+"""A restricted Python REPL for exact arithmetic, string, and date work."""
+
+from __future__ import annotations
+
+import contextlib
+import io
+import math
+
+from langchain_core.tools import tool
+
+# A small, deliberately limited set of safe builtins/modules.
+_SAFE_BUILTINS = {
+    "abs": abs, "all": all, "any": any, "bin": bin, "bool": bool, "chr": chr,
+    "dict": dict, "divmod": divmod, "enumerate": enumerate, "filter": filter,
+    "float": float, "hex": hex, "int": int, "len": len, "list": list, "map": map,
+    "max": max, "min": min, "oct": oct, "ord": ord, "pow": pow, "range": range,
+    "reversed": reversed, "round": round, "set": set, "sorted": sorted, "str": str,
+    "sum": sum, "tuple": tuple, "zip": zip, "print": print,
+}
+
+
+@tool
+def python_repl(code: str) -> str:
+    """Execute a short Python snippet and return its stdout (and `result` if set).
+
+    Use for exact arithmetic, string manipulation, sorting, and date math. The
+    ``math``, ``statistics``, ``datetime``, ``re``, ``itertools``, and ``collections``
+    modules are available. Assign to a variable named ``result`` to return a value.
+
+    Args:
+        code: Python source to execute.
+    """
+    import collections
+    import datetime
+    import itertools
+    import re
+    import statistics
+
+    env = {
+        "__builtins__": _SAFE_BUILTINS,
+        "math": math,
+        "statistics": statistics,
+        "datetime": datetime,
+        "re": re,
+        "itertools": itertools,
+        "collections": collections,
+    }
+    buf = io.StringIO()
+    try:
+        with contextlib.redirect_stdout(buf):
+            exec(code, env)  # noqa: S102 - sandboxed builtins, internal use only
+    except Exception as exc:  # noqa: BLE001
+        return f"Error: {exc}\nOutput so far:\n{buf.getvalue()}"
+
+    out = buf.getvalue()
+    if "result" in env:
+        out += ("\n" if out else "") + f"result = {env['result']!r}"
+    return out or "(no output)"

gaia_agent/tools/search.py

@@ -0,0 +1,54 @@
+"""Web search tools: Tavily (primary) and Wikipedia (free fallback)."""
+
+from __future__ import annotations
+
+from langchain_core.tools import tool
+
+from gaia_agent.config import get_settings
+
+
+@tool
+def tavily_search(query: str) -> str:
+    """Search the web with Tavily and return the top results as text.
+
+    Args:
+        query: A focused natural-language search query.
+    """
+    from langchain_tavily import TavilySearch
+
+    s = get_settings()
+    try:
+        searcher = TavilySearch(max_results=5, tavily_api_key=s.tavily_api_key)
+        result = searcher.invoke({"query": query})
+    except Exception as exc:  # noqa: BLE001
+        return f"Tavily search failed: {exc}"
+
+    # TavilySearch returns a dict with a "results" list.
+    if isinstance(result, dict):
+        rows = result.get("results", [])
+        lines = []
+        for r in rows:
+            lines.append(f"- {r.get('title', '')}\n  {r.get('url', '')}\n  {r.get('content', '')}")
+        answer = result.get("answer")
+        head = f"Answer: {answer}\n\n" if answer else ""
+        return head + "\n".join(lines) if lines else "No results."
+    return str(result)
+
+
+@tool
+def wikipedia_search(query: str) -> str:
+    """Look up a topic on Wikipedia and return a summary of the best-matching page.
+
+    Args:
+        query: The topic or page title to look up.
+    """
+    from langchain_community.tools import WikipediaQueryRun
+    from langchain_community.utilities import WikipediaAPIWrapper
+
+    try:
+        runner = WikipediaQueryRun(
+            api_wrapper=WikipediaAPIWrapper(top_k_results=3, doc_content_chars_max=4000)
+        )
+        return runner.run(query)
+    except Exception as exc:  # noqa: BLE001
+        return f"Wikipedia lookup failed: {exc}"

langgraph.json

@@ -0,0 +1,8 @@
+{
+  "$schema": "https://langgra.ph/schema.json",
+  "dependencies": ["."],
+  "graphs": {
+    "gaia": "./gaia_agent/graph.py:graph"
+  },
+  "env": ".env"
+}

pyproject.toml

@@ -0,0 +1,46 @@
+[project]
+name = "gaia-agent"
+version = "0.1.0"
+description = "LangGraph multi-agent for the HF Agents Course GAIA final assignment."
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+    "gradio>=5.25.2",
+    "requests>=2.32.0",
+    "pandas>=2.2.0",
+    "openpyxl>=3.1.0",
+    "tabulate>=0.9.0",
+    "langgraph>=1.0.0",
+    "langchain>=0.3.0",
+    "langchain-core>=0.3.0",
+    "langchain-groq>=0.2.0",
+    "langchain-tavily>=0.1.0",
+    "langchain-community>=0.3.0",
+    "wikipedia>=1.4.0",
+    "groq>=0.13.0",
+    "pydantic>=2.9.0",
+    "pydantic-settings>=2.6.0",
+    "python-dotenv>=1.0.1",
+]
+
+[project.optional-dependencies]
+dev = [
+    "langgraph-cli[inmem]>=0.3.0",
+    "pytest>=8.3.0",
+    "ruff>=0.8.0",
+]
+
+[build-system]
+requires = ["setuptools>=73.0.0", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools]
+packages = ["gaia_agent", "gaia_agent.tools"]
+
+[tool.ruff]
+line-length = 100
+lint.select = ["E", "F", "I", "UP"]
+lint.ignore = ["E501"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]

requirements.txt

@@ -1,2 +1,16 @@
 gradio
-requests
+requests
+python-dotenv
+pandas
+openpyxl
+tabulate
+langgraph>=1.0.0
+langchain>=0.3.0
+langchain-core>=0.3.0
+langchain-groq>=0.2.0
+langchain-tavily>=0.1.0
+langchain-community>=0.3.0
+wikipedia
+groq>=0.13.0
+pydantic>=2.9.0
+pydantic-settings>=2.6.0

scripts/dry_run.py

@@ -0,0 +1,39 @@
+"""Fetch all GAIA questions, run the agent locally, print answers — DO NOT submit.
+
+Usage (from the template dir, venv active):
+    python scripts/dry_run.py            # all questions
+    python scripts/dry_run.py 3          # first 3 only
+"""
+
+from __future__ import annotations
+
+import sys
+
+import requests
+from dotenv import load_dotenv
+
+load_dotenv()
+
+from gaia_agent import GaiaAgent  # noqa: E402
+from gaia_agent.config import get_settings  # noqa: E402
+
+
+def main(limit: int | None) -> None:
+    api = get_settings().gaia_api_url
+    questions = requests.get(f"{api}/questions", timeout=30).json()
+    if limit:
+        questions = questions[:limit]
+
+    agent = GaiaAgent()
+    for i, item in enumerate(questions, 1):
+        tid = item.get("task_id", "")
+        qtext = item.get("question", "")
+        answer = agent(qtext, tid)
+        print(f"\n[{i}/{len(questions)}] task={tid}")
+        print(f"  Q: {qtext[:120]}...")
+        print(f"  A: {answer!r}")
+
+
+if __name__ == "__main__":
+    n = int(sys.argv[1]) if len(sys.argv) > 1 else None
+    main(n)

tests/conftest.py

@@ -0,0 +1,7 @@
+"""Shared test fixtures."""
+
+import os
+
+# Ensure settings never pick up real keys during tests.
+os.environ.setdefault("GROQ_API_KEY", "test-key")
+os.environ.setdefault("TAVILY_API_KEY", "test-key")

tests/test_graph.py

@@ -0,0 +1,60 @@
+"""Tests for graph wiring, routing, and the judge evaluation loop."""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from gaia_agent.nodes import (
+    judge,
+    route_after_judge,
+    route_after_planner,
+    route_after_research,
+)
+from gaia_agent.schemas import JudgeVerdict, Plan
+
+
+def test_graph_compiles():
+    from gaia_agent.graph import graph
+
+    assert graph.name == "GAIA Agent"
+    nodes = set(graph.get_graph().nodes.keys())
+    assert {"planner", "research", "tools", "judge", "formatter"} <= nodes
+
+
+def test_route_after_planner():
+    assert route_after_planner({"plan": Plan(needs_file=True)}) == "ingest_file"
+    assert route_after_planner({"plan": Plan(needs_file=False)}) == "research"
+
+
+def test_route_after_research_tools_vs_evidence():
+    with_calls = SimpleNamespace(tool_calls=[{"name": "tavily_search"}])
+    without = SimpleNamespace(tool_calls=[])
+    assert route_after_research({"messages": [with_calls]}) == "tools"
+    assert route_after_research({"messages": [without]}) == "evidence"
+
+
+def test_route_after_judge_pass():
+    state = {"verdict": JudgeVerdict(verdict="PASS"), "attempts": 0}
+    assert route_after_judge(state) == "formatter"
+
+
+def test_route_after_judge_revise_within_budget():
+    # max_judge_retries default is 2; one attempt used -> still loop back.
+    state = {"verdict": JudgeVerdict(verdict="REVISE"), "attempts": 1}
+    assert route_after_judge(state) == "research"
+
+
+def test_route_after_judge_revise_budget_exhausted():
+    state = {"verdict": JudgeVerdict(verdict="REVISE"), "attempts": 2}
+    assert route_after_judge(state) == "formatter"
+
+
+def test_judge_increments_attempts_on_revise():
+    fake_llm = MagicMock()
+    fake_llm.with_structured_output.return_value.invoke.return_value = JudgeVerdict(
+        verdict="REVISE", feedback="wrong format", missing=["units"]
+    )
+    with patch("gaia_agent.nodes.get_text_llm", return_value=fake_llm):
+        out = judge({"question": "q", "candidate": None, "evidence": None, "attempts": 0})
+    assert out["attempts"] == 1
+    assert out["verdict"].verdict == "REVISE"
+    assert "messages" in out  # feedback injected back into the loop

tests/test_tools.py

@@ -0,0 +1,58 @@
+"""Unit tests for tools with mocked HTTP / external calls."""
+
+from unittest.mock import MagicMock, patch
+
+from gaia_agent.tools import (
+    classify_file,
+    python_repl,
+    read_spreadsheet,
+    read_text_file,
+)
+from gaia_agent.tools.files import fetch_task_file
+
+
+def test_classify_file():
+    assert classify_file("a/b/data.xlsx") == "spreadsheet"
+    assert classify_file("clip.MP3") == "audio"
+    assert classify_file("pic.jpeg") == "image"
+    assert classify_file("script.py") == "code"
+    assert classify_file("notes.txt") == "text"
+    assert classify_file("weird.xyz") == "other"
+
+
+def test_python_repl_arithmetic():
+    out = python_repl.invoke({"code": "result = sum(range(1, 11))"})
+    assert "55" in out
+
+
+def test_python_repl_handles_error():
+    out = python_repl.invoke({"code": "1/0"})
+    assert "Error" in out
+
+
+def test_read_text_file(tmp_path):
+    p = tmp_path / "f.txt"
+    p.write_text("hello gaia", encoding="utf-8")
+    assert "hello gaia" in read_text_file.invoke({"path": str(p)})
+
+
+def test_read_spreadsheet_csv(tmp_path):
+    p = tmp_path / "t.csv"
+    p.write_text("a,b\n1,2\n3,4\n", encoding="utf-8")
+    out = read_spreadsheet.invoke({"path": str(p)})
+    assert "2 rows" in out and "['a', 'b']" in out
+
+
+def test_fetch_task_file_no_file():
+    """A non-200 response yields None rather than a path."""
+    with patch("gaia_agent.tools.files.requests.get") as mock_get:
+        mock_get.return_value = MagicMock(status_code=404, content=b"")
+        assert fetch_task_file("missing") is None
+
+
+def test_fetch_task_file_writes(tmp_path):
+    resp = MagicMock(status_code=200, content=b"col\n1\n",
+                     headers={"content-disposition": 'attachment; filename="x.csv"'})
+    with patch("gaia_agent.tools.files.requests.get", return_value=resp):
+        path = fetch_task_file("task1")
+    assert path is not None and path.endswith("x.csv")