커스텀 콜백 (Custom Callbacks)

콜백을 "어떤 프로바이더로 보낼지" 고르는 것에서 더 나아가, "내가 직접 원하는 로직을 특정 시점에 실행"하고 싶을 때가 있어요. 커스텀 콜백은 LiteLLM에서 이벤트가 발생하는 정확한 순간에 내 코드가 실행되게 하는 방법이에요. 특히 성공·실패 이벤트를 잡아내거나, 응답에 커스텀 헤더를 심거나, 호출별 비용을 계산해 로깅하는 데 강력합니다.

출처: 공식문서

:::info 프록시용 커스텀 콜백 가이드는 프록시 로깅 문서에서 확인할 수 있어요. :::

Callback 클래스

CustomLogger를 상속한 클래스를 만들면 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="{{openai_small}}", 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="{{openai_small}}", messages=[{ "role": "user", "content": "Hi 👋 - i'm openai"}],
                              stream=True)
    async for chunk in response:
        continue
asyncio.run(completion())

자주 쓰는 훅

async_log_success_event — 성공한 API 호출 기록
async_log_failure_event — 실패한 API 호출 기록
log_pre_api_call — API 호출 전 기록
log_post_api_call — API 호출 후 기록

프록시 전용 훅(LiteLLM Proxy에서만 동작): async_post_call_success_hook(사용자 데이터 접근 + 응답 수정), async_pre_call_hook(전송 전 요청 수정).

시도 단위 배포 훅 (Per-Attempt Deployment Hooks)

async_log_success_event와 async_log_failure_event는 "클라이언트 요청 한 번에 한 번" 발화하는 요청 레벨 훅이에요. 반면 아래의 배포 훅은 계약이 완전히 달라요. 실제 배포 호출 한 번마다 정확히 한 번 발화하는 게 보장되며, 원래 시도·재시도·폴백 단계 모두를 포함하고 디듀프 로직도 없어요. "논리적 요청이 아니라 배포 시도 하나하나"에 대한 신호가 필요할 때는 요청 레벨 훅 대신 이 훅을 쓰세요. Prometheus 스타일의 배포별 카운터나, 개별 시도 실패에 걸린 회로 차단기(circuit breaker)가 정확히 이 용도예요.

이 훅들은 SDK와 프록시 양쪽에서 동작해요.

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) — 배포 호출이 실패한 뒤 실행

예를 들어 폴백 체인에서 처음 두 배포는 실패하고 세 번째가 성공한다면, 실패 훅이 2번, 성공 훅이 1번 그 순서대로 호출돼요. request_data는 그 시도 자체의 요청 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 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

이렇게 하면 아래에서 소비하는 쪽(클라이언트·프록시·관측성 도구)으로 커스텀 메타데이터나 헤더를 전달할 수 있어요.

콜백 함수 (Callbacks Functions)

특정 이벤트(예: 입력 시점)에서만 로깅하고 싶다면 콜백 함수를 쓸 수 있어요. 다음 세 가지로 트리거를 지정합니다.

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]

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", {})

model — 모델 이름
messages — 입력 메시지
response_cost — 계산된 비용
cache_hit — 응답이 캐시되었는지 여부
litellm_params.metadata — 직접 넘긴 커스텀 메타데이터

:::info 전체 로깅 페이로드 명세는 Standard Logging Payload Spec을 참고하세요. :::

실전 예제

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="{{openai_small}}", 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="{{anthropic}}", 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]

자주 겪는 문제

콜백이 호출되지 않을 때

콜백을 올바르게 등록했는지 확인 — litellm.callbacks = [MyHandler()]
훅 이름이 정확한지(오타 확인)
라이브러리 모드에서 프록시 전용 훅을 쓰지 않았는지 확인

성능 문제

I/O 작업에는 async 훅을 쓰기
콜백 함수 안에서 블로킹하지 않기
예외를 적절히 처리하기 — 콜백에서 에러가 나도 메인 흐름을 깨지 않게요.

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

더 알아보기

콜백의 개념과 외부 프로바이더 연동 목록은 콜백 문서를 봐요.
프록시 환경에서 커스텀 콜백 클래스를 쓰는 방법은 프록시 로깅을 참고하세요.
표준 로깅 페이로드 필드 전체는 StandardLoggingPayload 명세에 있어요.