Refactor: Make system_prompt_path and content configurable in BaseTelegramInferenceBot

This commit is contained in:
cyclop-bot
2025-06-02 16:35:08 -05:00
parent 0ffafe3bc5
commit 75381f5f11
+49 -23
View File
@@ -7,26 +7,47 @@ from abc import ABC, abstractmethod
from tools.base_tool import BaseTool from tools.base_tool import BaseTool
class BaseTelegramInferenceBot(ABC): class BaseTelegramInferenceBot(ABC):
def __init__(self): def __init__(self, system_prompt_content: str | None = None, system_prompt_path: str | None = None): # MODIFIED
self.conversation_history = {} self.conversation_history = {}
self.processing_status = {} self.processing_status = {}
self.system_prompt = self.load_system_prompt() # MODIFIED to pass arguments
self.system_prompt = self.load_system_prompt(
direct_content=system_prompt_content,
file_path=system_prompt_path
)
self.tools, self.functions = self.load_functions() self.tools, self.functions = self.load_functions()
logging.info(f'System Prompt: {os.environ.get("SYSTEM_PROMPT_PATH")}') # Logging the actual source of the system prompt might be more complex now,
# but we can log the final prompt or indicate if it's custom/default.
# We'll also log the source of the prompt inside load_system_prompt.
logging.info(f'System Prompt (effective): {"Custom" if self.system_prompt != "You are a helpful AI assistant." else "Default"}')
logging.info(f'Github Repository: {os.environ.get("GITHUB_REPOSITORY")}') logging.info(f'Github Repository: {os.environ.get("GITHUB_REPOSITORY")}')
def load_system_prompt(self): def load_system_prompt(self, direct_content: str | None = None, file_path: str | None = None) -> str: # MODIFIED
system_prompt_path = os.getenv("SYSTEM_PROMPT_PATH") default_prompt = "You are a helpful AI assistant."
if system_prompt_path and os.path.isfile(system_prompt_path):
try: if direct_content:
with open(system_prompt_path, "r", encoding="utf-8") as file: logging.info("Using direct content for system prompt.")
return file.read().strip() return direct_content.strip()
except IOError as e:
logging.warning(f"Could not read system prompt file {system_prompt_path}: {e}") prompt_path_to_try = file_path or os.getenv("SYSTEM_PROMPT_PATH")
return "You are a helpful AI assistant."
if prompt_path_to_try:
if os.path.isfile(prompt_path_to_try):
try:
with open(prompt_path_to_try, "r", encoding="utf-8") as file:
content = file.read().strip()
logging.info(f"Successfully loaded system prompt from {prompt_path_to_try}.")
return content
except IOError as e:
logging.warning(f"Could not read system prompt file {prompt_path_to_try}: {e}. Using default.")
return default_prompt
else:
# This condition now also covers if 'file_path' argument was given but invalid
logging.warning(f"System prompt file {prompt_path_to_try} not found. Using default system prompt.")
return default_prompt
else: else:
logging.warning("SYSTEM_PROMPT_PATH is not set or file does not exist. Using default system prompt.") logging.info("No system prompt path provided (argument or ENV) or direct content. Using default system prompt.")
return "You are a helpful AI assistant." return default_prompt
def load_functions(self): def load_functions(self):
tools = [] tools = []
@@ -44,7 +65,7 @@ class BaseTelegramInferenceBot(ABC):
for name, obj in inspect.getmembers(module): for name, obj in inspect.getmembers(module):
if inspect.isclass(obj) and issubclass(obj, BaseTool) and obj != BaseTool: if inspect.isclass(obj) and issubclass(obj, BaseTool) and obj != BaseTool:
try: try:
tools.append(obj()) tools.append(obj()) # This instantiation might be an issue for tools needing config
except Exception as e: except Exception as e:
logging.error(f"Error instantiating tool {name} from {filename}: {e}") logging.error(f"Error instantiating tool {name} from {filename}: {e}")
except Exception as e: except Exception as e:
@@ -87,9 +108,9 @@ class BaseTelegramInferenceBot(ABC):
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
logging.error(f"Error decoding function call arguments (string) for {function_call_name}: {e}. Arguments: {function_call_arguments}") logging.error(f"Error decoding function call arguments (string) for {function_call_name}: {e}. Arguments: {function_call_arguments}")
return f"Error: Malformed arguments for tool call: {e}" return f"Error: Malformed arguments for tool call: {e}"
else: # Handle cases where arguments might be None or other unexpected types else:
if function_call_arguments is None: if function_call_arguments is None:
function_args = {} # Default to empty dict if arguments are None function_args = {}
else: else:
logging.error(f"Unexpected type for function_call_arguments for {function_call_name}: {type(function_call_arguments)}. Arguments: {function_call_arguments}") logging.error(f"Unexpected type for function_call_arguments for {function_call_name}: {type(function_call_arguments)}. Arguments: {function_call_arguments}")
return f"Error: Invalid argument type for tool call: {type(function_call_arguments)}" return f"Error: Invalid argument type for tool call: {type(function_call_arguments)}"
@@ -98,7 +119,6 @@ class BaseTelegramInferenceBot(ABC):
for function in tool.get_functions(): for function in tool.get_functions():
if function["function"]["name"] == function_name: if function["function"]["name"] == function_name:
try: try:
# Ensure function_args is a dictionary before unpacking
if not isinstance(function_args, dict): if not isinstance(function_args, dict):
logging.error(f"Internal error: function_args not a dict for {function_name} before execution. Args: {function_args}") logging.error(f"Internal error: function_args not a dict for {function_name} before execution. Args: {function_args}")
return f"Internal error preparing arguments for tool {function_name}." return f"Internal error preparing arguments for tool {function_name}."
@@ -110,16 +130,23 @@ class BaseTelegramInferenceBot(ABC):
return f"Error: Tool function {function_name} not found." return f"Error: Tool function {function_name} not found."
def get_system_prompt_description(self) -> str: def get_system_prompt_description(self) -> str:
"""Returns a description of the system prompt being used.""" # This method could be updated to be more specific about the prompt source if needed.
return f"System Prompt: {'Custom' if os.getenv('SYSTEM_PROMPT_PATH') else 'Default'}" # For now, it still reflects custom vs default based on the original ENV var logic's spirit.
# A more accurate reflection would require storing how the prompt was loaded.
# For simplicity, let's assume if it's not the default, it's "Custom".
if self.system_prompt != "You are a helpful AI assistant.":
return "System Prompt: Custom"
# Check original ENV var for backward compatibility in description only
elif os.getenv('SYSTEM_PROMPT_PATH'):
return "System Prompt: Custom (via ENV)"
return "System Prompt: Default"
@abstractmethod @abstractmethod
def get_llm_description(self) -> str: def get_llm_description(self) -> str:
"""Returns a description of the LLM being used."""
pass pass
async def get_bot_status(self) -> str: async def get_bot_status(self) -> str:
"""Provides a status message including prompt and LLM information."""
prompt_desc = self.get_system_prompt_description() prompt_desc = self.get_system_prompt_description()
llm_desc = self.get_llm_description() llm_desc = self.get_llm_description()
return f"{prompt_desc}\n{llm_desc}" return f"{prompt_desc}\n{llm_desc}"
@@ -134,5 +161,4 @@ class BaseTelegramInferenceBot(ABC):
@abstractmethod @abstractmethod
async def switch_model(self): async def switch_model(self):
"""Switches the underlying model if supported by the bot."""
pass pass