Update app.py

app.py

@@ -15,15 +15,16 @@ import aiohttp
 import asyncio
 import json
 from agent import MagAgent  
-import base64
+from token_bucket import TokenBucket
 
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
 # Rate limiting configuration
-MAX_CONCURRENT_REQUESTS = 1  # Adjust based on performance needs
-REQUEST_DELAY = 9.0  # 2 seconds delay to meet 30 RPM
+RATE_LIMIT = 15  # Requests per minute
+TOKEN_BUCKET_CAPACITY = RATE_LIMIT
+TOKEN_BUCKET_REFILL_RATE = RATE_LIMIT / 60.0  # Tokens per second
 
 async def fetch_questions(session: aiohttp.ClientSession, questions_url: str) -> list:
     """Fetch questions asynchronously."""
@@ -44,7 +45,8 @@ async def fetch_questions(session: aiohttp.ClientSession, questions_url: str) ->
         print(f"An unexpected error occurred fetching questions: {e}")
         return None
 
-async def submit_answers(session: aiohttp.ClientSession, submit_url: str, submission_data: dict) -> dict:
+async def submit_answers(session: aiohttp.ClientSession, submit_url: str,
+ submission_data: dict) -> dict:
     """Submit answers asynchronously."""
     try:
         async with session.post(submit_url, json=submission_data, timeout=60) as response:
@@ -60,20 +62,28 @@ async def submit_answers(session: aiohttp.ClientSession, submit_url: str, submis
         print(f"An unexpected error occurred during submission: {e}")
         return None
 
-async def process_question(agent, question_text: str, task_id: str, semaphore: asyncio.Semaphore, results_log: list):
-    """Process a single question with rate limiting."""
-    async with semaphore:
-        try:
+async def process_question(agent, question_text: str, task_id: str, results_log: list):
+    """Process a single question with global rate limiting."""
+    try:
+        # Wait for a token before proceeding
+        await token_bucket.consume(1)
+        submitted_answer = await agent(question_text)
+        results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer})
+        return {"task_id": task_id, "submitted_answer": submitted_answer}
+    except aiohttp.ClientResponseError as e:
+        if e.status == 429:
+            print(f"Rate limit hit for task {task_id}. Retrying after delay...")
+            await asyncio.sleep(60 / RATE_LIMIT)  # Wait before retry
+            await token_bucket.consume(1)
             submitted_answer = await agent(question_text)
             results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer})
             return {"task_id": task_id, "submitted_answer": submitted_answer}
-            await asyncio.sleep(REQUEST_DELAY)  # Enforce delay after each request
-        except Exception as e:
-            print(f"Error running agent on task {task_id}: {e}")
-            results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"})
-            return None
-        finally:
-            await asyncio.sleep(REQUEST_DELAY)  # Enforce delay after each request
+        else:
+            raise
+    except Exception as e:
+        print(f"Error running agent on task {task_id}: {e}")
+        results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"})
+        return None
 
 async def run_and_submit_all(profile: gr.OAuthProfile | None):
     """
@@ -113,26 +123,21 @@ async def run_and_submit_all(profile: gr.OAuthProfile | None):
             return "Fetched questions list is empty or invalid format.", None
 
         # 3. Run Agent on Questions
-        
-        # Initialize semaphore and results log
-        semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
+        # Process questions sequentially with rate limiting
         results_log = []
         answers_payload = []
-        
         print(f"Running agent on {len(questions_data)} questions...")
 
-        tasks = [
-            process_question(agent, item["question"], item["task_id"], semaphore, results_log)
-            for item in questions_data
-            if item.get("task_id") and item.get("question") is not None
-        ]
-        results = await asyncio.gather(*tasks)
-        answers_payload = [r for r in results if r is not None]
-        
+        for item in questions_data:
+            if item.get("task_id") and item.get("question"):
+                result = await process_question(agent, item["question"], item["task_id"], results_log)
+                if result:
+                    answers_payload.append(result)
+
         if not answers_payload:
             print("Agent did not produce any answers to submit.")
             return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)
-
+            
         # 4. Prepare Submission
         submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}
         status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..."