Add GAIA agent pipeline without committing secrets.

Ignore .env and local artifacts; use Space secrets for API keys.

Co-authored-by: Cursor <cursoragent@cursor.com>

.env.example

@@ -0,0 +1,4 @@
+# Copy to .env locally and set Space secrets on Hugging Face (never commit .env)
+GOOGLE_API_KEY=your_google_api_key_here
+GEMINI_MODEL=gemini-2.0-flash
+QUESTION_DELAY_SECONDS=2

.gitignore

@@ -0,0 +1,10 @@
+.env
+.env.*
+!.env.example
+
+.DS_Store
+__pycache__/
+*.py[cod]
+.ipynb_checkpoints/
+.venv/
+*.png

README.md

@@ -1,15 +0,0 @@
----
-title: Template Final Assignment
-emoji: 🕵🏻‍♂️
-colorFrom: indigo
-colorTo: indigo
-sdk: gradio
-sdk_version: 5.25.2
-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
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

agent.py

@@ -0,0 +1,301 @@
+"""
+This file contains the agent that will be used to evaluate GAIA benchmark questions.
+"""
+
+import os
+import re
+import json
+import time
+from dotenv import load_dotenv
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain_google_genai.chat_models import ChatGoogleGenerativeAIError
+from typing_extensions import TypedDict
+from prompts import PLANNER_PROMPT, SYNTHESIS_ANSWER_PROMPT, ANSWER_PROMPT
+from langgraph.graph import StateGraph, START, END
+from tools import search_web
+
+load_dotenv()
+
+MAX_SEARCH_QUERIES = 3
+MIN_SEARCH_QUERIES = 1
+MAX_LLM_RETRIES = 6
+DEFAULT_RETRY_SECONDS = 15
+QUESTION_DELAY_SECONDS = float(os.getenv("QUESTION_DELAY_SECONDS", "2"))
+
+#--- Model --------------------------------------------------------------#
+
+model = ChatGoogleGenerativeAI(
+    model=os.getenv("GEMINI_MODEL", "gemini-2.0-flash"),
+    api_key=os.getenv("GOOGLE_API_KEY"),
+    temperature=0,
+)
+
+#--- State --------------------------------------------------------------#
+
+class AgentState(TypedDict):
+    question: str
+    plan: list[str]
+    search_queries: list[str]
+    search_results: list[str]
+    synthesis_notes: str
+    answer: str
+
+#--- Helpers ------------------------------------------------------------#
+
+def _normalize(text: str) -> str:
+    return " ".join(text.lower().split())
+
+
+def _strip_json_fences(text: str) -> str:
+    text = text.strip()
+    match = re.search(r"```(?:json)?\s*([\s\S]*?)\s*```", text, re.IGNORECASE)
+    if match:
+        return match.group(1).strip()
+    return text
+
+
+def _retry_delay_seconds(error: Exception, attempt: int) -> float:
+    msg = str(error)
+    match = re.search(r"retry in (\d+(?:\.\d+)?)s", msg, re.IGNORECASE)
+    if match:
+        return float(match.group(1)) + 1.0
+    return min(DEFAULT_RETRY_SECONDS * (attempt + 1), 60.0)
+
+
+def _is_rate_limit_error(error: Exception) -> bool:
+    msg = str(error).lower()
+    return "429" in msg or "resource_exhausted" in msg or "quota" in msg
+
+
+def invoke_llm(prompt: str) -> str:
+    """Call Gemini with retries on rate-limit / quota errors."""
+    last_error: Exception | None = None
+    for attempt in range(MAX_LLM_RETRIES):
+        try:
+            response = model.invoke(prompt)
+            return response.content.strip()
+        except ChatGoogleGenerativeAIError as e:
+            last_error = e
+            if _is_rate_limit_error(e) and attempt < MAX_LLM_RETRIES - 1:
+                delay = _retry_delay_seconds(e, attempt)
+                print(f"Rate limited; retrying in {delay:.0f}s (attempt {attempt + 1}/{MAX_LLM_RETRIES})...")
+                time.sleep(delay)
+                continue
+            raise
+    if last_error:
+        raise last_error
+    raise RuntimeError("invoke_llm failed without an exception")
+
+
+def _is_question_copy(text: str, question: str) -> bool:
+    t, q = _normalize(text), _normalize(question)
+    if not t:
+        return True
+    if t == q:
+        return True
+    if len(t) >= len(q) * 0.85 and q in t:
+        return True
+    return False
+
+
+def _keyword_fallback_query(question: str) -> str:
+    stop = {
+        "what", "which", "when", "where", "who", "whom", "whose", "why", "how",
+        "the", "a", "an", "is", "are", "was", "were", "of", "in", "on", "at",
+        "to", "for", "and", "or", "does", "do", "did", "many", "much",
+    }
+    words = [
+        w.strip(".,?!\"'()[]")
+        for w in question.split()
+        if w.lower().strip(".,?!\"'()[]") not in stop and len(w) > 2
+    ]
+    return " ".join(words[:8]) if words else question[:80]
+
+
+def _parse_planner_response(content: str, question: str) -> dict:
+    raw = _strip_json_fences(content)
+    data = json.loads(raw)
+    steps = data.get("steps") or []
+    queries = data.get("search_queries") or []
+
+    if not isinstance(steps, list):
+        steps = [str(steps)]
+    if not isinstance(queries, list):
+        queries = [str(queries)]
+
+    steps = [str(s).strip() for s in steps if str(s).strip()]
+    steps = [s for s in steps if not _is_question_copy(s, question)]
+    if not steps:
+        steps = [
+            "Identify entities and constraints in the question",
+            "Determine what facts must be verified externally",
+            "Cross-check candidates against authoritative sources",
+        ]
+
+    queries = [str(q).strip() for q in queries if str(q).strip()]
+    queries = [q for q in queries if not _is_question_copy(q, question)]
+    seen: set[str] = set()
+    deduped: list[str] = []
+    for q in queries:
+        key = _normalize(q)
+        if key in seen or len(q.split()) > 10:
+            continue
+        seen.add(key)
+        deduped.append(q)
+        if len(deduped) >= MAX_SEARCH_QUERIES:
+            break
+
+    if len(deduped) < MIN_SEARCH_QUERIES:
+        deduped = [_keyword_fallback_query(question)]
+
+    return {"steps": steps, "search_queries": deduped}
+
+
+def _format_plan(plan: list[str]) -> str:
+    return "\n".join(f"- {step}" for step in plan)
+
+
+def _format_search_results(results: list[str], max_chars: int = 6000) -> str:
+    blocks = []
+    total = 0
+    for i, block in enumerate(results, 1):
+        chunk = f"### Query {i}\n{block}"
+        if total + len(chunk) > max_chars:
+            chunk = chunk[: max_chars - total]
+        blocks.append(chunk)
+        total += len(chunk)
+        if total >= max_chars:
+            break
+    return "\n\n".join(blocks) if blocks else "(no search results)"
+
+
+def _heuristic_synthesis(search_results: list[str]) -> str:
+    bullets: list[str] = []
+    for block in search_results:
+        for line in block.split("\n"):
+            line = line.strip()
+            if not line or line == "No results found.":
+                continue
+            bullets.append(f"- {line[:400]}")
+            if len(bullets) >= 12:
+                break
+        if len(bullets) >= 12:
+            break
+    return "\n".join(bullets) if bullets else "- No usable search evidence."
+
+
+def _parse_synthesis_answer(content: str) -> dict:
+    data = json.loads(_strip_json_fences(content))
+    notes = str(data.get("synthesis_notes", "")).strip()
+    answer = str(data.get("answer", "")).strip()
+    answer = re.sub(r'^["\']|["\']$', "", answer)
+    if not notes:
+        raise ValueError("missing synthesis_notes")
+    if not answer:
+        raise ValueError("missing answer")
+    return {"synthesis_notes": notes, "answer": answer}
+
+
+def _answer_from_notes(question: str, plan: list[str], synthesis_notes: str) -> str:
+    prompt = ANSWER_PROMPT.format(
+        question=question,
+        plan=_format_plan(plan),
+        synthesis_notes=synthesis_notes,
+    )
+    content = invoke_llm(prompt)
+    answer = re.sub(r'^["\']|["\']$', "", content.strip())
+    return answer
+
+
+#--- Nodes --------------------------------------------------------------#
+
+def plan_node(state: AgentState) -> AgentState:
+    """Plan steps and capped search queries."""
+    question = state["question"]
+    prompt = PLANNER_PROMPT.format(question=question)
+
+    try:
+        content = invoke_llm(prompt)
+        data = _parse_planner_response(content, question)
+    except (json.JSONDecodeError, TypeError, KeyError, ChatGoogleGenerativeAIError):
+        data = {
+            "steps": [
+                "Identify entities and constraints in the question",
+                "Determine what external facts are required",
+                "Verify the answer against reliable sources",
+            ],
+            "search_queries": [_keyword_fallback_query(question)],
+        }
+
+    return {
+        "plan": data["steps"],
+        "search_queries": data["search_queries"],
+    }
+
+
+def search_node(state: AgentState) -> AgentState:
+    """Run 1–3 web searches."""
+    search_queries = state["search_queries"][:MAX_SEARCH_QUERIES]
+    results = [search_web(query) for query in search_queries]
+    return {"search_results": results}
+
+
+def finalize_node(state: AgentState) -> AgentState:
+    """Synthesize evidence and produce answer in one LLM call (saves quota)."""
+    question = state["question"]
+    plan = state["plan"]
+    search_results = state["search_results"]
+    formatted_results = _format_search_results(search_results)
+
+    try:
+        prompt = SYNTHESIS_ANSWER_PROMPT.format(
+            question=question,
+            plan=_format_plan(plan),
+            search_results=formatted_results,
+        )
+        parsed = _parse_synthesis_answer(invoke_llm(prompt))
+        return parsed
+    except (json.JSONDecodeError, TypeError, KeyError, ValueError, ChatGoogleGenerativeAIError) as e:
+        print(f"Finalize LLM failed ({e}); using heuristic synthesis.")
+        notes = _heuristic_synthesis(search_results)
+        try:
+            answer = _answer_from_notes(question, plan, notes)
+        except ChatGoogleGenerativeAIError:
+            first_line = notes.split("\n", 1)[0].lstrip("- ").strip()
+            answer = first_line[:200] if first_line else "Unknown"
+        return {"synthesis_notes": notes, "answer": answer}
+
+
+#--- Graph --------------------------------------------------------------#
+
+agent_builder = StateGraph(AgentState)
+agent_builder.add_node("plan", plan_node)
+agent_builder.add_node("search", search_node)
+agent_builder.add_node("finalize", finalize_node)
+agent_builder.add_edge(START, "plan")
+agent_builder.add_edge("plan", "search")
+agent_builder.add_edge("search", "finalize")
+agent_builder.add_edge("finalize", END)
+
+agent = agent_builder.compile()
+
+
+def run_agent(question: str) -> AgentState:
+    """Run the full pipeline for a single question (2 LLM calls: plan + finalize)."""
+    return agent.invoke({"question": question})
+
+
+if __name__ == "__main__":
+    # Free tier: ~20 requests/day per model — use 1 question while testing.
+    questions = [
+        "What is the capital of France?",
+    ]
+    for i, question in enumerate(questions):
+        if i > 0:
+            time.sleep(QUESTION_DELAY_SECONDS)
+        result = run_agent(question)
+        print(f"Question: {result['question']}")
+        print(f"Plan: {result['plan']}")
+        print(f"Queries: {result['search_queries']}")
+        print(f"Synthesis:\n{result['synthesis_notes']}")
+        print(f"Answer: {result['answer']}\n")

