Merge pull request #596 from macrocosm-os/staging

v2.17.3

Author: bkb2135

.env.api.example

@@ -1,6 +1,5 @@
 API_PORT = "42170" # Port for the API server
 API_HOST = "0.0.0.0" # Host for the API server
 SCORING_KEY = "123" # The scoring key for the validator (must match the scoring key in the .env.validator file)
-SCORE_ORGANICS = True # Whether to score organics
 VALIDATOR_API = "0.0.0.0:8094" # The validator API to forward responses to for scoring
 WORKERS=4

.env.validator.example

@@ -26,7 +26,6 @@ HF_TOKEN = "your_huggingface_token_here"
 
 # Scoring API (optional).
 DEPLOY_SCORING_API = true
-SCORING_ADMIN_KEY = "123456"
 SCORING_API_PORT = 8094
 # Scoring key must match the scoring key in the .env.api.
 # SCORING_KEY="..."

data/top100k_domains.csv

@@ -99997,4 +99997,4 @@
 "99996","tankspotter.com","4.51"
 "99997","targetshootingapp.com","4.51"
 "99998","tastytalegame.com","4.51"
-"99999","tbscan.com","4.51"
+"99999","tbscan.com","4.51"

neurons/miners/epistula_miner/web_retrieval.py

@@ -3,7 +3,6 @@
 
 import numpy as np
 import trafilatura
-from loguru import logger
 from openai import OpenAI
 
 from prompting.base.duckduckgo_patch import PatchedDDGS
