Spaces:
Sleeping
Sleeping
| import os | |
| import gradio as gr | |
| import requests | |
| import pandas as pd | |
| from smolagents import ToolCallingAgent, OpenAIServerModel | |
| from smolagents import DuckDuckGoSearchTool, PythonInterpreterTool | |
| from agentsTools.toolVisitWebpage import visit_webpage | |
| from agentsTools.tool_fetch_task_file import fetch_task_file | |
| from agentsTools.tool_read_excel_as_json import read_excel_as_json | |
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" | |
| class GeminiReActAgent: | |
| def __init__(self): | |
| self.model = OpenAIServerModel( | |
| model_id="gemini-2.5-pro-preview-03-25", | |
| api_base="https://generativelanguage.googleapis.com/v1beta/", | |
| api_key=os.getenv("GEMINI_API_KEY_1") | |
| ) | |
| self.agent = ToolCallingAgent( | |
| tools=[ | |
| DuckDuckGoSearchTool(), | |
| PythonInterpreterTool(), | |
| visit_webpage, | |
| fetch_task_file, | |
| read_excel_as_json, | |
| ], | |
| model=self.model, | |
| max_steps=8 | |
| ) | |
| def __call__(self, question: str, taskid: str) -> str: | |
| prompt = """ | |
| You are an intelligent assistant answering questions step-by-step. | |
| Use tools only when necessary. You can take up to 8 steps. | |
| You have access to tools such as a web search, Python execution, and file reading. | |
| Use the task ID {tid} if you need to fetch additional data. | |
| Think aloud, justify each step. Conclude your process with: | |
| FINAL ANSWER = [your answer here] | |
| Question: {q} | |
| """.format(tid=taskid, q=question) | |
| result = self.agent.run(prompt) | |
| return result | |
| def submit_eval_run(profile: gr.OAuthProfile | None): | |
| space_id = os.getenv("SPACE_ID") | |
| if profile: | |
| username = profile.username | |
| else: | |
| return "Please login to Hugging Face first.", None | |
| agent = GeminiReActAgent() | |
| agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" | |
| try: | |
| response = requests.get(f"{DEFAULT_API_URL}/questions", timeout=15) | |
| response.raise_for_status() | |
| questions_data = response.json() | |
| except Exception as e: | |
| return f"Failed to fetch 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, 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: | |
| results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"ERROR: {e}"}) | |
| if not answers_payload: | |
| return "Agent did not return any answers.", pd.DataFrame(results_log) | |
| submission_data = { | |
| "username": username, | |
| "agent_code": agent_code, | |
| "answers": answers_payload | |
| } | |
| try: | |
| response = requests.post(f"{DEFAULT_API_URL}/submit", json=submission_data, timeout=60) | |
| response.raise_for_status() | |
| result = response.json() | |
| summary = ( | |
| f"Submission Successful!\n" | |
| f"User: {result.get('username')}\n" | |
| f"Score: {result.get('score')}% ({result.get('correct_count')}/{result.get('total_attempted')})\n" | |
| f"Message: {result.get('message', 'No message received.')}" | |
| ) | |
| return summary, pd.DataFrame(results_log) | |
| except Exception as e: | |
| return f"Submission failed: {e}", pd.DataFrame(results_log) | |
| with gr.Blocks() as demo: | |
| gr.Markdown("# Gemini Agent Submission Tool") | |
| gr.LoginButton() | |
| btn = gr.Button("Submit Answers") | |
| status = gr.Textbox(label="Submission Status") | |
| table = gr.DataFrame(label="Answers Log") | |
| btn.click(fn=submit_eval_run, outputs=[status, table]) | |
| demo.launch(debug=True, share=False) |