phoenix/litellm-mirror
0
0
Fork 1
mirror of https://github.com/BerriAI/litellm.git synced 2025-04-25 18:54:30 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(proxy_server.py): fix pydantic version errors

Browse source
This commit is contained in:
Krrish Dholakia 2023-12-09 12:09:49 -08:00
parent 0294e1119e
commit ed50522863
5 changed files with 36 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

28
litellm/proxy/custom_callbacks.py
View file

@ -4,44 +4,48 @@ import inspect
# This file includes the custom callbacks for LiteLLM Proxy
# Once defined, these can be passed in proxy_config.yaml
def print_verbose(print_statement):
if litellm.set_verbose:
print(print_statement) # noqa
class MyCustomHandler(CustomLogger):
def __init__(self):
blue_color_code = "\033[94m"
reset_color_code = "\033[0m"
print(f"{blue_color_code}Initialized LiteLLM custom logger")
print_verbose(f"{blue_color_code}Initialized LiteLLM custom logger")
try:
print(f"Logger Initialized with following methods:")
print_verbose(f"Logger Initialized with following methods:")
methods = [method for method in dir(self) if inspect.ismethod(getattr(self, method))]
# Pretty print the methods
# Pretty print_verbose the methods
for method in methods:
print(f" - {method}")
print(f"{reset_color_code}")
print_verbose(f" - {method}")
print_verbose(f"{reset_color_code}")
except:
pass
def log_pre_api_call(self, model, messages, kwargs):
print(f"Pre-API Call")
print_verbose(f"Pre-API Call")
def log_post_api_call(self, kwargs, response_obj, start_time, end_time):
print(f"Post-API Call")
print_verbose(f"Post-API Call")
def log_stream_event(self, kwargs, response_obj, start_time, end_time):
print(f"On Stream")
print_verbose(f"On Stream")
def log_success_event(self, kwargs, response_obj, start_time, end_time):
print("On Success!")
print_verbose("On Success!")
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
print(f"On Async Success!")
print_verbose(f"On Async Success!")
return
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
try:
print(f"On Async Failure !")
print_verbose(f"On Async Failure !")
except Exception as e:
print(f"Exception: {e}")
print_verbose(f"Exception: {e}")
proxy_handler_instance = MyCustomHandler()