| import json |
| import os |
| import time |
| from pathlib import Path |
| from tempfile import TemporaryDirectory |
| from urllib.parse import quote |
|
|
| import gradio as gr |
| import pandas as pd |
| import requests |
|
|
| try: |
| from dotenv import load_dotenv |
| except ImportError: |
| load_dotenv = None |
|
|
| try: |
| import spaces |
| except ImportError: |
| spaces = None |
|
|
| from agent import BasicAgent |
|
|
|
|
| if load_dotenv: |
| load_dotenv() |
|
|
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" |
| CACHE_PATH = Path("answers_cache.json") |
|
|
|
|
| def log(message: str) -> None: |
| print(f"[{time.strftime('%H:%M:%S')}] {message}", flush=True) |
|
|
|
|
| if spaces: |
| @spaces.GPU(duration=1) |
| def dummy_gpu(): |
| return None |
|
|
|
|
| def load_cache() -> dict: |
| if not CACHE_PATH.exists(): |
| return {} |
| try: |
| return json.loads(CACHE_PATH.read_text(encoding="utf-8")) |
| except (json.JSONDecodeError, OSError): |
| return {} |
|
|
|
|
| def save_cache(cache: dict) -> None: |
| CACHE_PATH.write_text(json.dumps(cache, indent=2, ensure_ascii=False), encoding="utf-8") |
|
|
|
|
| def fetch_questions(api_url: str) -> list[dict]: |
| questions_url = f"{api_url}/questions" |
| log(f"Fetching questions from {questions_url}") |
| response = requests.get(questions_url, timeout=30) |
| response.raise_for_status() |
| questions = response.json() |
| if not isinstance(questions, list) or not questions: |
| raise ValueError("Questions endpoint returned an empty or invalid payload.") |
| return questions |
|
|
|
|
| def download_attachment(item: dict, api_url: str, temp_dir: str) -> tuple[str | None, str | None]: |
| file_name = item.get("file_name") |
| task_id = item.get("task_id") |
| if not file_name or not task_id: |
| return None, None |
|
|
| safe_name = Path(str(file_name)).name |
| encoded_task_id = quote(str(task_id), safe="") |
| attachment_path = Path(temp_dir) / f"{encoded_task_id}_{safe_name}" |
| file_url = f"{api_url}/files/{encoded_task_id}" |
|
|
| try: |
| log(f"Downloading attachment for task {task_id}: {safe_name}") |
| response = requests.get(file_url, timeout=60) |
| response.raise_for_status() |
| except requests.exceptions.HTTPError as exc: |
| if exc.response is not None and exc.response.status_code == 404: |
| log(f"Attachment 404 for task {task_id}: {safe_name}") |
| return None, f"Attachment listed as {safe_name}, but the scoring API returned 404 for {file_url}." |
| raise |
|
|
| attachment_path.write_bytes(response.content) |
| log(f"Attachment saved for task {task_id}: {attachment_path}") |
| return str(attachment_path), None |
|
|
|
|
| def rows_to_dataframe(rows: list[dict]) -> pd.DataFrame: |
| return pd.DataFrame( |
| rows, |
| columns=["Task ID", "Question", "File", "Submitted Answer", "Source", "Status"], |
| ) |
|
|
|
|
| def cache_to_payload(cache: dict) -> list[dict]: |
| payload = [] |
| for task_id, item in cache.items(): |
| submitted_answer = item.get("submitted_answer") |
| if submitted_answer and not str(submitted_answer).startswith("AGENT ERROR:"): |
| payload.append({"task_id": task_id, "submitted_answer": submitted_answer}) |
| return payload |
|
|
|
|
| def run_agent(profile: gr.OAuthProfile | None, run_all_questions: bool, use_cache: bool): |
| if not profile: |
| return "Please log in to Hugging Face first.", [], rows_to_dataframe([]) |
|
|
| api_url = DEFAULT_API_URL |
| cache = load_cache() |
| answers_payload = [] |
| results_log = [] |
|
|
| try: |
| questions_data = fetch_questions(api_url) |
| except Exception as exc: |
| return f"Error fetching questions: {exc}", [], rows_to_dataframe([]) |
|
|
| if not run_all_questions: |
| questions_data = questions_data[:1] |
|
|
| try: |
| agent = BasicAgent() |
| except Exception as exc: |
| return f"Error initializing agent: {exc}", [], rows_to_dataframe([]) |
|
|
| with TemporaryDirectory() as temp_dir: |
| for index, item in enumerate(questions_data, start=1): |
| task_id = item.get("task_id") |
| question_text = item.get("question") |
| file_name = item.get("file_name") or "" |
|
|
| if not task_id or question_text is None: |
| results_log.append( |
| { |
| "Task ID": task_id or "missing", |
| "Question": question_text or "", |
| "File": file_name, |
| "Submitted Answer": "", |
| "Source": "skip", |
| "Status": "Missing task_id or question.", |
| } |
| ) |
| continue |
|
|
| cache_key = str(task_id) |
| log(f"Task {index}/{len(questions_data)} started: {task_id}") |
| if use_cache and cache_key in cache: |
| submitted_answer = cache[cache_key]["submitted_answer"] |
| source = "cache" |
| status = "Loaded from answers_cache.json." |
| log(f"Task {index}/{len(questions_data)} cache hit: {task_id} -> {submitted_answer}") |
| else: |
| try: |
| attachment_path, attachment_error = download_attachment(item, api_url, temp_dir) |
| question_for_agent = question_text |
| if attachment_error: |
| question_for_agent += f"\n\n{attachment_error}" |
| submitted_answer = agent( |
| question=question_for_agent, |
| task_id=str(task_id), |
| file_path=attachment_path, |
| file_name=file_name or None, |
| ) |
| source = "agent" |
| status = "Generated." |
| cache[cache_key] = { |
| "question": question_text, |
| "file_name": file_name, |
| "submitted_answer": submitted_answer, |
| } |
| save_cache(cache) |
| log(f"Task {index}/{len(questions_data)} cached answer: {task_id}") |
| except Exception as exc: |
| submitted_answer = f"AGENT ERROR: {exc}" |
| source = "error" |
| status = str(exc) |
| log(f"Task {index}/{len(questions_data)} error: {task_id}: {exc}") |
|
|
| answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer}) |
| results_log.append( |
| { |
| "Task ID": task_id, |
| "Question": question_text, |
| "File": file_name, |
| "Submitted Answer": submitted_answer, |
| "Source": source, |
| "Status": status, |
| } |
| ) |
| log(f"[{index}/{len(questions_data)}] {task_id}: {submitted_answer}") |
|
|
| status = ( |
| f"Generated {len(answers_payload)} answer(s) for {profile.username}. " |
| "Review the table, then submit when ready." |
| ) |
| return status, answers_payload, rows_to_dataframe(results_log) |
|
|
|
|
| def submit_answers(profile: gr.OAuthProfile | None, answers_payload: list[dict] | None): |
| if not profile: |
| return "Please log in to Hugging Face first." |
| source_note = "current run" |
| if not answers_payload: |
| answers_payload = cache_to_payload(load_cache()) |
| if not answers_payload: |
| return "No answers to submit. Run the agent first." |
| source_note = "answers_cache.json" |
|
|
| space_id = os.getenv("SPACE_ID") |
| if space_id: |
| agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" |
| else: |
| agent_code = f"https://huggingface.co/spaces/{profile.username}/YOUR_SPACE_NAME/tree/main" |
|
|
| submission_data = { |
| "username": profile.username.strip(), |
| "agent_code": agent_code, |
| "answers": answers_payload, |
| } |
|
|
| try: |
| response = requests.post(f"{DEFAULT_API_URL}/submit", json=submission_data, timeout=90) |
| response.raise_for_status() |
| result_data = response.json() |
| return ( |
| "Submission Successful!\n" |
| f"User: {result_data.get('username')}\n" |
| f"Submitted: {len(answers_payload)} answer(s) from {source_note}\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" |
| f"Code: {agent_code}" |
| ) |
| except requests.exceptions.HTTPError as exc: |
| detail = exc.response.text |
| try: |
| detail = exc.response.json().get("detail", detail) |
| except requests.exceptions.JSONDecodeError: |
| pass |
| return f"Submission failed: {exc.response.status_code}. {detail}" |
| except Exception as exc: |
| return f"Submission failed: {exc}" |
|
|
|
|
| def run_and_submit_all(profile: gr.OAuthProfile | None): |
| status, answers_payload, results_df = run_agent( |
| profile=profile, |
| run_all_questions=True, |
| use_cache=True, |
| ) |
| if not answers_payload: |
| return status, results_df |
|
|
| submit_status = submit_answers(profile, answers_payload) |
| return f"{status}\n\n{submit_status}", results_df |
|
|
|
|
| with gr.Blocks() as demo: |
| gr.Markdown("# Basic Agent Evaluation Runner") |
| gr.Markdown( |
| """ |
| **Instructions:** |
| |
| 1. Log in to your Hugging Face account using the button below. |
| 2. Click 'Run Evaluation & Submit All Answers' to fetch questions, run the agent, submit answers, and see the score. |
| |
| --- |
| **Disclaimers:** |
| This can take quite some time while the agent works through all questions. |
| """ |
| ) |
|
|
| gr.LoginButton() |
|
|
| run_button = gr.Button("Run Evaluation & Submit All Answers") |
|
|
| status_output = gr.Textbox(label="Run Status / Submission Result", lines=8, interactive=False) |
| results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True) |
|
|
| run_button.click( |
| fn=run_and_submit_all, |
| outputs=[status_output, results_table], |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| print("\n" + "-" * 30 + " App Starting " + "-" * 30) |
| space_host_startup = os.getenv("SPACE_HOST") |
| space_id_startup = os.getenv("SPACE_ID") |
|
|
| if space_host_startup: |
| print(f"SPACE_HOST found: {space_host_startup}") |
| print(f"Runtime URL should be: https://{space_host_startup}.hf.space") |
| else: |
| print("SPACE_HOST environment variable not found (running locally?).") |
|
|
| if space_id_startup: |
| print(f"SPACE_ID found: {space_id_startup}") |
| print(f"Repo URL: https://huggingface.co/spaces/{space_id_startup}") |
| else: |
| print("SPACE_ID environment variable not found. Repo URL cannot be determined.") |
|
|
| print("-" * 74 + "\n") |
| demo.launch(debug=True, share=False) |
|
|
