수신 요청 수정 / 거부

수신 요청 수정 / 거부 (Modify / Reject Incoming Requests)

프록시에서 LLM API 호출을 수행하기 전에 데이터를 수정하거나, LLM API 호출 전/응답 반환 전에 데이터를 거부하는 방법을 알려드려요. 모든 openai 엔드포인트 호출에 'user' 파라미터를 강제할 수도 있어요.

출처: 문서

본문

tip

콜백 훅 이해하기? async_pre_call_hook 같은 프록시 특화 훅과 async_log_success_event 같은 일반 로깅 훅의 차이를 보려면 Callback Guide를 확인하세요.

어떤 훅을 사용해야 하나요? (Which Hook Should I Use?)

훅	사용 사례	실행 시점
async_pre_call_hook	모델로 보내기 전에 수신 요청 수정	LLM API 호출이 이루어지기 전
async_moderation_hook	LLM API 호출과 병렬로 입력 확인 실행	LLM API 호출과 병행
async_post_call_success_hook	나가는 응답 수정 (비스트리밍)	성공적인 LLM API 호출 이후, 비스트리밍 응답
async_post_call_failure_hook	클라이언트로 보내는 오류 응답 변환	실패한 LLM API 호출 이후
async_post_call_streaming_hook	나가는 응답 수정 (스트리밍)	성공적인 LLM API 호출 이후, 스트리밍 응답
async_post_call_response_headers_hook	커스텀 HTTP 응답 헤더 주입	LLM API 호출 이후 (성공과 실패 모두)

병렬 요청 요율 제한기로 완전한 예시를 보세요.

빠른 시작 (Quick Start)

Custom Handler에 새로운 async_pre_call_hook 함수를 추가해요.

이 함수는 litellm completion 호출이 이루어지기 직전에 호출되며, litellm 호출로 들어가는 데이터를 수정할 수 있게 해줘요. 코드 보기

from litellm.integrations.custom_logger import CustomLogger
import litellm
from litellm.proxy.proxy_server import UserAPIKeyAuth, DualCache
from litellm.types.utils import ModelResponseStream
from typing import Any, AsyncGenerator, Optional, Literal

# This file includes the custom callbacks for LiteLLM Proxy
# Once defined, these can be passed in proxy_config.yaml

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

    #### CALL HOOKS - proxy only ####

    async def async_pre_call_hook(self, user_api_key_dict: UserAPIKeyAuth, cache: DualCache, data: dict, call_type: Literal[
            "completion",
            "text_completion",
            "embeddings",
            "image_generation",
            "moderation",
            "audio_transcription",
        ]):
        data["model"] = "my-new-model"
        return data

    async def async_post_call_failure_hook(
        self,
        request_data: dict,
        original_exception: Exception,
        user_api_key_dict: UserAPIKeyAuth,
        traceback_str: Optional[str] = None,
    ) -> Optional[HTTPException]:
        """
        Transform error responses sent to clients.

        Return an HTTPException to replace the original error with a user-friendly message.
        Return None to use the original exception.

        Example:
            if isinstance(original_exception, litellm.ContextWindowExceededError):
                return HTTPException(
                    status_code=400,
                    detail="Your prompt is too long. Please reduce the length and try again."
                )
            return None  # Use original exception
        """
        pass

    async def async_post_call_success_hook(
        self,
        data: dict,
        user_api_key_dict: UserAPIKeyAuth,
        response,
    ):
        pass

    async def async_moderation_hook( # call made in parallel to llm api call
        self,
        ...

마지막 줄이 중요해요: callbacks는 인스턴스의 dotted path를 받으므로, 파일이 인스턴스를 하나 만들어야 해요.

이 파일을 프록시 콘피그에 추가하세요:

model_list:
  - model_name: gpt-5.6-luna
    litellm_params:
      model: gpt-5.6-luna
litellm_settings:
  callbacks: custom_callbacks.proxy_handler_instance # sets litellm.callbacks = [proxy_handler_instance]

warning

callbacks를 클래스(custom_callbacks.MyCustomHandler)가 아니라 인스턴스로 가리켜야 해요. 클래스로 가리키면 프록시가 콘피그 로드에 실패하며 해당 항목과 해석된 대상을 명명하는 오류를 냅니다. 프록시는 CustomLogger 인스턴스만 디스패치하므로, 그 검사 이전 버전에서는 오류나 로그 줄 없이 깨끗하게 시작되어 트래픽을 서빙하고 훅은 절대 실행하지 않았어요.

서버 시작 + 요청 테스트:

$ litellm /path/to/config.yaml

curl --location 'http://0.0.0.0:4000/chat/completions' \
    --data ' {
    "model": "gpt-5.6-luna",
    "messages": [
        {
        "role": "user",
        "content": "good morning good sir"
        }
    ],
    "user": "ishaan-app",
    "temperature": 0.2
    }'

