KyMAN/AnP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#wip(py): Building AI Sessions and PseudoLoRAs interaction

Browse Source
This commit is contained in:
KyMAN 2026-06-01 08:02:00 +02:00
parent 6a2a778110
commit 8bd5f46c3a
27 changed files with 754 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
JSON/AnP.pseudoloras.json Normal file
View File

@ -0,0 +1,4 @@
[
["Qué es AnP", "/MD/PseudoLoRAs/AnP.md", [], false],
["Gestión de configuración", "/MD/AnP.Settings.md", [], false]
]

4
JSON/AnP.routes.json
View File

@ -1,4 +1,4 @@
[
"[anp]get:/ /Public",
"[anp]post:/ai/new_message ai@new_message"
"[anp]get:/ai/test test@ai",
"[anp]get:/ /Public"
]

23
JSON/AnP.settings.json
View File

@ -78,6 +78,29 @@
},
"AnP_HTTPServersManager_end" : null,
"AnP_PseudoLoRAsManager_start" : null,
"default_pseudoloras_files" : "/JSON/AnP.pseudoloras.json",
"AnP_PseudoLoRAsManager_end" : null,
"AnP_AIInterpretersManager_start" : null,
"default_ai_interpreters" : {
"anp_titles" : {
"type" : "OllamaDriver",
"host" : "localhost",
"port" : 11434,
"model" : "gemma3:1b",
"pool" : "anp"
},
"anp_responses" : {
"type" : "OllamaDriver",
"host" : "localhost",
"port" : 11434,
"model" : "gemma4:e4b",
"pool" : "anp"
}
},
"AnP_AIInterpretersManager_end" : null,
"AnP_TitlesManager_start" : null,
"default_titles_files" : [
"/JSON/AnP.titles.json",

13
MD/AnP.Settings.md Normal file
View File

@ -0,0 +1,13 @@
## Gestión de configuración
Para gestionar las configuraciones de AnP hay que mandarle dichos parámetros por diccionarios, ya sea JSON como un diccionario a nivel de código. Se pueden apilar en Arrays.
Los parámetros de configuración son los siguientes:
- **autostart**: Booleano. Determina si se autoinicia o no.
- **end_print_types**: Array de Strings. Determina los tipos de impresión que integra el valor `end` en el `print`.
- **root_projects_paths**: Array de Strings. Determina las rutas absolutas de los proyectos. Con esto se suprime esta parte del Path en la impresión de los ficheros.
- **print_format**: String. Formato de impresión.
- **exception_format**: String. Formato de impresión de la parte que define una excepción.
A mayores, cada librería, Driver o Gestor entre otros, tiene sus propios parámetros de configuración.

9
MD/AnP.md Normal file
View File

@ -0,0 +1,9 @@
## ¿Qué es AnP?
AnP, de Attach & Play en inglés, es un Framework basado en Gestores y su consumo sobre un sistema de Controladores.
Está desarrollado en Python para el servidor; y JavaScript para el entorno cliente Web.
Para crear un objecto AnP simplemente hemos de crearlo sin ningún argumento si queremos dejar todo por defecto o si queremos personalizar algo, podemos hacer uso del argumento `inputs`.
> Para saber qué puede contener el argumento `inputs` mirar la configuración.

31
Python/Abstracts/AIInterpretersAbstract.py
View File

@ -1,22 +1,33 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Any, Self, Sequence
from typing import Any, Self, Sequence, Optional, Callable
from abc import ABC, abstractmethod
from Interfaces.Application.AnPInterface import AnPInterface
from Models.AIResponseModel import AIResponseModel
from Utils.Checks import Check
class AIInterpretersAbstract:
class AIInterpretersAbstract(ABC):
def __init__(self:Self, anp:AnPInterface, inputs:str|dict[str, Any|None]|Sequence[Any|None]) -> None:
def __init__(self:Self, anp:AnPInterface, key:str, inputs:str|dict[str, Any|None]|Sequence[Any|None]) -> None:
if Check.is_string(inputs):
inputs = {"url" : inputs.strip()}
self.anp:AnPInterface = anp
self.key:str = key
self.url:str = self.anp.settings.get(("ai_interpreter_url", "ai_url", "url"), inputs, "")
self.stream:bool = self.anp.settings.get(("ai_interpreter_stream", "ai_stream", "stream"), inputs, False)
self.orders:list[str] = self.anp.settings.get(("ai_interpreter_orders", "ai_orders", "orders"), inputs, [])
self.maximum_tokens_per_session:int = self.anp.settings.get((
"ai_interpreter_maximum_tokens_per_session", "ai_maximum_tokens_per_session", "maximum_tokens_per_session"
), inputs, 2048)
), inputs, None)
self.maximum_response_tokens:int = self.anp.settings.get((
"ai_interpreter_maximum_response_tokens", "ai_maximum_response_tokens", "maximum_response_tokens"
), inputs, None)
self.format:Any|None = self.anp.settings.get(("ai_interpreter_format", "ai_format", "format"), inputs, None)
self.model:str = self.anp.settings.get(("ai_interpreter_model", "ai_model", "model"), inputs, "gemma3:1b")
self.pool:str = self.anp.settings.get(("ai_interpreter_pool", "ai_pool", "pool"), inputs, self.key)
self.sessions:dict[int, list[int]] = {}
self.sessions_i:int = 0
@ -35,8 +46,20 @@ class AIInterpretersAbstract:
return id, self.sessions[id]
def save_context(self:Self, id:int, context:list[int]) -> None:
if id in self.sessions:
self.sessions[id] = context
def close_session(self:Self, id:int) -> bool:
if id in self.sessions:
del self.sessions[id]
return True
return False
@abstractmethod
def request(self:Self,
session:int|None,
message:str,
orders:list[str] = [],
callback:Optional[Callable[[int, AIResponseModel], None]] = None
) -> tuple[int|None, AIResponseModel]:pass

