v2.15.4

v2.15.4

Author: bkb2135

neurons/validator.py

@@ -1,25 +1,20 @@
-# ruff: noqa: E402
-from shared import settings
-
-settings.shared_settings = settings.SharedSettings.load(mode="validator")
-shared_settings = settings.shared_settings
-
 import asyncio
 import multiprocessing as mp
+import sys
 import time
 
 import loguru
 import torch
+import wandb
+
+# ruff: noqa: E402
+from shared import settings
+
+shared_settings = settings.shared_settings
+settings.shared_settings = settings.SharedSettings.load(mode="validator")
+
 
-from prompting.api.api import start_scoring_api
-from prompting.llms.model_manager import model_scheduler
 from prompting.llms.utils import GPUInfo
-from prompting.miner_availability.miner_availability import availability_checking_loop
-from prompting.rewards.scoring import task_scorer
-from prompting.tasks.task_creation import task_loop
-from prompting.tasks.task_sending import task_sender
-from prompting.weight_setting.weight_setter import weight_setter
-from shared.profiling import profiler
 
 # Add a handler to write logs to a file
 loguru.logger.add("logfile.log", rotation="1000 MB", retention="10 days", level="DEBUG")
@@ -32,8 +27,34 @@
 
 def create_loop_process(task_queue, scoring_queue, reward_events):
     async def spawn_loops(task_queue, scoring_queue, reward_events):
+        # ruff: noqa: E402
+        wandb.setup()
+        from shared import settings
+
+        settings.shared_settings = settings.SharedSettings.load(mode="validator")
+
+        from prompting.llms.model_manager import model_scheduler
+        from prompting.miner_availability.miner_availability import availability_checking_loop
+        from prompting.rewards.scoring import task_scorer
+        from prompting.tasks.task_creation import task_loop
+        from prompting.tasks.task_sending import task_sender
+        from prompting.weight_setting.weight_setter import weight_setter
+        from shared.profiling import profiler
+
         logger.info("Starting Profiler...")
         asyncio.create_task(profiler.print_stats(), name="Profiler"),
+
+        # -------- Duplicate of create_task_loop ----------
+        logger.info("Starting AvailabilityCheckingLoop...")
+        asyncio.create_task(availability_checking_loop.start())
+
+        logger.info("Starting TaskSender...")
+        asyncio.create_task(task_sender.start(task_queue, scoring_queue))
+
+        logger.info("Starting TaskLoop...")
+        asyncio.create_task(task_loop.start(task_queue, scoring_queue))
+        # -------------------------------------------------
+
         logger.info("Starting ModelScheduler...")
         asyncio.create_task(model_scheduler.start(scoring_queue), name="ModelScheduler"),
         logger.info("Starting TaskScorer...")
@@ -62,6 +83,8 @@ async def spawn_loops(task_queue, scoring_queue, reward_events):
 
 def start_api():
     async def start():
+        from prompting.api.api import start_scoring_api  # noqa: F401
+
         await start_scoring_api()
         while True:
             await asyncio.sleep(10)
@@ -70,21 +93,21 @@ async def start():
     asyncio.run(start())
 
 
-def create_task_loop(task_queue, scoring_queue):
-    async def start(task_queue, scoring_queue):
-        logger.info("Starting AvailabilityCheckingLoop...")
-        asyncio.create_task(availability_checking_loop.start())
+# def create_task_loop(task_queue, scoring_queue):
+#     async def start(task_queue, scoring_queue):
+#         logger.info("Starting AvailabilityCheckingLoop...")
+#         asyncio.create_task(availability_checking_loop.start())
 
-        logger.info("Starting TaskSender...")
-        asyncio.create_task(task_sender.start(task_queue, scoring_queue))
+#         logger.info("Starting TaskSender...")
+#         asyncio.create_task(task_sender.start(task_queue, scoring_queue))
 
-        logger.info("Starting TaskLoop...")
-        asyncio.create_task(task_loop.start(task_queue, scoring_queue))
-        while True:
-            await asyncio.sleep(10)
-            logger.debug("Running task loop...")
+#         logger.info("Starting TaskLoop...")
+#         asyncio.create_task(task_loop.start(task_queue, scoring_queue))
+#         while True:
+#             await asyncio.sleep(10)
+#             logger.debug("Running task loop...")
 
