Skip to content

Dummy Demo Server

Bases: BaseServerObject

This is a Test Server used for Debugging Orchestrator flow.

Source code in app/servers/dummy_server.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
class DummyLocalOrchestrator(BaseServerObject):
    """
        This is a Test Server used for Debugging Orchestrator flow.
    """
    def __init__(self,**kwargs):
        """
        """
        self.config = kwargs
        self.model_id = self.config.get("model_id")
        self.instance_name = self.config.get("model_id")
        self.config.update(dict(logs=[]))

    def start_server(self):
        self.config['logs'].append(f'Dummy Orchestrator server {self.model_id}')

    def start_inference_endpoint(self, max_wait_time=120):
        self.config['logs'].append(f'Dummy Orchestrator inference endpoint {self.model_id}')

    def stop_server(self):
        return "server stopped"

    def check_servers_state(self):
        return (True, 'running')

    def get_response(self, message, stream=False):
        """Since this is a dummy orchestrator for testing, 
        it would return random expert id for each of the input query"""
        return random.randint(0,2)

__init__(**kwargs)

Source code in app/servers/dummy_server.py 
11
12
13
14
15
16
17
def __init__(self,**kwargs):
    """
    """
    self.config = kwargs
    self.model_id = self.config.get("model_id")
    self.instance_name = self.config.get("model_id")
    self.config.update(dict(logs=[]))

get_response(message, stream=False)

Since this is a dummy orchestrator for testing, it would return random expert id for each of the input query

Source code in app/servers/dummy_server.py 
31
32
33
34
def get_response(self, message, stream=False):
    """Since this is a dummy orchestrator for testing, 
    it would return random expert id for each of the input query"""
    return random.randint(0,2)