cyclop-bot
96 lines
3.4 KiB
Python
96 lines
3.4 KiB
Python
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()
|
|
print(f'System Prompt: {os.environ.get("SYSTEM_PROMPT_PATH")}')
|
|
print(f'Github Repository: {os.environ.get("GITHUB_REPOSITORY")}')
|
|
|
|
@staticmethod
|
|
def load_system_prompt():
|
|
system_prompt_path = os.getenv("SYSTEM_PROMPT_PATH")
|
|
if system_prompt_path and os.path.isfile(system_prompt_path):
|
|
with open(system_prompt_path, "r", encoding="utf-8") as file:
|
|
return file.read().strip()
|
|
else:
|
|
raise FileNotFoundError("SYSTEM_PROMPT_PATH is not set or file does not exist.")
|
|
|
|
@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:
|
|
for function in tool.get_functions():
|
|
if function["function"]["name"] == function_name:
|
|
return tool.execute(function_name, **function_args)
|
|
|
|
@abstractmethod
|
|
def get_system_prompt_description(self) -> str:
|
|
"""Returns a description of the system prompt being used."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_llm_description(self) -> str:
|
|
"""Returns a description of the LLM being used."""
|
|
pass
|
|
|
|
async def status(self) -> str: # Changed from abstract to concrete
|
|
"""Provides a status message including prompt and LLM information."""
|
|
prompt_desc = self.get_system_prompt_description()
|
|
llm_desc = self.get_llm_description()
|
|
# Consider potential async calls if get_... methods were async
|
|
# For now, assuming they are synchronous as per design
|
|
return f"{prompt_desc}\n{llm_desc}"
|
|
|
|
@abstractmethod
|
|
async def start(self):
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def clear(self, user_id):
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def abort_processing(self, user_id):
|
|
pass
|