22
Python/Application/AnP.py
View File

@ -10,13 +10,16 @@ from Managers.SettingsManager import SettingsManager
from Managers.PrintTypesManager import PrintTypesManager
from Managers.TerminalManager import TerminalManager
from Managers.ModelsManager import ModelsManager
from Managers.UniqueKeysManager import UniqueKeysManager
from Managers.SessionsManager import SessionsManager
from Managers.ControllersManager import ControllersManager
from Managers.DispatchesManager import DispatchesManager
from Managers.IndexesManager import IndexesManager
from Managers.RoutesManager import RoutesManager
from Managers.WebSocketServersManager import WebSocketServersManager
from Managers.HTTPServersManager import HTTPServersManager
from Drivers.HTTPDriver import HTTPDriver
from Managers.PseudoLoRAsManager import PseudoLoRAsManager
from Managers.AIInterpretersManager import AIInterpretersManager
from Utils.Common import Common
from Utils.Patterns import RE
@ -43,13 +46,16 @@ class AnP:
self.__own_update()
self.terminal:TerminalManager = TerminalManager(self)
self.models:ModelsManager = ModelsManager(self)
self.unique_keys:UniqueKeysManager = UniqueKeysManager(self)
self.sessions:SessionsManager = SessionsManager(self)
self.controllers:ControllersManager = ControllersManager(self)
self.dispatches:DispatchesManager = DispatchesManager(self)
self.indexes:IndexesManager = IndexesManager(self)
self.routes:RoutesManager = RoutesManager(self)
self.web_socket_servers:WebSocketServersManager = WebSocketServersManager(self)
self.http_servers:HTTPServersManager = HTTPServersManager(self)
# self.http_server:HTTPDriver = HTTPDriver(self)
self.pseudoloras:PseudoLoRAsManager = PseudoLoRAsManager(self)
self.ai_interpreters:AIInterpretersManager = AIInterpretersManager(self)
def update(self:Self) -> None:
self.settings.update()
@ -58,34 +64,38 @@ class AnP:
self.__own_update()
self.terminal.update()
self.models.update()
self.unique_keys.update()
self.sessions.update()
self.controllers.update()
self.dispatches.update()
self.indexes.update()
self.routes.update()
self.web_socket_servers.update()
self.http_servers.update()
# self.http_server.update()
self.pseudoloras.update()
self.ai_interpreters.update()
def reset(self:Self) -> None:
self.settings.reset()
self.i18n.reset()
self.print_types.reset()
self.__own_update()
self.terminal.reset()
self.models.reset()
self.unique_keys.reset()
self.sessions.reset()
self.controllers.reset()
self.dispatches.reset()
self.indexes.reset()
self.routes.reset()
self.web_socket_servers.reset()
self.http_servers.reset()
# self.http_server.reset()
self.ai_interpreters.reset()
def close(self:Self) -> None:
self.__working = False
self.ai_interpreters.close()
self.web_socket_servers.close()
self.http_servers.close()
# self.http_server.close()
def working(self:Self) -> bool:
return self.__working

