GAIA_Evaluation / questions.py
denven
Leverage LangGraph & LLM models to implement
bd62c9a
"""
Questions API Module
This module provides a client for interacting with the agents-course-unit4-scoring API.
It implements methods for the 4 endpoints available in the API.
"""
import requests
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
import json
@dataclass
class Question:
"""Data class representing a question"""
id: str
question: str
level: int
metadata: Optional[Dict[str, Any]] = None
class QuestionsAPI:
"""Client for the agents-course-unit4-scoring API"""
def __init__(self, base_url: str = "https://agents-course-unit4-scoring.hf.space"):
"""
Initialize the QuestionsAPI client.
Args:
base_url: The base URL of the API
"""
self.base_url = base_url.rstrip('/')
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"Accept": "application/json"
})
def get_questions(self, level: Optional[int] = None, limit: int = 10) -> List[Dict[str, Any]]:
"""
Get a list of questions from the API.
Args:
level: Optional level filter (1, 2, or 3)
limit: Maximum number of questions to retrieve
Returns:
List of question dictionaries
Raises:
requests.RequestException: If the API request fails
"""
url = f"{self.base_url}/questions"
params = {"limit": limit}
if level is not None:
params["level"] = level
try:
response = self.session.get(url, params=params)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
print(f"Error getting questions: {e}")
raise
def get_question_by_id(self, question_id: str) -> Dict[str, Any]:
"""
Get a specific question by its ID.
Args:
question_id: The ID of the question to retrieve
Returns:
Question dictionary
Raises:
requests.RequestException: If the API request fails
"""
url = f"{self.base_url}/questions/{question_id}"
try:
response = self.session.get(url)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
print(f"Error getting question {question_id}: {e}")
raise
def submit_answer(self, question_id: str, answer: str, hf_token: Optional[str] = None) -> Dict[str, Any]:
"""
Submit an answer for a specific question.
Args:
question_id: The ID of the question being answered
answer: The answer to submit
hf_token: Optional Hugging Face token for authentication
Returns:
Response dictionary with score and feedback
Raises:
requests.RequestException: If the API request fails
"""
url = f"{self.base_url}/submit"
data = {
"question_id": question_id,
"answer": answer
}
headers = {}
if hf_token:
headers["Authorization"] = f"Bearer {hf_token}"
try:
response = self.session.post(url, json=data, headers=headers)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
print(f"Error submitting answer for question {question_id}: {e}")
raise
def get_leaderboard(self, limit: int = 50) -> List[Dict[str, Any]]:
"""
Get the current leaderboard.
Args:
limit: Maximum number of leaderboard entries to retrieve
Returns:
List of leaderboard entries
Raises:
requests.RequestException: If the API request fails
"""
url = f"{self.base_url}/leaderboard"
params = {"limit": limit}
try:
response = self.session.get(url, params=params)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
print(f"Error getting leaderboard: {e}")
raise
def close(self):
"""Close the session"""
self.session.close()
def __enter__(self):
"""Context manager entry"""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit"""
self.close()
# Example usage
if __name__ == "__main__":
# Initialize the API client
api = QuestionsAPI()
try:
# Example 1: Get questions
print("Getting questions...")
questions = api.get_questions(level=1, limit=5)
print(f"Retrieved {len(questions)} questions")
for q in questions[:2]: # Print first 2 questions
print(f"- ID: {q.get('id', 'N/A')}, Question: {q.get('question', 'N/A')[:100]}...")
# Example 2: Get a specific question
if questions:
question_id = questions[0].get('id', '')
if question_id:
print(f"\nGetting question {question_id}...")
question = api.get_question_by_id(question_id)
print(f"Question: {question.get('question', 'N/A')}")
# Example 3: Submit an answer (example only - would need real answer)
# Note: This would typically require authentication
# result = api.submit_answer(
# question_id="example-id",
# answer="Example answer",
# hf_token="your-hf-token"
# )
# print(f"Score: {result.get('score')}")
# Example 4: Get leaderboard
print("\nGetting leaderboard...")
leaderboard = api.get_leaderboard(limit=10)
print(f"Top {len(leaderboard)} entries:")
for i, entry in enumerate(leaderboard[:3]): # Print top 3
print(f"{i+1}. {entry.get('username', 'Anonymous')} - Score: {entry.get('score', 0)}")
except Exception as e:
print(f"Error: {e}")
finally:
api.close()