Skip to content

Commit

Permalink
Refactor code spaced to tabs
Browse files Browse the repository at this point in the history
  • Loading branch information
@haseeb-heaven
haseeb-heaven committed Oct 26, 2024
1 parent 79e4f60 commit f7a61e2
Show file tree
Hide file tree
Showing 8 changed files with 1,888 additions and 1,887 deletions.
366 changes: 183 additions & 183 deletions libs/code_interpreter.py
View file

Large diffs are not rendered by default.

124 changes: 62 additions & 62 deletions libs/gemini_vision.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,71 +4,71 @@
import litellm

class GeminiVision:
def __init__(self, api_key=None) -> None:
self.logger = Logger.initialize_logger('logs/vision_interpreter.log')
self.logger.info(f"Initializing Gemini Vision")
self.api_key = api_key
if self.api_key is None:
self.logger.warning("API key is not initialized")
def __init__(self, api_key=None) -> None:
self.logger = Logger.initialize_logger('logs/vision_interpreter.log')
self.logger.info("Initializing Gemini Vision")
self.api_key = api_key
if self.api_key is None:
self.logger.warning("API key is not initialized")

# load the key from the .env file
load_dotenv()
api_key = os.getenv("GEMINI_API_KEY")
if not api_key:
self.logger.error("No API key found in the .env file")
raise ValueError("No API key found in the .env file")
self.logger.info(f"Gemini Vision configured success")
self.logger.info(f"Model setup success")
# load the key from the .env file
load_dotenv()
api_key = os.getenv("GEMINI_API_KEY")
if not api_key:
self.logger.error("No API key found in the .env file")
raise ValueError("No API key found in the .env file")
self.logger.info("Gemini Vision configured success")
self.logger.info("Model setup success")

def generate_text(self, prompt, image_url):
self.logger.info(f"Generating contents")
# Create the messages payload according to the documentation
messages = [
{
"role": "user",
"content": [
{
"type": "text",
"text": prompt
},
{
"type": "image_url",
"image_url": {"url": image_url}
}
]
}
]
def generate_text(self, prompt, image_url):
self.logger.info("Generating contents")
# Create the messages payload according to the documentation
messages = [
{
"role": "user",
"content": [
{
"type": "text",
"text": prompt
},
{
"type": "image_url",
"image_url": {"url": image_url}
}
]
}
]

# Make the API call to Gemini model
response = litellm.completion(
model="gemini/gemini-pro-vision",
messages=messages,
)
# Make the API call to Gemini model
response = litellm.completion(
model="gemini/gemini-pro-vision",
messages=messages,
)

# Extract the response content
return response.get('choices', [{}])[0].get('message', {}).get('content')
# Extract the response content
return response.get('choices', [{}])[0].get('message', {}).get('content')

def gemini_vision_url(self, prompt, image_url):
self.logger.info(f"Generating text from URL: {image_url}")
try:
return self.generate_text(prompt, image_url)
except Exception as exception:
self.logger.error(f"Error generating text from URL: {exception}")
raise
def gemini_vision_url(self, prompt, image_url):
self.logger.info(f"Generating text from URL: {image_url}")
try:
return self.generate_text(prompt, image_url)
except Exception as exception:
self.logger.error(f"Error generating text from URL: {exception}")
raise

def gemini_vision_path(self, prompt, image_path):
self.logger.info(f"Generating text from image path: '{image_path}'")
try:
self.logger.info(f"Checking if image path exists for: '{image_path}'")
# check if the image path exists
if not os.path.exists(image_path):
raise ValueError(f"Image path does not exist: {image_path}")
return self.generate_text(prompt,image_path)
except Exception as exception:
self.logger.error(f"Error generating text from image path: {exception}")
raise
def gemini_vision_path(self, prompt, image_path):
self.logger.info(f"Generating text from image path: '{image_path}'")
try:
self.logger.info(f"Checking if image path exists for: '{image_path}'")
# check if the image path exists
if not os.path.exists(image_path):
raise ValueError(f"Image path does not exist: {image_path}")
return self.generate_text(prompt,image_path)
except Exception as exception:
self.logger.error(f"Error generating text from image path: {exception}")
raise
190 changes: 95 additions & 95 deletions libs/history_manager.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,108 +5,108 @@
from libs.logger import Logger

class History:
def __init__(self, history_file: str):
self.history_file = history_file
self.logger = Logger.initialize_logger("logs/interpreter.log")
def __init__(self, history_file: str):
self.history_file = history_file
self.logger = Logger.initialize_logger("logs/interpreter.log")

def save_history_json(self, task, mode, os_name, language, prompt, code_snippet, code_output, model_name):
try:
history_entry = {
"assistant": {
"task": task,
"mode": mode,
"os": os_name,
"language": language,
"model": model_name
},
"user": prompt,
"system": {
"code": code_snippet,
"output": code_output
}
}
def save_history_json(self, task, mode, os_name, language, prompt, code_snippet, code_output, model_name):
try:
history_entry = {
"assistant": {
"task": task,
"mode": mode,
"os": os_name,
"language": language,
"model": model_name
},
"user": prompt,
"system": {
"code": code_snippet,
"output": code_output
}
}

data = []
if os.path.isfile(self.history_file) and os.path.getsize(self.history_file) > 0:
with open(self.history_file, "r") as history_file: # Open the file in read mode
data = json.load(history_file)
data = []
if os.path.isfile(self.history_file) and os.path.getsize(self.history_file) > 0:
with open(self.history_file, "r") as history_file: # Open the file in read mode
data = json.load(history_file)

