|
|
import os |
|
|
import gradio as gr |
|
|
import requests |
|
|
import pandas as pd |
|
|
from smolagents import ToolCallingAgent, tool |
|
|
from duckduckgo_search import DDGS |
|
|
import math |
|
|
|
|
|
|
|
|
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" |
|
|
|
|
|
|
|
|
@tool |
|
|
def duck_search(query: str) -> str: |
|
|
""" |
|
|
Searches the web using DuckDuckGo and returns a short summary. |
|
|
|
|
|
Args: |
|
|
query: The search query string. |
|
|
|
|
|
Returns: |
|
|
A string summarizing the top search results. |
|
|
""" |
|
|
try: |
|
|
results = ddg(query, max_results=3) |
|
|
if results: |
|
|
return "\n\n".join([f"{r['title']}: {r['body']}" for r in results]) |
|
|
else: |
|
|
return "No results found." |
|
|
except Exception as e: |
|
|
return f"Search error: {e}" |
|
|
|
|
|
@tool |
|
|
def calculator(expression: str) -> str: |
|
|
""" |
|
|
Safely evaluates math expressions using Python's math module. |
|
|
|
|
|
Args: |
|
|
expression: A valid math expression as a string (e.g., 'sqrt(16) + 10'). |
|
|
|
|
|
Returns: |
|
|
The result of the evaluated expression or an error message. |
|
|
""" |
|
|
try: |
|
|
result = eval(expression, {"__builtins__": {}}, math.__dict__) |
|
|
return str(result) |
|
|
except Exception as e: |
|
|
return f"Calculation error: {e}" |
|
|
|
|
|
|
|
|
class WebSearchAgent: |
|
|
def __init__(self): |
|
|
self.agent = ToolCallingAgent( |
|
|
name="GAIAWebToolAgent", |
|
|
description="Agent that answers questions using web search and calculator tools.", |
|
|
tools=[duck_search, calculator], |
|
|
step_limit=5, |
|
|
system_prompt=( |
|
|
"You're a helpful reasoning agent. Use available tools like web search " |
|
|
"and calculator to answer the user's question accurately and concisely." |
|
|
), |
|
|
) |
|
|
print("β
Agent initialized.") |
|
|
|
|
|
def __call__(self, question: str) -> str: |
|
|
print(f"π Question: {question}") |
|
|
try: |
|
|
return self.agent.run(question) |
|
|
except Exception as e: |
|
|
print(f"β Agent error: {e}") |
|
|
return f"Error: {e}" |
|
|
|
|
|
|
|
|
def run_and_submit_all(profile: gr.OAuthProfile | None): |
|
|
space_id = os.getenv("SPACE_ID") |
|
|
if profile: |
|
|
username = profile.username |
|
|
print(f"π€ User: {username}") |
|
|
else: |
|
|
return "Please login to Hugging Face.", None |
|
|
|
|
|
agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" |
|
|
questions_url = f"{DEFAULT_API_URL}/questions" |
|
|
submit_url = f"{DEFAULT_API_URL}/submit" |
|
|
|
|
|
try: |
|
|
agent = WebSearchAgent() |
|
|
except Exception as e: |
|
|
return f"Agent initialization error: {e}", None |
|
|
|
|
|
try: |
|
|
print("π₯ Fetching questions...") |
|
|
response = requests.get(questions_url, timeout=15) |
|
|
response.raise_for_status() |
|
|
questions = response.json() |
|
|
if not questions: |
|
|
return "No questions received.", None |
|
|
print(f"β
Retrieved {len(questions)} questions.") |
|
|
except Exception as e: |
|
|
return f"Failed to fetch questions: {e}", None |
|
|
|
|
|
results_log = [] |
|
|
answers_payload = [] |
|
|
|
|
|
for item in questions: |
|
|
task_id = item.get("task_id") |
|
|
question = item.get("question") |
|
|
if not task_id or not question: |
|
|
continue |
|
|
try: |
|
|
answer = agent(question) |
|
|
results_log.append({ |
|
|
"Task ID": task_id, |
|
|
"Question": question, |
|
|
"Submitted Answer": answer |
|
|
}) |
|
|
answers_payload.append({ |
|
|
"task_id": task_id, |
|
|
"submitted_answer": answer |
|
|
}) |
|
|
except Exception as e: |
|
|
error_msg = f"Agent error: {e}" |
|
|
results_log.append({ |
|
|
"Task ID": task_id, |
|
|
"Question": question, |
|
|
"Submitted Answer": error_msg |
|
|
}) |
|
|
|
|
|
if not answers_payload: |
|
|
return "No answers were generated.", pd.DataFrame(results_log) |
|
|
|
|
|
print("π€ Submitting answers...") |
|
|
try: |
|
|
response = requests.post(submit_url, json={ |
|
|
"username": username.strip(), |
|
|
"agent_code": agent_code, |
|
|
"answers": answers_payload |
|
|
}, timeout=60) |
|
|
response.raise_for_status() |
|
|
result = response.json() |
|
|
status = ( |
|
|
f"β
Submission Successful!\n" |
|
|
f"User: {result.get('username')}\n" |
|
|
f"Score: {result.get('score', 'N/A')}% " |
|
|
f"({result.get('correct_count', '?')}/{result.get('total_attempted', '?')} correct)\n" |
|
|
f"Message: {result.get('message', 'No message')}" |
|
|
) |
|
|
return status, pd.DataFrame(results_log) |
|
|
except Exception as e: |
|
|
return f"β Submission failed: {e}", pd.DataFrame(results_log) |
|
|
|
|
|
|
|
|
with gr.Blocks() as demo: |
|
|
gr.Markdown("# π€ GAIA Agent with Web Search & Calculator Tools") |
|
|
gr.Markdown(""" |
|
|
- β
Log in to your Hugging Face account |
|
|
- π Click the button to run and submit your agent |
|
|
- π§ Agent uses DuckDuckGo search + calculator |
|
|
""") |
|
|
gr.LoginButton() |
|
|
run_btn = gr.Button("Run Evaluation & Submit All Answers") |
|
|
status_box = gr.Textbox(label="Status", lines=5) |
|
|
result_table = gr.DataFrame(label="Agent Answers") |
|
|
|
|
|
run_btn.click(fn=run_and_submit_all, outputs=[status_box, result_table]) |
|
|
|
|
|
if __name__ == "__main__": |
|
|
print("π Starting Gradio App...") |
|
|
demo.launch(debug=True, share=False) |
|
|
|