diff --git a/.github/workflows/docker_build.yaml b/.github/workflows/docker_build.yaml index 2effd247..ddda3a98 100644 --- a/.github/workflows/docker_build.yaml +++ b/.github/workflows/docker_build.yaml @@ -6,6 +6,7 @@ on: - dev - staging - main + - simulator jobs: docker_publish: @@ -28,7 +29,7 @@ jobs: echo "BRANCH_NAME=$SANITIZED_BRANCH_NAME" >> $GITHUB_ENV - name: Build and Push Docker Image with Branch Tag - if: github.ref == 'refs/heads/dev' || github.ref == 'refs/heads/staging' || github.ref == 'refs/heads/main' + if: github.ref == 'refs/heads/dev' || github.ref == 'refs/heads/staging' || github.ref == 'refs/heads/main' || github.ref == 'refs/heads/simulator' uses: macbre/push-to-ghcr@master with: image_name: ${{ github.repository }} diff --git a/commons/objects.py b/commons/objects.py index 114c49e1..1d8ae1c9 100644 --- a/commons/objects.py +++ b/commons/objects.py @@ -9,18 +9,26 @@ class ObjectManager: @classmethod def get_miner(cls): - from neurons.miner import Miner - - if cls._miner is None: - cls._miner = Miner() + if get_config().simulation: + from simulator.miner import MinerSim + if cls._miner is None: + cls._miner = MinerSim() + else: + from neurons.miner import Miner + if cls._miner is None: + cls._miner = Miner() return cls._miner @classmethod def get_validator(cls): - from neurons.validator import Validator - - if cls._validator is None: - cls._validator = Validator() + if get_config().simulation: + from simulator.validator import ValidatorSim + if cls._validator is None: + cls._validator = ValidatorSim() + else: + from neurons.validator import Validator + if cls._validator is None: + cls._validator = Validator() return cls._validator @classmethod diff --git a/dojo/__init__.py b/dojo/__init__.py index 11625674..5674629d 100644 --- a/dojo/__init__.py +++ b/dojo/__init__.py @@ -59,4 +59,4 @@ def get_dojo_api_base_url() -> str: if base_url is None: raise ValueError("DOJO_API_BASE_URL is not set in the environment.") - return base_url + return base_url \ No newline at end of file diff --git a/dojo/utils/config.py b/dojo/utils/config.py index a0022d54..2c399997 100644 --- a/dojo/utils/config.py +++ b/dojo/utils/config.py @@ -178,6 +178,18 @@ def add_args(parser): help="Whether to run in fast mode, for developers to test locally.", ) + parser.add_argument( + "--simulation", + action="store_true", + help="Whether to run the validator in simulation mode", + ) + + parser.add_argument( + "--simulation_bad_miner", + action="store_true", + help="Set miner simluation to a bad one", + ) + epoch_length = 100 known_args, _ = parser.parse_known_args() if known_args := vars(known_args): diff --git a/entrypoints.sh b/entrypoints.sh index 7f64dfe7..325fc7ad 100755 --- a/entrypoints.sh +++ b/entrypoints.sh @@ -21,6 +21,17 @@ if [ "$1" = 'miner' ]; then echo "SUBTENSOR_ENDPOINT: ${SUBTENSOR_ENDPOINT}" echo "NETUID: ${NETUID}" + EXTRA_ARGS="" + if [ "${SIMULATION}" = "true" ]; then + EXTRA_ARGS="${EXTRA_ARGS} --simulation" + fi + if [ "${FAST_MODE}" = "true" ]; then + EXTRA_ARGS="${EXTRA_ARGS} --fast_mode" + fi + if [ "${SIMULATION_BAD_MINER}" = "true" ]; then + EXTRA_ARGS="${EXTRA_ARGS} --simulation_bad_miner" + fi + python main_miner.py \ --netuid ${NETUID} \ --subtensor.network ${SUBTENSOR_NETWORK} \ @@ -29,7 +40,8 @@ if [ "$1" = 'miner' ]; then --wallet.name ${WALLET_COLDKEY} \ --wallet.hotkey ${WALLET_HOTKEY} \ --axon.port ${AXON_PORT} \ - --neuron.type miner + --neuron.type miner \ + ${EXTRA_ARGS} fi # If the first argument is 'validator', run the validator script @@ -43,6 +55,14 @@ if [ "$1" = 'validator' ]; then echo "NETUID: ${NETUID}" echo "WANDB_PROJECT_NAME: ${WANDB_PROJECT_NAME}" + EXTRA_ARGS="" + if [ "${SIMULATION}" = "true" ]; then + EXTRA_ARGS="${EXTRA_ARGS} --simulation" + fi + if [ "${FAST_MODE}" = "true" ]; then + EXTRA_ARGS="${EXTRA_ARGS} --fast_mode" + fi + python main_validator.py \ --netuid ${NETUID} \ --subtensor.network ${SUBTENSOR_NETWORK} \ @@ -51,5 +71,6 @@ if [ "$1" = 'validator' ]; then --wallet.name ${WALLET_COLDKEY} \ --wallet.hotkey ${WALLET_HOTKEY} \ --neuron.type validator \ - --wandb.project_name ${WANDB_PROJECT_NAME} -fi + --wandb.project_name ${WANDB_PROJECT_NAME} \ + ${EXTRA_ARGS} +fi \ No newline at end of file diff --git a/simulator/__init__.py b/simulator/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/simulator/miner.py b/simulator/miner.py new file mode 100644 index 00000000..a3e891c9 --- /dev/null +++ b/simulator/miner.py @@ -0,0 +1,173 @@ +import os +import redis +import traceback +import asyncio +import random +import json +from datetime import datetime, timezone +from dojo.utils.config import get_config +from bittensor.btlogging import logging as logger +from neurons.miner import Miner +from dojo.protocol import ( + FeedbackRequest, + TaskResultRequest, + TaskResult, + Result +) +from commons.utils import get_new_uuid + + +class MinerSim(Miner): + def __init__(self): + super().__init__() + try: + # Initialize Redis connection + host = os.getenv("REDIS_HOST", "localhost") + port = int(os.getenv("REDIS_PORT", 6379)) + self.redis_client = redis.Redis( + host=host, + port=port, + db=0, + decode_responses=True + ) + logger.info("Redis connection established") + + self._configure_simulation() + + self.is_bad_miner = get_config().simulation_bad_miner + logger.info(f"Miner role set to: {'bad' if self.is_bad_miner else 'good'}") + + logger.info("Starting Miner Simulator") + except Exception as e: + logger.error(f"Failed to connect to Redis: {e}") + raise + + def _configure_simulation(self): + """Configure simulation parameters with environment variables or defaults.""" + self.response_behaviors = { + 'normal': float(os.getenv("SIM_NORMAL_RESP_PROB", 0.8)), + 'no_response': float(os.getenv("SIM_NO_RESP_PROB", 0.1)), + 'timeout': float(os.getenv("SIM_TIMEOUT_PROB", 0.1)) + } + + async def forward_feedback_request(self, synapse: FeedbackRequest) -> FeedbackRequest: + try: + # Validate that synapse, dendrite, dendrite.hotkey, and response are not None + if not synapse or not synapse.dendrite or not synapse.dendrite.hotkey: + logger.error("Invalid synapse: dendrite or dendrite.hotkey is None.") + return synapse + + if not synapse.completion_responses: + logger.error("Invalid synapse: response field is None.") + return synapse + + # Empty out completion response since not needed in simulator + new_synapse = synapse.model_copy(deep=True) + new_synapse.completion_responses = [] + + synapse.dojo_task_id = synapse.request_id + self.hotkey_to_request[synapse.dendrite.hotkey] = synapse + + redis_key = f"feedback:{synapse.request_id}" + self.redis_client.set( + redis_key, + new_synapse.model_dump_json(), + ex=86400 # expire after 24 hours + ) + logger.info(f"Stored feedback request {synapse.request_id}") + + synapse.ground_truth = {} + return synapse + + except Exception as e: + logger.error(f"Error handling FeedbackRequest: {e}") + traceback.print_exc() + return synapse + + async def forward_task_result_request(self, synapse: TaskResultRequest) -> TaskResultRequest | None: + try: + logger.info(f"Received TaskResultRequest for task id: {synapse.task_id}") + if not synapse or not synapse.task_id: + logger.error("Invalid TaskResultRequest: missing task_id") + return None + + # Simulate different response behaviors + behavior = self._get_response_behavior() + + if behavior in ['no_response', 'timeout']: + logger.debug(f"Simulating {behavior} for task {synapse.task_id}") + if behavior == 'timeout': + await asyncio.sleep(30) + return None + + redis_key = f"feedback:{synapse.task_id}" + request_data = self.redis_client.get(redis_key) + + request_dict = json.loads(request_data) if request_data else None + feedback_request = FeedbackRequest(**request_dict) if request_dict else None + + if not feedback_request: + logger.debug(f"No task result found for task id: {synapse.task_id}") + return None + + current_time = datetime.now(timezone.utc).isoformat() + + task_results = [] + for criteria_type in feedback_request.criteria_types: + result = Result( + type=criteria_type.type, + value=self._generate_scores(feedback_request.ground_truth) + ) + + task_result = TaskResult( + id=get_new_uuid(), + status='COMPLETED', + created_at=current_time, + updated_at=current_time, + result_data=[result], + worker_id=get_new_uuid(), + task_id=synapse.task_id + ) + task_results.append(task_result) + + synapse.task_results = task_results + logger.info(f"TaskResultRequest: {synapse}") + + self.redis_client.delete(redis_key) + logger.debug(f"Processed task result for task {synapse.task_id}") + + return synapse + + except Exception as e: + traceback.print_exc() + logger.error(f"Error handling TaskResultRequest: {e}") + return None + + def _get_response_behavior(self) -> str: + """Determine the response behavior based on configured probabilities.""" + return random.choices( + list(self.response_behaviors.keys()), + weights=list(self.response_behaviors.values()) + )[0] + + def _generate_scores(self, ground_truth: dict) -> dict: + scores = {} + + for k, v in ground_truth.items(): + if self.is_bad_miner: + deviation = random.randint(-5, 5) + else: + deviation = random.randint(-2, 2) + random_score = max(1, min(10, v + deviation)) + score = int((random_score / (10 - 1)) * (100 - 1) + 1) + scores[k] = score + + return scores + + # def __del__(self): + # """Cleanup Redis connection on object destruction""" + # try: + # self.redis_client.close() + # logger.info("Redis connection closed") + # except: + # pass diff --git a/simulator/validator.py b/simulator/validator.py new file mode 100644 index 00000000..8fc9f2e6 --- /dev/null +++ b/simulator/validator.py @@ -0,0 +1,254 @@ +import logging +import traceback +from typing import List +import aiohttp +import dojo +from commons.dataset.synthetic import SyntheticAPI +from commons.orm import ORM +from commons.utils import get_epoch_time, get_new_uuid, set_expire_time, ttl_get_block +from dojo.protocol import FeedbackRequest, TaskType, MultiScoreCriteria, DendriteQueryResponse +from neurons.validator import Validator +from bittensor.btlogging import logging as logger +from tenacity import RetryError +import bittensor as bt +import asyncio + + +class ValidatorSim(Validator): + def __init__(self): + super().__init__() + logger.info("Starting Validator Simulator") + self._last_block = None + self._block_check_attempts = 0 + self.MAX_BLOCK_CHECK_ATTEMPTS = 3 + + async def _try_reconnect_subtensor(self): + """Attempt to reconnect to the subtensor network""" + try: + logger.info("Attempting to reconnect to subtensor...") + self.subtensor = bt.subtensor(self.subtensor.config) + self._block_check_attempts = 0 + return True + except Exception as e: + logger.error(f"Failed to reconnect to subtensor: {e}") + return False + + @property + def block(self): + try: + self._last_block = ttl_get_block(self.subtensor) + self._block_check_attempts = 0 + return self._last_block + except BrokenPipeError: + self._block_check_attempts += 1 + if self._block_check_attempts >= self.MAX_BLOCK_CHECK_ATTEMPTS: + logger.error("Multiple failed attempts to get block number, attempting reconnection") + if asyncio.get_event_loop().run_until_complete(self._try_reconnect_subtensor()): + return self.block + + return self._last_block if self._last_block is not None else 0 + except Exception as e: + logger.error(f"Error getting block number: {e}") + return self._last_block if self._last_block is not None else 0 + + def check_registered(self): + new_subtensor = bt.subtensor(self.subtensor.config) + if not new_subtensor.is_hotkey_registered( + netuid=self.config.netuid, + hotkey_ss58=self.wallet.hotkey.ss58_address, + ): + logger.error( + f"Wallet: {self.wallet} is not registered on netuid {self.config.netuid}." + f" Please register the hotkey using `btcli s register` before trying again" + ) + exit() + + async def send_request( + self, + synapse: FeedbackRequest | None = None, + external_user: bool = False, + ): + start = get_epoch_time() + # typically the request may come from an external source however, + # initially will seed it with some data for miners to get started + if len(self._active_miner_uids) == 0: + logger.info("No active miners to send request to... skipping") + return + + request_id = get_new_uuid() + # sel_miner_uids = await self.get_miner_uids(external_user, request_id) + sel_miner_uids = sorted(list(self._active_miner_uids)) + + axons = [ + self.metagraph.axons[uid] + for uid in sel_miner_uids + if self.metagraph.axons[uid].hotkey.casefold() + != self.wallet.hotkey.ss58_address.casefold() + ] + if not len(axons): + logger.warning("🤷 No axons to query ... skipping") + return + + obfuscated_model_to_model = {} + + if synapse is None: + try: + data = await SyntheticAPI.get_qa() + except RetryError as e: + logger.error( + f"Exhausted all retry attempts for synthetic data generation: {e}" + ) + return + except ValueError as e: + logger.error(f"Invalid response from synthetic data API: {e}") + return + except aiohttp.ClientError as e: + logger.error(f"Network error when calling synthetic data API: {e}") + return + except Exception as e: + logger.error(f"Unexpected error during synthetic data generation: {e}") + logger.debug(f"Traceback: {traceback.format_exc()}") + return + + if not data: + logger.error("No data returned from synthetic data API") + return + + obfuscated_model_to_model = self.obfuscate_model_names(data.responses) + expire_at = set_expire_time(dojo.TASK_DEADLINE) + synapse = FeedbackRequest( + request_id=request_id, + task_type=str(TaskType.CODE_GENERATION), + criteria_types=[ + MultiScoreCriteria( + options=list(obfuscated_model_to_model.keys()), + min=1.0, + max=100.0, + ), + ], + prompt=data.prompt, + completion_responses=data.responses, + expire_at=expire_at, + ground_truth=data.ground_truth # Added ground truth!!!!! + ) + elif external_user: + obfuscated_model_to_model = self.obfuscate_model_names( + synapse.completion_responses + ) + + logger.info( + f"⬆️ Sending feedback request for request id: {synapse.request_id}, miners uids:{sel_miner_uids} with expire_at: {synapse.expire_at}" + ) + + miner_responses: List[FeedbackRequest] = await self._send_shuffled_requests( + self.dendrite, axons, synapse + ) + + valid_miner_responses: List[FeedbackRequest] = [] + try: + for miner_response in miner_responses: + miner_hotkey = ( + miner_response.axon.hotkey if miner_response.axon else "??" + ) + # map obfuscated model names back to the original model names + real_model_ids = [] + + for i, completion in enumerate(miner_response.completion_responses): + found_model_id = obfuscated_model_to_model.get( + completion.model, None + ) + real_model_ids.append(found_model_id) + if found_model_id: + miner_response.completion_responses[i].model = found_model_id + synapse.completion_responses[i].model = found_model_id + + if any(c is None for c in real_model_ids): + logger.warning("Failed to map obfuscated model to original model") + continue + + if miner_response.dojo_task_id is None: + logger.debug(f"Miner {miner_hotkey} must provide the dojo task id") + continue + + # update the miner response with the real model ids + valid_miner_responses.append(miner_response) + except Exception as e: + logger.error(f"Failed to map obfuscated model to original model: {e}") + pass + + logger.info(f"⬇️ Received {len(valid_miner_responses)} valid responses") + if valid_miner_responses is None or len(valid_miner_responses) == 0: + logger.info("No valid miner responses to process... skipping") + return + + # include the ground_truth to keep in data manager + synapse.ground_truth = data.ground_truth + synapse.dendrite.hotkey = self.wallet.hotkey.ss58_address + response_data = DendriteQueryResponse( + request=synapse, + miner_responses=valid_miner_responses, + ) + + logger.debug("Attempting to saving dendrite response") + vali_request_model = await ORM.save_task( + validator_request=synapse, + miner_responses=valid_miner_responses, + ground_truth=data.ground_truth, + ) + + if vali_request_model is None: + logger.error("Failed to save dendrite response") + return + + # saving response + logger.success( + f"Saved dendrite response for request id: {response_data.request.request_id}" + ) + logger.info( + f"Sending request to miners & processing took {get_epoch_time() - start}" + ) + return + + @staticmethod + async def _send_shuffled_requests( + dendrite: bt.dendrite, axons: List[bt.AxonInfo], synapse: FeedbackRequest + ) -> list[FeedbackRequest]: + """Send the same request to all miners without shuffling the order. + WARNING: This should only be used for testing/debugging as it could allow miners to game the system. + + Args: + dendrite (bt.dendrite): Communication channel to send requests + axons (List[bt.AxonInfo]): List of miner endpoints + synapse (FeedbackRequest): The feedback request to send + + Returns: + list[FeedbackRequest]: List of miner responses + """ + all_responses = [] + batch_size = 10 + + for i in range(0, len(axons), batch_size): + batch_axons = axons[i: i + batch_size] + tasks = [] + + for axon in batch_axons: + tasks.append( + dendrite.forward( + axons=[axon], + synapse=synapse, + deserialize=False, + timeout=12, + ) + ) + + batch_responses = await asyncio.gather(*tasks) + flat_batch_responses = [ + response for sublist in batch_responses for response in sublist + ] + all_responses.extend(flat_batch_responses) + + logger.info( + f"Processed batch {i // batch_size + 1} of {(len(axons) - 1) // batch_size + 1}" + ) + + return all_responses