reallocates and refactors validator code to internal package

validators/__init__.py

@@ -0,0 +1,2 @@
+from base import QueryValidatorParams, ValidatorWrapper, MockValidator
+from sn1_validator_wrapper import S1ValidatorWrapper

validators/base.py

@@ -0,0 +1,35 @@
+from abc import ABC, abstractmethod
+from typing import List
+from dataclasses import dataclass
+from aiohttp.web_response import Response
+
+@dataclass
+class QueryValidatorParams:
+    k_miners: int
+    exclude: List[str]
+    roles: List[str]
+    messages: List[str]
+    timeout: int
+    prefer: str
+    
+    @staticmethod
+    def from_dict(data: dict):
+        return QueryValidatorParams(            
+            k_miners=data.get('k', 10),
+            exclude=data.get('exclude', []),
+            roles=data['roles'],
+            messages=data['messages'],
+            timeout=data.get('timeout', 10),
+            prefer=data.get('prefer', 'longest')
+        )
+
+class ValidatorWrapper(ABC):
+    @abstractmethod
+    async def query_validator(self, params:QueryValidatorParams) -> Response:
+        pass
+    
+    
+class MockValidator(ValidatorWrapper):    
+    async def query_validator(self, params:QueryValidatorParams) -> Response:
+        ...
+        

validators/sn1_validator_wrapper.py

@@ -0,0 +1,106 @@
+import json
+import utils
+import traceback
+import bittensor as bt
+import asyncio
+from prompting.forward import handle_response
+from prompting.validator import Validator
+from prompting.utils.uids import get_random_uids
+from prompting.protocol import PromptingSynapse, StreamPromptingSynapse
+from prompting.dendrite import DendriteResponseEvent
+from base import QueryValidatorParams, ValidatorWrapper
+from aiohttp.web_response import Response, StreamResponse
+from deprecated import deprecated
+
+class S1ValidatorWrapper(ValidatorWrapper):
+    def __init__(self):
+        self.validator = Validator()    
+    
+                    
+    @deprecated(reason="This function is deprecated. Validators use stream synapse now, use get_stream_response instead.")
+    async def get_response(self, params:QueryValidatorParams) -> Response:
+        try:
+            # Guess the task name of current request
+            task_name = utils.guess_task_name(params.messages[-1])
+
+            # Get the list of uids to query for this step.
+            uids = get_random_uids(self.validator, k=params.k_miners, exclude=params.exclude or []).tolist()
+            axons = [self.validator.metagraph.axons[uid] for uid in uids]
+
+            # Make calls to the network with the prompt.
+            bt.logging.info(f'Calling dendrite')
+            responses = await self.validator.dendrite(
+                axons=axons,
+                synapse=PromptingSynapse(roles=params.roles, messages=params.messages),
+                timeout=params.timeout,
+            )
+
+            bt.logging.info(f"Creating DendriteResponseEvent:\n {responses}")
+            # Encapsulate the responses in a response event (dataclass)
+            response_event = DendriteResponseEvent(responses, uids)
+
+            # convert dict to json
+            response = response_event.__state_dict__()
+
+            response['completion_is_valid'] = valid = list(map(utils.completion_is_valid, response['completions']))
+            valid_completions = [response['completions'][i] for i, v in enumerate(valid) if v]
+
+            response['task_name'] = task_name
+            response['ensemble_result'] = utils.ensemble_result(valid_completions, task_name=task_name, prefer=params.prefer)
+
+            bt.logging.info(f"Response:\n {response}")
+            return Response(status=200, reason="I can't believe it's not butter!", text=json.dumps(response))
+
+        except Exception:
+            bt.logging.error(f'Encountered in {self.__class__.__name__}:get_response:\n{traceback.format_exc()}')
+            return Response(status=500, reason="Internal error")
+        
+        
+    async def get_stream_response(self, params:QueryValidatorParams) -> StreamResponse:
+        response = StreamResponse(status=200, reason="OK")
+        response.headers['Content-Type'] = 'application/json'
+
+        await response.prepare()  # Prepare and send the headers
+        
+        try:
+            # Guess the task name of current request
+            task_name = utils.guess_task_name(params.messages[-1])
+
+            # Get the list of uids to query for this step.
+            uids = get_random_uids(self.validator, k=params.k_miners, exclude=params.exclude or []).tolist()
+            axons = [self.validator.metagraph.axons[uid] for uid in uids]
+
+            # Make calls to the network with the prompt.
+            bt.logging.info(f'Calling dendrite')
+            streams_responses = await self.validator.dendrite(
+                axons=axons,
+                synapse=StreamPromptingSynapse(roles=params.roles, messages=params.messages),
+                timeout=params.timeout,
+                deserialize=False,
+                streaming=True,
+            )
+
+            # Asynchronous iteration over streaming responses
+            async for stream_result in streams_responses:
+                if stream_result is not None:
+                    # Convert stream result to JSON and write to the response stream
+                    json_data = json.dumps(stream_result)
+                    await response.write(json_data.encode('utf-8'))
+
+        except Exception as e:
+            bt.logging.error(f'Encountered an error in {self.__class__.__name__}:get_stream_response:\n{traceback.format_exc()}')
+            response.set_status(500, reason="Internal error")
+            await response.write(json.dumps({'error': str(e)}).encode('utf-8'))
+        finally:
+            await response.write_eof()  # Ensure to close the response properly
+
+        return response
+            
+
+    async def query_validator(self, params:QueryValidatorParams, stream: bool = True) -> Response:        
+        if stream:
+            return await self.get_stream_response(params)
+        else:
+            # DEPRECATED
+            return await self.get_response(params)
+