diff --git a/ai_providers.py b/ai_providers.py deleted file mode 100644 index 2b983f0..0000000 --- a/ai_providers.py +++ /dev/null @@ -1,136 +0,0 @@ -import os -import json -import logging -import anthropic -from openai import OpenAI -from abc import ABC, abstractmethod -from tools.github_tool import GitHubTool - -# Initialize GitHubTool and get functions -github_tool = GitHubTool() -functions = github_tool.get_functions() - -class AIProvider(ABC): - @abstractmethod - def get_chat_response(self, messages): - pass - - @abstractmethod - def format_messages(self, messages): - pass - - @abstractmethod - def format_tool_calls(self, response): - pass - - @abstractmethod - def format_tool_result(self, tool_call, tool_response): - pass - -class AnthropicProvider(AIProvider): - def __init__(self): - self.client = anthropic.Anthropic( - api_key=os.environ.get("ANTHROPIC_API_KEY"), - default_headers={"anthropic-beta": "max-tokens-3-5-sonnet-2024-07-15"} - ) - self.model = "claude-3-5-sonnet-20240620" - - def get_chat_response(self, messages): - try: - response = self.client.messages.create( - model=self.model, - system=messages[0]['content'], - messages=self.format_messages(messages[1:]), - max_tokens=8192, - tools=self.format_tools() - ) - return response - except Exception as e: - logging.error(f"An error occurred: {str(e)}") - return None - - def format_messages(self, messages): - return messages - - def format_tool_calls(self, response): - tool_calls = [] - for message in response.content: - if message.type == "tool_use": - tool_calls.append(message) - return tool_calls - - def format_tools(self): - return [ - { - "name": function['name'], - "description": function['description'], - "input_schema": function['parameters'] if function['parameters'] not in [None, {}] else {"type": "object", "properties": {"param1": {"type": "string", "description": "Unnecessary"}}, "required": []} - } - for function in functions - ] - - def format_assistant_reply(self, response): - for message in response.content: - if message.type == "text": - return message.text - return "" - - def get_reply_text(self, response): - return self.format_assistant_reply(response) - - def get_model(self): - return self.model - - def format_tool_result(self, tool_call, tool_response): - return { - "role": "function", - "name": tool_call.name, - "content": json.dumps(tool_response) - } - -class OpenAIProvider(AIProvider): - def __init__(self, use_smart_model=True): - self.client = OpenAI() - self.use_smart_model = use_smart_model - self.model = self.get_model() - - def get_model(self): - return "gpt-4o" if self.use_smart_model else "gpt-4o-mini" - - def get_chat_response(self, messages): - response = self.client.chat.completions.create( - model=self.model, - messages=self.format_messages(messages), - functions=functions, - function_call="auto", - max_tokens=self.get_max_tokens() - ) - return response - - def format_messages(self, messages): - return messages - - def format_tool_calls(self, response): - tool_calls = [] - assistant_message = response.choices[0].message - if hasattr(assistant_message, 'function_call') and assistant_message.function_call is not None: - tool_calls.append(assistant_message.function_call) - return tool_calls - - def get_max_tokens(self): - return 4096 if self.model == "gpt-4o" else 16384 - - def format_tool_result(self, tool_call, tool_response): - return { - "role": "function", - "name": tool_call.name, - "content": json.dumps(tool_response) - } - -def create_ai_provider(provider_name="anthropic", use_smart_model=True): - if provider_name.lower() == "anthropic": - return AnthropicProvider() - elif provider_name.lower() == "openai": - return OpenAIProvider(use_smart_model) - else: - raise ValueError(f"Unknown provider: {provider_name}") \ No newline at end of file diff --git a/persona_tool.py b/persona_tool.py deleted file mode 100644 index e69de29..0000000 diff --git a/telegram_inference_bot.py b/telegram_inference_bot.py index df35994..eba6345 100644 --- a/telegram_inference_bot.py +++ b/telegram_inference_bot.py @@ -8,13 +8,14 @@ from telegram.ext import Application, CommandHandler, MessageHandler, filters, C from openai import OpenAI from dotenv import load_dotenv from tools.base_tool import BaseTool +from anthropic import Anthropic # Load environment variables load_dotenv() openai_client = OpenAI() -anthropic_client = anthropic.Anthropic( +anthropic_client = Anthropic( api_key=os.environ.get("ANTHROPIC_API_KEY"), default_headers={"anthropic-beta": "max-tokens-3-5-sonnet-2024-07-15"} ) diff --git a/tools/camera_pretool.prepy b/tools/camera_pretool.prepy deleted file mode 100644 index b7e5f98..0000000 --- a/tools/camera_pretool.prepy +++ /dev/null @@ -1,31 +0,0 @@ -# tools/camera_tool.py -from .base_tool import BaseTool -import picamera -import time -import base64 -from io import BytesIO - -class CameraTool(BaseTool): - def get_functions(self): - return [{ - "name": "take_picture", - "description": "Take a picture using the Raspberry Pi camera", - "parameters": { - "type": "object", - "properties": {}, - "required": [] - } - }] - - def execute(self, function_name, **kwargs): - if function_name == "take_picture": - with picamera.PiCamera() as camera: - camera.resolution = (1024, 768) - camera.start_preview() - # Camera warm-up time - time.sleep(2) - image_stream = BytesIO() - camera.capture(image_stream, 'jpeg') - image_stream.seek(0) - image_base64 = base64.b64encode(image_stream.getvalue()).decode('utf-8') - return f"data:image/jpeg;base64,{image_base64}" diff --git a/tools/persona_tool.prepy b/tools/persona_tool.prepy deleted file mode 100644 index cee6ea8..0000000 --- a/tools/persona_tool.prepy +++ /dev/null @@ -1,79 +0,0 @@ -import os -import openai -import json -from tools.base_tool import BaseTool - -class PersonaTool(BaseTool): - - - def __init__(self): - super().__init__() - self.api_key = os.environ.get("OPENAI_API_KEY") - - GPT_4O = "gpt-4o" - GPT_4O_MINI = "gpt-4o-mini" - - def generate_response(self, persona_description: str, query: str) -> str: - """ - Makes a call to the OpenAI API using the persona as a system prompt. - - Parameters: - persona_description (str): Description of the persona. - query (str): Query to be processed. - - Returns: - str: The response generated by the OpenAI API. - """ - response = openai.ChatCompletion.create( - model="gpt-3.5-turbo", - messages=[ - {"role": "system", "content": persona_description}, - {"role": "user", "content": query} - ] - ) - return response.choices[0].message['content'] - - def get_functions(self): - return json.dumps({ - "functions": [ - { - "name": "generate_response", - "description": "Generates a response based on a persona description and a user query.", - "parameters": { - "type": "object", - "properties": { - "persona_description": { - "type": "string", - "description": "Description of the persona." - }, - "query": { - "type": "string", - "description": "User's query to be processed." - } - }, - "required": ["persona_description", "query"] - } - } - ] - }) - - def execute(self, function_name, **kwargs): - if function_name == "generate_response": - return self.generate_response(kwargs.get("persona_description"), kwargs.get("query")) - else: - raise ValueError(f"Function {function_name} not found") - - def get_chat_response(client, messages, model): - - model_max_tokens = { - GPT_4O: 4096, - GPT_4O_MINI: 16384 - } - - response = client.chat.completions.create( - model=model, - messages=messages, - function_call="none", - max_tokens=model_max_tokens[model] - ) - return response \ No newline at end of file diff --git a/tools/telegram_keyboard_tool.prepy b/tools/telegram_keyboard_tool.prepy deleted file mode 100644 index aca4c0e..0000000 --- a/tools/telegram_keyboard_tool.prepy +++ /dev/null @@ -1,49 +0,0 @@ -# tools/telegram_keyboard_tool.py - -from .base_tool import BaseTool - -class TelegramKeyboardTool(BaseTool): - - def get_functions(self): - return [ - { - "name": "create_keyboard", - "description": "Create a Telegram keyboard layout.", - "parameters": { - "type": "object", - "properties": { - "buttons": { - "type": "array", - "items": { - "type": "array", - "items": { - "type": "string", - "description": "Text on the button." - }, - "description": "A row of buttons." - }, - "description": "List of rows of buttons." - } - }, - "required": ["buttons"] - } - } - ] - - def execute(self, function_name, **kwargs): - if function_name == "create_keyboard": - return self._create_keyboard(kwargs.get("buttons")) - else: - return f"Unknown function: {function_name}" - - def _create_keyboard(self, buttons): - if not isinstance(buttons, list): - return "Invalid input: 'buttons' must be a list of lists." - - keyboard = { - "keyboard": buttons, - "resize_keyboard": True, - "one_time_keyboard": True - } - - return keyboard