Update app.py

app.py

@@ -12,25 +12,30 @@ import base64
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
-LANGFUSE_PUBLIC_KEY="pk-lf-7f0677a5-09f7-4508-88a2-bf7f44ee172c"
-LF_SECRET_KEY = os.environ.get("LANGFUSE_SECRET_KEY")
-if LF_SECRET_KEY is None:
-    raise EnvironmentError("LANGFUSE_SECRET_KEY environment variable is not set.")
-LANGFUSE_AUTH=base64.b64encode(f"{LANGFUSE_PUBLIC_KEY}:{LF_SECRET_KEY}".encode()).decode()
 
-from opentelemetry.sdk.trace import TracerProvider
-from openinference.instrumentation.smolagents import SmolagentsInstrumentor
-from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
-from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+# Rate limiting configuration
+MAX_CONCURRENT_REQUESTS = 5  # Adjust based on performance needs
+REQUEST_DELAY = 2.0  # 2 seconds delay to meet 30 RPM
 
-os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "https://cloud.langfuse.com/api/public/otel" # EU data region
-os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"
+# LANGFUSE_PUBLIC_KEY="pk-lf-7f0677a5-09f7-4508-88a2-bf7f44ee172c"
+# LF_SECRET_KEY = os.environ.get("LANGFUSE_SECRET_KEY")
+# if LF_SECRET_KEY is None:
+#    raise EnvironmentError("LANGFUSE_SECRET_KEY environment variable is not set.")
+# LANGFUSE_AUTH=base64.b64encode(f"{LANGFUSE_PUBLIC_KEY}:{LF_SECRET_KEY}".encode()).decode()
 
+# from opentelemetry.sdk.trace import TracerProvider
+# from openinference.instrumentation.smolagents import SmolagentsInstrumentor
+# from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+# from opentelemetry.sdk.trace.export import SimpleSpanProcessor
 
-trace_provider = TracerProvider()
-trace_provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter(endpoint="https://cloud.langfuse.com/api/public/otel")))
+# os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "https://cloud.langfuse.com/api/public/otel" # EU data region
+# os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"
 
-SmolagentsInstrumentor().instrument(tracer_provider=trace_provider)
+
+# trace_provider = TracerProvider()
+# trace_provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter(endpoint="https://cloud.langfuse.com/api/public/otel")))
+
+# SmolagentsInstrumentor().instrument(tracer_provider=trace_provider)
 
 async def fetch_questions(session: aiohttp.ClientSession, questions_url: str) -> list:
     """Fetch questions asynchronously."""
@@ -66,6 +71,20 @@ 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:
+            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 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
+
 async def run_and_submit_all(profile: gr.OAuthProfile | None):
     """
     Fetches all questions asynchronously, runs the MagAgent on them, submits all answers,
@@ -104,22 +123,35 @@ 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)
         results_log = []
         answers_payload = []
+        
         print(f"Running agent on {len(questions_data)} questions...")
-        for item in questions_data:
-            task_id = item.get("task_id")
-            question_text = item.get("question")
-            if not task_id or question_text is None:
-                print(f"Skipping item with missing task_id or question: {item}")
-                continue
-            try:
-                submitted_answer = await agent(question_text)
-                answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
-                results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer})            
-            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}"})
+
+        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:
+#            task_id = item.get("task_id")
+#            question_text = item.get("question")
+#            if not task_id or question_text is None:
+#                print(f"Skipping item with missing task_id or question: {item}")
+#                continue
+#            try:
+#                submitted_answer = await agent(question_text)
+#                answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
+#                results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer})            
+#            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}"})
 
         if not answers_payload:
             print("Agent did not produce any answers to submit.")