[BETA] 새로운 async_moderation_hook

실제 LLM API 호출과 병렬로 중재 검사를 실행해요.

CustomGuardrail을 서브클래싱하고 async_moderation_hook 함수를 정의하세요.

guardrails: 아래 mode: during_call로 가드레일을 등록하세요. 훅은 data, user_api_key_dict, call_type을 받아야 해요. 옛 두 인자 시그니처는 매 요청에서 TypeError로 실패해요.

이 함수는 실제 LLM API 호출과 병렬로 실행돼요.

async_moderation_hook이 Exception을 발생시키면 이를 사용자에게 반환해요.

Llama Guard 콘텐츠 중재 훅과 custom guardrail 문서에서 완전한 예시를 보세요.

from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.utils import CallTypesLiteral
from fastapi import HTTPException

class MyCustomGuardrail(CustomGuardrail):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    async def async_moderation_hook( ### 👈 KEY CHANGE ###
        self,
        data: dict,
        user_api_key_dict: UserAPIKeyAuth,
        call_type: CallTypesLiteral,
    ):
        messages = data["messages"]
        print(messages)
        if messages[0]["content"] == "hello world":
            raise HTTPException(
                    status_code=400, detail={"error": "Violated content safety policy"}
                )

이 파일을 프록시 콘피그에 추가하세요:

model_list:
  - model_name: gpt-5.6-luna
    litellm_params:
      model: gpt-5.6-luna
guardrails:
  - guardrail_name: "my-moderation-guardrail"
    litellm_params:
      guardrail: custom_guardrail.MyCustomGuardrail # {file_name}.{class_name}
      mode: "during_call"
      default_on: true

서버 시작 + 요청 테스트:

$ litellm /path/to/config.yaml

curl --location 'http://0.0.0.0:4000/chat/completions' \
    --data ' {
    "model": "gpt-5.6-luna",
    "messages": [
        {
        "role": "user",
        "content": "Hello world"
        }
    ],
    }'

고급 - 'user' 파라미터 강제 (Advanced - Enforce 'user' param)

enforce_user_param을 true로 설정하면 openai 엔드포인트에 대한 모든 호출에 'user' 파라미터가 필요해져요.

코드 보기

general_settings:
  enforce_user_param: True

결과 (Result)

고급 - 거부 메시지를 응답으로 반환 (Advanced - Return rejected message as response)

chat completions과 text completion 호출의 경우, 거부 메시지를 사용자 응답으로 반환할 수 있어요.

문자열을 반환하면 돼요. LiteLLM이 엔드포인트와 스트리밍 여부에 따라 올바른 형식으로 응답을 반환해요.

비-chat/text completion 엔드포인트에서는 이 응답이 400 상태 코드 예외로 반환돼요.

1. Custom Handler 생성

from litellm.integrations.custom_logger import CustomLogger
import litellm
from litellm.utils import get_formatted_prompt
# This file includes the custom callbacks for LiteLLM Proxy
# Once defined, these can be passed in proxy_config.yaml

