Add BaseTelegramInferenceBot class
This commit is contained in:
@@ -0,0 +1,56 @@
|
||||
import os
|
||||
import json
|
||||
import logging
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
class BaseTelegramInferenceBot(ABC):
|
||||
def __init__(self):
|
||||
self.conversation_history = {}
|
||||
self.processing_status = {}
|
||||
self.system_prompt = self.load_system_prompt()
|
||||
self.functions = self.load_functions()
|
||||
|
||||
@staticmethod
|
||||
def load_system_prompt():
|
||||
with open("prompts/developer_prompt.txt", "r") as file:
|
||||
return file.read().strip()
|
||||
|
||||
@staticmethod
|
||||
def load_functions():
|
||||
# Implement function loading logic here
|
||||
return []
|
||||
|
||||
@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]
|
||||
|
||||
def call_tool(self, function_call):
|
||||
function_name = function_call.name
|
||||
function_args = json.loads(function_call.arguments)
|
||||
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
|
||||
Reference in New Issue
Block a user