Spaces:
Sleeping
Sleeping
import requests, traceback, json, os | |
import types | |
class LiteDebugger: | |
user_email = None | |
dashboard_url = None | |
def __init__(self, email=None): | |
self.api_url = "https://api.litellm.ai/debugger" | |
self.validate_environment(email) | |
pass | |
def validate_environment(self, email): | |
try: | |
self.user_email = ( | |
email or os.getenv("LITELLM_TOKEN") or os.getenv("LITELLM_EMAIL") | |
) | |
if ( | |
self.user_email == None | |
): # if users are trying to use_client=True but token not set | |
raise ValueError( | |
"litellm.use_client = True but no token or email passed. Please set it in litellm.token" | |
) | |
self.dashboard_url = "https://admin.litellm.ai/" + self.user_email | |
try: | |
print( | |
f"\033[92mHere's your LiteLLM Dashboard 👉 \033[94m\033[4m{self.dashboard_url}\033[0m" | |
) | |
except: | |
print(f"Here's your LiteLLM Dashboard 👉 {self.dashboard_url}") | |
if self.user_email == None: | |
raise ValueError( | |
"[Non-Blocking Error] LiteLLMDebugger: Missing LITELLM_TOKEN. Set it in your environment. Eg.: os.environ['LITELLM_TOKEN']= <your_email>" | |
) | |
except Exception as e: | |
raise ValueError( | |
"[Non-Blocking Error] LiteLLMDebugger: Missing LITELLM_TOKEN. Set it in your environment. Eg.: os.environ['LITELLM_TOKEN']= <your_email>" | |
) | |
def input_log_event( | |
self, | |
model, | |
messages, | |
end_user, | |
litellm_call_id, | |
call_type, | |
print_verbose, | |
litellm_params, | |
optional_params, | |
): | |
print_verbose( | |
f"LiteDebugger: Pre-API Call Logging for call id {litellm_call_id}" | |
) | |
try: | |
print_verbose( | |
f"LiteLLMDebugger: Logging - Enters input logging function for model {model}" | |
) | |
def remove_key_value(dictionary, key): | |
new_dict = dictionary.copy() # Create a copy of the original dictionary | |
new_dict.pop(key) # Remove the specified key-value pair from the copy | |
return new_dict | |
updated_litellm_params = remove_key_value(litellm_params, "logger_fn") | |
if call_type == "embedding": | |
for ( | |
message | |
) in ( | |
messages | |
): # assuming the input is a list as required by the embedding function | |
litellm_data_obj = { | |
"model": model, | |
"messages": [{"role": "user", "content": message}], | |
"end_user": end_user, | |
"status": "initiated", | |
"litellm_call_id": litellm_call_id, | |
"user_email": self.user_email, | |
"litellm_params": updated_litellm_params, | |
"optional_params": optional_params, | |
} | |
print_verbose( | |
f"LiteLLMDebugger: Logging - logged data obj {litellm_data_obj}" | |
) | |
response = requests.post( | |
url=self.api_url, | |
headers={"content-type": "application/json"}, | |
data=json.dumps(litellm_data_obj), | |
) | |
print_verbose(f"LiteDebugger: embedding api response - {response.text}") | |
elif call_type == "completion": | |
litellm_data_obj = { | |
"model": model, | |
"messages": messages | |
if isinstance(messages, list) | |
else [{"role": "user", "content": messages}], | |
"end_user": end_user, | |
"status": "initiated", | |
"litellm_call_id": litellm_call_id, | |
"user_email": self.user_email, | |
"litellm_params": updated_litellm_params, | |
"optional_params": optional_params, | |
} | |
print_verbose( | |
f"LiteLLMDebugger: Logging - logged data obj {litellm_data_obj}" | |
) | |
response = requests.post( | |
url=self.api_url, | |
headers={"content-type": "application/json"}, | |
data=json.dumps(litellm_data_obj), | |
) | |
print_verbose( | |
f"LiteDebugger: completion api response - {response.text}" | |
) | |
except: | |
print_verbose( | |
f"[Non-Blocking Error] LiteDebugger: Logging Error - {traceback.format_exc()}" | |
) | |
pass | |
def post_call_log_event( | |
self, original_response, litellm_call_id, print_verbose, call_type, stream | |
): | |
print_verbose( | |
f"LiteDebugger: Post-API Call Logging for call id {litellm_call_id}" | |
) | |
try: | |
if call_type == "embedding": | |
litellm_data_obj = { | |
"status": "received", | |
"additional_details": { | |
"original_response": str( | |
original_response["data"][0]["embedding"][:5] | |
) | |
}, # don't store the entire vector | |
"litellm_call_id": litellm_call_id, | |
"user_email": self.user_email, | |
} | |
elif call_type == "completion" and not stream: | |
litellm_data_obj = { | |
"status": "received", | |
"additional_details": {"original_response": original_response}, | |
"litellm_call_id": litellm_call_id, | |
"user_email": self.user_email, | |
} | |
elif call_type == "completion" and stream: | |
litellm_data_obj = { | |
"status": "received", | |
"additional_details": { | |
"original_response": "Streamed response" | |
if isinstance(original_response, types.GeneratorType) | |
else original_response | |
}, | |
"litellm_call_id": litellm_call_id, | |
"user_email": self.user_email, | |
} | |
print_verbose(f"litedebugger post-call data object - {litellm_data_obj}") | |
response = requests.post( | |
url=self.api_url, | |
headers={"content-type": "application/json"}, | |
data=json.dumps(litellm_data_obj), | |
) | |
print_verbose(f"LiteDebugger: api response - {response.text}") | |
except: | |
print_verbose( | |
f"[Non-Blocking Error] LiteDebugger: Logging Error - {traceback.format_exc()}" | |
) | |
def log_event( | |
self, | |
end_user, | |
response_obj, | |
start_time, | |
end_time, | |
litellm_call_id, | |
print_verbose, | |
call_type, | |
stream=False, | |
): | |
print_verbose( | |
f"LiteDebugger: Success/Failure Call Logging for call id {litellm_call_id}" | |
) | |
try: | |
print_verbose( | |
f"LiteLLMDebugger: Success/Failure Logging - Enters handler logging function for function {call_type} and stream set to {stream} with response object {response_obj}" | |
) | |
total_cost = 0 # [TODO] implement cost tracking | |
response_time = (end_time - start_time).total_seconds() | |
if call_type == "completion" and stream == False: | |
litellm_data_obj = { | |
"response_time": response_time, | |
"total_cost": total_cost, | |
"response": response_obj["choices"][0]["message"]["content"], | |
"litellm_call_id": litellm_call_id, | |
"status": "success", | |
} | |
print_verbose( | |
f"LiteDebugger: Logging - final data object: {litellm_data_obj}" | |
) | |
response = requests.post( | |
url=self.api_url, | |
headers={"content-type": "application/json"}, | |
data=json.dumps(litellm_data_obj), | |
) | |
elif call_type == "embedding": | |
litellm_data_obj = { | |
"response_time": response_time, | |
"total_cost": total_cost, | |
"response": str(response_obj["data"][0]["embedding"][:5]), | |
"litellm_call_id": litellm_call_id, | |
"status": "success", | |
} | |
response = requests.post( | |
url=self.api_url, | |
headers={"content-type": "application/json"}, | |
data=json.dumps(litellm_data_obj), | |
) | |
elif call_type == "completion" and stream == True: | |
if len(response_obj["content"]) > 0: # don't log the empty strings | |
litellm_data_obj = { | |
"response_time": response_time, | |
"total_cost": total_cost, | |
"response": response_obj["content"], | |
"litellm_call_id": litellm_call_id, | |
"status": "success", | |
} | |
print_verbose( | |
f"LiteDebugger: Logging - final data object: {litellm_data_obj}" | |
) | |
response = requests.post( | |
url=self.api_url, | |
headers={"content-type": "application/json"}, | |
data=json.dumps(litellm_data_obj), | |
) | |
elif "error" in response_obj: | |
if "Unable to map your input to a model." in response_obj["error"]: | |
total_cost = 0 | |
litellm_data_obj = { | |
"response_time": response_time, | |
"model": response_obj["model"], | |
"total_cost": total_cost, | |
"error": response_obj["error"], | |
"end_user": end_user, | |
"litellm_call_id": litellm_call_id, | |
"status": "failure", | |
"user_email": self.user_email, | |
} | |
print_verbose( | |
f"LiteDebugger: Logging - final data object: {litellm_data_obj}" | |
) | |
response = requests.post( | |
url=self.api_url, | |
headers={"content-type": "application/json"}, | |
data=json.dumps(litellm_data_obj), | |
) | |
print_verbose(f"LiteDebugger: api response - {response.text}") | |
except: | |
print_verbose( | |
f"[Non-Blocking Error] LiteDebugger: Logging Error - {traceback.format_exc()}" | |
) | |
pass | |