| """
|
| Hugging Face Agents Course — Final Assignment Agent
|
| ===================================================
|
| A production-ready CodeAgent built with `smolagents`, wrapped in Gradio.
|
|
|
| Features
|
| --------
|
| 1. Interactive chat (try the agent yourself)
|
| 2. Official Unit 4 evaluation + leaderboard submission (needed for the certificate)
|
|
|
| Free inference uses Hugging Face Inference Providers via `InferenceClientModel`
|
| (the current name for what older docs called `HfApiModel` / Hub server models).
|
| """
|
|
|
| from __future__ import annotations
|
|
|
| import ast
|
| import math
|
| import operator
|
| import os
|
| import re
|
| from typing import Any
|
|
|
| import requests
|
| from dotenv import load_dotenv
|
|
|
| from smolagents import (
|
| CodeAgent,
|
| DuckDuckGoSearchTool,
|
| InferenceClientModel,
|
| VisitWebpageTool,
|
| tool,
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| load_dotenv()
|
|
|
|
|
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
|
|
|
|
|
| MODEL_ID = os.getenv("MODEL_ID", "meta-llama/Llama-3.3-70B-Instruct")
|
|
|
| MODEL_PROVIDER = os.getenv("MODEL_PROVIDER", "together")
|
|
|
|
|
| HF_TOKEN = os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACE_HUB_TOKEN")
|
|
|
|
|
|
|
|
|
|
|
|
|
| _SAFE_OPS = {
|
| ast.Add: operator.add,
|
| ast.Sub: operator.sub,
|
| ast.Mult: operator.mul,
|
| ast.Div: operator.truediv,
|
| ast.FloorDiv: operator.floordiv,
|
| ast.Mod: operator.mod,
|
| ast.Pow: operator.pow,
|
| ast.USub: operator.neg,
|
| ast.UAdd: operator.pos,
|
| }
|
|
|
| _SAFE_FUNCS = {
|
| "sqrt": math.sqrt,
|
| "abs": abs,
|
| "round": round,
|
| "min": min,
|
| "max": max,
|
| "ceil": math.ceil,
|
| "floor": math.floor,
|
| "log": math.log,
|
| "log10": math.log10,
|
| "sin": math.sin,
|
| "cos": math.cos,
|
| "tan": math.tan,
|
| "pi": math.pi,
|
| "e": math.e,
|
| }
|
|
|
|
|
| def _eval_ast(node: ast.AST) -> Any:
|
| """Recursively evaluate a parsed AST using only safe operators."""
|
| if isinstance(node, ast.Expression):
|
| return _eval_ast(node.body)
|
| if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
|
| return node.value
|
| if isinstance(node, ast.Num):
|
| return node.n
|
| if isinstance(node, ast.BinOp):
|
| op_type = type(node.op)
|
| if op_type not in _SAFE_OPS:
|
| raise ValueError(f"Unsupported operator: {op_type.__name__}")
|
| return _SAFE_OPS[op_type](_eval_ast(node.left), _eval_ast(node.right))
|
| if isinstance(node, ast.UnaryOp):
|
| op_type = type(node.op)
|
| if op_type not in _SAFE_OPS:
|
| raise ValueError(f"Unsupported unary operator: {op_type.__name__}")
|
| return _SAFE_OPS[op_type](_eval_ast(node.operand))
|
| if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
|
| name = node.func.id
|
| if name not in _SAFE_FUNCS or not callable(_SAFE_FUNCS[name]):
|
| raise ValueError(f"Unsupported function: {name}")
|
| args = [_eval_ast(a) for a in node.args]
|
| return _SAFE_FUNCS[name](*args)
|
| if isinstance(node, ast.Name) and node.id in _SAFE_FUNCS:
|
| return _SAFE_FUNCS[node.id]
|
| raise ValueError(f"Unsafe or unsupported expression node: {type(node).__name__}")
|
|
|
|
|
| @tool
|
| def calculator(expression: str) -> str:
|
| """Evaluate a mathematical expression safely and return the numeric result.
|
|
|
| Use this for arithmetic, percentages, unit conversions expressed as math,
|
| roots, powers, and basic trigonometry. Do NOT use it for symbolic algebra.
|
|
|
| Args:
|
| expression: A math expression such as "2 + 2 * 3", "(10 - 3) / 7",
|
| "17 * 250 / 100", or "sqrt(144) + 5". For percentages write
|
| "17/100 * 250" (do not use a % sign).
|
| """
|
| cleaned = expression.strip().replace("^", "**")
|
|
|
| cleaned = re.sub(r"(\d+(?:\.\d+)?)\s*%", r"(\1/100)", cleaned)
|
| try:
|
| tree = ast.parse(cleaned, mode="eval")
|
| result = _eval_ast(tree)
|
|
|
| if isinstance(result, float) and result.is_integer():
|
| result = int(result)
|
| return str(result)
|
| except Exception as exc:
|
| return f"Calculator error: {exc}"
|
|
|
|
|
| @tool
|
| def summarize_text(text: str, max_sentences: int = 3) -> str:
|
| """Create a short plain-text summary of a longer passage.
|
|
|
| Useful after visiting a webpage or reading search snippets when you only
|
| need the key facts.
|
|
|
| Args:
|
| text: The full text to summarize.
|
| max_sentences: Maximum number of sentences to keep (default 3).
|
| """
|
| if not text or not text.strip():
|
| return "No text provided to summarize."
|
|
|
|
|
| sentences = re.split(r"(?<=[.!?])\s+", text.strip())
|
| useful = [s.strip() for s in sentences if len(s.strip()) > 40]
|
| if not useful:
|
| useful = [s.strip() for s in sentences if s.strip()]
|
|
|
| selected = useful[: max(1, int(max_sentences))]
|
| summary = " ".join(selected)
|
|
|
| if len(summary) > 1200:
|
| summary = summary[:1197] + "..."
|
| return summary
|
|
|
|
|
|
|
|
|
|
|
| SYSTEM_HINT = """
|
| You are a careful research assistant solving GAIA-style questions.
|
|
|
| Rules:
|
| 1. Use tools when you need fresh facts (search, visit webpage) or exact math (calculator).
|
| 2. Think step by step, but your FINAL reply must contain ONLY the answer itself —
|
| no preamble, no "the answer is", no markdown, unless the question clearly asks for explanation.
|
| 3. Prefer short, exact answers (numbers, names, short phrases) that would match a benchmark key.
|
| 4. If a webpage is needed, search first, then visit the most relevant URL.
|
| 5. If calculation is involved, use the calculator tool — do not guess arithmetic.
|
| """.strip()
|
|
|
|
|
| def build_agent() -> CodeAgent:
|
| """Create a CodeAgent wired to tools + the best available free model."""
|
| model = _build_model()
|
| agent = CodeAgent(
|
| tools=[
|
| DuckDuckGoSearchTool(),
|
| VisitWebpageTool(),
|
| calculator,
|
| summarize_text,
|
| ],
|
| model=model,
|
| max_steps=12,
|
| additional_authorized_imports=["math", "re", "json", "datetime", "statistics"],
|
| )
|
| return agent
|
|
|
|
|
| def _build_model():
|
| """
|
| Model selection (first match wins):
|
| 1. MODEL_BACKEND=ollama → local Ollama (no HF credits)
|
| 2. LiteLLM provider keys (GROQ_API_KEY / GEMINI_API_KEY / TOGETHER_API_KEY)
|
| 3. Hugging Face Inference Providers (uses monthly HF credits)
|
| """
|
| backend = (os.getenv("MODEL_BACKEND") or "auto").lower().strip()
|
|
|
|
|
| if backend in {"ollama", "auto"}:
|
| ollama_model = os.getenv("OLLAMA_MODEL", "qwen2.5:14b")
|
| if backend == "ollama" or _ollama_available(ollama_model):
|
| from smolagents import LiteLLMModel
|
|
|
| print(f"Using local Ollama model: {ollama_model}")
|
| return LiteLLMModel(
|
| model_id=f"ollama_chat/{ollama_model}",
|
| api_base=os.getenv("OLLAMA_API_BASE", "http://127.0.0.1:11434"),
|
| num_ctx=8192,
|
| )
|
|
|
|
|
| litellm_key_map = [
|
| ("GROQ_API_KEY", os.getenv("GROQ_MODEL", "groq/llama-3.3-70b-versatile")),
|
| ("GEMINI_API_KEY", os.getenv("GEMINI_MODEL", "gemini/gemini-2.0-flash")),
|
| ("GOOGLE_API_KEY", os.getenv("GEMINI_MODEL", "gemini/gemini-2.0-flash")),
|
| ("TOGETHER_API_KEY", os.getenv("TOGETHER_MODEL", "together_ai/meta-llama/Llama-3.3-70B-Instruct-Turbo")),
|
| ]
|
| for env_name, model_id in litellm_key_map:
|
| if os.getenv(env_name):
|
| from smolagents import LiteLLMModel
|
|
|
| print(f"Using LiteLLM via {env_name}: {model_id}")
|
| return LiteLLMModel(model_id=model_id, api_key=os.getenv(env_name))
|
|
|
| if not HF_TOKEN:
|
| print(
|
| "⚠️ HF_TOKEN not found. Set it as a Space secret or in a local .env file. "
|
| "Inference may fail without authentication."
|
| )
|
|
|
| print(f"Using Hugging Face Inference Providers: {MODEL_ID} ({MODEL_PROVIDER})")
|
| return InferenceClientModel(
|
| model_id=MODEL_ID,
|
| provider=MODEL_PROVIDER if MODEL_PROVIDER.lower() != "auto" else None,
|
| token=HF_TOKEN,
|
| timeout=120,
|
| )
|
|
|
|
|
| def _ollama_available(model_name: str) -> bool:
|
| try:
|
| resp = requests.get("http://127.0.0.1:11434/api/tags", timeout=2)
|
| resp.raise_for_status()
|
| names = {m.get("name", "").split(":")[0] for m in resp.json().get("models", [])}
|
| base = model_name.split(":")[0]
|
| full_names = {m.get("name", "") for m in resp.json().get("models", [])}
|
| return model_name in full_names or base in names
|
| except Exception:
|
| return False
|
|
|
|
|
| class CourseAgent:
|
| """Thin wrapper used by both the chat UI and the Unit 4 evaluation loop."""
|
|
|
| def __init__(self) -> None:
|
| self.agent = build_agent()
|
| print("CourseAgent ready")
|
|
|
| def __call__(self, question: str) -> str:
|
| prompt = f"{SYSTEM_HINT}\n\nQuestion:\n{question}"
|
| try:
|
| result = self.agent.run(prompt, reset=True)
|
| answer = result if isinstance(result, str) else str(result)
|
| return _clean_final_answer(answer)
|
| except Exception as exc:
|
| print(f"Agent error: {exc}")
|
| return f"ERROR: {exc}"
|
|
|
|
|
| def _clean_final_answer(text: str) -> str:
|
| """Strip common wrappers so GAIA exact-match scoring works better."""
|
| answer = text.strip()
|
|
|
| if (answer.startswith('"') and answer.endswith('"')) or (
|
| answer.startswith("'") and answer.endswith("'")
|
| ):
|
| answer = answer[1:-1].strip()
|
|
|
| lines = [ln.strip() for ln in answer.splitlines() if ln.strip()]
|
| if len(lines) > 1:
|
| for prefix in ("final answer:", "answer:", "the answer is"):
|
| for ln in reversed(lines):
|
| low = ln.lower()
|
| if low.startswith(prefix):
|
| return ln[len(prefix) :].strip(" :.-")
|
|
|
| answer = min(lines[-3:], key=len)
|
| return answer.strip()
|
|
|
|
|
|
|
|
|
|
|
| _chat_agent: CourseAgent | None = None
|
|
|
|
|
| def get_chat_agent() -> CourseAgent:
|
| global _chat_agent
|
| if _chat_agent is None:
|
| _chat_agent = CourseAgent()
|
| return _chat_agent
|
|
|
|
|
| def chat_respond(message: str, history: list) -> str:
|
| """Gradio ChatInterface callback — one user message in, one reply out."""
|
| if not message or not message.strip():
|
| return "Please enter a question."
|
| agent = get_chat_agent()
|
| return agent(message.strip())
|
|
|
|
|
|
|
|
|
|
|
| def run_and_submit_all(profile: Any = None):
|
| """
|
| Fetch all GAIA subset questions, run the agent, submit answers,
|
| and return (status_text, results_rows).
|
| """
|
| space_id = os.getenv("SPACE_ID")
|
|
|
| if profile is not None and getattr(profile, "username", None):
|
| username = profile.username
|
| print(f"User logged in: {username}")
|
| else:
|
| print("User not logged in.")
|
| return "Please log in to Hugging Face with the button above, then try again.", None
|
|
|
| api_url = DEFAULT_API_URL
|
| questions_url = f"{api_url}/questions"
|
| submit_url = f"{api_url}/submit"
|
|
|
| try:
|
| agent = CourseAgent()
|
| except Exception as e:
|
| print(f"Error instantiating agent: {e}")
|
| return f"Error initializing agent: {e}", None
|
|
|
| agent_code = (
|
| f"https://huggingface.co/spaces/{space_id}/tree/main"
|
| if space_id
|
| else "local-run"
|
| )
|
| print(f"Agent code URL: {agent_code}")
|
|
|
|
|
| print(f"Fetching questions from: {questions_url}")
|
| try:
|
| response = requests.get(questions_url, timeout=30)
|
| response.raise_for_status()
|
| questions_data = response.json()
|
| if not questions_data:
|
| return "Fetched questions list is empty.", None
|
| print(f"Fetched {len(questions_data)} questions.")
|
| except requests.exceptions.RequestException as e:
|
| return f"Error fetching questions: {e}", None
|
| except Exception as e:
|
| return f"Unexpected error fetching questions: {e}", None
|
|
|
|
|
| results_log: list[dict] = []
|
| answers_payload: list[dict] = []
|
| print(f"Running agent on {len(questions_data)} questions...")
|
|
|
| for item in questions_data:
|
| task_id = item.get("task_id")
|
| question_text = item.get("question")
|
| if not task_id or question_text is None:
|
| print(f"Skipping incomplete item: {item}")
|
| continue
|
| try:
|
| submitted_answer = agent(question_text)
|
| 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 on task {task_id}: {e}")
|
| results_log.append(
|
| {
|
| "Task ID": task_id,
|
| "Question": question_text,
|
| "Submitted Answer": f"AGENT ERROR: {e}",
|
| }
|
| )
|
|
|
| if not answers_payload:
|
| return "Agent did not produce any answers to submit.", results_log
|
|
|
|
|
| submission_data = {
|
| "username": username.strip(),
|
| "agent_code": agent_code,
|
| "answers": answers_payload,
|
| }
|
| print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
|
|
|
| try:
|
| response = requests.post(submit_url, json=submission_data, timeout=120)
|
| response.raise_for_status()
|
| result_data = response.json()
|
| final_status = (
|
| f"Submission Successful!\n"
|
| f"User: {result_data.get('username')}\n"
|
| f"Overall Score: {result_data.get('score', 'N/A')}% "
|
| f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
|
| f"Message: {result_data.get('message', 'No message received.')}\n\n"
|
| f"🎯 You need ≥ 30% to earn the Certificate of Completion."
|
| )
|
| return final_status, results_log
|
| except requests.exceptions.HTTPError as e:
|
| detail = f"HTTP {e.response.status_code}"
|
| try:
|
| detail += f" — {e.response.json().get('detail', e.response.text)}"
|
| except Exception:
|
| detail += f" — {e.response.text[:500]}"
|
| return f"Submission Failed: {detail}", results_log
|
| except requests.exceptions.RequestException as e:
|
| return f"Submission Failed: {e}", results_log
|
| except Exception as e:
|
| return f"Unexpected submission error: {e}", results_log
|
|
|
|
|
|
|
|
|
|
|
| CHAT_EXAMPLES = [
|
| "What is 17% of 250?",
|
| "Who is the current Secretary-General of the United Nations?",
|
| "Convert 72 degrees Fahrenheit to Celsius (answer with one decimal).",
|
| ]
|
|
|
|
|
| def build_demo():
|
| import gradio as gr
|
|
|
| with gr.Blocks(title="HF Agents Course — Smolagents Final Agent") as demo:
|
| gr.Markdown(
|
| f"""
|
| # 🕵️♂️ Hugging Face Agents Course — Final Agent
|
|
|
| Built with **smolagents** `CodeAgent` + free Hub inference (`{MODEL_ID}`).
|
|
|
| **Tools:** DuckDuckGo search · Visit webpage · Calculator · Text summarizer
|
|
|
| Use the **Chat** tab to try the agent. Use the **Certificate Evaluation** tab to
|
| run the official Unit 4 GAIA subset and submit your score (need **≥ 30%**).
|
| """
|
| )
|
|
|
| with gr.Tab("💬 Chat with the Agent"):
|
| gr.ChatInterface(
|
| fn=chat_respond,
|
| examples=CHAT_EXAMPLES,
|
| title=None,
|
| description="Ask a question. The agent may search the web, visit pages, or calculate.",
|
| )
|
|
|
| with gr.Tab("🎓 Certificate Evaluation (Unit 4)"):
|
| gr.Markdown(
|
| """
|
| ### Instructions
|
| 1. Make sure this Space is **public** and your `HF_TOKEN` secret is set.
|
| 2. Click **Log in with Hugging Face** below (required for leaderboard submission).
|
| 3. Click **Run Evaluation & Submit All Answers**.
|
| 4. Wait — answering all questions can take several minutes.
|
| 5. Aim for **30% or higher** to unlock your Certificate of Completion.
|
| """
|
| )
|
| gr.LoginButton()
|
| run_button = gr.Button(
|
| "Run Evaluation & Submit All Answers", variant="primary"
|
| )
|
| status_output = gr.Textbox(
|
| label="Run Status / Submission Result", lines=8, interactive=False
|
| )
|
| results_table = gr.Dataframe(
|
| label="Questions and Agent Answers",
|
| headers=["Task ID", "Question", "Submitted Answer"],
|
| wrap=True,
|
| )
|
| run_button.click(
|
| fn=run_and_submit_all,
|
| outputs=[status_output, results_table],
|
| )
|
|
|
| gr.Markdown(
|
| """
|
| ---
|
| **Secrets / env:** `HF_TOKEN` (required) · optional `MODEL_ID`
|
| **Docs:** [Agents Course Unit 4](https://huggingface.co/learn/agents-course/unit4/introduction) · [smolagents](https://huggingface.co/docs/smolagents)
|
| """
|
| )
|
|
|
| return demo
|
|
|
|
|
| if __name__ == "__main__":
|
| print("\n" + "-" * 30 + " App Starting " + "-" * 30)
|
| space_host = os.getenv("SPACE_HOST")
|
| space_id = os.getenv("SPACE_ID")
|
| if space_host:
|
| print(f"✅ SPACE_HOST: {space_host} → https://{space_host}.hf.space")
|
| else:
|
| print("ℹ️ Running locally (SPACE_HOST not set).")
|
| if space_id:
|
| print(f"✅ SPACE_ID: {space_id}")
|
| if not HF_TOKEN:
|
| print("⚠️ HF_TOKEN missing — create a free token at https://huggingface.co/settings/tokens")
|
| print("-" * 60 + "\n")
|
| demo = build_demo()
|
|
|
| demo.launch(debug=True, share=True)
|
|
|
