ollama_request.extensions.ollama_request_extension 

 1from typing import Type
 2
 3from ollama import AsyncClient
 4from semantic_version import Version, SimpleSpec
 5
 6from aaambos.core.module.extension import Extension, ExtensionSetup
 7from aaambos.core.module.feature import ExtensionFeature
 8
 9from ollama_request.conventions.definitions import OllamaRequest
10from ollama_request.utils import get_response
11
12OLLAMA_REQUEST_EXTENSION = "OllamaRequestExtension"
13
14
15class OllamaRequestExtension(Extension):
16    """
17
18    """
19    name = OLLAMA_REQUEST_EXTENSION
20
21    async def before_module_initialize(self, module):
22        self.module = module
23        self.ollama_clients = {}
24    
25    async def after_module_initialize(self, module):
26        ...
27    
28    def add_ollama_client(self, host: str = None, ip: str = "localhost", port: int = 11434, use_ssl: bool = False, ref=None):
29        if not host:
30            host = f"http{'s' if use_ssl else ''}://{ip}:{port}"
31        self.ollama_clients[host] = AsyncClient(host=host)
32        self.ollama_clients[ref] = self.ollama_clients[host]
33        
34    async def request_ollama(self, request: OllamaRequest, ref=None):
35        if ref and ref in self.ollama_clients:
36            client = self.ollama_clients[ref]
37            return await get_response(client, request)
38        if request.ollama_host:
39            if request.ollama_host not in self.ollama_clients:
40                self.add_ollama_client(host=request.ollama_host, ref=ref)
41            return await get_response(self.ollama_clients[request.ollama_host], request)
42    
43    @classmethod
44    def get_extension_setup(cls) -> Type[ExtensionSetup]:
45        return OllamaRequestExtensionSetup
46
47
48class OllamaRequestExtensionSetup(ExtensionSetup):
49    
50    def before_module_start(self, module_config) -> Extension:
51        return OllamaRequestExtension()
52
53
54OllamaRequestExtensionExtensionFeature = ExtensionFeature(
55    name=OLLAMA_REQUEST_EXTENSION,
56    version=Version("0.1.0"),
57    requirements=[],
58    version_compatibility=SimpleSpec(">=0.1.0"),
59    extension_setup_class=OllamaRequestExtensionSetup,
60    as_kwarg=True,
61    kwarg_alias="ollama_request",
62)
OLLAMA_REQUEST_EXTENSION = 'OllamaRequestExtension'
class OllamaRequestExtension(aaambos.core.module.extension.Extension): 
16class OllamaRequestExtension(Extension):
17    """
18
19    """
20    name = OLLAMA_REQUEST_EXTENSION
21
22    async def before_module_initialize(self, module):
23        self.module = module
24        self.ollama_clients = {}
25    
26    async def after_module_initialize(self, module):
27        ...
28    
29    def add_ollama_client(self, host: str = None, ip: str = "localhost", port: int = 11434, use_ssl: bool = False, ref=None):
30        if not host:
31            host = f"http{'s' if use_ssl else ''}://{ip}:{port}"
32        self.ollama_clients[host] = AsyncClient(host=host)
33        self.ollama_clients[ref] = self.ollama_clients[host]
34        
35    async def request_ollama(self, request: OllamaRequest, ref=None):
36        if ref and ref in self.ollama_clients:
37            client = self.ollama_clients[ref]
38            return await get_response(client, request)
39        if request.ollama_host:
40            if request.ollama_host not in self.ollama_clients:
41                self.add_ollama_client(host=request.ollama_host, ref=ref)
42            return await get_response(self.ollama_clients[request.ollama_host], request)
43    
44    @classmethod
45    def get_extension_setup(cls) -> Type[ExtensionSetup]:
46        return OllamaRequestExtensionSetup
name = 'OllamaRequestExtension'
async def before_module_initialize(self, module): 
22    async def before_module_initialize(self, module):
23        self.module = module
24        self.ollama_clients = {}

Is called before the module 'initialze' call. The com is already initialized.

Args: module: the assigned module of the extension.

Returns: None

async def after_module_initialize(self, module): 
26    async def after_module_initialize(self, module):
27        ...

Is called after the module 'initialize' call.

Args: module: the assigned module of the extension.

Returns: None

def add_ollama_client( self, host: str = None, ip: str = 'localhost', port: int = 11434, use_ssl: bool = False, ref=None): 
29    def add_ollama_client(self, host: str = None, ip: str = "localhost", port: int = 11434, use_ssl: bool = False, ref=None):
30        if not host:
31            host = f"http{'s' if use_ssl else ''}://{ip}:{port}"
32        self.ollama_clients[host] = AsyncClient(host=host)
33        self.ollama_clients[ref] = self.ollama_clients[host]
async def request_ollama( self, request: ollama_request.conventions.definitions.OllamaRequest, ref=None): 
35    async def request_ollama(self, request: OllamaRequest, ref=None):
36        if ref and ref in self.ollama_clients:
37            client = self.ollama_clients[ref]
38            return await get_response(client, request)
39        if request.ollama_host:
40            if request.ollama_host not in self.ollama_clients:
41                self.add_ollama_client(host=request.ollama_host, ref=ref)
42            return await get_response(self.ollama_clients[request.ollama_host], request)
@classmethod
def get_extension_setup(cls) -> Type[aaambos.core.module.extension.ExtensionSetup]: 
44    @classmethod
45    def get_extension_setup(cls) -> Type[ExtensionSetup]:
46        return OllamaRequestExtensionSetup

Provides the extension setup class.

Maybe delete in the future.

Returns: the extension setup class that can init the extension in the before_module_start method.

class OllamaRequestExtensionSetup(aaambos.core.module.extension.ExtensionSetup): 
49class OllamaRequestExtensionSetup(ExtensionSetup):
50    
51    def before_module_start(self, module_config) -> Extension:
52        return OllamaRequestExtension()

Setup of one or more extensions.

def before_module_start(self, module_config) -> aaambos.core.module.extension.Extension: 
51    def before_module_start(self, module_config) -> Extension:
52        return OllamaRequestExtension()

Before the module start the extension setup provides the extension.

Args: module_config: the module config.

Returns: The initialized extension.

OllamaRequestExtensionExtensionFeature = ExtensionFeature(name='OllamaRequestExtension', version=Version('0.1.0'), requirements=[], version_compatibility=<SimpleSpec: '>=0.1.0'>, behavior_to_same_provided_feature=None, extension_setup_class=<class 'OllamaRequestExtensionSetup'>, as_kwarg=True, kwarg_alias='ollama_request', shared_extension_setup=True)