Spaces:
Sleeping
Sleeping
File size: 3,229 Bytes
1094f14 96f6b21 1094f14 b719109 1094f14 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 |
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")
|