SN1-13: Organic Scoring implementation

README.md

@@ -47,6 +47,12 @@ If you are running a miner, you will also need to uninstall uvloop.
 pip uninstall uvloop -y
 ```
 
+If you are running a validator, logging in to Hugging Face is required:
+```shell
+huggingface-cli login
+```
+You also need to accept the License Agreement for the LMSYS-Chat-1M dataset: https://huggingface.co/datasets/lmsys/lmsys-chat-1m
+
 </div>
 
 # Compute Requirements

prompting/agent.py

@@ -18,10 +18,11 @@
 import time
 import bittensor as bt
 from dataclasses import asdict
+from typing import Optional
+
 from prompting.tasks import Task
-from prompting.llms import HuggingFaceLLM, vLLM_LLM
+from prompting.llms import vLLM_LLM
 from prompting.cleaners.cleaner import CleanerPipeline
-
 from prompting.persona import Persona, create_persona
 
 from transformers import Pipeline
@@ -42,7 +43,7 @@ def finished(self):
         """This is a roleplaying game where you are impersonating {mood} human user with a specific persona. As a human, you are using AI assistant to {desc} related to {topic} ({subtopic}) in a {tone} tone. You don't need to greet the assistant or be polite, unless this is part of your persona. The spelling and grammar of your messages should also reflect your persona.
 
         Your singular focus is to use the assistant to {goal}: {query}
-    """
+        """
     )
 
     def __init__(
@@ -52,22 +53,24 @@ def __init__(
         system_template: str = None,
         persona: Persona = None,
         begin_conversation=True,
+        system_prompt: Optional[str] = None,
     ):
-        if persona is None:
-            persona = create_persona()
-
         self.persona = persona
         self.task = task
         self.llm_pipeline = llm_pipeline
 
         if system_template is not None:
             self.system_prompt_template = system_template
 
-        self.system_prompt = self.system_prompt_template.format(
-            mood=self.persona.mood,
-            tone=self.persona.tone,
-            **self.task.__state_dict__(),  # Adds desc, subject, topic
-        )
+        self.system_prompt = system_prompt
+        if self.system_prompt is None:
+            if self.persona is None:
+                self.persona = create_persona()
+            self.system_prompt = self.system_prompt_template.format(
+                mood=self.persona.mood,
+                tone=self.persona.tone,
+                **self.task.__state_dict__(),  # Adds desc, subject, topic
+            )
 
         super().__init__(
             llm_pipeline=llm_pipeline,

prompting/base/neuron.py

@@ -17,6 +17,7 @@
 
 import copy
 import sys
+import threading
 
 import bittensor as bt
 

prompting/base/validator.py

@@ -15,19 +15,21 @@
 # OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 # DEALINGS IN THE SOFTWARE.
 
-import sys
-import copy
-import torch
-import asyncio
 import argparse
+import asyncio
+import copy
+import sys
 import threading
-import bittensor as bt
-
-from typing import List
 from traceback import print_exception
+from typing import Optional
+
+import bittensor as bt
+import torch
+from organic_scoring.synth_dataset import SynthDatasetConversation
 
 from prompting.base.neuron import BaseNeuron
 from prompting.mock import MockDendrite
+from prompting.organic.organic_scoring_prompting import OrganicScoringPrompting
 from prompting.utils.config import add_validator_args
 from prompting.utils.exceptions import MaxRetryError
 
@@ -64,9 +66,9 @@ def __init__(self, config=None):
         # Init sync with the network. Updates the metagraph.
         self.sync()
 
-        # Serve axon to enable external connections.
+        self.axon: Optional[bt.axon] = None
         if not self.config.neuron.axon_off:
-            self.serve_axon()
+            self.axon = bt.axon(wallet=self.wallet, config=self.config)
         else:
             bt.logging.warning("axon off, not serving ip to chain.")
 
@@ -79,23 +81,39 @@ def __init__(self, config=None):
         self.thread: threading.Thread = None
         self.lock = asyncio.Lock()
 
-    def serve_axon(self):
-        """Serve axon to enable external connections."""
-
-        bt.logging.info("serving ip to chain...")
-        try:
-            self.axon = bt.axon(wallet=self.wallet, config=self.config)
-
-            try:
-                self.subtensor.serve_axon(
-                    netuid=self.config.netuid,
-                    axon=self.axon,
+        self._organic_scoring: Optional[OrganicScoringPrompting] = None
+        if self.axon is not None and not self.config.neuron.organic_disabled:
+            dataset = SynthDatasetConversation()
+            if dataset.exception is not None:
+                bt.logging.error(
+                    f"Organic scoring on synthetic data is disabled. Failed to load dataset: {dataset.exception}"
                 )
-            except Exception as e:
-                bt.logging.error(f"Failed to serve Axon with exception: {e}")
+                dataset = None
+            self._organic_scoring = OrganicScoringPrompting(
+                axon=self.axon,
+                synth_dataset=dataset,
+                trigger_frequency=self.config.neuron.organic_trigger_frequency,
+                trigger_frequency_min=self.config.neuron.organic_trigger_frequency_min,
+                trigger=self.config.neuron.organic_trigger,
+                trigger_scaling_factor=self.config.neuron.organic_scaling_factor,
+                validator=self,
+            )
+        else:
+            bt.logging.warning(
+                "Organic scoring is not enabled. To enable, remove '--neuron.axon_off' and '--neuron.organic_disabled'"
+            )
+
+        if self.axon is not None:
+            self._serve_axon()
 
-        except Exception as e:
-            bt.logging.error(f"Failed to create Axon initialize with exception: {e}")
+        if self._organic_scoring is not None:
+            self.loop.create_task(self._organic_scoring.start_loop())
+
+    def _serve_axon(self):
+        """Serve axon to enable external connections"""
+        validator_uid = self.metagraph.hotkeys.index(self.wallet.hotkey.ss58_address)
+        bt.logging.info(f"Serving validator IP of UID {validator_uid} to chain...")
+        self.axon.serve(netuid=self.config.netuid, subtensor=self.subtensor).start()
 
     def run(self):
         """
@@ -116,7 +134,6 @@ def run(self):
             KeyboardInterrupt: If the miner is stopped by a manual interruption.
             Exception: For unforeseen errors during the miner's operation, which are logged for diagnosis.
         """
-
         # Check that validator is registered on the network.
         self.sync()
 
@@ -313,7 +330,7 @@ def resync_metagraph(self):
         # Update the hotkeys.
         self.hotkeys = copy.deepcopy(self.metagraph.hotkeys)
 
-    def update_scores(self, rewards: torch.FloatTensor, uids: List[int]):
+    def update_scores(self, rewards: torch.FloatTensor, uids: list[int]):
         """Performs exponential moving average on the scores based on the rewards received from the miners."""
 
         # Check if rewards contains NaN values.
@@ -327,6 +344,7 @@ def update_scores(self, rewards: torch.FloatTensor, uids: List[int]):
         step_rewards = self.scores.scatter(
             0, torch.tensor(uids).to(self.device), rewards.to(self.device)
         ).to(self.device)
+
         bt.logging.debug(f"Scattered rewards: {rewards}")
 
         # Update scores with rewards produced by this step.

prompting/forward.py

@@ -23,6 +23,7 @@
 import numpy as np
 import bittensor as bt
 from typing import List, Dict, Awaitable
+
 from prompting.agent import HumanAgent
 from prompting.dendrite import DendriteResponseEvent, SynapseStreamResult
 from prompting.conversation import create_task
@@ -33,16 +34,9 @@
 from prompting.utils.logging import log_event
 from prompting.utils.misc import async_log, serialize_exception_to_string
 from transformers import PreTrainedTokenizerFast as Tokenizer
-from prompting.utils.uids import get_random_uids
-from dataclasses import dataclass
 
-SINGLE_TURN_TASKS = ['sentiment', 'translation']
+SINGLE_TURN_TASKS = ('sentiment', 'translation')
 
-@async_log
-async def generate_reference(agent):
-    loop = asyncio.get_running_loop()
-    result = await loop.run_in_executor(None, agent.task.generate_reference, agent.llm_pipeline)
-    return result
 
 @async_log
 async def execute_dendrite_call(dendrite_call):
@@ -59,7 +53,8 @@ async def process_stream(uid: int, async_iterator: Awaitable, tokenizer: Tokeniz
     accumulated_tokens_per_chunk = []
     start_time = time.time()
 
-    try:                
+    try:
+        chunk = None
         async for chunk in async_iterator:  # most important loop, as this is where we acquire the final synapse.
             if isinstance(chunk, str):
                 accumulated_chunks.append(chunk)
@@ -76,7 +71,7 @@ async def process_stream(uid: int, async_iterator: Awaitable, tokenizer: Tokeniz
             raise ValueError(
                 f"Something went wrong with miner uid {uid}, Synapse is not StreamPromptingSynapse."
             )
-    except Exception as e:        
+    except Exception as e:
         exception = e
         traceback_details = traceback.format_exc()
         bt.logging.error(
@@ -204,10 +199,11 @@ async def run_step(
     handle_stream_responses_task = asyncio.create_task(handle_response(stream_results_dict, tokenizer))
 
     if not agent.task.static_reference:
-        reference_generation_task = generate_reference(agent)
-        _, stream_results = await asyncio.gather(
-            reference_generation_task, handle_stream_responses_task
-        )
+        async with self.lock:
+            reference_generation_task = generate_reference(agent)
+            _, stream_results = await asyncio.gather(
+                reference_generation_task, handle_stream_responses_task
+            )
     else:
         stream_results = await handle_stream_responses_task
 
@@ -244,7 +240,7 @@ async def run_step(
         "best": best_response,
         "block": self.block,
         "step": self.step,
-        "step_time": time.time() - start_time,        
+        "step_time": time.time() - start_time,
         **agent.__state_dict__(full=self.config.neuron.log_full),
         **reward_result.__state_dict__(full=self.config.neuron.log_full),
         **response_event.__state_dict__(),
@@ -292,7 +288,7 @@ async def forward(self):
 
     turn = 0
     exclude_uids = []
-    roles = ['user']
+    roles = ["user"]
     messages = [agent.challenge]
     while True:
         # Note: The try catch is a safe clause to ensure that the forward loop continues even if an error occurs in run_step.
@@ -314,13 +310,13 @@ async def forward(self):
             event["turn"] = turn
             log_event(self, event)
             task.complete = True
-            
+
             accepted_answer = event["best"] if random.random() < 0.5 else agent.task.reference
             roles.append("assistant")
             messages.append(accepted_answer)
 
-            # 50% chance of single turn conversation, 25% of two turns, 12.5% chance of 3 turns, 6.25% chance of 4 turns, 3.63% chance of 5...
-            if random.random()<0.5 or turn>=1:
+            # 50% chance of single turn conversation, 25% of two turns.
+            if random.random() < 0.5 or turn >= 1:
                 break
 
             if task.name in SINGLE_TURN_TASKS:
@@ -341,13 +337,16 @@ async def forward(self):
         except BaseException as e:
             unexpected_errors = serialize_exception_to_string(e)
             bt.logging.error(
-                f"Error in run_step: Skipping to next round. \n {unexpected_errors}"
+                f"Error in run_step: Skipping to next round.\n"
+                f"Task: {task_name}\nMessages: {messages}\nRoles: {roles}\nTurn: {turn}.\n"
+                f"{unexpected_errors}\n"
             )
 
             event = {"unexpected_errors": unexpected_errors}
 
             log_event(self, event)
-            continue
 
+            await asyncio.sleep(1)
+            continue
     del agent
     del task

prompting/llms/vllm_llm.py

@@ -14,11 +14,10 @@
 # THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
 # OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 # DEALINGS IN THE SOFTWARE.
-import gc
+import threading
 import time
-import torch
 import bittensor as bt
-from typing import List, Dict
+from typing import List, Dict, Optional, Any
 from vllm import LLM, SamplingParams
 from prompting.cleaners.cleaner import CleanerPipeline
 from prompting.llms import BasePipeline, BaseLLM
@@ -55,6 +54,8 @@ def load_vllm_pipeline(model_id: str, device: str, gpus: int, max_allowed_memory
 
 
 class vLLMPipeline(BasePipeline):
+    _LOCK = threading.Lock()
+
     def __init__(
         self,
         model_id: str,
@@ -81,7 +82,8 @@ def __call__(self, composed_prompt: str, **model_kwargs: Dict) -> str:
         sampling_params = SamplingParams(
             temperature=temperature, top_p=top_p, max_tokens=max_tokens
         )
-        output = self.llm.generate(composed_prompt, sampling_params, use_tqdm=True)
+        with self._LOCK:
+            output = self.llm.generate(composed_prompt, sampling_params, use_tqdm=True)
         response = output[0].outputs[0].text
         return response
 
@@ -112,6 +114,30 @@ def __init__(
             "end": "<|start_header_id|>assistant<|end_header_id|>",
         }
 
+    def query_conversation(
+        self,
+        messages: list[str],
+        roles: list[str],
+        cleaner: Optional[CleanerPipeline] = None,
+    ):
+        """Query LLM with the given lists of conversation history and roles
+
+        Args:
+            messages (list[str]): List of messages in the conversation.
+            roles (list[str]): List of roles for each message.
+            cleaner (Optional[CleanerPipeline], optional): Cleaner pipeline to use, if any.
+        """
+        assert len(messages) == len(roles), "Length of messages and roles must be the same"
+        inputs: list[dict[str, Any]] = [{"content": self.system_prompt, "role": "system"}]
+        for role, message in zip(roles, messages):
+            inputs.append({"content": message, "role": role})
+
+        t0 = time.perf_counter()
+        response = self.forward(messages=inputs)
+        response = self.clean_response(cleaner, response)
+        self.times.extend((0, time.perf_counter() - t0))
+        return response
+
     def query(
         self,
         message: str,