Merge pull request #2 from macrocosm-os/features/mock-validator-integration

Back-end development: code refactoring, middlewares and validator abstraction

.gitignore

@@ -0,0 +1,169 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+.DS_Store
+**/.DS_Store
+
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+
+testing/
+core
+app.config.js
+wandb
+.vscode

README.md

@@ -1,4 +1,18 @@
 # chattensor-backend
 Backend for Chattensor app
 
-To run, you will need a bittensor wallet which is registered to the relevant subnet (1@mainnet or 61@testnet).
+To run, you will need a bittensor wallet which is registered to the relevant subnet (1@mainnet or 61@testnet).
+
+
+
+
+## Install
+Create a new python environment and install the dependencies with the command
+
+```bash
+pip install -r requirements.txt
+```
+
+> Note: Currently the prompting library is only installable on machines with cuda devices (NVIDIA-GPU).
+
+

forward.py

@@ -15,17 +15,22 @@ from prompting.utils.logging import log_event
 from prompting.utils.misc import async_log, serialize_exception_to_string
 from dataclasses import dataclass
 
+
 @async_log
-async def generate_reference(agent):    
+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    
+    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):
     responses = await dendrite_call
     return responses
 
+
 @dataclass
 class StreamResult:
     synapse: StreamPromptingSynapse = None
