Bittensor shared request layer

[ifrit98]

Creates a generic shared request layer for bittensor subnets to facilitate cross-communication. Introduces the bittensor.subnets.SubnetsAPI abstract class to achieve this.

What does Bittensor communication entail? Typically two processes, (1) preparing data for transit (creating and filling synapses) and (2), processing the responses received from the axon(s).

This protocol uses a handler registry system to associate bespoke interfaces for subnets by implementing two simple abstract functions:

• prepare_synapse
• process_responses

These can be implemented as extensions of the generic SubnetsAPI interface. E.g.:

This is abstract, generic, and takes(*args, **kwargs) for flexibility. See the extremely simple base class:

class SubnetsAPI(ABC):
    def __init__(self, wallet: "bt.wallet"):
        self.wallet = wallet
        self.dendrite = bt.dendrite(wallet=wallet)

    async def __call__(self, *args, **kwargs):
        return await self.query_api(*args, **kwargs)

    @abstractmethod
    def prepare_synapse(self, *args, **kwargs) -> Any:
        """
        Prepare the synapse-specific payload.
        """
        ...

    @abstractmethod
    def process_responses(self, responses: List[Union["bt.Synapse", Any]]) -> Any:
        """
        Process the responses from the network.
        """
        ...

Here is a toy example:

from bittensor.subnets import SubnetsAPI
from MySubnet import MySynapse

class MySynapseAPI(SubnetsAPI):
    def __init__(self, wallet: "bt.wallet"):
        super().__init__(wallet)
        self.netuid = 99

    def prepare_synapse(self, prompt: str) -> MySynapse:
        # Do any preparatory work to fill the synapse
        data = do_prompt_injection(prompt)

        # Fill the synapse for transit
        synapse = StoreUser(
            messages=[data],
        )
        # Send it along
        return synapse

    def process_responses(self, responses: List[Union["bt.Synapse", Any]]) -> str:
        # Look through the responses for information required by your application
        for response in responses:
            if response.dendrite.status_code != 200:
                continue
            # potentially apply post processing
            result_data = postprocess_data_from_response(response)
        # return data to the client
        return result_data

You can use a subnet API to the registry by doing the following:
(1) Download and install the specific repo you want
(2) Import the appropriate API handler from bespoke subnets
(3) Make the query given the subnet specific API

See simplified example for subnet 21 below:

# Subnet 21 Interface

class RetrieveUserAPI(SubnetsAPI):
    def __init__(self, wallet: "bt.wallet"):
        super().__init__(wallet)
        self.netuid = 21

    def prepare_synapse(self, cid: str) -> RetrieveUser:
        synapse = RetrieveUser(data_hash=cid)
        return synapse

    def process_responses(self, responses: List[Union["bt.Synapse", Any]]) -> bytes:
        success = False
        decrypted_data = b""
        for response in responses:
            .... # retrieve the data
        return data

 

# import the bespoke subnet API
from storage import StoreUserAPI, RetrieveUserAPI

wallet = bt.wallet()
metagraph = ...
query_axon = ... # logic to retrieve desired axons (e.g. validator set)

bt.logging.info(f"Initiating store_handler: {store_handler}")
cid = await StoreUserAPI(
      axons=query_axons,
      data=b"some data",
      encrypt=False,
      ttl=60 * 60 * 24 * 30,
      encoding="utf-8",
      uid=None,
)

# Now retrieve data from SN21 (storage)
retrieve_response = await RetrieveUserAPI(axons=query_axons, cid=cid)