140 lines
5.1 KiB
Python
140 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from typing import Self, Any, Callable
|
|
from Abstracts.ModelAbstract import ModelAbstract
|
|
from Abstracts.ControllerAbstract import ControllerAbstract
|
|
from Models.RequestModel import RequestModel
|
|
from Models.PseudoLoRAModel import PseudoLoRAModel
|
|
|
|
class AIController(ControllerAbstract, ModelAbstract):
|
|
|
|
# def __message_execution(self:Self, end:Callable[[], None], request:RequestModel) -> None:
|
|
|
|
# session:str|None = request.session.get("anp_ai_session")
|
|
# message_id:str|None = request.get("message_id")
|
|
# client_id:str|None = request.get("client_id")
|
|
|
|
# session, _ = self.anp.ai_interpreters.request(
|
|
# "anp_responses",
|
|
# session,
|
|
# request.get("conversation"),
|
|
# request.get("message"),
|
|
# [],
|
|
# lambda id, response: self.anp.web_socket_servers.send("anp", "ai", "message", {
|
|
# "id" : id,
|
|
# "conversation" : response.conversation,
|
|
# "response" : response.response,
|
|
# "ok" : response.ok,
|
|
# "done" : response.done,
|
|
# "data_id" : message_id
|
|
# }, client_id)
|
|
# )
|
|
|
|
# if session is not None:
|
|
# request.session.set("anp_ai_session", session)
|
|
|
|
# end()
|
|
|
|
def __analyse_loras(self:Self, data:dict[str, Any], request:RequestModel, loras:list[str]) -> list[tuple[str, str]]:
|
|
|
|
session:str|None = request.session.get("anp_loras_session")
|
|
results:list[tuple[str, str]] = []
|
|
|
|
print(self.anp.i18n.get("ai_controller_loras_system", {
|
|
"items" : len(loras),
|
|
"list" : "".join("\n" + str(i) + ". " + title for i, title in enumerate(loras))
|
|
}))
|
|
|
|
# session, results = self.anp.ai_interpreters.request(
|
|
# "anp_loras",
|
|
# session,
|
|
# data["conversation"],
|
|
# data["message"],
|
|
# self.anp.i18n.get("ai_controller_loras_system", {
|
|
# "items" : len(loras),
|
|
# "list" : "".join("\n" + str(i) + ". " + title for i, title in enumerate(loras))
|
|
# })
|
|
# )
|
|
data["i"] += 1
|
|
|
|
# if session is not None or data["i"] == 1:
|
|
# request.session.set("anp_loras_session", session)
|
|
|
|
return results
|
|
|
|
def __message_execution(self:Self, end:Callable[[], None], request:RequestModel) -> None:
|
|
|
|
session:str|None = request.session.get("anp_ai_session")
|
|
message_id:str|None = request.get("message_id")
|
|
client_id:str|None = request.get("client_id")
|
|
conversation:str = request.get("conversation")
|
|
message:str = request.get("message")
|
|
loras:list[tuple[str, str]]
|
|
loras_data:dict[str, Any] = {
|
|
"conversation" : conversation,
|
|
"message" : message,
|
|
"i" : 0
|
|
}
|
|
|
|
loras = self.anp.pseudoloras.get(lambda loras:self.__analyse_loras(loras_data, request, loras))
|
|
|
|
print([title for title, _ in loras])
|
|
|
|
# session, _ = self.anp.ai_interpreters.request(
|
|
# "anp_responses",
|
|
# session,
|
|
# conversation,
|
|
# message,
|
|
# ("Usa las siguientes guías para responder y dar soporte.\n\n" + "".join(
|
|
# "# " + title + "\n\n" + text + "\n\n" for title, text in loras
|
|
# ) if len(loras) else []),
|
|
# lambda id, response: self.anp.web_socket_servers.send("anp", "ai", "message", {
|
|
# "id" : id,
|
|
# "conversation" : response.conversation,
|
|
# "response" : response.response,
|
|
# "ok" : response.ok,
|
|
# "done" : response.done,
|
|
# "data_id" : message_id
|
|
# }, client_id)
|
|
# )
|
|
|
|
# if session is not None:
|
|
# request.session.set("anp_ai_session", session)
|
|
|
|
end()
|
|
|
|
# def __message_by_themes_execution(self:Self, end:Callable[[], None], request:RequestModel) -> None:
|
|
|
|
# session:str|None = request.session.get("anp_ai_session")
|
|
# message_id:str|None = request.get("message_id")
|
|
# client_id:str|None = request.get("client_id")
|
|
|
|
# session, _ = self.anp.ai_interpreters.request(
|
|
# "anp_responses",
|
|
# session,
|
|
# request.get("conversation"),
|
|
# request.get("message"),
|
|
# [],
|
|
# lambda id, response: self.anp.web_socket_servers.send("anp", "ai", "message", {
|
|
# "id" : id,
|
|
# "conversation" : response.conversation,
|
|
# "response" : response.response,
|
|
# "ok" : response.ok,
|
|
# "done" : response.done,
|
|
# "data_id" : message_id
|
|
# }, client_id)
|
|
# )
|
|
|
|
# if session is not None:
|
|
# request.session.set("anp_ai_session", session)
|
|
|
|
# end()
|
|
|
|
def message(self:Self, request:RequestModel) -> None:
|
|
self.anp.queues.add("anp", self.__message_execution, request)
|
|
request.set_response({
|
|
"ok" : True,
|
|
"code" : 200,
|
|
"message" : "ok"
|
|
}) |