Benutzerdefinierte Callbacks

Info

FÜR PROXY Hier klicken

Callback-Klasse

Sie können eine benutzerdefinierte Callback-Klasse erstellen, um Ereignisse präzise zu protokollieren, sobald sie in litellm auftreten.

import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm import completion, acompletion

class MyCustomHandler(CustomLogger):
    def log_pre_api_call(self, model, messages, kwargs): 
        print(f"Pre-API Call")
    
    def log_post_api_call(self, kwargs, response_obj, start_time, end_time): 
        print(f"Post-API Call")
    

    def log_success_event(self, kwargs, response_obj, start_time, end_time): 
        print(f"On Success")

    def log_failure_event(self, kwargs, response_obj, start_time, end_time): 
        print(f"On Failure")
    
    #### ASYNC #### - for acompletion/aembeddings

    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
        print(f"On Async Success")

    async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
        print(f"On Async Failure")

customHandler = MyCustomHandler()

litellm.callbacks = [customHandler]

## sync 
response = completion(model="gpt-3.5-turbo", messages=[{ "role": "user", "content": "Hi 👋 - i'm openai"}],
                              stream=True)
for chunk in response: 
    continue


## async
import asyncio 

def async completion():
    response = await acompletion(model="gpt-3.5-turbo", messages=[{ "role": "user", "content": "Hi 👋 - i'm openai"}],
                              stream=True)
    async for chunk in response: 
        continue
asyncio.run(completion())

Callback-Funktionen

Wenn Sie nur bei einem bestimmten Ereignis protokollieren möchten (z. B. bei Eingaben) - können Sie Callback-Funktionen verwenden.

Sie können benutzerdefinierte Callbacks festlegen, die ausgelöst werden für

• litellm.input_callback - Eingaben/transformierte Eingaben verfolgen, bevor der LLM-API-Aufruf erfolgt
• litellm.success_callback - Eingaben/Ausgaben verfolgen, nachdem der LLM-API-Aufruf erfolgt ist
• litellm.failure_callback - Eingaben/Ausgaben + Ausnahmen für litellm-Aufrufe verfolgen

Definieren einer benutzerdefinierten Callback-Funktion

Erstellen Sie eine benutzerdefinierte Callback-Funktion, die spezifische Argumente entgegennimmt

def custom_callback(
    kwargs,                 # kwargs to completion
    completion_response,    # response from completion
    start_time, end_time    # start/end time
):
    # Your custom code here
    print("LITELLM: in custom callback function")
    print("kwargs", kwargs)
    print("completion_response", completion_response)
    print("start_time", start_time)
    print("end_time", end_time)

Festlegen der benutzerdefinierten Callback-Funktion

import litellm
litellm.success_callback = [custom_callback]

Verwenden Ihrer benutzerdefinierten Callback-Funktion

import litellm
from litellm import completion

# Assign the custom callback function
litellm.success_callback = [custom_callback]

response = completion(
    model="gpt-3.5-turbo",
    messages=[
        {
            "role": "user",
            "content": "Hi 👋 - i'm openai"
        }
    ]
)

print(response)

Asynchrone Callback-Funktionen

Wir empfehlen die Verwendung der benutzerdefinierten Logger-Klasse für asynchrone Vorgänge.

from litellm.integrations.custom_logger import CustomLogger
from litellm import acompletion 

class MyCustomHandler(CustomLogger):
    #### ASYNC #### 
    


    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
        print(f"On Async Success")

    async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
        print(f"On Async Failure")

import asyncio 
customHandler = MyCustomHandler()

litellm.callbacks = [customHandler]

def async completion():
    response = await acompletion(model="gpt-3.5-turbo", messages=[{ "role": "user", "content": "Hi 👋 - i'm openai"}],
                              stream=True)
    async for chunk in response: 
        continue
asyncio.run(completion())

Funktionen

Wenn Sie nur eine asynchrone Funktion für die Protokollierung übergeben möchten.

LiteLLM unterstützt derzeit nur asynchrone Erfolgs-Callback-Funktionen für asynchrone Abschluss-/Embedding-Aufrufe.

import asyncio, litellm 

async def async_test_logging_fn(kwargs, completion_obj, start_time, end_time):
    print(f"On Async Success!")