data.append(history_entry)
data.append(history_entry)

with open(self.history_file, "w") as history_file:
json.dump(data, history_file)
except Exception as exception:
self.logger.error(f"Error in saving history to JSON: {str(exception)}")
raise
with open(self.history_file, "w") as history_file:
json.dump(data, history_file)
except Exception as exception:
self.logger.error(f"Error in saving history to JSON: {str(exception)}")
raise

def _get_data_for_key(self, key: str) -> List[Any]:
"""Returns a list of all values for the specified key in the history data."""
try:
if os.path.getsize(self.history_file) > 0:
with open(self.history_file, 'r') as file:
history_data = json.load(file)
else:
return []
specific_data = []
for entry in history_data:
if key in entry['assistant']:
specific_data.append(entry['assistant'].get(key))
elif key in entry['system']:
specific_data.append(entry['system'].get(key))
self.logger.info(f'Successfully retrieved {key} data from history')
return specific_data
except Exception as exception:
self.logger.error(f'Error getting {key} data from history: {exception}')
raise
def _get_data_for_key(self, key: str) -> List[Any]:
"""Returns a list of all values for the specified key in the history data."""
try:
if os.path.getsize(self.history_file) > 0:
with open(self.history_file, 'r') as file:
history_data = json.load(file)
else:
return []
specific_data = []
for entry in history_data:
if key in entry['assistant']:
specific_data.append(entry['assistant'].get(key))
elif key in entry['system']:
specific_data.append(entry['system'].get(key))
self.logger.info(f'Successfully retrieved {key} data from history')
return specific_data
except Exception as exception:
self.logger.error(f'Error getting {key} data from history: {exception}')
raise

def _get_last_entries(self, count: int) -> List[dict]:
"""Returns the last n entries from the history data."""
try:
with open(self.history_file, 'r') as file:
history_data = json.load(file)
last_entries = history_data[-count:]
self.logger.info(f'Successfully retrieved last {count} entries from history')
return last_entries
except Exception as exception:
self.logger.error(f'Error getting last {count} entries from history: {exception}')
raise
def _get_last_entries(self, count: int) -> List[dict]:
"""Returns the last n entries from the history data."""
try:
with open(self.history_file, 'r') as file:
history_data = json.load(file)
last_entries = history_data[-count:]
self.logger.info(f'Successfully retrieved last {count} entries from history')
return last_entries
except Exception as exception:
self.logger.error(f'Error getting last {count} entries from history: {exception}')
raise

def _get_last_entries_for_key(self, key: str, count: int) -> List[Any]:
"""Returns the values of the specified key for the last n entries in the history data."""
try:
specific_key_data = self._get_data_for_key(key)
last_specific_data = specific_key_data[-count:]
if last_specific_data:
self.logger.info(f'Successfully retrieved {key} data for last {count} entries from history')
self.logger.info(f"\n'{last_specific_data}'\n")
return last_specific_data
else:
self.logger.info(f'No {key} data found in history')
return []
except Exception as exception:
self.logger.error(f'Error getting {key} data for last {count} entries from history: {exception}')
raise
def _get_last_entries_for_keys(self, count: int, *keys: str) -> List[dict]:
last_entries = []
try:
entries = {key: self._get_last_entries_for_key(key, count) for key in keys}
for index in range(count):
session = {key: entries[key][index] if index < len(entries[key]) else None for key in keys}
last_entries.append(session)
return last_entries
except Exception as exception:
self.logger.error(f'Error getting last {count} entries for keys {keys} from history: {exception}')
raise
def _get_last_entries_for_key(self, key: str, count: int) -> List[Any]:
"""Returns the values of the specified key for the last n entries in the history data."""
try:
specific_key_data = self._get_data_for_key(key)
last_specific_data = specific_key_data[-count:]
if last_specific_data:
self.logger.info(f'Successfully retrieved {key} data for last {count} entries from history')
self.logger.info(f"\n'{last_specific_data}'\n")
return last_specific_data
else:
self.logger.info(f'No {key} data found in history')
return []
except Exception as exception:
self.logger.error(f'Error getting {key} data for last {count} entries from history: {exception}')
raise
def _get_last_entries_for_keys(self, count: int, *keys: str) -> List[dict]:
last_entries = []
try:
entries = {key: self._get_last_entries_for_key(key, count) for key in keys}
for index in range(count):
session = {key: entries[key][index] if index < len(entries[key]) else None for key in keys}
last_entries.append(session)
return last_entries
except Exception as exception:
self.logger.error(f'Error getting last {count} entries for keys {keys} from history: {exception}')
raise

def get_chat_history(self, count: int) -> List[dict]:
return self._get_last_entries_for_keys(count, "task", "output")
def get_chat_history(self, count: int) -> List[dict]:
return self._get_last_entries_for_keys(count, "task", "output")

def get_code_history(self, count: int) -> List[dict]:
return self._get_last_entries_for_keys(count, "code", "output")
def get_code_history(self, count: int) -> List[dict]:
return self._get_last_entries_for_keys(count, "code", "output")

def get_full_history(self, count: int) -> List[dict]:
return self._get_last_entries_for_keys(count, "task", "code", "output")
def get_full_history(self, count: int) -> List[dict]:
return self._get_last_entries_for_keys(count, "task", "code", "output")
Loading

0 comments on commit f7a61e2

Please sign in to comment.