-    asyncio.run(start(task_queue, scoring_queue))
+#     asyncio.run(start(task_queue, scoring_queue))
 
 
 async def main():
@@ -109,23 +132,38 @@ async def main():
             loop_process = mp.Process(
                 target=create_loop_process, args=(task_queue, scoring_queue, reward_events), name="LoopProcess"
             )
-            task_loop_process = mp.Process(
-                target=create_task_loop, args=(task_queue, scoring_queue), name="TaskLoopProcess"
-            )
+            # task_loop_process = mp.Process(
+            #     target=create_task_loop, args=(task_queue, scoring_queue), name="TaskLoopProcess"
+            # )
             loop_process.start()
-            task_loop_process.start()
+            # task_loop_process.start()
             processes.append(loop_process)
-            processes.append(task_loop_process)
+            # processes.append(task_loop_process)
             GPUInfo.log_gpu_info()
 
+            step = 0
             while True:
-                await asyncio.sleep(10)
-                logger.debug("Running...")
+                await asyncio.sleep(30)
+                if (
+                    shared_settings.SUBTENSOR.get_current_block()
+                    - shared_settings.METAGRAPH.last_update[shared_settings.UID]
+                    > 500
+                    and step > 120
+                ):
+                    logger.warning(
+                        f"UPDATES HAVE STALED FOR {shared_settings.SUBTENSOR.get_current_block() - shared_settings.METAGRAPH.last_update[shared_settings.UID]} BLOCKS AND {step} STEPS"
+                    )
+                    logger.warning(
+                        f"STALED: {shared_settings.SUBTENSOR.get_current_block()}, {shared_settings.METAGRAPH.block}"
+                    )
+                    sys.exit(1)
+                step += 1
 
         except Exception as e:
             logger.error(f"Main loop error: {e}")
             raise
         finally:
+            wandb.teardown()
             # Clean up processes
             for process in processes:
                 if process.is_alive():

prompting/rewards/scoring.py

@@ -1,11 +1,11 @@
 import asyncio
 import threading
-from dataclasses import dataclass
 
 from loguru import logger
 from pydantic import ConfigDict
 
 from prompting.llms.model_manager import model_manager, model_scheduler
+from prompting.rewards.scoring_config import ScoringConfig
 from prompting.tasks.base_task import BaseTextTask
 from prompting.tasks.task_registry import TaskRegistry
 from shared.base import DatasetEntry
@@ -14,16 +14,6 @@
 from shared.loop_runner import AsyncLoopRunner
 
 