async def test_chat_openai():
    try:
        # litellm.set_verbose = True
        litellm.success_callback = [async_test_logging_fn]
        response = await litellm.acompletion(model="gpt-3.5-turbo",
                              messages=[{
                                  "role": "user",
                                  "content": "Hi 👋 - i'm openai"
                              }],
                              stream=True)
        async for chunk in response: 
            continue
    except Exception as e:
        print(e)
        pytest.fail(f"An error occurred - {str(e)}")

asyncio.run(test_chat_openai())

Info

Wir arbeiten aktiv daran, dies auf andere Ereignistypen auszuweiten. Sagen Sie uns Bescheid, wenn Sie dies benötigen!

Was ist in kwargs?

Beachten Sie, dass wir ein kwargs-Argument an den benutzerdefinierten Callback übergeben.

def custom_callback(
    kwargs,                 # kwargs to completion
    completion_response,    # response from completion
    start_time, end_time    # start/end time
):
    # Your custom code here
    print("LITELLM: in custom callback function")
    print("kwargs", kwargs)
    print("completion_response", completion_response)
    print("start_time", start_time)
    print("end_time", end_time)

Dies ist ein Dictionary, das alle Details des Modellaufrufs enthält (die Parameter, die wir erhalten, die Werte, die wir an den HTTP-Endpunkt senden, die Antwort, die wir erhalten, den Stacktrace im Fehlerfall usw.).

Dies wird alles im model_call_details über unseren Logger protokolliert.

Hier sehen Sie genau, was Sie im kwargs-Dictionary erwarten können

### DEFAULT PARAMS ### 
"model": self.model,
"messages": self.messages,
"optional_params": self.optional_params, # model-specific params passed in
"litellm_params": self.litellm_params, # litellm-specific params passed in (e.g. metadata passed to completion call)
"start_time": self.start_time, # datetime object of when call was started

### PRE-API CALL PARAMS ### (check via kwargs["log_event_type"]="pre_api_call")
"input" = input # the exact prompt sent to the LLM API
"api_key" = api_key # the api key used for that LLM API 
"additional_args" = additional_args # any additional details for that API call (e.g. contains optional params sent)

### POST-API CALL PARAMS ### (check via kwargs["log_event_type"]="post_api_call")
"original_response" = original_response # the original http response received (saved via response.text)

### ON-SUCCESS PARAMS ### (check via kwargs["log_event_type"]="successful_api_call")
"complete_streaming_response" = complete_streaming_response # the complete streamed response (only set if `completion(..stream=True)`)
"end_time" = end_time # datetime object of when call was completed

### ON-FAILURE PARAMS ### (check via kwargs["log_event_type"]="failed_api_call")
"exception" = exception # the Exception raised
"traceback_exception" = traceback_exception # the traceback generated via `traceback.format_exc()`
"end_time" = end_time # datetime object of when call was completed

Cache-Treffer

Cache-Treffer werden bei Erfolgsereignissen als kwarg["cache_hit"] protokolliert.

Hier ist ein Beispiel für den Zugriff darauf

import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm import completion, acompletion, Cache

class MyCustomHandler(CustomLogger):
  async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): 
      print(f"On Success")
      print(f"Value of Cache hit: {kwargs['cache_hit']"})

async def test_async_completion_azure_caching():
  customHandler_caching = MyCustomHandler()
  litellm.cache = Cache(type="redis", host=os.environ['REDIS_HOST'], port=os.environ['REDIS_PORT'], password=os.environ['REDIS_PASSWORD'])
  litellm.callbacks = [customHandler_caching]
  unique_time = time.time()
  response1 = await litellm.acompletion(model="azure/chatgpt-v-2",
                          messages=[{
                              "role": "user",
                              "content": f"Hi 👋 - i'm async azure {unique_time}"
                          }],
                          caching=True)
  await asyncio.sleep(1)
  print(f"customHandler_caching.states pre-cache hit: {customHandler_caching.states}")
  response2 = await litellm.acompletion(model="azure/chatgpt-v-2",
                          messages=[{
                              "role": "user",
                              "content": f"Hi 👋 - i'm async azure {unique_time}"
                          }],
                          caching=True)
  await asyncio.sleep(1) # success callbacks are done in parallel
  print(f"customHandler_caching.states post-cache hit: {customHandler_caching.states}")
  assert len(customHandler_caching.errors) == 0
  assert len(customHandler_caching.states) == 4 # pre, post, success, success