23
Python/Controllers/AIController.py
View File

@ -1,15 +1,26 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Self
from typing import Self, Any, Callable
from Abstracts.ModelAbstract import ModelAbstract
from Abstracts.ControllerAbstract import ControllerAbstract
from Interfaces.Application.AnPInterface import AnPInterface
from Models.RequestModel import RequestModel
class AIController(ModelAbstract):
class AIController(ControllerAbstract, ModelAbstract):
def __init__(self:Self, anp:AnPInterface) -> None:
self.anp: AnPInterface = anp
# def __init__(self:Self, anp:AnPInterface) -> None:
# self.anp: AnPInterface = anp
def new_message(self:Self, request:RequestModel) -> None:
pass
# def __temp(self:Self)
# def __get_data(self:Self, request:RequestModel, callback:Callable[..., Any|None]) -> None:
# self.anp.pseudoloras.get("anp_titles")
def test(self:Self, request:RequestModel) -> None:
self.anp.ai_interpreters.request("anp_titles", None, request.get("message", "Hola"), lambda id, response: print((id, response.response)))
request.set_response({
"ok" : True,
"code" : 200,
"message" : "ok"
})

84
Python/Drivers/OllamaDriver.py
View File

@ -1,14 +1,90 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Any, Optional, Self, Sequence
from typing import Any, Optional, Self, Sequence, Callable
from requests import post as Post, Response
# from pydantic import BaseModel
from Interfaces.Application.AnPInterface import AnPInterface
from requests import post as Post
from Abstracts.AIInterpretersAbstract import AIInterpretersAbstract
from Abstracts.ModelAbstract import ModelAbstract
from Models.AIResponseModel import AIResponseModel
from Utils.Checks import Check
from Utils.Common import Common
class OllamaDriver:
class OllamaDriver(AIInterpretersAbstract, ModelAbstract):
def __init__(self:Self,
anp:AnPInterface,
key:str,
inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None
) -> None:
self.anp:AnPInterface = anp
super().__init__(anp, key, inputs)
def request(self:Self,
session:int|None,
message:str,
orders:list[str] = [],
callback:Optional[Callable[[int, AIResponseModel], None]] = None
) -> tuple[int|None, AIResponseModel]:
response:Response
context:list[int]
options:dict[str, Any] = {}
results:AIResponseModel = AIResponseModel()
if self.maximum_response_tokens is not None:
options["num_predict"] = self.maximum_response_tokens
if self.maximum_tokens_per_session is not None:
options["num_ctx"] = self.maximum_tokens_per_session
orders += self.orders
session, context = self.get_session(session)
with Post(self.url, json = {
"model" : self.model,
"prompt": message,
**({"system": "\\n".join(
str(i + 1) + ". " + order for i, order in enumerate(orders, 1)
)} if len(orders) else {}),
"stream": self.stream,
**(
{"format" : self.format} if (
self.format == "json" or
# Check.is_array(self.format) or
Check.is_dictionary(self.format)
) else
# {"format" : self.format.model_json_schema()} if isinstance(self.format, BaseModel) else
{}),
**({"context" : context} if len(context) else {}),
**({"options" : options} if len(options) else {})
}, stream = self.stream) as response:
results.http_code = response.status_code
if results.http_code == 200:
results.ok = True
try:
chunk:bytes
for chunk in response.iter_lines():
if chunk:
print(Common.json_decode(chunk))
results.update(Common.json_decode(chunk))
if results.done:
self.save_context(session, results.context)
break
Common.execute(callback, session, results)
except Exception as exception:
self.anp.exception(exception, "anp_ollama_driver_request")
results.ok = False
else:
try:
results.http_message = response.json().get("error", "unknown_error")
except Exception as _:
results.http_message = "unknown_error"
Common.execute(callback, session, results)
return session, results

8
Python/Interfaces/Application/AnPInterface.py
View File

