Merge pull request #576 from macrocosm-os/staging

v2.17.1

Author: dbobrenko

README.md

@@ -35,7 +35,7 @@ Validators and miners are based on large language models (LLM). The validation p
 
 <div align="center">
 
-**[For Validators](./assets/validator.md)** · **[For Miners](./assets/miner.md)** · **[API Documentation](./validator_api/API_docs.md)**
+**[For Validators](./docs/validator.md)** · **[For Miners](./docs/epistula_miner.md)** · **[API Documentation]((./docs/API_docs.md))**
 
 
 </div>
@@ -66,7 +66,7 @@ The miner is given a complex problem that requires multiple steps to solve. Each
 
 # API Documentation
 
-For detailed information on the available API endpoints, request/response formats, and usage examples, please refer to the [API Documentation](./validator_api/API_docs.md).
+For detailed information on the available API endpoints, request/response formats, and usage examples, please refer to the [API Documentation](./docs/API_docs.md).
 
 # Contribute
 <div align="center">

validator_api/API_docs.md renamed to docs/API_docs.md

@@ -7,7 +7,7 @@
 from openai import OpenAI
 
 from prompting.base.duckduckgo_patch import PatchedDDGS
-from shared.settings import shared_settings
+from shared import settings
 
 # Import the patched DDGS and use that
 
