src/base_service.py

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,
                            ]
 
    @property
    @abstractmethod
    def system_prompt(self) -> str:
        """Возвращает системный промпт для сервиса"""
        pass
    
    @property
    @abstractmethod
    def tools(self) -> List[BaseTool]:
        """Возвращает список инструментов сервиса"""
        pass

    @property
    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))]
    
    @tool
    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)

    
    @tool
    def closing_task(question: str) -> None:
        """Задача выполнена, сохраняем результаты"""
        print("### Задача выполнена, сохраняем результаты")
        time.sleep(2)
        logger.info("task closed")

    @tool
    def make_jira_task(question: str) -> None:
        """Задача не может быть выполнена с помощью ии, создаем задачу в jira для того,
        чтобы её выполнил человек"""
        time.sleep(2)
        print("### Скриптом создается задача в jira с соответствующими деталями")
        logger.info("jira task was make")