@ -8,12 +8,16 @@ from Interfaces.Managers.I18NManagerInterface import I18NManagerInterface
from Interfaces.Managers.PrintTypesManagerInterface import PrintTypesManagerInterface
from Interfaces.Managers.TerminalManagerInterface import TerminalManagerInterface
from Interfaces.Managers.ModelsManagerInterface import ModelsManagerInterface
from Interfaces.Managers.UniqueKeysManagerInterface import UniqueKeysManagerInterface
from Interfaces.Managers.SessionsManagerInterface import SessionsManagerInterface
from Interfaces.Managers.ControllersManagerInterface import ControllersManagerInterface
from Interfaces.Managers.DispatchesManagerInterface import DispatchesManagerInterface
from Interfaces.Managers.IndexesManagerInterface import IndexesManagerInterface
from Interfaces.Managers.RoutesManagerInterface import RoutesManagerInterface
from Interfaces.Managers.WebSocketServersManagerInterface import WebSocketServersManagerInterface
from Interfaces.Managers.HTTPServersManagerInterface import HTTPServersManagerInterface
from Interfaces.Managers.PseudoLoRAsManagerInterface import PseudoLoRAsManagerInterface
from Interfaces.Managers.AIInterpretersManagerInterface import AIInterpretersManagerInterface
class AnPInterface(ABC):
@ -23,12 +27,16 @@ class AnPInterface(ABC):
self.print_types:PrintTypesManagerInterface = None
self.terminal:TerminalManagerInterface = None
self.models:ModelsManagerInterface = None
self.unique_keys:UniqueKeysManagerInterface = None
self.sessions:SessionsManagerInterface = None
self.controllers:ControllersManagerInterface = None
self.dispatches:DispatchesManagerInterface = None
self.indexes:IndexesManagerInterface = None
self.routes:RoutesManagerInterface = None
self.web_socket_servers:WebSocketServersManagerInterface = None
self.http_servers:HTTPServersManagerInterface = None
self.pseudoloras:PseudoLoRAsManagerInterface = None
self.ai_interpreters:AIInterpretersManagerInterface = None
@abstractmethod
def update(self:Self) -> None:pass

29
Python/Interfaces/Managers/AIInterpretersManagerInterface.py Normal file
View File

@ -0,0 +1,29 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Any, Callable, Self, Optional
from abc import ABC, abstractmethod
from Models.AIResponseModel import AIResponseModel
class AIInterpretersManagerInterface(ABC):
@abstractmethod
def update(self:Self) -> None:pass
@abstractmethod
def reset(self:Self) -> None:pass
@abstractmethod
def close(self:Self) -> None:pass
@abstractmethod
def add(self:Self, inputs:Any|None, overwrite:bool = False) -> None:pass
@abstractmethod
def remove(self:Self, keys:Optional[str|list[str]] = None) -> None:pass
@abstractmethod
def request(self:Self, key:str, session:int|None, message:str, callback:Callable[[int, AIResponseModel], None]) -> int|None:pass
@abstractmethod
def cancel_request(self:Self, key:str, id:int) -> None:pass

20
Python/Interfaces/Managers/PseudoLoRAsManagerInterface.py Normal file
View File

@ -0,0 +1,20 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Self, Any, Callable
from abc import ABC, abstractmethod
from Models.PseudoLoRAModel import PseudoLoRAModel
class PseudoLoRAsManagerInterface(ABC):
@abstractmethod
def update(self:Self) -> None:pass
@abstractmethod
def add(self:Self, inputs:Any|None) -> None:pass
@abstractmethod
def clean_cache(self:Self) -> None:pass
@abstractmethod
def get(self:Self, callback:Callable[[list[PseudoLoRAModel]], bool]) -> list[tuple[str, str]]:pass

20
Python/Interfaces/Managers/SessionsManagerInterface.py Normal file
View File

@ -0,0 +1,20 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Self, Optional
from abc import ABC, abstractmethod
from Models.SessionModel import SessionModel
class SessionsManagerInterface(ABC):
@abstractmethod
def update(self:Self) -> None:pass
@abstractmethod
def reset(self:Self) -> None:pass
@abstractmethod
def get(self:Self, id:Optional[str] = None) -> SessionModel|None:pass
@abstractmethod
def remove(self:Self, id:str) -> bool:pass

19
Python/Interfaces/Managers/UniqueKeysManagerInterface.py Normal file
View File

@ -0,0 +1,19 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Self
from abc import ABC, abstractmethod
class UniqueKeysManagerInterface(ABC):
@abstractmethod
def update(self:Self) -> None:pass
@abstractmethod
def reset(self:Self) -> None:pass
@abstractmethod
def get(self:Self) -> str:pass
@abstractmethod
def remove(self:Self, key:str) -> None:pass

