Spaces:
Running
Running
| """ | |
| Gradio interface for the smolagents CodeAgent. | |
| By default this script will NOT launch a local Gradio server. To allow | |
| local runs (for testing) set the environment variable `RUN_LOCAL=1` or pass | |
| `--run-local` on the command line. This prevents accidental local launches | |
| when you intended to deploy to Hugging Face Spaces. | |
| """ | |
| import os | |
| import sys | |
| import logging | |
| import json | |
| from typing import List, Dict, Any | |
| from code_agent import run_agent | |
| # (Gradio update/analytics env flags removed per user request) | |
| logger = logging.getLogger(__name__) | |
| logging.basicConfig(level=logging.INFO) | |
| def respond(prompt: str) -> str: | |
| if not prompt or not prompt.strip(): | |
| return "Please provide a prompt." | |
| try: | |
| return run_agent(prompt) | |
| except Exception as e: | |
| logger.error("Agent failed: %s", e) | |
| return f"Agent error: {type(e).__name__}: {str(e)[:200]}" | |
| def extract_prompt_from_question(q: Dict[str, Any]) -> str: | |
| """Extract the actual question/prompt from a question dict.""" | |
| for key in ("question", "prompt", "input", "text", "task"): | |
| if key in q and isinstance(q[key], str): | |
| return q[key] | |
| return str(q) | |
| def fetch_all_questions() -> List[Dict[str, Any]]: | |
| """Fetch all questions from the scoring API.""" | |
| try: | |
| from evaluation_client import ScoringAPIClient | |
| client = ScoringAPIClient() | |
| questions = client.get_questions() | |
| return questions if questions else [] | |
| except Exception as e: | |
| logger.error("Failed to fetch questions: %s", e) | |
| return [] | |
| def answer_all_questions(questions: List[Dict[str, Any]]) -> List[Dict[str, Any]]: | |
| """Answer all questions and return answers in submission format. | |
| Args: | |
| questions: List of question dicts from the scoring API. | |
| Returns: | |
| List of dicts with keys: task_id, submitted_answer. | |
| """ | |
| answers = [] | |
| total = len(questions) | |
| for idx, q in enumerate(questions, start=1): | |
| task_id = q.get("task_id") or q.get("id") or q.get("taskId") | |
| if not task_id: | |
| logger.warning("Question %d/%d: Missing task_id, skipping", idx, total) | |
| continue | |
| prompt = extract_prompt_from_question(q) | |
| logger.info("Question %d/%d: task_id=%s, prompt_len=%d", idx, total, task_id, len(prompt)) | |
| try: | |
| ans = run_agent(prompt) | |
| ans = ans.strip() | |
| logger.info(" ✓ Answer: %s", ans[:60]) | |
| answers.append({"task_id": task_id, "submitted_answer": ans}) | |
| except Exception as e: | |
| logger.error(" ❌ Failed to answer: %s", type(e).__name__) | |
| answers.append({"task_id": task_id, "submitted_answer": f"(error) {type(e).__name__}"}) | |
| logger.info("✓ Answered %d/%d questions", len(answers), total) | |
| return answers | |
| def submit_answers(username: str, agent_code_url: str, answers: List[Dict[str, Any]]) -> str: | |
| """Submit answers to the scoring API.""" | |
| try: | |
| from evaluation_client import ScoringAPIClient | |
| client = ScoringAPIClient() | |
| resp = client.submit(username=username, agent_code=agent_code_url, answers=answers) | |
| return f"✓ Submission successful! Response: {resp}" | |
| except Exception as e: | |
| logger.error("Submission failed: %s", e) | |
| return f"ERROR: Submission failed: {e}" | |
| _demo = None | |
| def _get_demo(): | |
| """Lazily import gradio and construct the demo to avoid network calls on import.""" | |
| global _demo | |
| if _demo is not None: | |
| return _demo | |
| try: | |
| import gradio as gr | |
| except Exception as e: | |
| logger.error("Failed to import gradio: %s", e) | |
| raise | |
| # We'll provide a readonly prompt field populated from the scoring API | |
| # and buttons to fetch a random task and run the agent on it. | |
| def fetch_random_task(): | |
| try: | |
| from evaluation_client import ScoringAPIClient | |
| client = ScoringAPIClient() | |
| q = client.get_random_question() | |
| if not q: | |
| return "(no question returned)" | |
| # extract prompt safely | |
| for key in ("question", "prompt", "input", "text", "task"): | |
| if key in q and isinstance(q[key], str): | |
| return q[key] | |
| return str(q) | |
| except Exception as e: | |
| logger.error("Failed to fetch random question: %s", e) | |
| return f"(fetch error) {type(e).__name__}: {str(e)[:200]}" | |
| def run_on_current(prompt_text: str): | |
| if not prompt_text or not prompt_text.strip(): | |
| return "(no prompt to run on)" | |
| try: | |
| return respond(prompt_text) | |
| except Exception as e: | |
| logger.error("Agent run failed: %s", e) | |
| return f"Agent error: {type(e).__name__}: {str(e)[:200]}" | |
| with gr.Blocks() as demo: | |
| gr.Markdown("# Agents Course — Final Agent Demo") | |
| # Single question interface | |
| with gr.Row(): | |
| prompt_box = gr.Textbox(label="Fetched Prompt (read-only)", lines=6) | |
| with gr.Row(): | |
| fetch_btn = gr.Button("Fetch Random Task") | |
| run_btn = gr.Button("Run Agent on Fetched Task") | |
| single_output = gr.Textbox(label="Agent Response", lines=6) | |
| # Batch processing interface | |
| gr.Markdown("## Batch Processing") | |
| questions_state = gr.State([]) | |
| answers_state = gr.State([]) | |
| status_box = gr.Textbox(label="Status", lines=4) | |
| with gr.Row(): | |
| fetch_all_btn = gr.Button("Fetch All Questions") | |
| answer_all_btn = gr.Button("Answer All Questions") | |
| submit_btn = gr.Button("Submit All Answers") | |
| with gr.Row(): | |
| username_box = gr.Textbox(label="Hugging Face Username", placeholder="your_hf_username") | |
| agent_url_box = gr.Textbox(label="Agent Code URL", placeholder="https://huggingface.co/spaces/...") | |
| batch_output = gr.Textbox(label="Batch Results", lines=12) | |
| # Wire up the buttons | |
| fetch_btn.click(fn=fetch_random_task, inputs=[], outputs=[prompt_box]) | |
| run_btn.click(fn=run_on_current, inputs=[prompt_box], outputs=[single_output]) | |
| def fetch_all_questions_wrapper(): | |
| questions = fetch_all_questions() | |
| status = f"Fetched {len(questions)} questions" | |
| return questions, [], status | |
| def answer_all_questions_wrapper(questions): | |
| if not questions: | |
| return [], "No questions to answer" | |
| answers = answer_all_questions(questions) | |
| status = f"Answered {len(answers)} questions" | |
| return answers, status | |
| def submit_answers_wrapper(username, agent_url, answers): | |
| if not answers: | |
| return "No answers to submit" | |
| if not username or not agent_url: | |
| return "Please provide both username and agent code URL" | |
| return submit_answers(username, agent_url, answers) | |
| fetch_all_btn.click( | |
| fn=fetch_all_questions_wrapper, | |
| inputs=[], | |
| outputs=[questions_state, answers_state, status_box] | |
| ) | |
| answer_all_btn.click( | |
| fn=answer_all_questions_wrapper, | |
| inputs=[questions_state], | |
| outputs=[answers_state, status_box] | |
| ) | |
| submit_btn.click( | |
| fn=submit_answers_wrapper, | |
| inputs=[username_box, agent_url_box, answers_state], | |
| outputs=[batch_output] | |
| ) | |
| _demo = demo | |
| return _demo | |
| if __name__ == "__main__": | |
| # Launch unconditionally when executed as a script. | |
| try: | |
| _get_demo().launch(server_name="0.0.0.0", server_port=7860, share=False) | |
| except Exception as e: | |
| logger.error("Failed to launch Gradio demo: %s", e) | |