@@ -56,7 +56,7 @@ async def get_websites_with_similarity(
         List of dictionaries containing website URLs and their best matching chunks
     """
     logger.debug("Getting results")
-    ddgs = PatchedDDGS(proxy=shared_settings.PROXY_URL, verify=False)
+    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]
@@ -66,7 +66,7 @@ async def get_websites_with_similarity(
     extracted = await asyncio.gather(*[extract_content(c) for c in content])
 
     # Create embeddings
-    client = OpenAI(api_key=shared_settings.OPENAI_API_KEY)
+    client = OpenAI(api_key=settings.shared_settings.OPENAI_API_KEY)
     query_embedding = client.embeddings.create(model="text-embedding-ada-002", input=query).data[0].embedding
     # Process each website
     results_with_similarity = []

neurons/miners/epistula_miner/README.md renamed to docs/epistula_miner.md

@@ -9,8 +9,8 @@
 
 # ruff: noqa: E402
 from shared import settings
+from shared.logging import init_wandb
 
-shared_settings = settings.shared_settings
 settings.shared_settings = settings.SharedSettings.load(mode="validator")
 
 
@@ -26,13 +26,12 @@
 
 
 def create_loop_process(task_queue, scoring_queue, reward_events):
+    settings.shared_settings = settings.SharedSettings.load(mode="validator")
+    if settings.shared_settings.WANDB_ON:
+        init_wandb(neuron="validator")
+
     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
@@ -88,10 +87,11 @@ 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())
 
-        await start_scoring_api(scoring_queue, reward_events)
+        await start_scoring_api(task_scorer, scoring_queue, reward_events)
 
         while True:
             await asyncio.sleep(10)
@@ -100,23 +100,6 @@ 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())
-
-#         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...")
-
-#     asyncio.run(start(task_queue, scoring_queue))
-
-
 async def main():
     # will start checking the availability of miners at regular intervals, needed for API and Validator
     with torch.multiprocessing.Manager() as manager:
@@ -130,7 +113,7 @@ async def main():
         try:
             # # Start checking the availability of miners at regular intervals
 
-            if shared_settings.DEPLOY_SCORING_API:
+            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")
                 api_process.start()
@@ -152,17 +135,17 @@ async def main():
             while True:
                 await asyncio.sleep(30)
                 if (
-                    shared_settings.SUBTENSOR.get_current_block()
-                    - shared_settings.METAGRAPH.last_update[shared_settings.UID]
+                    settings.shared_settings.SUBTENSOR.get_current_block()
+                    - settings.shared_settings.METAGRAPH.last_update[settings.shared_settings.UID]
                     > 500
                     and step > 120
                 ):
+                    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 {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}"
+                        f"UPDATES HAVE STALED FOR {current_block - last_update_block} BLOCKS AND {step} STEPS"
                     )
+                    logger.warning(f"STALED: {current_block}, {settings.shared_settings.METAGRAPH.block}")
                     sys.exit(1)
                 step += 1
 

assets/validator.md renamed to docs/validator.md

@@ -4,8 +4,7 @@
 
 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.settings import shared_settings
+from shared import settings
 
 app = FastAPI()
 app.include_router(miner_availabilities_router, prefix="/miner_availabilities", tags=["miner_availabilities"])
@@ -18,10 +17,17 @@ def health():
     return {"status": "healthy"}
 
 
-async def start_scoring_api(scoring_queue, reward_events):
-    task_scorer.scoring_queue = scoring_queue
-    task_scorer.reward_events = reward_events
-    logger.info(f"Starting Scoring API on https://0.0.0.0:{shared_settings.SCORING_API_PORT}")
+async def start_scoring_api(task_scorer, scoring_queue, reward_events):
+    # We pass an object of task scorer then override it's attributes to ensure that they are managed by mp
+    app.state.task_scorer = task_scorer
+    app.state.task_scorer.scoring_queue = scoring_queue
+    app.state.task_scorer.reward_events = reward_events
+
+    logger.info(f"Starting Scoring API on https://0.0.0.0:{settings.shared_settings.SCORING_API_PORT}")
     uvicorn.run(
-        "prompting.api.api:app", host="0.0.0.0", port=shared_settings.SCORING_API_PORT, loop="asyncio", reload=False
+        "prompting.api.api:app",
+        host="0.0.0.0",
+        port=settings.shared_settings.SCORING_API_PORT,
+        loop="asyncio",
+        reload=False,
     )

neurons/miners/epistula_miner/web_retrieval.py

@@ -6,28 +6,33 @@
 
 from prompting.datasets.random_website import DDGDatasetEntry
 from prompting.llms.model_zoo import ModelZoo
-from prompting.rewards.scoring import task_scorer
 from prompting.tasks.inference import InferenceTask
 from prompting.tasks.web_retrieval import WebRetrievalTask
+from shared import settings
 from shared.base import DatasetEntry
 from shared.dendrite import DendriteResponseEvent
 from shared.epistula import SynapseStreamResult
-from shared.settings import shared_settings
 
 router = APIRouter()
 
 
 def validate_scoring_key(api_key: str = Header(...)):
-    if api_key != shared_settings.SCORING_KEY:
+    if api_key != settings.shared_settings.SCORING_KEY:
         raise HTTPException(status_code=403, detail="Invalid API key")
 
 
+def get_task_scorer(request: Request):
+    return request.app.state.task_scorer
+
+
 @router.post("/scoring")
-async def score_response(request: Request, api_key_data: dict = Depends(validate_scoring_key)):
+async def score_response(
+    request: Request, api_key_data: dict = Depends(validate_scoring_key), task_scorer=Depends(get_task_scorer)
+):
     model = None
     payload: dict[str, Any] = await request.json()
     body = payload.get("body")
-    timeout = payload.get("timeout", shared_settings.NEURON_TIMEOUT)
+    timeout = payload.get("timeout", settings.shared_settings.NEURON_TIMEOUT)
     uids = payload.get("uid", [])
     chunks = payload.get("chunks", {})
     if not uids or not chunks:
@@ -54,7 +59,7 @@ async def score_response(request: Request, api_key_data: dict = Depends(validate
             llm_model=llm_model,
             llm_model_id=body.get("model"),
             seed=int(body.get("seed", 0)),
-            sampling_params=body.get("sampling_parameters", shared_settings.SAMPLING_PARAMS),
+            sampling_params=body.get("sampling_parameters", settings.shared_settings.SAMPLING_PARAMS),
             query=body.get("messages"),
         )
         logger.info(f"Task created: {organic_task}")
@@ -66,7 +71,7 @@ async def score_response(request: Request, api_key_data: dict = Depends(validate
                 timeout=timeout,
             ),
             dataset_entry=DatasetEntry(),
-            block=shared_settings.METAGRAPH.block,
+            block=settings.shared_settings.METAGRAPH.block,
             step=-1,
             task_id=str(uuid.uuid4()),
         )
@@ -90,10 +95,10 @@ async def score_response(request: Request, api_key_data: dict = Depends(validate
                 stream_results=[
                     SynapseStreamResult(accumulated_chunks=[chunk for chunk in chunks if chunk is not None])
                 ],
-                timeout=body.get("timeout", shared_settings.NEURON_TIMEOUT),
+                timeout=body.get("timeout", settings.shared_settings.NEURON_TIMEOUT),
             ),
             dataset_entry=DDGDatasetEntry(search_term=search_term),
-            block=shared_settings.METAGRAPH.block,
+            block=settings.shared_settings.METAGRAPH.block,
             step=-1,
             task_id=str(uuid.uuid4()),
         )

neurons/validator.py

@@ -1,14 +1,14 @@
 import random
+from functools import lru_cache
 from typing import Optional
 
 import trafilatura
 from loguru import logger
 
-# from duckduckgo_search import DDGS
 from prompting.base.duckduckgo_patch import PatchedDDGS
 from prompting.datasets.utils import ENGLISH_WORDS
+from shared import settings
 from shared.base import BaseDataset, Context, DatasetEntry
-from shared.settings import shared_settings
 
 MAX_CHARS = 5000
 
@@ -25,7 +25,7 @@ class DDGDataset(BaseDataset):
     english_words: list[str] = None
 
     def search_random_term(self, retries: int = 3) -> tuple[Optional[str], Optional[list[dict[str, str]]]]:
-        ddg = PatchedDDGS(proxy=shared_settings.PROXY_URL, verify=False)
+        ddg = PatchedDDGS(proxy=settings.shared_settings.PROXY_URL, verify=False)
         for _ in range(retries):
             random_words = " ".join(random.sample(ENGLISH_WORDS, 3))
             try:
@@ -38,6 +38,7 @@ def search_random_term(self, retries: int = 3) -> tuple[Optional[str], Optional[
         return None, None
 
     @staticmethod
+    @lru_cache(maxsize=1000)
     def extract_website_content(url: str) -> Optional[str]:
         try:
             website = trafilatura.fetch_url(url)

prompting/api/api.py

@@ -5,7 +5,9 @@
 from pydantic import BaseModel
 
 from prompting.llms.apis.llm_messages import LLMMessage, LLMMessages
-from shared.settings import shared_settings
+from shared import settings
+
+shared_settings = settings.shared_settings
 
 
 class GPT(BaseModel):