87
Python/Managers/AIInterpretersManager.py
View File

@ -1,25 +1,40 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Any, Self, Optional
from typing import Any, Callable, Self, Optional
from Interfaces.Application.AnPInterface import AnPInterface
from Abstracts.AIInterpretersAbstract import AIInterpretersAbstract
from Models.AIResponseModel import AIResponseModel
from Models.AIPoolRequestsModel import AIPoolRequestsModel
from Utils.Common import Common
class AIInterpretersManager:
def __init__(self:Self, anp:AnPInterface) -> None:
self.anp:AnPInterface = anp
self.__interpreters:dict[str, Any] = {}
self.__interpreters:dict[str, AIInterpretersAbstract] = {}
self.__pool_requests:dict[str, AIPoolRequestsModel] = {}
self.update()
def update(self:Self) -> None:
pass
key:str
for key in ("default_ai_interpreters_files", "ai_interpreters_files", "default_ai_interpreters", "ai_interpreters"):
self.add(self.anp.settings.get(key), True)
def reset(self:Self) -> None:
self.__interpreters = {}
self.update()
def close(self:Self) -> None:
self.__interpreters = {}
key:str
for key in list(self.__interpreters.keys()):
self.remove(key)
def add(self:Self, inputs:Any|None, overwrite:bool = False) -> None:
@ -32,12 +47,64 @@ class AIInterpretersManager:
for key, value in subinputs.items():
if overwrite or key not in self.__interpreters:
self.__interpreters[key] = value
def remove(self:Self, names:Optional[str|list[str]] = None) -> None:
interpreter:AIInterpretersAbstract|None = None
name:str
if isinstance(value, AIInterpretersAbstract):
interpreter = value
elif isinstance(value, dict) and "type" in value:
for name in Common.get_keys(names) if names else list(self.__interpreters.keys()):
if name in self.__interpreters:
del self.__interpreters[name]
Model:AIInterpretersAbstract|None = self.anp.models.get(AIInterpretersAbstract, value["type"])
if Model is not None:
interpreter = Model(self.anp, key, value)
if interpreter is not None:
if key in self.__interpreters:
self.remove(key)
self.__interpreters[key] = interpreter
self.__pool_requests[interpreter.pool] = AIPoolRequestsModel()
def remove(self:Self, keys:Optional[str|list[str]] = None) -> None:
key:str
for key in Common.get_keys(keys) if keys else list(self.__interpreters.keys()):
if key in self.__interpreters:
pool:str = self.__interpreters[key].pool
has_interpreters:bool = False
interpreter:AIInterpretersAbstract
self.__pool_requests.get(pool).remove(self.__interpreters[key].key)
self.__interpreters[key].close()
del self.__interpreters[key]
for interpreter in self.__interpreters.values():
if interpreter.pool == pool:
has_interpreters = True
break
if not has_interpreters and pool in self.__pool_requests:
del self.__pool_requests[pool]
def request(self:Self, key:str, session:int|None, message:str, callback:Callable[[int, AIResponseModel], None]) -> int|None:
i:int|None = None
if key in self.__interpreters:
pool:str = self.__interpreters[key].pool
i = self.__pool_requests[pool].add(self.__interpreters[key], session, message, callback)
self.__pool_requests[pool].execute()
return i
def cancel_request(self:Self, key:str, id:int) -> None:
if key in self.__interpreters:
pool:str = self.__interpreters[key].pool
if id in self.__pool_requests[pool].pool:
del self.__pool_requests[pool].pool[id]

1
Python/Managers/ControllersManager.py
View File

@ -58,6 +58,7 @@ class ControllersManager:
None)
def execute(self:Self, key:str, method:str, request:RequestModel) -> bool:
print([self.__controllers, key, method])
if key in self.__controllers and hasattr(self.__controllers[key], method):
getattr(self.__controllers[key], method)(request)
return True

126
Python/Managers/PseudoLoRAsManager.py
View File