app.py

@@ -1,9 +1,16 @@
 import os
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
 import gradio as gr
 import requests
 import inspect
 import pandas as pd
 
+from agent import run_agent
+
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
@@ -13,11 +20,13 @@ DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 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
+        result = run_agent(question)
+        answer = result["answer"]
+        print(f"Agent answer (first 80 chars): {answer[:80]}...")
+        return answer
 
 def run_and_submit_all( profile: gr.OAuthProfile | None):
     """
@@ -176,6 +185,12 @@ if __name__ == "__main__":
     # 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
+    GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
+    
+    if GOOGLE_API_KEY:
+        print(f"✅ GOOGLE_API_KEY found: {GOOGLE_API_KEY}")
+    else:
+        print("ℹ️  GOOGLE_API_KEY environment variable not found (running locally?).")
 
     if space_host_startup:
         print(f"✅ SPACE_HOST found: {space_host_startup}")

prompts.py

@@ -0,0 +1,81 @@
+PLANNER_PROMPT = """
+You are a planning agent for a GAIA-style question-answering benchmark.
+
+Break the question into:
+1. A step-by-step reasoning plan (3–6 atomic steps)
+2. Web search queries to gather missing facts
+
+Hard rules:
+- Output ONLY valid JSON. No markdown, no code fences, no commentary.
+- Use EXACTLY this schema (array lengths: steps 3–6, search_queries 1–3):
+{{
+  "steps": ["..."],
+  "search_queries": ["..."]
+}}
+- NEVER copy or paraphrase the full question as a step or as a search query.
+- NEVER use the question text verbatim in search_queries.
+- Steps must be actionable reasoning actions (e.g. "Identify the entity", "Verify the date from an authoritative source").
+- search_queries: 1–3 items only. Short keyword phrases (2–8 words). Each query must target a distinct fact gap.
+- Prefer precision over breadth: one excellent query beats three redundant ones.
+- If the question is answerable without search, still return 1 minimal disambiguation query.
+
+Penalties (invalid output — avoid):
+- steps or search_queries containing the full question string
+- search_queries longer than 10 words
+- more than 3 search_queries or fewer than 1 search_queries
+- placeholder text like "step1", "query1", "..."
+
+Question:
+{question}
+"""
+
+SYNTHESIS_ANSWER_PROMPT = """
+You are a GAIA research and answer agent. Do two tasks in one response.
+
+Task 1 — Synthesis: compress search results into bullet notes (max 12, prefix "- ").
+Task 2 — Answer: produce the final GAIA answer from those notes.
+
+Answer rules:
+- Return ONLY valid JSON. No markdown fences.
+- Schema:
+{{
+  "synthesis_notes": "- fact1\\n- fact2",
+  "answer": "final answer only"
+}}
+- synthesis_notes: bullets only, no preamble.
+- answer: concise; no quotes, meta-commentary, or search artifacts (no broken fragments like "Rome., is good?").
+- Do NOT repeat the question text in the answer.
+
+Question:
+{question}
+
+Plan:
+{plan}
+
+Search results:
+{search_results}
+"""
+
+ANSWER_PROMPT = """
+You are a GAIA benchmark answer agent. Produce the final answer only.
+
+Rules:
+- Use the question, plan, and synthesis notes — not raw search snippets.
+- Return ONLY the final answer text. No preamble, labels, quotes, or explanation unless the question requires it.
+- Be concise: prefer a short phrase, name, number, or yes/no when appropriate.
+- Do NOT echo search artifacts, typos, or broken fragments (e.g. "Rome., is good?", trailing punctuation glitches).
+- Do NOT repeat the question, plan, or notes.
+- Do NOT hedge with "based on the search results" or similar meta-commentary.
+- For factual questions: give the single best-supported answer.
+- For lists: use the format the question implies (comma-separated or one item per line).
+- If evidence is insufficient, give the most likely short answer; do not invent citations.
+
+Question:
+{question}
+
+Plan:
+{plan}
+
+Synthesis notes:
+{synthesis_notes}
+"""

requirements.txt

@@ -1,6 +1,11 @@
 gradio
+python-dotenv
 requests
 langgraph
 langchain
 langchain-google-genai
-duckduckgo-search
+ddgs
+google-generativeai
+google-auth
+gradio[oauth]
+ipython

tools.py

@@ -0,0 +1,13 @@
+from ddgs import DDGS
+
+def search_web(query: str) -> str:
+    """Search the web for information related to the query."""
+    results = DDGS().text(query, max_results=10)
+    if not results:
+        return "No results found."
+    return "\n\n".join(
+        f"{r.get('title', '')}: {r.get('body', r.get('href', ''))}"
+        for r in results
+    )
+
+