Merge pull request #523 from macrocosm-os/staging

v2.16.0

Author: bkb2135

.github/workflows/python-package.yml

@@ -12,7 +12,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.9", "3.10"]
+        python-version: ["3.10"]
 
     steps:
     - uses: actions/checkout@v3

neurons/miners/epistula_miner/miner.py

@@ -1,8 +1,8 @@
 # ruff: noqa: E402
-from prompting import settings
+from shared import settings
 
-settings.settings = settings.Settings.load(mode="miner")
-settings = settings.settings
+settings.shared_settings = settings.SharedSettings.load(mode="miner")
+shared_settings = settings.shared_settings
 
 import asyncio
 import json
@@ -39,7 +39,7 @@ def __init__(self):
         self.client = httpx.AsyncClient(
             base_url="https://api.openai.com/v1",
             headers={
-                "Authorization": f"Bearer {settings.OPENAI_API_KEY}",
+                "Authorization": f"Bearer {shared_settings.OPENAI_API_KEY}",
                 "Content-Type": "application/json",
             },
         )
@@ -107,14 +107,14 @@ async def verify_request(
 
         signed_by = request.headers.get("Epistula-Signed-By")
         signed_for = request.headers.get("Epistula-Signed-For")
-        if signed_for != settings.WALLET.hotkey.ss58_address:
+        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 not in settings.METAGRAPH.hotkeys:
+        if signed_by not in shared_settings.METAGRAPH.hotkeys:
             raise HTTPException(status_code=401, detail="Signer not in metagraph")
 
-        uid = settings.METAGRAPH.hotkeys.index(signed_by)
-        stake = settings.METAGRAPH.S[uid].item()
-        if not settings.NETUID == 61 and stake < 10000:
+        uid = shared_settings.METAGRAPH.hotkeys.index(signed_by)
+        stake = shared_settings.METAGRAPH.S[uid].item()
+        if not shared_settings.NETUID == 61 and stake < 10000:
             logger.warning(f"Blacklisting request from {signed_by} [uid={uid}], not enough stake -- {stake}")
             raise HTTPException(status_code=401, detail="Stake below minimum: {stake}")
 
@@ -133,7 +133,7 @@ async def verify_request(
             raise HTTPException(status_code=400, detail=err)
 
     def run(self):
-        external_ip = None  # settings.EXTERNAL_IP
+        external_ip = None  # shared_settings.EXTERNAL_IP
         if not external_ip or external_ip == "[::]":
             try:
                 external_ip = requests.get("https://checkip.amazonaws.com").text.strip()
@@ -142,16 +142,16 @@ def run(self):
                 logger.error("Failed to get external IP")
 
         logger.info(
-            f"Serving miner endpoint {external_ip}:{settings.AXON_PORT} on network: {settings.SUBTENSOR_NETWORK} with netuid: {settings.NETUID}"
+            f"Serving miner endpoint {external_ip}:{shared_settings.AXON_PORT} on network: {shared_settings.SUBTENSOR_NETWORK} with netuid: {shared_settings.NETUID}"
         )
 
         serve_success = serve_extrinsic(
-            subtensor=settings.SUBTENSOR,
-            wallet=settings.WALLET,
+            subtensor=shared_settings.SUBTENSOR,
+            wallet=shared_settings.WALLET,
             ip=external_ip,
-            port=settings.AXON_PORT,
+            port=shared_settings.AXON_PORT,
             protocol=4,
-            netuid=settings.NETUID,
+            netuid=shared_settings.NETUID,
         )
         if not serve_success:
             logger.error("Failed to serve endpoint")
@@ -174,15 +174,15 @@ def run(self):
         fast_config = uvicorn.Config(
             app,
             host="0.0.0.0",
-            port=settings.AXON_PORT,
+            port=shared_settings.AXON_PORT,
             log_level="info",
             loop="asyncio",
             workers=4,
         )
         self.fast_api = FastAPIThreadedServer(config=fast_config)
         self.fast_api.start()
 
-        logger.info(f"Miner starting at block: {settings.SUBTENSOR.block}")
+        logger.info(f"Miner starting at block: {shared_settings.SUBTENSOR.block}")
 
         # Main execution loop.
         try:

neurons/test_vanilla_post.py

@@ -1,10 +1,10 @@
 import openai
 from httpx import Timeout
 
-from prompting import settings
+from shared import settings
 
-settings.settings = settings.Settings.load(mode="validator")
-settings = settings.settings
+settings.shared_settings = settings.SharedSettings.load(mode="validator")
+shared_settings = settings.shared_settings
 
 from shared.epistula import create_header_hook
 

neurons/validator.py

@@ -81,11 +81,12 @@ async def spawn_loops(task_queue, scoring_queue, reward_events):
     asyncio.run(spawn_loops(task_queue, scoring_queue, reward_events))
 
 
-def start_api():
+def start_api(scoring_queue, reward_events):
     async def start():
         from prompting.api.api import start_scoring_api  # noqa: F401
 
-        await start_scoring_api()
+        await start_scoring_api(scoring_queue, reward_events)
+
         while True:
             await asyncio.sleep(10)
             logger.debug("Running API...")
@@ -125,7 +126,7 @@ async def main():
 
             if shared_settings.DEPLOY_SCORING_API:
                 # Use multiprocessing to bypass API blocking issue
-                api_process = mp.Process(target=start_api, name="API_Process")
+                api_process = mp.Process(target=start_api, args=(scoring_queue, reward_events), name="API_Process")
                 api_process.start()
                 processes.append(api_process)
 

prompting/api/api.py

@@ -4,6 +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
 
 app = FastAPI()
@@ -17,7 +18,9 @@ def health():
     return {"status": "healthy"}
 
 
-async def start_scoring_api():
+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}")
     uvicorn.run(
         "prompting.api.api:app", host="0.0.0.0", port=shared_settings.SCORING_API_PORT, loop="asyncio", reload=False

prompting/api/scoring/api.py

@@ -4,9 +4,11 @@
 from fastapi import APIRouter, Depends, Header, HTTPException, Request
 from loguru import logger
 
+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.base import DatasetEntry
 from shared.dendrite import DendriteResponseEvent
 from shared.epistula import SynapseStreamResult
@@ -37,22 +39,54 @@ async def score_response(request: Request, api_key_data: dict = Depends(validate
     uid = int(payload.get("uid"))
     chunks = payload.get("chunks")
     llm_model = ModelZoo.get_model_by_id(model) if (model := body.get("model")) else None
-    task_scorer.add_to_queue(
-        task=InferenceTask(
-            messages=[msg["content"] for msg in body.get("messages")],
-            llm_model=llm_model,
-            llm_model_id=body.get("model"),
-            seed=int(body.get("seed", 0)),
-            sampling_params=body.get("sampling_params", {}),
-        ),
-        response=DendriteResponseEvent(
-            uids=[uid],
-            stream_results=[SynapseStreamResult(accumulated_chunks=[chunk for chunk in chunks if chunk is not None])],
-            timeout=shared_settings.NEURON_TIMEOUT,
-        ),
-        dataset_entry=DatasetEntry(),
-        block=shared_settings.METAGRAPH.block,
-        step=-1,
-        task_id=str(uuid.uuid4()),
-    )
-    logger.info("Organic tas appended to scoring queue")
+    task = body.get("task")
+    if task == "InferenceTask":
+        logger.info(f"Received Organic InferenceTask with body: {body}")
+        task_scorer.add_to_queue(
+            task=InferenceTask(
+                messages=[msg["content"] for msg in body.get("messages")],
+                llm_model=llm_model,
+                llm_model_id=body.get("model"),
+                seed=int(body.get("seed", 0)),
+                sampling_params=body.get("sampling_params", {}),
+            ),
+            response=DendriteResponseEvent(
+                uids=[uid],
+                stream_results=[
+                    SynapseStreamResult(accumulated_chunks=[chunk for chunk in chunks if chunk is not None])
+                ],
+                timeout=shared_settings.NEURON_TIMEOUT,
+            ),
+            dataset_entry=DatasetEntry(),
+            block=shared_settings.METAGRAPH.block,
+            step=-1,
+            task_id=str(uuid.uuid4()),
+        )
+    elif task == "WebRetrievalTask":
+        logger.info(f"Received Organic WebRetrievalTask with body: {body}")
+        try:
+            search_term = body.get("messages")[0].get("content")
+        except Exception as ex:
+            logger.error(f"Failed to get search term from messages: {ex}, can't score WebRetrievalTask")
+            return
+
+        task_scorer.add_to_queue(
+            task=WebRetrievalTask(
+                messages=[msg["content"] for msg in body.get("messages")],
+                seed=int(body.get("seed", 0)),
+                sampling_params=body.get("sampling_params", {}),
+                query=search_term,
+            ),
+            response=DendriteResponseEvent(
+                uids=[uid],
+                stream_results=[
+                    SynapseStreamResult(accumulated_chunks=[chunk for chunk in chunks if chunk is not None])
+                ],
+                timeout=shared_settings.NEURON_TIMEOUT,
+            ),
+            dataset_entry=DDGDatasetEntry(search_term=search_term),
+            block=shared_settings.METAGRAPH.block,
+            step=-1,
+            task_id=str(uuid.uuid4()),
+        )
+    logger.info("Organic task appended to scoring queue")

prompting/datasets/random_website.py

@@ -15,8 +15,8 @@
 
 class DDGDatasetEntry(DatasetEntry):
     search_term: str
-    website_url: str
-    website_content: str
+    website_url: str = None
+    website_content: str = None
 
 
 class DDGDataset(BaseDataset):

prompting/settings.py

@@ -102,7 +102,7 @@ def set_weights(
                     "weights": processed_weights.flatten(),
                     "raw_weights": str(list(weights.flatten())),
                     "averaged_weights": str(list(averaged_weights.flatten())),
-                    "block": ttl_get_block(),
+                    "block": ttl_get_block(subtensor=subtensor),
                 }
             )
             step_filename = "weights.csv"