@ -1,7 +1,7 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Self, Any, Sequence
from typing import Self, Any, Sequence, Callable
from Interfaces.Application.AnPInterface import AnPInterface
from Models.PseudoLoRAModel import PseudoLoRAModel
from Utils.Common import Common
@ -10,10 +10,37 @@ from Utils.Checks import Check
class PseudoLoRAsManager:
def __init__(self:Self, anp:AnPInterface) -> None:
self.anp:AnPInterface = anp
self.__maximum_cache:int = (1 << 20) * 100
self.__memory_cached:int = 0
self.__cache:dict[int, PseudoLoRAModel] = {}
self.__loras:list[PseudoLoRAModel] = []
self.update()
def update(self:Self) -> None:
key:str
self.clean_cache()
self.__loras.clear()
self.__maximum_cache = self.anp.settings.get((
"pseudoloras_maximum_cache_size", "maximum_cache_size"
), None, self.__maximum_cache)
for key in ("default_pseudoloras_files", "pseudoloras_files", "default_pseudoloras", "pseudoloras"):
self.add(self.anp.settings.get(key))
def __try_load_file(self:Self, path:str) -> bool:
data:list[dict[str, Any|None]|Sequence[Any|None]] = Common.load_json(path)
if len(data):
self.add(data)
return True
return False
def add(self:Self, inputs:Any|None) -> None:
if isinstance(inputs, PseudoLoRAModel):
self.__loras.append(inputs)
@ -22,5 +49,98 @@ class PseudoLoRAsManager:
subinputs:dict[str, Any|None]|Sequence[Any|None]
for subinputs in Common.load_json(inputs, False):
if Check.is_array(inputs):
pass
if Check.is_string(subinputs):
self.__try_load_file(subinputs)
elif Check.is_array(subinputs):
if all(Check.is_string(item) for item in subinputs):
item:str
ok:bool = True
for item in subinputs:
if not self.__try_load_file(item):
ok = False
break
if not ok:
self.__loras.append(PseudoLoRAModel(*subinputs))
elif (
len(subinputs) >= 2 and len(subinputs) <= 4 and
Check.is_string(subinputs[0]) and Check.is_string(subinputs[1]) and
(len(subinputs) < 3 or subinputs[2] is None or (Check.is_array(subinputs[2]) and all(Check.is_string(item) for item in subinputs[2]))) and
(len(subinputs) < 4 or subinputs[3] is None or Check.is_boolean(subinputs[3]))
):
self.__loras.append(PseudoLoRAModel(*subinputs))
else:
item:Any|None
for item in subinputs:
self.add(item)
def clean_cache(self:Self) -> None:
lora:PseudoLoRAModel
self.__memory_cached = 0
self.__cache.clear()
for lora in self.__loras:
lora.clean_cache()
def get(self:Self, callback:Callable[[list[PseudoLoRAModel]], bool], keys:list[str] = []) -> list[tuple[str, str]]:
next:list[PseudoLoRAModel] = []
results:list[tuple[str, str]] = []
i:int
ok:bool
has_keys:bool = len(keys) > 0
for i, ok in enumerate(callback(
lora.title for lora in self.__loras if not has_keys or any(
key in lora.keys for key in keys
)
)):
if not ok:
continue
lora:PseudoLoRAModel = self.__loras[i]
if lora.path is not None:
data:str|None
if lora.cacheable:
if not lora.cache:
lora.cache = Common.load_file(lora.path, "r")
lora.memory = len(lora.cache)
self.__memory_cached += lora.memory
if self.__cache[lora.i]:
del self.__cache[lora.i]
lora.i += 1
if lora.i in self.__cache:
self.__cache[lora.i].append(lora)
else:
self.__cache[lora.i] = [lora]
if self.__memory_cached > self.__maximum_cache:
i:int = min(self.__cache.keys())
for lora in self.__cache[i]:
lora.clean_cache()
del self.__cache[i]
if (data := lora.cache or Common.load_file(lora.path, "r")):
results.append((lora.title, lora.path))
else:
next.extend(lora.nested)
if len(next):
results.extend(callback(next))
return results

5
Python/Managers/RoutesManager.py
View File

