Code clean up and optimizations

neurons/validator.py

@@ -46,21 +46,21 @@ async def spawn_loops(task_queue, scoring_queue, reward_events):
 
         # -------- Duplicate of create_task_loop ----------
         logger.info("Starting AvailabilityCheckingLoop...")
-        asyncio.create_task(availability_checking_loop.start())
+        await availability_checking_loop.start(name="AvailabilityCheckingLoop")
 
         logger.info("Starting TaskSender...")
-        asyncio.create_task(task_sender.start(task_queue, scoring_queue))
+        await task_sender.start(task_queue, scoring_queue)
 
         logger.info("Starting TaskLoop...")
-        asyncio.create_task(task_loop.start(task_queue, scoring_queue))
+        await task_loop.start(task_queue, scoring_queue)
         # -------------------------------------------------
 
         logger.info("Starting ModelScheduler...")
-        asyncio.create_task(model_scheduler.start(scoring_queue), name="ModelScheduler"),
+        await model_scheduler.start(scoring_queue, name="ModelScheduler")
         logger.info("Starting TaskScorer...")
-        asyncio.create_task(task_scorer.start(scoring_queue, reward_events), name="TaskScorer"),
+        await task_scorer.start(scoring_queue, reward_events, name="TaskScorer")
         logger.info("Starting WeightSetter...")
-        asyncio.create_task(weight_setter.start(reward_events))
+        await weight_setter.start(reward_events, name="WeightSetter")
 
         # Main monitoring loop
         start = time.time()
@@ -122,7 +122,7 @@ async def main():
         processes = []
 
         try:
-            # # Start checking the availability of miners at regular intervals
+            # Start checking the availability of miners at regular intervals
 
             if shared_settings.DEPLOY_SCORING_API:
                 # Use multiprocessing to bypass API blocking issue

prompting/llms/model_manager.py

@@ -159,9 +159,9 @@ class AsyncModelScheduler(AsyncLoopRunner):
     interval: int = 14400
     scoring_queue: list | None = None
 
-    async def start(self, scoring_queue: list):
+    async def start(self, scoring_queue: list, name: str | None = None):
         self.scoring_queue = scoring_queue
-        return await super().start()
+        return await super().start(name=name)
 
     async def initialise_loop(self):
         model_manager.load_always_active_models()

prompting/rewards/scoring.py

@@ -27,10 +27,10 @@ class TaskScorer(AsyncLoopRunner):
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
-    async def start(self, scoring_queue, reward_events):
+    async def start(self, scoring_queue, reward_events, name: str | None = None):
         self.scoring_queue = scoring_queue
         self.reward_events = reward_events
-        return await super().start()
+        return await super().start(name=name)
 
     def add_to_queue(
         self,

prompting/weight_setting/weight_setter.py

@@ -146,7 +146,7 @@ class WeightSetter(AsyncLoopRunner):
     class Config:
         arbitrary_types_allowed = True
 
-    async def start(self, reward_events):
+    async def start(self, reward_events, name: str | None = None):
         self.reward_events = reward_events
         global PAST_WEIGHTS
 
@@ -159,7 +159,7 @@ async def start(self, reward_events):
             PAST_WEIGHTS = []
         except Exception as ex:
             logger.error(f"Couldn't load weights from file: {ex}")
-        return await super().start()
+        return await super().start(name=name)
 
     async def run_step(self):
         await asyncio.sleep(0.01)

scripts/test_api.py

@@ -50,11 +50,12 @@ async def main():
     # Example API key, replace with yours:
     API_KEY = "0566dbe21ee33bba9419549716cd6f1f"
     client = openai.AsyncOpenAI(
-        base_url=f"http://localhost:{PORT}/v1",
+        base_url=f"http://0.0.0.0:{PORT}/v1",
         max_retries=0,
         timeout=Timeout(90, connect=30, read=60),
         api_key=API_KEY,
     )
+    client._client.headers["api-key"] = API_KEY
     response = await make_completion(client=client, prompt="Say 10 random numbers between 1 and 100", stream=True)
     print(response)
 

shared/loop_runner.py

@@ -104,14 +104,14 @@ async def run_loop(self):
             logger.info("Loop has been cleaned up.")
         logger.debug("Exiting run_loop")
 
-    async def start(self):
+    async def start(self, name: str | None = None):
         """Start the loop."""
         if self.running:
             logger.warning("Loop is already running.")
             return
         self.running = True
         logger.debug(f"{self.name}: Starting loop with {'synchronized' if self.sync else 'non-synchronized'} mode")
-        self._task = asyncio.create_task(self.run_loop())
+        self._task = asyncio.create_task(self.run_loop(), name=name)
 
     async def stop(self):
         """Stop the loop."""

shared/settings.py

@@ -87,19 +87,23 @@ class SharedSettings(BaseSettings):
     HF_TOKEN: Optional[str] = Field(None, env="HF_TOKEN")
     DEPLOY_VALIDATOR: bool = Field(True, env="DEPLOY_VALDITAOR")
 
+    # Validator scoring API (.env.validator).
     DEPLOY_SCORING_API: bool = Field(False, env="DEPLOY_SCORING_API")
     SCORING_API_PORT: int = Field(8094, env="SCORING_API_PORT")
     SCORING_ADMIN_KEY: str | None = Field(None, env="SCORING_ADMIN_KEY")
+    SCORE_ORGANICS: bool = Field(False, env="SCORE_ORGANICS")
+
+    # API Management (.env.api).
     API_PORT: int = Field(8005, env="API_PORT")
     API_HOST: str = Field("0.0.0.0", env="API_HOST")
-
-    # API Management.
+    # File with keys used to access API.
     API_KEYS_FILE: str = Field("api_keys.json", env="API_KEYS_FILE")
+    # Admin key used to generate API keys.
     ADMIN_KEY: str | None = Field(None, env="ADMIN_KEY")
+    # API key used to access validator organic scoring mechanism.
     SCORING_KEY: str | None = Field(None, env="SCORING_KEY")
-    SCORE_ORGANICS: bool = Field(False, env="SCORE_ORGANICS")
 
-    # Additional Fields.
+    # Additional Validator Fields.
     NETUID: Optional[int] = Field(61, env="NETUID")
     TEST: bool = False
     OPENAI_API_KEY: Optional[str] = Field(None, env="OPENAI_API_KEY")