@@ -55,10 +54,8 @@ async def get_websites_with_similarity(
     Returns:
         List of dictionaries containing website URLs and their best matching chunks
     """
-    logger.debug("Getting results")
     ddgs = PatchedDDGS(proxy=settings.shared_settings.PROXY_URL, verify=False)
     results = list(ddgs.text(query))
-    logger.debug(f"Got {len(results)} results")
     urls = [r["href"] for r in results][:n_results]
 
     # Fetch and extract content
@@ -74,7 +71,6 @@ async def get_websites_with_similarity(
         if not text:  # Skip if extraction failed
             continue
 
-        # logger.debug(f"TEXTS: {text}")
         chunks = create_chunks(text)
         chunk_embeddings = client.embeddings.create(model="text-embedding-ada-002", input=chunks).data
 

neurons/validator.py

@@ -1,11 +1,15 @@
 import asyncio
 import multiprocessing as mp
 import sys
-import time
 
 import loguru
+import netaddr
+import requests
 import torch
 import wandb
+from bittensor.core.extrinsics.serving import serve_extrinsic
+
+from prompting.rewards.scoring import task_scorer
 
 # ruff: noqa: E402
 from shared import settings
@@ -34,7 +38,6 @@ async def spawn_loops(task_queue, scoring_queue, reward_events):
         # ruff: noqa: E402
         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
@@ -61,18 +64,10 @@ async def spawn_loops(task_queue, scoring_queue, reward_events):
         logger.info("Starting WeightSetter...")
         asyncio.create_task(weight_setter.start(reward_events))
 
-        # Main monitoring loop
-        start = time.time()
-
-        logger.info("Starting Main Monitoring Loop...")
         while True:
             await asyncio.sleep(5)
-            current_time = time.time()
-            time_diff = current_time - start
-            start = current_time
 
             # Check if all tasks are still running
-            logger.debug(f"Running {time_diff:.2f} seconds")
             logger.debug(f"Number of tasks in Task Queue: {len(task_queue)}")
             logger.debug(f"Number of tasks in Scoring Queue: {len(scoring_queue)}")
             logger.debug(f"Number of tasks in Reward Events: {len(reward_events)}")
@@ -87,15 +82,29 @@ async def start():
         # TODO: We should not use 2 availability loops for each process, in reality
         # we should only be sharing the miner availability data between processes.
         from prompting.miner_availability.miner_availability import availability_checking_loop
-        from prompting.rewards.scoring import task_scorer
 
         asyncio.create_task(availability_checking_loop.start())
 
+        try:
+            external_ip = requests.get("https://checkip.amazonaws.com").text.strip()
+            netaddr.IPAddress(external_ip)
+
+            serve_success = serve_extrinsic(
+                subtensor=settings.shared_settings.SUBTENSOR,
+                wallet=settings.shared_settings.WALLET,
+                ip=external_ip,
+                port=settings.shared_settings.SCORING_API_PORT,
+                protocol=4,
+                netuid=settings.shared_settings.NETUID,
+            )
+
+            logger.debug(f"Serve success: {serve_success}")
+        except Exception as e:
+            logger.warning(f"Failed to serve scoring api to chain: {e}")
         await start_scoring_api(task_scorer, scoring_queue, reward_events)
 
         while True:
             await asyncio.sleep(10)
-            logger.debug("Running API...")
 
     asyncio.run(start())
 
@@ -112,7 +121,6 @@ async def main():
 
         try:
             # # Start checking the availability of miners at regular intervals
-
             if settings.shared_settings.DEPLOY_SCORING_API:
                 # Use multiprocessing to bypass API blocking issue
                 api_process = mp.Process(target=start_api, args=(scoring_queue, reward_events), name="API_Process")
@@ -122,13 +130,9 @@ 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"
-            # )
+
             loop_process.start()
-            # task_loop_process.start()
             processes.append(loop_process)
-            # processes.append(task_loop_process)
             GPUInfo.log_gpu_info()
 
             step = 0
@@ -143,9 +147,9 @@ async def main():
                     current_block = settings.shared_settings.SUBTENSOR.get_current_block()
                     last_update_block = settings.shared_settings.METAGRAPH.last_update[settings.shared_settings.UID]
                     logger.warning(
-                        f"UPDATES HAVE STALED FOR {current_block - last_update_block} BLOCKS AND {step} STEPS"
+                        f"Metagraph hasn't been updated for {current_block - last_update_block} blocks. "
+                        f"Staled block: {current_block}, Last update: {last_update_block}"
                     )
-                    logger.warning(f"STALED: {current_block}, {settings.shared_settings.METAGRAPH.block}")
                     sys.exit(1)
                 step += 1
 

prompting/api/api.py

@@ -4,6 +4,8 @@
 
 from prompting.api.miner_availabilities.api import router as miner_availabilities_router
 from prompting.api.scoring.api import router as scoring_router
+
+# from prompting.rewards.scoring import task_scorer
 from shared import settings
 
 app = FastAPI()
@@ -13,7 +15,6 @@
 
 @app.get("/health")
 def health():
-    logger.info("Health endpoint accessed.")
     return {"status": "healthy"}
 
 

prompting/api/miner_availabilities/api.py

@@ -1,7 +1,6 @@
 from typing import Literal
 
 from fastapi import APIRouter
-from loguru import logger
 
 from prompting.miner_availability.miner_availability import miner_availabilities
 from prompting.tasks.task_registry import TaskRegistry
@@ -13,7 +12,6 @@
 async def get_miner_availabilities(uids: list[int] | None = None):
     if uids:
         return {uid: miner_availabilities.miners.get(uid) for uid in uids}
-    logger.info(f"Returning all miner availabilities for {len(miner_availabilities.miners)} miners")
     return miner_availabilities.miners
 
 
@@ -23,7 +21,6 @@ async def get_available_miners(
     model: str | None = None,
     k: int = 10,
 ):
-    logger.info(f"Getting {k} available miners for task {task} and model {model}")
     task_configs = [config for config in TaskRegistry.task_configs if config.task.__name__ == task]
     task_config = task_configs[0] if task_configs else None
     return miner_availabilities.get_available_miners(task=task_config, model=model, k=k)

prompting/api/scoring/api.py

@@ -1,7 +1,8 @@
+import time
 import uuid
 from typing import Any
 
-from fastapi import APIRouter, Depends, Header, HTTPException, Request
+from fastapi import APIRouter, Depends, HTTPException, Request
 from loguru import logger
 
 from prompting.datasets.random_website import DDGDatasetEntry
@@ -11,13 +12,38 @@
 from shared import settings
 from shared.base import DatasetEntry
 from shared.dendrite import DendriteResponseEvent
-from shared.epistula import SynapseStreamResult
+from shared.epistula import SynapseStreamResult, verify_signature
+from shared.settings import shared_settings
 
 router = APIRouter()
 
 
-def validate_scoring_key(api_key: str = Header(...)):
-    if api_key != settings.shared_settings.SCORING_KEY:
+async def verify_scoring_signature(request: Request):
+    signed_by = request.headers.get("Epistula-Signed-By")
+    signed_for = request.headers.get("Epistula-Signed-For")
+    if signed_for != shared_settings.WALLET.hotkey.ss58_address:
+        raise HTTPException(status_code=400, detail="Bad Request, message is not intended for self")
+    if signed_by != shared_settings.API_HOTKEY:
+        raise HTTPException(status_code=401, detail="Signer not the expected ss58 address")
+
+    body = await request.body()
+    now = time.time()
+    err = verify_signature(
+        request.headers.get("Epistula-Request-Signature"),
+        body,
+        request.headers.get("Epistula-Timestamp"),
+        request.headers.get("Epistula-Uuid"),
+        signed_for,
+        signed_by,
+        now,
+    )
+    if err:
+        logger.error(err)
+        raise HTTPException(status_code=400, detail=err)
+
+
+def validate_scoring_key(request: Request):
+    if request.headers.api_key != settings.shared_settings.SCORING_KEY:
         raise HTTPException(status_code=403, detail="Invalid API key")
 
 
@@ -27,56 +53,59 @@ def get_task_scorer(request: Request):
 
 @router.post("/scoring")
 async def score_response(
-    request: Request, api_key_data: dict = Depends(validate_scoring_key), task_scorer=Depends(get_task_scorer)
+    request: Request, api_key_data: dict = Depends(verify_scoring_signature), task_scorer=Depends(get_task_scorer)
 ):
+    logger.debug("Scoring Request received!!!!!!!!!!!!!!!!")
     model = None
+    logger.debug("Setted Model to None")
     payload: dict[str, Any] = await request.json()
+    logger.debug(f"Awaited body: {payload}")
     body = payload.get("body")
-    timeout = payload.get("timeout", settings.shared_settings.NEURON_TIMEOUT)
-    uids = payload.get("uid", [])
+    timeout = payload.get("timeout", shared_settings.NEURON_TIMEOUT)
+    uids = payload.get("uids", [])
     chunks = payload.get("chunks", {})
+    timings = payload.get("timings", {})
+    logger.debug("About to check chunks and uids")
     if not uids or not chunks:
         logger.error(f"Either uids: {uids} or chunks: {chunks} is not valid, skipping scoring")
         return
     uids = [int(uid) for uid in uids]
     model = body.get("model")
-    if model:
-        try:
-            llm_model = ModelZoo.get_model_by_id(model)
-        except Exception:
-            logger.warning(
-                f"Organic request with model {body.get('model')} made but the model cannot be found in model zoo. Skipping scoring."
-            )
+    logger.debug("About to check model")
+    if model and model != shared_settings.LLM_MODEL:
+        logger.error(f"Model {model} not available for scoring on this validator.")
         return
-    else:
-        llm_model = None
+    logger.debug("Model has been checked")
+    llm_model = ModelZoo.get_model_by_id(model)
+    logger.debug("Got LLM Model from ModelZoo")
     task_name = body.get("task")
+    logger.debug(f"Task name set: {task_name}")
+    logger.debug(f"Length pre-insertion: {len(task_scorer.scoring_queue)}")
     if task_name == "InferenceTask":
-        logger.info(f"Received Organic InferenceTask with body: {body}")
-        logger.info(f"With model of type {type(body.get('model'))}")
         organic_task = InferenceTask(
             messages=body.get("messages"),
             llm_model=llm_model,
-            llm_model_id=body.get("model"),
+            llm_model_id=llm_model,
             seed=int(body.get("seed", 0)),
-            sampling_params=body.get("sampling_parameters", settings.shared_settings.SAMPLING_PARAMS),
+            sampling_params=body.get("sampling_parameters", shared_settings.SAMPLING_PARAMS),
             query=body.get("messages"),
+            organic=True,
         )
-        logger.info(f"Task created: {organic_task}")
         task_scorer.add_to_queue(
             task=organic_task,
             response=DendriteResponseEvent(
                 uids=uids,
                 stream_results=[SynapseStreamResult(accumulated_chunks=chunks.get(str(uid), None)) for uid in uids],
                 timeout=timeout,
+                stream_results_all_chunks_timings=[timings.get(str(uid), None) for uid in uids],
             ),
             dataset_entry=DatasetEntry(),
-            block=settings.shared_settings.METAGRAPH.block,
+            block=shared_settings.METAGRAPH.block,
             step=-1,
             task_id=str(uuid.uuid4()),
         )
+
     elif task_name == "WebRetrievalTask":
-        logger.info(f"Received Organic WebRetrievalTask with body: {body}")
         try:
             search_term = body.get("messages")[0].get("content")
         except Exception as ex:
@@ -91,15 +120,14 @@ async def score_response(
                 query=search_term,
             ),
             response=DendriteResponseEvent(
-                uids=[uids],
-                stream_results=[
-                    SynapseStreamResult(accumulated_chunks=[chunk for chunk in chunks if chunk is not None])
-                ],
-                timeout=body.get("timeout", settings.shared_settings.NEURON_TIMEOUT),
+                uids=uids,
+                stream_results=[SynapseStreamResult(accumulated_chunks=chunks.get(str(uid), [])) for uid in uids],
+                timeout=body.get("timeout", shared_settings.NEURON_TIMEOUT),
             ),
             dataset_entry=DDGDatasetEntry(search_term=search_term),
-            block=settings.shared_settings.METAGRAPH.block,
+            block=shared_settings.METAGRAPH.block,
             step=-1,
             task_id=str(uuid.uuid4()),
         )
+    logger.debug(f"Current Queue: {len(task_scorer.scoring_queue)}")
     logger.info("Organic task appended to scoring queue")

prompting/datasets/huggingface_github.py

@@ -1,5 +1,4 @@
 from datasets import load_dataset
-from loguru import logger
 from pydantic import ConfigDict, model_validator
 
 from shared.base import BaseDataset, DatasetEntry
@@ -61,7 +60,6 @@ def next(self) -> HuggingFaceGithubDatasetEntry:
                 entry = next(self.iterator)
                 return self._process_entry(entry)  # Throws failed to get a valid file after multiple attempts
             except StopIteration:
-                logger.warning("Reached end of dataset. Resetting iterator.")
                 self.reset()
         raise Exception("Failed to get a valid file after multiple attempts")