104 lines
4.5 KiB
Python
104 lines
4.5 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
|
|
self.think:bool = self.anp.settings.get(("ai_interpreter_think", "ai_think", "think"), inputs, False)
|
|
self.allow_contexts:bool = self.anp.settings.get(("ai_interpreter_allow_contexts", "ai_allow_contexts", "allow_contexts"), inputs, True)
|
|
self.temperature:float = self.anp.settings.get(("ai_interpreter_temperature", "ai_temperature", "temperature"), inputs, 0.7)
|
|
|
|
def start(self:Self) -> None:
|
|
pass
|
|
|
|
def close(self:Self) -> None:
|
|
self.sessions = {}
|
|
|
|
def get_session(self:Self, id:int|None = None, conversation:str|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][conversation] if conversation in self.sessions[id] else []
|
|
|
|
def save_context(self:Self, id:int, conversation:str, context:list[int]) -> None:
|
|
if self.allow_contexts and id in self.sessions:
|
|
if conversation not in self.sessions[id]:
|
|
self.sessions[id][conversation] = context
|
|
self.sessions[id][conversation] = context
|
|
|
|
def close_conversation(self:Self, id:int, conversation:str) -> bool:
|
|
if id in self.sessions and conversation in self.sessions[id]:
|
|
del self.sessions[id][conversation]
|
|
return True
|
|
return False
|
|
|
|
def close_session(self:Self, id:int) -> bool:
|
|
if id in self.sessions:
|
|
del self.sessions[id]
|
|
return True
|
|
return False
|
|
|
|
def get_context(self:Self, id:int) -> list[int]:
|
|
return self.sessions[id] if id in self.sessions else []
|
|
|
|
def get_orders(self:Self, orders:Optional[str|list[str]] = None) -> str:
|
|
return (
|
|
orders if Check.is_string(orders) else
|
|
"\n".join(orders) if Check.is_array(orders) else
|
|
"")
|
|
|
|
# results:str = ""
|
|
# i:int = 0
|
|
|
|
# for block in (self.orders, 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()
|
|
|
|
# return results
|
|
|
|
@abstractmethod
|
|
def request(self:Self,
|
|
session:int|None,
|
|
conversation:str|None,
|
|
message:str,
|
|
orders:str|list[str] = [],
|
|
callback:Optional[Callable[[int, AIResponseModel], None]] = None,
|
|
custom_context:list[int] = []
|
|
) -> tuple[int|None, AIResponseModel]:pass |