Vollständige Streaming-Antwort erhalten

LiteLLM übergibt Ihnen die vollständige Streaming-Antwort im letzten Streaming-Chunk als Teil der kwargs für Ihre benutzerdefinierte Callback-Funktion.

# litellm.set_verbose = False
        def custom_callback(
            kwargs,                 # kwargs to completion
            completion_response,    # response from completion
            start_time, end_time    # start/end time
        ):
            # print(f"streaming response: {completion_response}")
            if "complete_streaming_response" in kwargs: 
                print(f"Complete Streaming Response: {kwargs['complete_streaming_response']}")
        
        # Assign the custom callback function
        litellm.success_callback = [custom_callback]

        response = completion(model="claude-instant-1", messages=messages, stream=True)
        for idx, chunk in enumerate(response): 
            pass

Zusätzliche Metadaten protokollieren

LiteLLM akzeptiert ein Metadaten-Dictionary im Completion-Aufruf. Sie können zusätzliche Metadaten über completion(..., metadata={"key": "value"}) an Ihren Completion-Aufruf übergeben.

Da dies ein litellm-spezifischer Parameter ist, ist er über kwargs zugänglich["litellm_params"]

from litellm import completion
import os, litellm

## set ENV variables
os.environ["OPENAI_API_KEY"] = "your-api-key"

messages = [{ "content": "Hello, how are you?","role": "user"}]

def custom_callback(
    kwargs,                 # kwargs to completion
    completion_response,    # response from completion
    start_time, end_time    # start/end time
):
    print(kwargs["litellm_params"]["metadata"])
    

# Assign the custom callback function
litellm.success_callback = [custom_callback]

response = litellm.completion(model="gpt-3.5-turbo", messages=messages, metadata={"hello": "world"})

Beispiele

Benutzerdefinierter Callback zur Nachverfolgung von Kosten für Streaming + Nicht-Streaming

Standardmäßig sind die Kosten der Antwort im Protokollierungsobjekt über kwargs["response_cost"] im Erfolgsfall zugänglich (synchron + asynchron)

# Step 1. Write your custom callback function
def track_cost_callback(
    kwargs,                 # kwargs to completion
    completion_response,    # response from completion
    start_time, end_time    # start/end time
):
    try:
        response_cost = kwargs["response_cost"] # litellm calculates response cost for you
        print("regular response_cost", response_cost)
    except:
        pass

# Step 2. Assign the custom callback function
litellm.success_callback = [track_cost_callback]

# Step 3. Make litellm.completion call
response = completion(
    model="gpt-3.5-turbo",
    messages=[
        {
            "role": "user",
            "content": "Hi 👋 - i'm openai"
        }
    ]
)

print(response)

Benutzerdefinierter Callback zur Protokollierung transformierter Eingaben an LLMs

def get_transformed_inputs(
    kwargs,
):
    params_to_model = kwargs["additional_args"]["complete_input_dict"]
    print("params to model", params_to_model)

litellm.input_callback = [get_transformed_inputs]

def test_chat_openai():
    try:
        response = completion(model="claude-2",
                              messages=[{
                                  "role": "user",
                                  "content": "Hi 👋 - i'm openai"
                              }])

        print(response)

    except Exception as e:
        print(e)
        pass

Ausgabe

params to model {'model': 'claude-2', 'prompt': "\n\nHuman: Hi 👋 - i'm openai\n\nAssistant: ", 'max_tokens_to_sample': 256}

Benutzerdefinierter Callback zur Protokollierung in Mixpanel

import mixpanel
import litellm
from litellm import completion

def custom_callback(
    kwargs,                 # kwargs to completion
    completion_response,    # response from completion
    start_time, end_time    # start/end time
):
    # Your custom code here
    mixpanel.track("LLM Response", {"llm_response": completion_response})


# Assign the custom callback function
litellm.success_callback = [custom_callback]

response = completion(
    model="gpt-3.5-turbo",
    messages=[
        {
            "role": "user",
            "content": "Hi 👋 - i'm openai"
        }
    ]
)

print(response)