File size: 4,122 Bytes
7db0ae4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 |
#### What this does ####
# On success + failure, log events to Supabase
import dotenv, os
import requests
dotenv.load_dotenv() # Loading env variables using dotenv
import traceback
import datetime, subprocess, sys
import litellm
class Supabase:
# Class variables or attributes
supabase_table_name = "request_logs"
def __init__(self):
# Instance variables
self.supabase_url = os.getenv("SUPABASE_URL")
self.supabase_key = os.getenv("SUPABASE_KEY")
try:
import supabase
except ImportError:
subprocess.check_call([sys.executable, "-m", "pip", "install", "supabase"])
import supabase
self.supabase_client = supabase.create_client(
self.supabase_url, self.supabase_key
)
def input_log_event(
self, model, messages, end_user, litellm_call_id, print_verbose
):
try:
print_verbose(
f"Supabase Logging - Enters input logging function for model {model}"
)
supabase_data_obj = {
"model": model,
"messages": messages,
"end_user": end_user,
"status": "initiated",
"litellm_call_id": litellm_call_id,
}
data, count = (
self.supabase_client.table(self.supabase_table_name)
.insert(supabase_data_obj)
.execute()
)
print_verbose(f"data: {data}")
except:
print_verbose(f"Supabase Logging Error - {traceback.format_exc()}")
pass
def log_event(
self,
model,
messages,
end_user,
response_obj,
start_time,
end_time,
litellm_call_id,
print_verbose,
):
try:
print_verbose(
f"Supabase Logging - Enters logging function for model {model}, response_obj: {response_obj}"
)
total_cost = litellm.completion_cost(completion_response=response_obj)
response_time = (end_time - start_time).total_seconds()
if "choices" in response_obj:
supabase_data_obj = {
"response_time": response_time,
"model": response_obj["model"],
"total_cost": total_cost,
"messages": messages,
"response": response_obj["choices"][0]["message"]["content"],
"end_user": end_user,
"litellm_call_id": litellm_call_id,
"status": "success",
}
print_verbose(
f"Supabase Logging - final data object: {supabase_data_obj}"
)
data, count = (
self.supabase_client.table(self.supabase_table_name)
.upsert(supabase_data_obj, on_conflict="litellm_call_id")
.execute()
)
elif "error" in response_obj:
if "Unable to map your input to a model." in response_obj["error"]:
total_cost = 0
supabase_data_obj = {
"response_time": response_time,
"model": response_obj["model"],
"total_cost": total_cost,
"messages": messages,
"error": response_obj["error"],
"end_user": end_user,
"litellm_call_id": litellm_call_id,
"status": "failure",
}
print_verbose(
f"Supabase Logging - final data object: {supabase_data_obj}"
)
data, count = (
self.supabase_client.table(self.supabase_table_name)
.upsert(supabase_data_obj, on_conflict="litellm_call_id")
.execute()
)
except:
# traceback.print_exc()
print_verbose(f"Supabase Logging Error - {traceback.format_exc()}")
pass
|