Spaces:
Sleeping
Sleeping
import os | |
import gradio as gr | |
import requests | |
import pandas as pd | |
import google.generativeai as genai | |
from smolagents import CodeAgent, DuckDuckGoSearchTool | |
# Define the system prompt | |
SYSTEM_PROMPT = """You are a general AI assistant. I will ask you a question. | |
Report your thoughts, and finish your answer with just the answer — no prefixes like "FINAL ANSWER:". | |
Your answer should be a number OR as few words as possible OR a comma-separated list of numbers and/or strings. | |
If you're asked for a number, don’t use commas or units like $ or %, unless specified. | |
If you're asked for a string, don’t use articles or abbreviations (e.g. for cities), and write digits in plain text unless told otherwise.""" | |
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" | |
# Gemini model wrapper (lightweight, no smolagents.model.base) | |
class GeminiFlashModel: | |
def __init__(self, model_name="gemini-1.5-flash", api_key=None): | |
self.model_name = model_name | |
self.api_key = api_key or os.getenv("GEMINI_API_KEY") | |
if not self.api_key: | |
raise ValueError("GEMINI_API_KEY is not set.") | |
genai.configure(api_key=self.api_key) | |
self.model = genai.GenerativeModel(model_name) | |
def generate(self, messages, stop_sequences=None, **kwargs): | |
# Insert system prompt if missing | |
if isinstance(messages, list): | |
if not any(m["role"] == "system" for m in messages): | |
messages = [{"role": "system", "content": SYSTEM_PROMPT}] + messages | |
else: | |
raise TypeError("Expected 'messages' to be a list of dicts.") | |
prompt = "\n".join(f"{m['role'].capitalize()}: {m['content']}" for m in messages) | |
try: | |
response = self.model.generate_content(prompt) | |
return response.text.strip() | |
except Exception as e: | |
return f"GENERATION ERROR: {e}" | |
# Agent using Gemini | |
class MyAgent: | |
def __init__(self): | |
self.model = GeminiFlashModel(model_name="gemini-1.5-flash") | |
self.agent = CodeAgent(tools=[DuckDuckGoSearchTool()], model=self.model) | |
def __call__(self, question: str) -> str: | |
return self.agent.run(question) | |
def run_and_submit_all(profile: gr.OAuthProfile | None): | |
space_id = os.getenv("SPACE_ID") | |
if profile: | |
username = profile.username | |
print(f"User logged in: {username}") | |
else: | |
print("User not logged in.") | |
return "Please login to Hugging Face.", None | |
questions_url = f"{DEFAULT_API_URL}/questions" | |
submit_url = f"{DEFAULT_API_URL}/submit" | |
try: | |
agent = MyAgent() | |
except Exception as e: | |
return f"Error initializing agent: {e}", None | |
try: | |
response = requests.get(questions_url, timeout=15) | |
response.raise_for_status() | |
questions_data = response.json() | |
except Exception as e: | |
return f"Error fetching questions: {e}", None | |
results_log = [] | |
answers_payload = [] | |
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: | |
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: | |
results_log.append({ | |
"Task ID": task_id, | |
"Question": question_text, | |
"Submitted Answer": f"AGENT ERROR: {e}" | |
}) | |
if not answers_payload: | |
return "Agent did not return any answers.", pd.DataFrame(results_log) | |
submission_data = { | |
"username": profile.username.strip(), | |
"agent_code": f"https://huggingface.co/spaces/{space_id}/tree/main", | |
"answers": answers_payload | |
} | |
try: | |
response = requests.post(submit_url, json=submission_data, timeout=60) | |
response.raise_for_status() | |
result_data = response.json() | |
final_status = ( | |
f"Submission Successful!\n" | |
f"User: {result_data.get('username')}\n" | |
f"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.')}" | |
) | |
return final_status, pd.DataFrame(results_log) | |
except Exception as e: | |
return f"Submission failed: {e}", pd.DataFrame(results_log) | |
# Gradio UI | |
with gr.Blocks() as demo: | |
gr.Markdown("# Basic Agent Evaluation Runner") | |
gr.Markdown(""" | |
**Instructions:** | |
1. Clone this space and configure your Gemini API key. | |
2. Log in to Hugging Face. | |
3. Run your agent on evaluation tasks and submit answers. | |
""") | |
gr.LoginButton() | |
run_button = gr.Button("Run Evaluation & Submit All Answers") | |
status_output = gr.Textbox(label="Submission Result", lines=5, interactive=False) | |
results_table = gr.DataFrame(label="Results", wrap=True) | |
run_button.click(fn=run_and_submit_all, outputs=[status_output, results_table]) | |
if __name__ == "__main__": | |
print("🔧 App starting...") | |
demo.launch(debug=True, share=False) | |