Spaces:
Sleeping
Sleeping
from abc import ABC, abstractmethod | |
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage | |
from system_prompt_template import DIALOG_PROMPT | |
from typing import List | |
from langchain_core.tools import BaseTool, tool | |
from service_by_name import get_service_by_name | |
import logging | |
import time | |
import os | |
from dotenv import load_dotenv | |
load_dotenv() | |
logger = logging.getLogger(__name__) | |
class BaseService(ABC): | |
def __init__(self): | |
self._base_tools = [self.closing_task, | |
self.make_jira_task, | |
] | |
def system_prompt(self) -> str: | |
"""Возвращает системный промпт для сервиса""" | |
pass | |
def tools(self) -> List[BaseTool]: | |
"""Возвращает список инструментов сервиса""" | |
pass | |
def log_path(self) -> str: | |
"""Возвращает путь до папки с логами""" | |
return os.path.join("/tmp", "service.log") | |
def get_initial_messages(self): | |
"""Возвращает начальные сообщения для графа""" | |
return [SystemMessage(content=self.system_prompt)] | |
def get_messages_from_redirect(self, messages): | |
messages_formatted = "" | |
for message in messages: | |
if isinstance(message, HumanMessage): | |
message += f"user: '{message.content}'\n" | |
elif isinstance(message, AIMessage): | |
message += f"assistant: '{message.content}'\n" | |
return [SystemMessage(content=self.system_prompt + DIALOG_PROMPT.format(messages_formatted))] | |
def make_redirect(service_name: str): | |
"""Когда оказывается, что пользователю нужен эксперт в другом сервисе, | |
пользователя нужно перенаправить на соответствующего эксперта. | |
Функция принимает аргумент service_name из списка: | |
['openvpn', 'corporate_email', 'internet_access'] и пересоздает класс эксперта | |
с соответствующими новому сервису знаниями.""" | |
logger.info("redirecting question to service {service_name}") | |
return get_service_by_name(service_name) | |
def closing_task(question: str) -> None: | |
"""Задача выполнена, сохраняем результаты""" | |
print("### Задача выполнена, сохраняем результаты") | |
time.sleep(2) | |
logger.info("task closed") | |
def make_jira_task(question: str) -> None: | |
"""Задача не может быть выполнена с помощью ии, создаем задачу в jira для того, | |
чтобы её выполнил человек""" | |
time.sleep(2) | |
print("### Скриптом создается задача в jira с соответствующими деталями") | |
logger.info("jira task was make") | |