-@dataclass
-class ScoringConfig:
-    task: BaseTextTask
-    response: DendriteResponseEvent
-    dataset_entry: DatasetEntry
-    block: int
-    step: int
-    task_id: str
-
-
 class TaskScorer(AsyncLoopRunner):
     """The scoring manager maintains a queue of tasks & responses to score and then runs a scoring loop in a background thread.
     This scoring loop will score the responses once the LLM needed is loaded in the model_manager and log the rewards.

prompting/rewards/scoring_config.py

@@ -0,0 +1,15 @@
+from dataclasses import dataclass
+
+from prompting.tasks.base_task import BaseTextTask
+from shared.base import DatasetEntry
+from shared.dendrite import DendriteResponseEvent
+
+
+@dataclass
+class ScoringConfig:
+    task: BaseTextTask
+    response: DendriteResponseEvent
+    dataset_entry: DatasetEntry
+    block: int
+    step: int
+    task_id: str

prompting/tasks/task_creation.py

@@ -6,7 +6,8 @@
 
 from prompting.miner_availability.miner_availability import miner_availabilities
 from prompting.tasks.task_registry import TaskRegistry
-from shared.logging import ErrorLoggingEvent, ValidatorLoggingEvent
+
+# from shared.logging import ErrorLoggingEvent, ValidatorLoggingEvent
 from shared.loop_runner import AsyncLoopRunner
 from shared.settings import shared_settings
 
@@ -26,7 +27,7 @@ async def start(self, task_queue, scoring_queue):
         self.scoring_queue = scoring_queue
         await super().start()
 
-    async def run_step(self) -> ValidatorLoggingEvent | ErrorLoggingEvent | None:
+    async def run_step(self):
         if len(self.task_queue) > shared_settings.TASK_QUEUE_LENGTH_THRESHOLD:
             logger.debug("Task queue is full. Skipping task generation.")
             return None

prompting/tasks/task_sending.py

@@ -9,7 +9,7 @@
 from prompting.miner_availability.miner_availability import miner_availabilities
 
 # from prompting.rewards.scoring import task_scorer
-from prompting.rewards.scoring import ScoringConfig
+from prompting.rewards.scoring_config import ScoringConfig
 from prompting.tasks.base_task import BaseTextTask
 from prompting.tasks.inference import InferenceTask
 from shared.dendrite import DendriteResponseEvent, SynapseStreamResult
@@ -66,6 +66,9 @@ async def collect_responses(task: BaseTextTask) -> DendriteResponseEvent | None:
     response_event = DendriteResponseEvent(
         stream_results=stream_results,
         uids=uids,
+        axons=[
+            shared_settings.METAGRAPH.axons[x].ip + ":" + str(shared_settings.METAGRAPH.axons[x].port) for x in uids
+        ],
         timeout=(
             shared_settings.INFERENCE_TIMEOUT if isinstance(task, InferenceTask) else shared_settings.NEURON_TIMEOUT
         ),

prompting/weight_setting/weight_setter.py

@@ -137,7 +137,7 @@ class WeightSetter(AsyncLoopRunner):
     """The weight setter looks at RewardEvents in the reward_events queue and sets the weights of the miners accordingly."""
 
     sync: bool = True
-    interval: int = 60 * 22  # set rewards every 20 minutes
+    interval: int = 60 * 25  # set rewards every 25 minutes
     reward_events: list[list[WeightedRewardEvent]] | None = None
     subtensor: bt.Subtensor | None = None
     metagraph: bt.Metagraph | None = None
@@ -240,10 +240,8 @@ async def run_step(self):
         set_weights(
             final_rewards, step=self.step, subtensor=shared_settings.SUBTENSOR, metagraph=shared_settings.METAGRAPH
         )
-        self.reward_events = []  # empty reward events queue
-        logger.debug(f"Pre-Refresh Metagraph Block: {shared_settings.METAGRAPH.block}")
-        shared_settings.refresh_metagraph()
-        logger.debug(f"Post-Refresh Metagraph Block: {shared_settings.METAGRAPH.block}")
+        # TODO: empty rewards queue only on weight setting success
+        self.reward_events[:] = []  # empty reward events queue
         await asyncio.sleep(0.01)
         return final_rewards
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "prompting"
-version = "2.15.3"
+version = "2.15.4"
 description = "Subnetwork 1 runs on Bittensor and is maintained by Macrocosmos. It's an effort to create decentralised AI"
 authors = ["Kalei Brady, Dmytro Bobrenko, Felix Quinque, Steffen Cruz, Richard Wardle"]
 readme = "README.md"

shared/dendrite.py

@@ -35,6 +35,7 @@ def model_dump(self):
 
 class DendriteResponseEvent(BaseModel):
     uids: np.ndarray | list[float]
+    axons: list[str]
     timeout: float
     stream_results: list[SynapseStreamResult]
     completions: list[str] = []

shared/epistula.py

@@ -124,7 +124,7 @@ async def query_miners(uids, body: dict[str, Any]):
         exceptions = [resp for resp in responses if isinstance(resp, Exception)]
         if exceptions:
             for exc in exceptions:
-                logger.error(f"Error in make_openai_query: {exc}")
+                logger.debug(f"Error in make_openai_query: {exc}")
 
         # 'responses' is a list of SynapseStreamResult objects
         results = []

shared/misc.py

@@ -138,3 +138,34 @@ def serialize_exception_to_string(e):
         return serialized_str
     else:
         return e
+
+
+def cached_property_with_expiration(expiration_seconds=1200):
+    """
+    Decorator that caches the property's value for `expiration_seconds` seconds.
+    After this duration, the cached value is refreshed.
+    """
+
+    def decorator(func):
+        attr_name = f"_cached_{func.__name__}"
+
+        @property
+        def wrapper(self):
+            now = time.time()
+
+            # Check if we have a cached value and if it's still valid
+            if hasattr(self, attr_name):
+                cached_value, timestamp = getattr(self, attr_name)
+
+                # If valid, return cached value
+                if now - timestamp < expiration_seconds:
+                    return cached_value
+
+            # Otherwise, compute the new value and cache it
+            value = func(self)
+            setattr(self, attr_name, (value, now))
+            return value
+
+        return wrapper
+
+    return decorator