removed unnecessary files

This commit is contained in:
2024-08-18 14:27:20 -05:00
parent 3acad2f861
commit a8bda7622e
6 changed files with 2 additions and 296 deletions
-136
View File
@@ -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}")
View File
+2 -1
View File
@@ -8,13 +8,14 @@ from telegram.ext import Application, CommandHandler, MessageHandler, filters, C
from openai import OpenAI from openai import OpenAI
from dotenv import load_dotenv from dotenv import load_dotenv
from tools.base_tool import BaseTool from tools.base_tool import BaseTool
from anthropic import Anthropic
# Load environment variables # Load environment variables
load_dotenv() load_dotenv()
openai_client = OpenAI() openai_client = OpenAI()
anthropic_client = anthropic.Anthropic( anthropic_client = Anthropic(
api_key=os.environ.get("ANTHROPIC_API_KEY"), api_key=os.environ.get("ANTHROPIC_API_KEY"),
default_headers={"anthropic-beta": "max-tokens-3-5-sonnet-2024-07-15"} default_headers={"anthropic-beta": "max-tokens-3-5-sonnet-2024-07-15"}
) )
-31
View File
@@ -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}"
-79
View File
@@ -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
-49
View File
@@ -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