class MyCustomHandler(CustomLogger):
    def __init__(self):
        pass

    #### CALL HOOKS - proxy only ####

    async def async_pre_call_hook(self, user_api_key_dict: UserAPIKeyAuth, cache: DualCache, data: dict, call_type: Literal[
            "completion",
            "text_completion",
            "embeddings",
            "image_generation",
            "moderation",
            "audio_transcription",
        ]) -> Optional[dict, str, Exception]:
        formatted_prompt = get_formatted_prompt(data=data, call_type=call_type)
        if "Hello world" in formatted_prompt:
            return "This is an invalid response"
        return data

proxy_handler_instance = MyCustomHandler()

2. config.yaml 업데이트

model_list:
  - model_name: gpt-5.6-luna
    litellm_params:
      model: gpt-5.6-luna
litellm_settings:
  callbacks: custom_callbacks.proxy_handler_instance # sets litellm.callbacks = [proxy_handler_instance]

3. 테스트!

$ litellm /path/to/config.yaml

curl --location 'http://0.0.0.0:4000/chat/completions' \
    --data ' {
    "model": "gpt-5.6-luna",
    "messages": [
        {
        "role": "user",
        "content": "Hello world"
        }
    ],
    }'

예상 응답 (Expected Response)

{
    "id": "chatcmpl-d00bbede-2d90-4618-bf7b-11a1c23cf360",
    "choices": [
        {
            "finish_reason": "stop",
            "index": 0,
            "message": {
                "content": "This is an invalid response.", # 👈 REJECTED RESPONSE
                "role": "assistant"
            }
        }
    ],
    "created": 1716234198,
    "model": null,
    "object": "chat.completion",
    "system_fingerprint": null,
    "usage": {}
}

고급 - 오류 응답 변환 (Advanced - Transform Error Responses)

async_post_call_failure_hook을 사용해 기술적인 API 오류를 사용자 친화적 메시지로 변환해요. 원래 오류를 교체하려면 HTTPException을 반환하고, 원래 예외를 사용하려면 None을 반환하세요.

from litellm.integrations.custom_logger import CustomLogger
from fastapi import HTTPException
from typing import Optional
import litellm

class MyErrorTransformer(CustomLogger):
    async def async_post_call_failure_hook(
        self,
        request_data: dict,
        original_exception: Exception,
        user_api_key_dict: UserAPIKeyAuth,
        traceback_str: Optional[str] = None,
    ) -> Optional[HTTPException]:
        if isinstance(original_exception, litellm.ContextWindowExceededError):
            return HTTPException(
                status_code=400,
                detail="Your prompt is too long. Please reduce the length and try again."
            )
        if isinstance(original_exception, litellm.RateLimitError):
            return HTTPException(
                status_code=429,
                detail="Rate limit exceeded. Please try again in a moment."
            )
        return None  # Use original exception

proxy_handler_instance = MyErrorTransformer()

결과: 클라이언트는 "ContextWindowExceededError: Prompt exceeds context window" 대신 "Your prompt is too long..."를 받아요.

고급 - 커스텀 HTTP 응답 헤더 주입 (Advanced - Inject Custom HTTP Response Headers)

async_post_call_response_headers_hook을 사용해 응답에 커스텀 HTTP 헤더를 주입해요. 이 훅은 성공과 실패한 LLM API 호출 모두에서 실행돼요.

from litellm.integrations.custom_logger import CustomLogger
from litellm.proxy.proxy_server import UserAPIKeyAuth
from typing import Any, Dict, Optional

class CustomHeaderLogger(CustomLogger):
    def __init__(self):
        super().__init__()

    async def async_post_call_response_headers_hook(
        self,
        data: dict,
        user_api_key_dict: UserAPIKeyAuth,
        response: Any,
        request_headers: Optional[Dict[str, str]] = None,
    ) -> Optional[Dict[str, str]]:
        """
        Inject custom headers into all responses (success and failure).
        """
        return {"x-custom-header": "custom-value"}

proxy_handler_instance = CustomHeaderLogger()