@@ -199,8 +204,8 @@ async def run_step(
 
     log_stream_results(stream_results)
 
-    all_synapses_results = [stream_result.synapse for stream_result in stream_results]   
-            
+    all_synapses_results = [stream_result.synapse for stream_result in stream_results]
+
     # Encapsulate the responses in a response event (dataclass)
     response_event = DendriteResponseEvent(
         responses=all_synapses_results, uids=uids, timeout=timeout

middlewares.py

@@ -0,0 +1,34 @@
+import os
+import json
+import bittensor as bt
+from aiohttp.web import Request, Response, middleware
+
+EXPECTED_ACCESS_KEY = os.environ.get("EXPECTED_ACCESS_KEY")
+
+
+@middleware
+async def api_key_middleware(request: Request, handler):
+    # Logging the request
+    bt.logging.info(f"Handling {request.method} request to {request.path}")
+
+    # Check access key
+    access_key = request.headers.get("api_key")
+    if EXPECTED_ACCESS_KEY is not None and access_key != EXPECTED_ACCESS_KEY:
+        bt.logging.error(f"Invalid access key: {access_key}")
+        return Response(status=401, reason="Invalid access key")
+
+    # Continue to the next handler if the API key is valid
+    return await handler(request)
+
+
+@middleware
+async def json_parsing_middleware(request: Request, handler):
+    try:
+        # Parsing JSON data from the request
+        request["data"] = await request.json()
+    except json.JSONDecodeError as e:
+        bt.logging.error(f"Invalid JSON data: {str(e)}")
+        return Response(status=400, text="Invalid JSON")
+
+    # Continue to the next handler if JSON is successfully parsed
+    return await handler(request)

requirements.txt

@@ -0,0 +1,3 @@
+git+https://github.com/opentensor/prompting.git@features/move-validator-into-prompting
+aiohttp
+deprecated

server.py

@@ -1,23 +1,10 @@
-
-
-
-import os
-import re
-import time
 import asyncio
-import json
-import traceback
+import utils
 import bittensor as bt
-
-from collections import Counter
-
-from neurons.validator import Validator
-from prompting.dendrite import DendriteResponseEvent
-from prompting.protocol import PromptingSynapse
-from prompting.utils.uids import get_random_uids
-from prompting.rewards import DateRewardModel, FloatDiffModel
 from aiohttp import web
 from aiohttp.web_response import Response
+from validators import S1ValidatorAPI, QueryValidatorParams, ValidatorAPI
+from middlewares import api_key_middleware, json_parsing_middleware
 
 """
 # test
@@ -43,270 +30,64 @@ EXPECTED_ACCESS_KEY="hey-michal" pm2 start app.py --interpreter python3 --name a
 
 basic testing
 ```
-EXPECTED_ACCESS_KEY="hey-michal" python app.py --neuron.model_id mock --wallet.name sn1 --wallet.hotkey v1 --netuid 1 --neuron.tasks math --neuron.task_p 1 --neuron.device cpu 
+EXPECTED_ACCESS_KEY="hey-michal" python app.py --neuron.model_id mock --wallet.name sn1 --wallet.hotkey v1 --netuid 1 --neuron.tasks math --neuron.task_p 1 --neuron.device cpu
 ```
 add --mock to test the echo stream
 """
 
-EXPECTED_ACCESS_KEY = os.environ.get('EXPECTED_ACCESS_KEY')
-
-validator = None
-reward_models = {
-    'date_qa': DateRewardModel(),
-    'math': FloatDiffModel(),
-}
-
-def completion_is_valid(completion: str):
-    """
-    Get the completion statuses from the completions.
-    """
-    patt = re.compile(r'I\'m sorry|unable to|I cannot|I can\'t|I am unable|I am sorry|I can not|don\'t know|not sure|don\'t understand')
-    if not len(re.findall(r'\w+',completion)) or patt.search(completion):
-        return False    
-    return True
-
-
-def ensemble_result(completions: list, task_name: str, prefer: str = 'longest'):
-    """
-    Ensemble completions from multiple models.
-    # TODO: Measure agreement
-    # TODO: Figure out how to mitigate the cabal effect (large groups will appear to be more credible)
-    # TODO: Reward pipeline
-    """
-    if not completions:
-        return None
-    
-    
-    answer = None
-    if task_name in ('qa', 'summarization'):
-        # No special handling for QA or summarization
-        supporting_completions = completions
-    
-    elif task_name == 'date_qa':
-        # filter the completions to be the ones that contain valid dates and if there are multiple dates, select the most common one (with support > 1)
-        dates = list(map(reward_models[task_name].parse_dates_from_text, completions))
-        bt.logging.info(f"Unprocessed dates: {dates}")
-        valid_date_indices = [i for i, d in enumerate(dates) if d]
-        valid_completions = [completions[i] for i in valid_date_indices]
-        valid_dates = [dates[i] for i in valid_date_indices]
-        dates = [f"{d[0].strftime('%-d %B')} {d[1]}" for d in valid_dates]
-        if not dates:
-            return None
-        
-        counter = Counter(dates)
-        most_common, count = counter.most_common()[0]
-        answer = most_common
-        if count == 1:
-            supporting_completions = valid_completions
-        else:
-            supporting_completions = [c for i, c in enumerate(valid_completions) if dates[i]==most_common]
-        
-    elif task_name == 'math':
-        # filter the completions to be the ones that contain valid numbers and if there are multiple values, select the most common one (with support > 1)
-        # TODO: use the median instead of the most common value
-        vals = list(map(reward_models[task_name].extract_number, completions))
-        vals = [val for val in vals if val]
-        if not vals:
-            return None
-        
-        most_common, count = Counter(dates).most_common()[0]
-        bt.logging.info(f"Most common value: {most_common}, count: {count}")
-        answer = most_common
-        if count == 1:
-            supporting_completions = completions
-        else:
-            supporting_completions = [c for i, c in enumerate(completions) if vals[i]==most_common]
-        
-    
-    bt.logging.info(f"Supporting completions: {supporting_completions}")
-    if prefer == 'longest':
-        preferred_completion = sorted(supporting_completions, key=len)[-1]
-    elif prefer == 'shortest':
-        preferred_completion = sorted(supporting_completions, key=len)[0]
-    elif prefer == 'most_common':
-        preferred_completion = max(set(supporting_completions), key=supporting_completions.count)
-    else:
-        raise ValueError(f"Unknown ensemble preference: {prefer}")
-        
-    return {
-        'completion': preferred_completion,
-        'accepted_answer': answer, 
-        'support': len(supporting_completions),
-        'support_indices': [completions.index(c) for c in supporting_completions],
-        'method': f'Selected the {prefer.replace("_", " ")} completion'
-    }
-    
-def guess_task_name(challenge: str):
-    categories = {
-        'summarization': re.compile('summar|quick rundown|overview'),
-        'date_qa': re.compile('exact date|tell me when|on what date|on what day|was born?|died?'),
-        'math': re.compile('math|solve|solution| sum |problem|geometric|vector|calculate|degrees|decimal|factorial'),
-    }
-    for task_name, patt in categories.items():
-        if patt.search(challenge):
-            return task_name
-
-    return 'qa'
 
 async def chat(request: web.Request) -> Response:
     """
     Chat endpoint for the validator.
-
-    Required headers:
-    - api_key: The access key for the validator.
-
-    Required body:
-    - roles: The list of roles to query.
-    - messages: The list of messages to query.
-    Optional body:
-    - k: The number of nodes to query.
-    - exclude: The list of nodes to exclude from the query.
-    - timeout: The timeout for the query.
     """
+    params = QueryValidatorParams.from_request(request)
 
-    bt.logging.info(f'chat()')
-    # Check access key
-    access_key = request.headers.get("api_key")
-    if EXPECTED_ACCESS_KEY is not None and access_key != EXPECTED_ACCESS_KEY:
-        bt.logging.error(f'Invalid access key: {access_key}')
-        return Response(status=401, reason="Invalid access key")
-
-    try:
-        request_data = await request.json()
-    except ValueError:
-        bt.logging.error(f'Invalid request data: {request_data}')
-        return Response(status=400)
-
-    bt.logging.info(f'Request data: {request_data}')
-    k = request_data.get('k', 10)
-    exclude = request_data.get('exclude', [])
-    timeout = request_data.get('timeout', 10)
-    prefer = request_data.get('prefer', 'longest')
-    try:
-        # Guess the task name of current request
-        task_name = guess_task_name(request_data['messages'][-1])
-        
-        # Get the list of uids to query for this step.
-        uids = get_random_uids(validator, k=k, exclude=exclude or []).to(validator.device)
-        axons = [validator.metagraph.axons[uid] for uid in uids]
-
-        # Make calls to the network with the prompt.
-        bt.logging.info(f'Calling dendrite')
-        responses = await validator.dendrite(
-            axons=axons,
-            synapse=PromptingSynapse(roles=request_data['roles'], messages=request_data['messages']),
-            timeout=timeout,
-        )
+    # Access the validator from the application context
+    validator: ValidatorAPI = request.app["validator"]
 
-        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(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'] = ensemble_result(valid_completions, task_name=task_name, prefer=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 {chat.__name__}:\n{traceback.format_exc()}')
-        return Response(status=500, reason="Internal error")
-
-
-
-async def echo_stream(request):
-    
-    bt.logging.info(f'echo_stream()')
-    # Check access key
-    access_key = request.headers.get("api_key")
-    if EXPECTED_ACCESS_KEY is not None and access_key != EXPECTED_ACCESS_KEY:
-        bt.logging.error(f'Invalid access key: {access_key}')
-        return Response(status=401, reason="Invalid access key")
-
-    try:
-        request_data = await request.json()
-    except ValueError:
-        bt.logging.error(f'Invalid request data: {request_data}')
-        return Response(status=400)
-
-    bt.logging.info(f'Request data: {request_data}')
-    k = request_data.get('k', 1)
-    exclude = request_data.get('exclude', [])
-    timeout = request_data.get('timeout', 0.2)    
-    message = '\n\n'.join(request_data['messages'])
-
-    # Create a StreamResponse
-    response = web.StreamResponse(status=200, reason='OK', headers={'Content-Type': 'text/plain'})
-    await response.prepare(request)
-
-    completion = ''
-    # Echo the message k times with a timeout between each chunk
-    for _ in range(k):
-        for word in message.split():
-            chunk = f'{word} '
-            await response.write(chunk.encode('utf-8'))
-            completion += chunk 
-            time.sleep(timeout)
-            bt.logging.info(f"Echoed: {chunk}")
+    response = await validator.query_validator(params)
+    return response
 
-    completion = completion.strip()
 
-    # Prepare final JSON chunk
-    json_chunk = json.dumps({
-        "uids": [0],
-        "completion": completion,
-        "completions": [completion.strip()],
-        "timings": [0],
-        "status_messages": ['Went well!'],
-        "status_codes": [200],
-        "completion_is_valid": [True],
-        "task_name": 'echo',
-        "ensemble_result": {}
-    })
-    
-    # Send the final JSON as part of the stream
-    await response.write(f"\n\nJSON_RESPONSE_BEGIN:\n{json_chunk}".encode('utf-8'))
+async def echo_stream(request, request_data):
+    request_data = request["data"]
+    return await utils.echo_stream(request_data)
 
-    # Finalize the response
-    await response.write_eof()
-    return response
 
 class ValidatorApplication(web.Application):
-    def __init__(self, *a, **kw):
-        super().__init__(*a, **kw)
-        # TODO: Enable rewarding and other features
+    def __init__(self, validator_instance=None, *args, **kwargs):
+        super().__init__(*args, **kwargs)
 
+        self["validator"] = (
+            validator_instance if validator_instance else S1ValidatorAPI()
+        )
 
-validator_app = ValidatorApplication()
-validator_app.add_routes([
-    web.post('/chat/', chat),
-    web.post('/echo/', echo_stream)
-    ])
+        # Add middlewares to application
+        self.add_routes([web.post("/chat/", chat), web.post("/echo/", echo_stream)])
+        self.setup_middlewares()
+        # TODO: Enable rewarding and other features
 
-bt.logging.info("Starting validator application.")
-bt.logging.info(validator_app)
+    def setup_middlewares(self):
+        self.middlewares.append(json_parsing_middleware)
+        self.middlewares.append(api_key_middleware)
 
 
 def main(run_aio_app=True, test=False) -> None:
-
     loop = asyncio.get_event_loop()
-
-    # port = validator.metagraph.axons[validator.uid].port
     port = 10000
     if run_aio_app:
+        # Instantiate the application with the actual validator
+        bt.logging.info("Starting validator application.")
+        validator_app = ValidatorApplication()
+        bt.logging.success(f"Validator app initialized successfully", validator_app)
+
         try:
             web.run_app(validator_app, port=port, loop=loop)
         except KeyboardInterrupt:
-            bt.logging.info("Keyboard interrupt detected. Exiting validator.")
+            print("Keyboard interrupt detected. Exiting validator.")
         finally:
             pass
 
+
 if __name__ == "__main__":
-    validator = Validator()
     main()

test.py

@@ -0,0 +1,17 @@
+import pytest
+
+
+def test_query_network():
+    pass
+
+
+def test_filter_completions():
+    pass
+
+
+def test_guess_task_name():
+    pass
+
+
+def test_ensemble_completions():
+    pass

utils.py

@@ -0,0 +1,181 @@
+import re
+import bittensor as bt
+import time
+import json
+from aiohttp import web
+from collections import Counter
+from prompting.rewards import DateRewardModel, FloatDiffModel
+
+UNSUCCESSFUL_RESPONSE_PATTERNS = [
+    "I'm sorry",
+    "unable to",
+    "I cannot",
+    "I can't",
+    "I am unable",
+    "I am sorry",
+    "I can not",
+    "don't know",
+    "not sure",
+    "don't understand",
+    "not capable",
+]
+
+reward_models = {
+    "date_qa": DateRewardModel(),
+    "math": FloatDiffModel(),
+}
+
+
+def completion_is_valid(completion: str):
+    """
+    Get the completion statuses from the completions.
+    """
+    if not completion.strip():
+        return False
+
+    patt = re.compile(
+        r"\b(?:" + "|".join(UNSUCCESSFUL_RESPONSE_PATTERNS) + r")\b", re.IGNORECASE
+    )
+    if not len(re.findall(r"\w+", completion)) or patt.search(completion):
+        return False
+    return True
+
+
+def ensemble_result(completions: list, task_name: str, prefer: str = "longest"):
+    """
+    Ensemble completions from multiple models.
+    # TODO: Measure agreement
+    # TODO: Figure out how to mitigate the cabal effect (large groups will appear to be more credible)
+    # TODO: Reward pipeline
+    """
+    if not completions:
+        return None
+
+    answer = None
+    if task_name in ("qa", "summarization"):
+        # No special handling for QA or summarization
+        supporting_completions = completions
+
+    elif task_name == "date_qa":
+        # filter the completions to be the ones that contain valid dates and if there are multiple dates, select the most common one (with support > 1)
+        dates = list(map(reward_models[task_name].parse_dates_from_text, completions))
+        bt.logging.info(f"Unprocessed dates: {dates}")
+        valid_date_indices = [i for i, d in enumerate(dates) if d]
+        valid_completions = [completions[i] for i in valid_date_indices]
+        valid_dates = [dates[i] for i in valid_date_indices]
+        dates = [f"{d[0].strftime('%-d %B')} {d[1]}" for d in valid_dates]
+        if not dates:
+            return None
+
+        counter = Counter(dates)
+        most_common, count = counter.most_common()[0]
+        answer = most_common
+        if count == 1:
+            supporting_completions = valid_completions
+        else:
+            supporting_completions = [
+                c for i, c in enumerate(valid_completions) if dates[i] == most_common
+            ]
+
+    elif task_name == "math":
+        # filter the completions to be the ones that contain valid numbers and if there are multiple values, select the most common one (with support > 1)
+        # TODO: use the median instead of the most common value
+        vals = list(map(reward_models[task_name].extract_number, completions))
+        vals = [val for val in vals if val]
+        if not vals:
+            return None
+
+        most_common, count = Counter(dates).most_common()[0]
+        bt.logging.info(f"Most common value: {most_common}, count: {count}")
+        answer = most_common
+        if count == 1:
+            supporting_completions = completions
+        else:
+            supporting_completions = [
+                c for i, c in enumerate(completions) if vals[i] == most_common
+            ]
+
+    bt.logging.info(f"Supporting completions: {supporting_completions}")
+    if prefer == "longest":
+        preferred_completion = sorted(supporting_completions, key=len)[-1]
+    elif prefer == "shortest":
+        preferred_completion = sorted(supporting_completions, key=len)[0]
+    elif prefer == "most_common":
+        preferred_completion = max(
+            set(supporting_completions), key=supporting_completions.count
+        )
+    else:
+        raise ValueError(f"Unknown ensemble preference: {prefer}")
+
+    return {
+        "completion": preferred_completion,
+        "accepted_answer": answer,
+        "support": len(supporting_completions),
+        "support_indices": [completions.index(c) for c in supporting_completions],
+        "method": f'Selected the {prefer.replace("_", " ")} completion',
+    }
+
+
+def guess_task_name(challenge: str):
+    # TODO: use a pre-trained classifier to guess the task name
+    categories = {
+        "summarization": re.compile("summar|quick rundown|overview"),
+        "date_qa": re.compile(
+            "exact date|tell me when|on what date|on what day|was born?|died?"
+        ),
+        "math": re.compile(
+            "math|solve|solution| sum |problem|geometric|vector|calculate|degrees|decimal|factorial"
+        ),
+    }
+    for task_name, patt in categories.items():
+        if patt.search(challenge):
+            return task_name
+
+    return "qa"
+
+
+async def echo_stream(request_data: dict):
+    k = request_data.get("k", 1)
+    exclude = request_data.get("exclude", [])
+    timeout = request_data.get("timeout", 0.2)
+    message = "\n\n".join(request_data["messages"])
+
+    # Create a StreamResponse
+    response = web.StreamResponse(
+        status=200, reason="OK", headers={"Content-Type": "text/plain"}
+    )
+    await response.prepare()
+
+    completion = ""
+    # Echo the message k times with a timeout between each chunk
+    for _ in range(k):
+        for word in message.split():
+            chunk = f"{word} "
+            await response.write(chunk.encode("utf-8"))
+            completion += chunk
+            time.sleep(timeout)
+            bt.logging.info(f"Echoed: {chunk}")
+
+    completion = completion.strip()
+
+    # Prepare final JSON chunk
+    json_chunk = json.dumps(
+        {
+            "uids": [0],
+            "completion": completion,
+            "completions": [completion.strip()],
+            "timings": [0],
+            "status_messages": ["Went well!"],
+            "status_codes": [200],
+            "completion_is_valid": [True],
+            "task_name": "echo",
+            "ensemble_result": {},
+        }
+    )
+
+    # Send the final JSON as part of the stream
+    await response.write(f"\n\nJSON_RESPONSE_BEGIN:\n{json_chunk}".encode("utf-8"))
+
+    # Finalize the response
+    await response.write_eof()
+    return response

validators/__init__.py

@@ -0,0 +1,2 @@
+from .base import QueryValidatorParams, ValidatorAPI, MockValidator
+from .sn1_validator_wrapper import S1ValidatorAPI

validators/base.py

@@ -0,0 +1,40 @@
+from abc import ABC, abstractmethod
+from typing import List
+from dataclasses import dataclass
+from aiohttp.web import Response, Request
+
+
+@dataclass
+class QueryValidatorParams:
+    k_miners: int
+    exclude: List[str]
+    roles: List[str]
+    messages: List[str]
+    timeout: int
+    prefer: str
+    request: Request
+
+    @staticmethod
+    def from_request(request: Request):
+        data = request["data"]
+
+        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"),
+            request=request,
+        )
+
+
+class ValidatorAPI(ABC):
+    @abstractmethod
+    async def query_validator(self, params: QueryValidatorParams) -> Response:
+        pass
+
+
+class MockValidator(ValidatorAPI):
+    async def query_validator(self, params: QueryValidatorParams) -> Response:
+        ...

validators/sn1_validator_wrapper.py

@@ -0,0 +1,154 @@
+import json
+import utils
+import torch
+import traceback
+import asyncio
+import bittensor as bt
+from typing import Awaitable
+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, ValidatorAPI
+from aiohttp.web_response import Response, StreamResponse
+from deprecated import deprecated
+
+
+class S1ValidatorAPI(ValidatorAPI):
+    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, torch.LongTensor(uids), params.timeout
+            )
+
+            # 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 process_response(
+        self, response: StreamResponse, uid: int, async_generator: Awaitable
+    ):
+        """Process a single response asynchronously."""
+        try:
+            chunk = None  # Initialize chunk with a default value
+            async for chunk in async_generator:  # most important loop, as this is where we acquire the final synapse.
+                bt.logging.debug(f"\nchunk for uid {uid}: {chunk}")
+
+                # TODO: SET PROPER IMPLEMENTATION TO RETURN CHUNK
+                if chunk is not None:
+                    json_data = json.dumps(chunk)
+                    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
+
+    async def get_stream_response(self, params: QueryValidatorParams) -> StreamResponse:
+        response = StreamResponse(status=200, reason="OK")
+        response.headers["Content-Type"] = "application/json"
+
+        await response.prepare(params.request)  # 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,
+            )
+
+            tasks = [
+                self.process_response(uid, res)
+                for uid, res in dict(zip(uids, streams_responses))
+            ]
+            results = await asyncio.gather(*tasks, return_exceptions=True)
+
+            # TODO: Continue implementation, business decision needs to be made on how to handle the results
+        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) -> Response:
+        # TODO: SET STREAM AS DEFAULT
+        stream = params.request.get("stream", False)
+
+        if stream:
+            return await self.get_stream_response(params)
+        else:
+            # DEPRECATED
+            return await self.get_response(params)