82 lines
3.2 KiB
Python
82 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
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(ABC):
|
|
|
|
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:str|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, 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
|
|
|
|
def start(self:Self) -> None:
|
|
pass
|
|
|
|
def close(self:Self) -> None:
|
|
self.sessions = {}
|
|
|
|
def get_session(self:Self, id:int|None = None) -> tuple[int, list[int]]:
|
|
|
|
if id is None or id not in self.sessions:
|
|
id = self.sessions_i
|
|
self.sessions_i += 1
|
|
self.sessions[id] = []
|
|
|
|
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
|
|
|
|
def get_orders(self:Self, orders:str|list[str]) -> str:
|
|
|
|
results:str = ""
|
|
i:int = 0
|
|
|
|
for block in (orders, self.orders):
|
|
if block:
|
|
if Check.is_array(block):
|
|
|
|
block_string:str = "".join(str(i + 1 + j) + ". " + str(order) + "\n" for j, order in enumerate(block))
|
|
|
|
i += len(block)
|
|
results += ("\n\n" if results else "") + block_string.strip()
|
|
|
|
else:
|
|
results += ("\n\n" if results else "") + str(block).strip()
|
|
|
|
@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 |