Spaces:

ka1kuk
/

litellm

Running

App Files Files Community

litellm / litellm /integrations /custom_logger.py

ka1kuk

Upload 235 files

7db0ae4 verified 5 months ago

raw history blame contribute delete

No virus

4.18 kB

	#### What this does ####
	# On success, logs events to Promptlayer
	import dotenv, os
	import requests
	from litellm.proxy._types import UserAPIKeyAuth
	from litellm.caching import DualCache
	from typing import Literal

	dotenv.load_dotenv() # Loading env variables using dotenv
	import traceback


	class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callback#callback-class
	# Class variables or attributes
	def __init__(self):
	pass

	def log_pre_api_call(self, model, messages, kwargs):
	pass

	def log_post_api_call(self, kwargs, response_obj, start_time, end_time):
	pass

	def log_stream_event(self, kwargs, response_obj, start_time, end_time):
	pass

	def log_success_event(self, kwargs, response_obj, start_time, end_time):
	pass

	def log_failure_event(self, kwargs, response_obj, start_time, end_time):
	pass

	#### ASYNC ####

	async def async_log_stream_event(self, kwargs, response_obj, start_time, end_time):
	pass

	async def async_log_pre_api_call(self, model, messages, kwargs):
	pass

	async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
	pass

	async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
	pass

	#### CALL HOOKS - proxy only ####
	"""
	Control the modify incoming / outgoung data before calling the model
	"""

	async def async_pre_call_hook(
	self,
	user_api_key_dict: UserAPIKeyAuth,
	cache: DualCache,
	data: dict,
	call_type: Literal["completion", "embeddings"],
	):
	pass

	async def async_post_call_failure_hook(
	self, original_exception: Exception, user_api_key_dict: UserAPIKeyAuth
	):
	pass

	#### SINGLE-USE #### - https://docs.litellm.ai/docs/observability/custom_callback#using-your-custom-callback-function

	def log_input_event(self, model, messages, kwargs, print_verbose, callback_func):
	try:
	kwargs["model"] = model
	kwargs["messages"] = messages
	kwargs["log_event_type"] = "pre_api_call"
	callback_func(
	kwargs,
	)
	print_verbose(f"Custom Logger - model call details: {kwargs}")
	except:
	traceback.print_exc()
	print_verbose(f"Custom Logger Error - {traceback.format_exc()}")

	async def async_log_input_event(
	self, model, messages, kwargs, print_verbose, callback_func
	):
	try:
	kwargs["model"] = model
	kwargs["messages"] = messages
	kwargs["log_event_type"] = "pre_api_call"
	await callback_func(
	kwargs,
	)
	print_verbose(f"Custom Logger - model call details: {kwargs}")
	except:
	traceback.print_exc()
	print_verbose(f"Custom Logger Error - {traceback.format_exc()}")

	def log_event(
	self, kwargs, response_obj, start_time, end_time, print_verbose, callback_func
	):
	# Method definition
	try:
	kwargs["log_event_type"] = "post_api_call"
	callback_func(
	kwargs, # kwargs to func
	response_obj,
	start_time,
	end_time,
	)
	print_verbose(f"Custom Logger - final response object: {response_obj}")
	except:
	# traceback.print_exc()
	print_verbose(f"Custom Logger Error - {traceback.format_exc()}")
	pass

	async def async_log_event(
	self, kwargs, response_obj, start_time, end_time, print_verbose, callback_func
	):
	# Method definition
	try:
	kwargs["log_event_type"] = "post_api_call"
	await callback_func(
	kwargs, # kwargs to func
	response_obj,
	start_time,
	end_time,
	)
	print_verbose(f"Custom Logger - final response object: {response_obj}")
	except:
	# traceback.print_exc()
	print_verbose(f"Custom Logger Error - {traceback.format_exc()}")
	pass