testing-mvp / src /base_service.py
Boris
log path
b719109
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")