File size: 8,190 Bytes
3202249 2fb7003 3202249 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 |
import os
import requests
import logging
import pandas as pd
import gradio as gr
# (Keep Constants as is)
# --- Constants ---
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger(__name__)
class GAIAHFAPIClient:
"""
A client for interacting with the GAIA HF API for the final assignment.
This client handles dealing with the API requests.
"""
def __init__(self, profile: gr.OAuthProfile | None, api_base_url=DEFAULT_API_URL):
# --- Determine HF Space Runtime URL and Repo URL ---
space_id = os.getenv("SPACE_ID") # Get the SPACE_ID for sending link to the code
if profile:
self.username = f"{profile.username}"
logger.info(f"User logged in: {self.username}")
else:
logger.warning("User not logged in.")
return "Please Login to Hugging Face with the button.", None
self.agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
self.questions_url = f"{api_base_url}/questions"
self.random_question_url = f"{api_base_url}/random-question"
self.file_task_url = f"{api_base_url}/files/" + "{}"
self.submit_url = f"{api_base_url}/submit"
self.api_base_url = api_base_url
def submit_answers(self, submission_data, results_log):
try:
response = requests.post(self.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"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.')}"
)
print("Submission successful.")
results_df = pd.DataFrame(results_log)
return final_status, results_df
except requests.exceptions.HTTPError as e:
error_detail = f"Server responded with status {e.response.status_code}."
try:
error_json = e.response.json()
error_detail += f" Detail: {error_json.get('detail', e.response.text)}"
except requests.exceptions.JSONDecodeError:
error_detail += f" Response: {e.response.text[:500]}"
status_message = f"Submission Failed: {error_detail}"
print(status_message)
results_df = pd.DataFrame(results_log)
return status_message, results_df
except requests.exceptions.Timeout:
status_message = "Submission Failed: The request timed out."
print(status_message)
results_df = pd.DataFrame(results_log)
return status_message, results_df
except requests.exceptions.RequestException as e:
status_message = f"Submission Failed: Network error - {e}"
print(status_message)
results_df = pd.DataFrame(results_log)
return status_message, results_df
except Exception as e:
status_message = f"An unexpected error occurred during submission: {e}"
print(status_message)
results_df = pd.DataFrame(results_log)
return status_message, results_df
def get_questions(self):
"""
Fetches the list of questions from the API.
Returns:
List of questions or an error message.
"""
logger.info(f"Fetching questions from: {self.questions_url}")
try:
response = requests.get(self.questions_url, timeout=15)
response.raise_for_status()
questions_data = response.json()
if not questions_data:
logger.warning("Fetched questions list is empty.")
return "Fetched questions list is empty or invalid format.", None
logger.info(f"Fetched {len(questions_data)} questions.")
return questions_data, 'success'
except requests.exceptions.RequestException as e:
logger.error(f"Error fetching questions: {e}")
return f"Error fetching questions: {e}", None
except requests.exceptions.JSONDecodeError as e:
logger.error(f"Error decoding JSON response from questions endpoint: {e}")
logger.error(f"Response text: {response.text[:500]}")
return f"Error decoding server response for questions: {e}", None
except Exception as e:
logger.error(f"An unexpected error occurred fetching questions: {e}")
return f"An unexpected error occurred fetching questions: {e}", None
def get_random_question(self):
"""
Fetches a random question from the API.
Returns:
A random question or an error message.
"""
logger.info(f"Fetching a random question from: {self.random_question_url}")
try:
response = requests.get(self.random_question_url, timeout=15)
response.raise_for_status()
question_data = response.json()
if not question_data:
logger.warning("No random question data received.")
return "No random question data received.", None
logger.info(f"Received random question: {question_data.get('question', 'No question text')}")
return question_data, None
except requests.exceptions.RequestException as e:
logger.error(f"Error fetching random question: {e}")
return f"Error fetching random question: {e}", None
def get_file_task(self, task_id):
"""
Fetches a file task by its ID from the API.
Args:
task_id (str): The ID of the task to fetch.
Returns:
The task data or an error message.
"""
if not task_id:
return "Task ID cannot be empty.", None
_url = self.file_task_url.format(task_id)
logger.info(f"Fetching file task from: {_url}")
try:
response = requests.get(_url, timeout=15)
response.raise_for_status()
# Handle the case where the response is not JSON / text or bytes. Inspect the response object
content_type = response.headers.get("Content-Type", "")
if "application/json" in content_type:
task_data = response.json()
elif "text/" in content_type:
task_data = response.text
else:
# Assume it's a binary file (e.g., CSV, image, etc.)
task_data = response.content
if not task_data:
logger.warning(f"No data found for task ID: {task_id}")
return f"No data found for task ID: {task_id}", None
logger.info(f"Received file task for ID {task_id}: {task_data}")
return task_data, None
except requests.exceptions.RequestException as e:
logger.error(f"Error fetching file task {task_id}: {e}")
return f"Error fetching file task {task_id}: {e}", None
if __name__ == "__main__":
# Example usage
demo = gr.Blocks(title="GAIA HF API Client Demo")
api_client = GAIAHFAPIClient(profile=demo)
random_question, error = api_client.get_random_question()
if error:
logger.error(f"Error fetching random question: {error}")
else:
logger.info(f"Random question: {random_question}")
questions, error = api_client.get_questions()
if not error:
logger.error(f"Error fetching questions: {error}")
else:
logger.info(f"Fetched questions: {questions}")
task_id = 'f918266a-b3e0-4914-865d-4faa564f1aef'
task_data, error = api_client.get_file_task(task_id)
if error:
logger.error(f"Error fetching file task {task_id}: {error}")
else:
logger.info(f"Fetched file task: {task_data}")
|