跳至主要內容

自訂回呼

資訊

適用於 PROXY 前往此處

回呼類別​

您可以建立自訂回呼類別，以精確地記錄 litellm 中發生的事件。

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-5.6-luna", messages=[{ "role": "user", "content": "Hi 👋 - i'm openai"}],
                              stream=True)
for chunk in response: 
    continue


## async
import asyncio 

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

常見 Hook​

  • async_log_success_event - 記錄成功的 API 呼叫
  • async_log_failure_event - 記錄失敗的 API 呼叫
  • log_pre_api_call - 在 API 呼叫前記錄
  • log_post_api_call - 在 API 呼叫後記錄

僅限 Proxy 的 hooks（僅適用於 LiteLLM Proxy）：

  • async_post_call_success_hook - 存取使用者資料 + 修改回應
  • async_pre_call_hook - 在送出前修改請求

每次嘗試的部署回呼​

async_log_success_event 和 async_log_failure_event 被文件說明為請求層級回呼：每個都 նախատես是每個邏輯用戶端請求觸發一次。下面的每次嘗試部署回呼則是完全不同的契約：每一個都保證會針對每一次實際的部署呼叫觸發一次，包括原始嘗試、每次重試，以及每個備援步驟，且不涉及任何去重邏輯。當您需要的是每次部署嘗試的訊號，而不是每個邏輯請求的訊號時，請使用這些，而不是請求層級回呼；以每個部署的 Prometheus 風格計數器，或是以個別嘗試失敗為鍵的 circuit breaker，正是它們的用途。

這些在 SDK 和 proxy 中都可使用：

  • async_pre_call_deployment_hook(kwargs, call_type) - 在每次部署呼叫前執行，可修改請求
  • async_post_call_success_deployment_hook(request_data, response, call_type) - 在每次成功的部署呼叫後執行
  • async_post_call_failure_deployment_hook(request_data, exception, call_type, fallback_depth=None) - 在每次失敗的部署呼叫後執行

當第一個和第二個部署都失敗、第三個成功時，備援鏈會依序呼叫失敗回呼兩次與成功回呼一次。request_data 是該次嘗試自身的 request kwargs，而不是從前一次嘗試沿用的共享狀態。fallback_depth 是失敗回呼上的 best-effort 欄位：None 用於原始嘗試，1 用於第一個備援跳轉，2 用於第二個，依此類推。

from litellm.integrations.custom_logger import CustomLogger

class DeploymentFailureCounter(CustomLogger):
    def __init__(self):
        super().__init__()
        self.failures_by_model = {}

    async def async_post_call_failure_deployment_hook(self, request_data, exception, call_type, fallback_depth=None):
        model = request_data.get("model", "unknown")
        self.failures_by_model[model] = self.failures_by_model.get(model, 0) + 1
        print(f"deployment failure: model={model} exception={type(exception).__name__} fallback_depth={fallback_depth} total={self.failures_by_model[model]}")

counter = DeploymentFailureCounter()
litellm.callbacks = [counter]

範例：在 async_post_call_success_hook 中修改回應​

您可以使用 async_post_call_success_hook 在回應傳回給用戶端之前，為其新增自訂標頭或中繼資料。範例如下：

async def async_post_call_success_hook(data, user_api_key_dict, response):
    # Add a custom header to the response
    additional_headers = getattr(response, "_hidden_params", {}).get("additional_headers", {}) or {}
    additional_headers["x-litellm-custom-header"] = "my-value"
    if not hasattr(response, "_hidden_params"):
        response._hidden_params = {}
    response._hidden_params["additional_headers"] = additional_headers
    return response

這可讓您將自訂中繼資料或標頭注入回應中，供下游消費者使用。您可以使用此模式將資訊傳遞給用戶端、Proxy 或可觀測性工具。

回呼函式​

如果您只想在特定事件（例如輸入時）記錄，則可以使用回呼函式。

您可以設定在以下情況觸發的自訂回呼：

  • litellm.input_callback - 在執行 LLM API 請求前追蹤輸入/轉換後的輸入
  • litellm.success_callback - 在執行 LLM API 請求後追蹤輸入/輸出
  • litellm.failure_callback - 追蹤 litellm 呼叫的輸入/輸出 + 例外狀況

定義自訂回呼函式​

建立一個接受特定引數的自訂回呼函式：

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)

設定自訂回呼函式​

import litellm
litellm.success_callback = [custom_callback]

使用您的自訂回呼函式​

import litellm
from litellm import completion

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

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

print(response)

非同步回呼函式​

我們建議在非同步情境使用 Custom Logger 類別。

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]

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

函式

如果您只想傳入一個非同步函式來進行記錄。

LiteLLM 目前僅支援用於非同步 completion/embedding 呼叫的非同步成功回呼函式。

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-5.6-luna",
                              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())

kwargs 中有哪些可用內容？​

kwargs 字典包含您 API 呼叫的所有詳細資訊。

資訊

如需完整的記錄負載規格，請參閱 標準記錄負載規格。

def custom_callback(kwargs, completion_response, start_time, end_time):
    # Access common data
    model = kwargs.get("model")
    messages = kwargs.get("messages", [])
    cost = kwargs.get("response_cost", 0)
    cache_hit = kwargs.get("cache_hit", False)
    
    # Access metadata you passed in
    metadata = kwargs.get("litellm_params", {}).get("metadata", {})

kwargs 中的關鍵欄位：

  • model - 模型名稱
  • messages - 輸入訊息
  • response_cost - 計算成本
  • cache_hit - 回應是否已被快取
  • litellm_params.metadata - 您的自訂中繼資料

實用範例​

追蹤 API 成本​

def track_cost_callback(kwargs, completion_response, start_time, end_time):
    cost = kwargs["response_cost"] # litellm calculates this for you
    print(f"Request cost: ${cost}")

litellm.success_callback = [track_cost_callback]

response = completion(model="gpt-5.6-luna", messages=[{"role": "user", "content": "Hello"}])

記錄輸入到 LLM​

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]

response = completion(model="claude-sonnet-5", messages=[{"role": "user", "content": "Hello"}])

傳送至外部服務​

import requests

def send_to_analytics(kwargs, completion_response, start_time, end_time):
    data = {
        "model": kwargs.get("model"),
        "cost": kwargs.get("response_cost", 0),
        "duration": (end_time - start_time).total_seconds()
    }
    requests.post("https://your-analytics.com/api", json=data)

litellm.success_callback = [send_to_analytics]

常見問題​

未呼叫回呼​

請確認您已：

  1. 正確註冊回呼：litellm.callbacks = [MyHandler()]
  2. 使用正確的 hook 名稱（檢查拼字）
  3. 不要在函式庫模式中使用僅限 Proxy 的 hooks

效能問題​

  • 對 I/O 作業使用非同步 hooks
  • 不要在回呼函式中阻塞
  • 正確處理例外狀況：
class SafeHandler(CustomLogger):
    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
        try:
            await external_service(response_obj)
        except Exception as e:
            print(f"Callback error: {e}")  # Log but don't break the flow
LiteLLM Enterprise
為正式環境打造的 SSO/SAML、稽核記錄、支出追蹤、多團隊管理與防護欄。
深入瞭解 →