@ -57,9 +57,8 @@ class RoutesManager:
"code" : 404,
"message" : "not_found"
})
elif route.controller and route.controller_method:
if not self.anp.controllers.execute(route.controller, route.controller_method, request):
request.set_response({
elif route.controller and route.action:
self.anp.controllers.execute(route.controller, route.action, request) or request.set_response({
"ok" : False,
"code" : 505,
"message" : "not_implemented"

39
Python/Managers/SessionsManager.py Normal file
View File

@ -0,0 +1,39 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Self, Optional
from Interfaces.Application.AnPInterface import AnPInterface
from Models.SessionModel import SessionModel
class SessionsManager:
def __init__(self:Self, anp:AnPInterface) -> None:
self.anp:AnPInterface = anp
self.__sessions:dict[str, SessionModel] = {}
self.update()
def update(self:Self) -> None:pass
def reset(self:Self) -> None:
self.__sessions = {}
self.update()
def get(self:Self, id:Optional[str] = None) -> SessionModel|None:
if id is None:
session:SessionModel = SessionModel(self.anp.unique_keys.get())
self.__sessions[session.id] = session
return session
return self.__sessions.get(id, None)
def remove(self:Self, id:str) -> bool:
if id in self.__sessions:
del self.__sessions[id]
return True
return False

44
Python/Managers/UniqueKeysManager.py Normal file
View File

@ -0,0 +1,44 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Self, Any
from Interfaces.Application.AnPInterface import AnPInterface
from Utils.Common import Common
class UniqueKeysManager:
def __init__(self:Self, anp:AnPInterface) -> None:
self.anp:AnPInterface = anp
self.__keys:list[str] = []
self.__alphabet:list[str] = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
self.__length:int = 13
self.update()
def update(self:Self) -> None:
self.__alphabet = Common.unique(self.anp.settings.get(("unique_keys_alphabet", "alphabet"), None, self.__alphabet))
self.__length = self.anp.settings.get(("unique_keys_length", "length"), None, self.__length)
def reset(self:Self) -> None:
self.__keys = []
self.__alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
self.__length = 13
self.update()
def get(self:Self) -> str:
key:str
while True:
key = ""
while len(key) < self.__length:
key += Common.random(self.__alphabet)
if key[0] not in "0123456789" and key not in self.__keys:
self.__keys.append(key)
return key
def remove(self:Self, key:str) -> None:
if key in self.__keys:
self.__keys.remove(key)

92
Python/Models/AIPoolRequestsModel.py Normal file
View File

@ -0,0 +1,92 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Callable, Self
from threading import Thread, Lock
from Models.AIResponseModel import AIResponseModel
from Abstracts.AIInterpretersAbstract import AIInterpretersAbstract
from Utils.Common import Common
class AIPoolRequestsItemsModel:
def __init__(self:Self,
key:str,
interpreter:AIInterpretersAbstract,
session:int|None,
message:str,
callback:Callable[[int, AIResponseModel], None],
orders:list[str] = []
) -> None:
self.key:str = key
self.interpreter:AIInterpretersAbstract = interpreter
self.session:int|None = session
self.message:str = message
self.callback:Callable[[int, AIResponseModel], None] = callback
self.orders:list[str] = orders
class AIPoolRequestsModel:
def __init__(self:Self) -> None:
self.pool:dict[int, AIPoolRequestsItemsModel] = {}
self.iterations:int = 0
self.maximum_iterations:int = 1
self.i:int = 0
self.j:int = 0
self.__lock:Lock = Lock()
def add(self:Self,
interpreter:AIInterpretersAbstract,
session:int|None,
message:str,
callback:Callable[[int, AIResponseModel], None],
orders:list[str] = []
) -> int:
id:int
with self.__lock:
self.i += 1
id = self.i
self.pool[self.i] = AIPoolRequestsItemsModel(
interpreter.key, interpreter, session, message, callback, orders
)
self.execute()
return id
def __next(self:Self, callback:Callable[[int, AIResponseModel], None], session:int, response:AIResponseModel) -> None:
Common.execute(callback, session, response)
if response.done or not response.ok:
with self.__lock:
self.iterations -= 1
self.execute()
def __execute(self:Self) -> None:
item:AIPoolRequestsItemsModel|None = None
with self.__lock:
if len(self.pool) and self.iterations != self.maximum_iterations and self.i != self.j:
self.iterations += 1
self.j = min(self.pool.keys())
item:AIPoolRequestsItemsModel = self.pool[self.j]
del self.pool[self.j]
item and item.interpreter.request(
item.session, item.message, item.orders, lambda session, response: self.__next(item.callback, session, response)
)
def execute(self:Self) -> None:
Thread(target = self.__execute).start()
def cancel(self:Self, ids:int|list[int]) -> None:
with self.__lock:
for id in Common.get_keys(ids):
if id in self.pool:
del self.pool[id]
def remove(self:Self, key:str) -> None:
with self.__lock:
for id, item in list(self.pool.items()):
if item.key == key:
del self.pool[id]

32
Python/Models/AIResponseModel.py Normal file
View File

@ -0,0 +1,32 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Any, Self
from time import time as timestamp
class AIResponseModel:
def __init__(self:Self) -> None:
self.start:float = timestamp()
self.model:str = ""
self.response:str = ""
self.total:str = ""
self.context:str|list[str] = ""
self.done:bool = False
self.end:float|None = None
self.chunks:int = 0
self.ok:bool = False
self.http_code:int = 0
self.http_message:str = ""
def update(self:Self, data:dict[str, Any|None]) -> None:
self.model = data.get("model", self.model)
self.response = data.get("response", "")
self.total += self.response
self.done = data.get("done", self.done)
self.chunks += 1
if self.done:
self.end = timestamp()
self.context = data.get("context", self.context)

8
Python/Models/PseudoLoRAModel.py
View File

@ -6,9 +6,10 @@ from Utils.Checks import Check
from Utils.Common import Common
class PseudoLoRAModel:
def __init__(self:Self,
title:str,
content:str|dict[str, str|Sequence[str, str|Sequence[Any], Optional[str|Sequence[str]], bool]],
content:str|Sequence[Any|None],
keys:Optional[str|Sequence[str]] = None,
cacheable:bool = False
) -> None:
@ -22,3 +23,8 @@ class PseudoLoRAModel:
self.nested:list[PseudoLoRAModel] = [
PseudoLoRAModel(subtitle, subcontent, subkeys, subcacheable) for subtitle, subcontent, subkeys, subcacheable in content
] if Check.is_array(content) else []
def clean_cache(self:Self) -> None:
self.cache = ""
self.memory = 0
self.i = 0

10
Python/Models/SessionModel.py Normal file
View File

@ -0,0 +1,10 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Self, Any
class SessionModel:
def __init__(self:Self, id:str) -> None:
self.id:str = id
self.variables:dict[str, Any|None] = {}

4
Python/Utils/Checks.py
View File

@ -23,6 +23,10 @@ class Check:
def is_dictionary(item:Any|None) -> bool:
return isinstance(item, dict)
@staticmethod
def is_boolean(item:Any|None) -> bool:
return isinstance(item, bool)
@staticmethod
def is_function(item:Any|None) -> bool:
return callable(item)

26
Python/Utils/Common.py
View File

@ -1,7 +1,7 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Any, Optional, Sequence, Self
from typing import Any, Callable, Optional, Sequence, Self
from re import Match as REMatch
from os.path import abspath as absolute_path, dirname as directory_name, exists as path_exists, isfile as is_file
from json import loads as json_decode
@ -10,6 +10,7 @@ from mimetypes import guess_type as get_mime_by_extension
from inspect import FrameInfo, stack as get_stack
from json import dumps as json_encode, loads as json_decode
from base64 import b64encode as base64_encode, b64decode as base64_decode
from random import choice as random_choice
from Utils.Checks import Check
from Utils.Patterns import RE
@ -326,3 +327,26 @@ class Common:
except Exception as exception:
return data
return None
@staticmethod
def execute(callback:Callable[..., Any|None], *arguments:Any|None) -> Any|None:
if Check.is_function(callback):
try:
return callback(*arguments)
except Exception as _:
pass
return None
@staticmethod
def unique(items:str|Sequence[Any|None]) -> str|list[Any|None]:
if Check.is_string(items):
return "".join(character for i, character in enumerate(items) if character not in items[:i])
elif Check.is_array(items):
return [item for i, item in enumerate(items) if item not in items[:i]]
return items
@staticmethod
def random(items:str|Sequence[Any|None]) -> Any|None:
return random_choice(items) if (
Check.is_string(items) or Check.is_array(items)
) and len(items) else None

4
Python/run.py
View File

@ -6,12 +6,14 @@ from Application.AnP import AnP
from Controllers.AIController import AIController
from Drivers.WebSocketServerDriver import WebSocketServerDriver
from Drivers.HTTPDriver import HTTPDriver
from Drivers.OllamaDriver import OllamaDriver
inputs:dict[str, dict[str, Any|None]] = {
"default_models" : {
"AIController" : AIController,
"WebSocketServerDriver" : WebSocketServerDriver,
"HTTPDriver" : HTTPDriver
"HTTPDriver" : HTTPDriver,
"OllamaDriver" : OllamaDriver
}
}