cyclop/base_telegram_inference_bot.py

import importlib
import os
import json
import inspect
from abc import ABC, abstractmethod
from tools.base_tool import BaseTool

class BaseTelegramInferenceBot(ABC):
    def __init__(self):
        self.conversation_history = {}
        self.processing_status = {}
        self.system_prompt = self.load_system_prompt()
        self.tools, self.functions = self.load_functions()

    @staticmethod
    def load_system_prompt():
        with open(os.environ.get("SYSTEM_PROMPT_PATH"), "r", encoding="utf-8") as file:
            return file.read().strip()

    @staticmethod
    def load_functions():
        tools = []
        tools_dir = os.path.join(os.path.dirname(__file__), 'tools')
        for filename in os.listdir(tools_dir):
            if filename.endswith('.py') and filename != '__init__.py' and filename != 'base_tool.py':
                module_name = f'tools.{filename[:-3]}'
                module = importlib.import_module(module_name)
                for name, obj in inspect.getmembers(module):
                    if inspect.isclass(obj) and issubclass(obj, BaseTool) and obj != BaseTool:
                        tools.append(obj())
        
        # Collect all function definitions
        functions = []
        for tool in tools:
            functions.extend(tool.get_functions())
        return tools, functions

    @abstractmethod
    def get_chat_response(self, messages):
        pass

    @abstractmethod
    async def handle_message(self, user_id, user_message):
        pass

    def clear_conversation(self, user_id):
        if user_id in self.conversation_history:
            del self.conversation_history[user_id]
        
        for tool in self.tools:
            tool.clear()

    def call_tool(self, function_call_name, function_call_arguments):
        function_name = function_call_name
        function_args = json.loads(function_call_arguments if function_call_arguments is not None else "{}")
        for tool in self.tools:
            if function_name in [f["name"] for f in tool.get_functions()]:
                return tool.execute(function_name, **function_args)

    @abstractmethod
    async def start(self):
        pass

    @abstractmethod
    async def clear(self, user_id):
        pass

    @abstractmethod
    async def status(self):
        pass

    @abstractmethod
    async def abort_processing(self, user_id):
        pass