LiteLLM 回调插件 10 分钟接入指南S3 日志、Prometheus 监控、内容护栏 3 个实战场景【免费下载链接】litellmThe fastest, litest AI Gateway. Rust core with Python SDK. Call 100 LLM APIs in OpenAI (or native) format with cost tracking, guardrails, load balancing, and logging [Bedrock, Azure, OpenAI, Anthropic, OpenAI, VertexAI, vLLM, Nvidia NIM]项目地址: https://gitcode.com/GitHub_Trending/li/litellm给 LLM 网关加 3 样东西请求日志落 S3、指标进 Prometheus、敏感内容在进模型前拦截。手写适配的话要维护 3 套逻辑用 LiteLLM 的回调callback与护栏guardrail体系每样只需实现 1 个类注册到litellm.callbacks即生效。它省掉的是每个工具各写一套埋点代码这件事。本文适合自研 LLM 网关、想给 LiteLLM 加可观测性或内容安全层的后端工程师。5 分钟接入路径所有插件的公共基类是 litellm/integrations/custom_logger.py 里的CustomLogger。你只重写关心的钩子不关心的方法基类里全是空实现class MyHook(CustomLogger): def log_pre_api_call(self, model, messages, kwargs): ... # 调模型 API 前 def log_stream_event(self, kwargs, response_obj, start_time, end_time): ... # 流式每块 def log_success_event(self, kwargs, response_obj, start_time, end_time): ... # 成功 def log_failure_event(self, kwargs, response_obj, start_time, end_time): ... # 失败注册有两种方式选一种即可import litellm litellm.callbacks [my_hook] # 全局之后所有调用都触发 # 或按单次调用litellm.completion(..., callbacks[my_hook])litellm/integrations/ 下有 47 个内置集成模块S3、Prometheus、Langfuse、Datadog、Slack 等都按同一套路注册无需自己写。插件生命周期注册、触发、回调主流程一条线注册litellm.callbacks.append(插件)→ 请求进来调度器先调log_pre_api_call→ 流式响应逐块调log_stream_event→ 结束调log_success_event或log_failure_event四个钩子统一带start_time/end_time延迟自己算不用另挂计时器。两个容易漏的点钩子有同步log_*和异步async_log_*两版签名完全一致异步链路实际走async_前缀方法除日志类钩子外基类还提供async_pre_call_hook、async_pre_request_hook改参等介入型钩子改请求要选后者别挂在日志钩子上做修改完整方法清单看源码custom_logger.py。把 LLM 请求日志落到 S3一句话场景每次请求的输入输出要留审计底档本地磁盘不够用落到 S3。配置要点litellm/integrations/s3.py 里的S3Logger构造参数包括s3_bucket_name、s3_path、s3_region_name以及兼容 MinIO 的s3_endpoint_urlfrom litellm.integrations.s3 import S3Logger litellm.callbacks [S3Logger( s3_bucket_namellm-logs, s3_pathlitellm/, s3_region_nameus-east-1, )]跑代理时不传实例也可以在 yaml /.env里配s3_callback_params字典值写成os.environ/XXX形式就能读环境变量见 s3.py 初始化逻辑。最终效果成功与失败事件都写入对象存储文件落在litellm/前缀下需要加密时加s3_server_side_encryption和s3_sse_kms_key_id两个参数即可。用 Prometheus 监控每个服务的延迟与失败数一句话场景要给 SRE 一个面板看到每次调用的延迟分布和失败量。配置要点先pip install prometheus-client缺失时__init__直接抛异常见 prometheus_services.pyfrom litellm.integrations.prometheus_services import PrometheusServicesLogger litellm.callbacks [PrometheusServicesLogger()]最终效果按服务自动注册三类指标——延迟 Histogram、total_requests与failed_requests两个 Counter面板直接拉/metrics不用自己定义指标名。内容护栏 自定义 Token 统计插件一句话场景A 类消息禁止进入模型同时想顺手统计线上 token 消耗。配置要点护栏基类是 custom_guardrail.py 的CustomGuardrailCustomLogger子类必须声明supported_event_hooksfrom litellm.integrations.custom_guardrail import CustomGuardrail from litellm.types.guardrails import GuardrailEventHooks from litellm.exceptions import GuardrailRaisedException class SensitiveBlock(CustomGuardrail): async def async_pre_call_hook(self, kwargs, call_type): for m in kwargs.get(messages, []): if 违禁词 in str(m.get(content, )): raise GuardrailRaisedException(status_code400, messageblocked) litellm.callbacks [SensitiveBlock( guardrail_namecontent-filter, supported_event_hooks[GuardrailEventHooks.pre_call], )]违规请求在 pre_call 阶段直接抛GuardrailRaisedException模型侧零消耗。自定义统计插件同理只写async_log_success_eventfrom litellm.integrations.custom_logger import CustomLogger class TokenCounter(CustomLogger): def __init__(self): self.total 0 async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): if hasattr(response_obj, usage): self.total response_obj.usage.total_tokens # 累计 token最终效果护栏 统计各 1 个类搞定都走同一个callbacks列表互不干扰。踩坑清单现象原因解决启动即Missing prometheus_client异常插件依赖没装pip install prometheus-client⚠️ 依赖装在跑 litellm 的同一环境异步请求里自定义钩子不触发重写了同步版log_success_event调度器在 async 链路调的是async_log_success_event把逻辑迁到async_log_*方法签名一致可直接搬日志里出现完整用户消息默认会把消息写进 StandardLoggingPayload构造CustomLogger时传turn_off_message_loggingTrue见基类__init__代理侧 S3 参数不生效只设了环境变量没走参数字典代理侧通过s3_callback_params传参值支持os.environ/XXX见 s3.pyguardrail 启动校验失败event_hook不在supported_event_hooks列表里严格模式LITELLM_STRICT_GUARDRAIL_MODES下直接报错对齐钩子类型或关闭严格模式延伸资源LiteLLM 的扩展点就是一个CustomLogger基类 一个callbacks列表日志、监控、安全三个方向共用这一套注册与调度机制。内置集成总目录47 个模块litellm/integrations/CustomLogger基类与全部钩子定义litellm/integrations/custom_logger.pyS3 日志完整参数litellm/integrations/s3.pyPrometheus 指标实现litellm/integrations/prometheus_services.py护栏基类与钩子枚举litellm/integrations/custom_guardrail.py、litellm/types/guardrails.py贡献新集成CONTRIBUTING.md【免费下载链接】litellmThe fastest, litest AI Gateway. Rust core with Python SDK. Call 100 LLM APIs in OpenAI (or native) format with cost tracking, guardrails, load balancing, and logging [Bedrock, Azure, OpenAI, Anthropic, OpenAI, VertexAI, vLLM, Nvidia NIM]项目地址: https://gitcode.com/GitHub_Trending/li/litellm创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 LiteLLM 回调插件 10 分钟接入指南：S3 日志、Prometheus 监控、内容护栏 3 个实战场景
LiteLLM 回调插件 10 分钟接入指南S3 日志、Prometheus 监控、内容护栏 3 个实战场景【免费下载链接】litellmThe fastest, litest AI Gateway. Rust core with Python SDK. Call 100 LLM APIs in OpenAI (or native) format with cost tracking, guardrails, load balancing, and logging [Bedrock, Azure, OpenAI, Anthropic, OpenAI, VertexAI, vLLM, Nvidia NIM]项目地址: https://gitcode.com/GitHub_Trending/li/litellm给 LLM 网关加 3 样东西请求日志落 S3、指标进 Prometheus、敏感内容在进模型前拦截。手写适配的话要维护 3 套逻辑用 LiteLLM 的回调callback与护栏guardrail体系每样只需实现 1 个类注册到litellm.callbacks即生效。它省掉的是每个工具各写一套埋点代码这件事。本文适合自研 LLM 网关、想给 LiteLLM 加可观测性或内容安全层的后端工程师。5 分钟接入路径所有插件的公共基类是 litellm/integrations/custom_logger.py 里的CustomLogger。你只重写关心的钩子不关心的方法基类里全是空实现class MyHook(CustomLogger): def log_pre_api_call(self, model, messages, kwargs): ... # 调模型 API 前 def log_stream_event(self, kwargs, response_obj, start_time, end_time): ... # 流式每块 def log_success_event(self, kwargs, response_obj, start_time, end_time): ... # 成功 def log_failure_event(self, kwargs, response_obj, start_time, end_time): ... # 失败注册有两种方式选一种即可import litellm litellm.callbacks [my_hook] # 全局之后所有调用都触发 # 或按单次调用litellm.completion(..., callbacks[my_hook])litellm/integrations/ 下有 47 个内置集成模块S3、Prometheus、Langfuse、Datadog、Slack 等都按同一套路注册无需自己写。插件生命周期注册、触发、回调主流程一条线注册litellm.callbacks.append(插件)→ 请求进来调度器先调log_pre_api_call→ 流式响应逐块调log_stream_event→ 结束调log_success_event或log_failure_event四个钩子统一带start_time/end_time延迟自己算不用另挂计时器。两个容易漏的点钩子有同步log_*和异步async_log_*两版签名完全一致异步链路实际走async_前缀方法除日志类钩子外基类还提供async_pre_call_hook、async_pre_request_hook改参等介入型钩子改请求要选后者别挂在日志钩子上做修改完整方法清单看源码custom_logger.py。把 LLM 请求日志落到 S3一句话场景每次请求的输入输出要留审计底档本地磁盘不够用落到 S3。配置要点litellm/integrations/s3.py 里的S3Logger构造参数包括s3_bucket_name、s3_path、s3_region_name以及兼容 MinIO 的s3_endpoint_urlfrom litellm.integrations.s3 import S3Logger litellm.callbacks [S3Logger( s3_bucket_namellm-logs, s3_pathlitellm/, s3_region_nameus-east-1, )]跑代理时不传实例也可以在 yaml /.env里配s3_callback_params字典值写成os.environ/XXX形式就能读环境变量见 s3.py 初始化逻辑。最终效果成功与失败事件都写入对象存储文件落在litellm/前缀下需要加密时加s3_server_side_encryption和s3_sse_kms_key_id两个参数即可。用 Prometheus 监控每个服务的延迟与失败数一句话场景要给 SRE 一个面板看到每次调用的延迟分布和失败量。配置要点先pip install prometheus-client缺失时__init__直接抛异常见 prometheus_services.pyfrom litellm.integrations.prometheus_services import PrometheusServicesLogger litellm.callbacks [PrometheusServicesLogger()]最终效果按服务自动注册三类指标——延迟 Histogram、total_requests与failed_requests两个 Counter面板直接拉/metrics不用自己定义指标名。内容护栏 自定义 Token 统计插件一句话场景A 类消息禁止进入模型同时想顺手统计线上 token 消耗。配置要点护栏基类是 custom_guardrail.py 的CustomGuardrailCustomLogger子类必须声明supported_event_hooksfrom litellm.integrations.custom_guardrail import CustomGuardrail from litellm.types.guardrails import GuardrailEventHooks from litellm.exceptions import GuardrailRaisedException class SensitiveBlock(CustomGuardrail): async def async_pre_call_hook(self, kwargs, call_type): for m in kwargs.get(messages, []): if 违禁词 in str(m.get(content, )): raise GuardrailRaisedException(status_code400, messageblocked) litellm.callbacks [SensitiveBlock( guardrail_namecontent-filter, supported_event_hooks[GuardrailEventHooks.pre_call], )]违规请求在 pre_call 阶段直接抛GuardrailRaisedException模型侧零消耗。自定义统计插件同理只写async_log_success_eventfrom litellm.integrations.custom_logger import CustomLogger class TokenCounter(CustomLogger): def __init__(self): self.total 0 async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): if hasattr(response_obj, usage): self.total response_obj.usage.total_tokens # 累计 token最终效果护栏 统计各 1 个类搞定都走同一个callbacks列表互不干扰。踩坑清单现象原因解决启动即Missing prometheus_client异常插件依赖没装pip install prometheus-client⚠️ 依赖装在跑 litellm 的同一环境异步请求里自定义钩子不触发重写了同步版log_success_event调度器在 async 链路调的是async_log_success_event把逻辑迁到async_log_*方法签名一致可直接搬日志里出现完整用户消息默认会把消息写进 StandardLoggingPayload构造CustomLogger时传turn_off_message_loggingTrue见基类__init__代理侧 S3 参数不生效只设了环境变量没走参数字典代理侧通过s3_callback_params传参值支持os.environ/XXX见 s3.pyguardrail 启动校验失败event_hook不在supported_event_hooks列表里严格模式LITELLM_STRICT_GUARDRAIL_MODES下直接报错对齐钩子类型或关闭严格模式延伸资源LiteLLM 的扩展点就是一个CustomLogger基类 一个callbacks列表日志、监控、安全三个方向共用这一套注册与调度机制。内置集成总目录47 个模块litellm/integrations/CustomLogger基类与全部钩子定义litellm/integrations/custom_logger.pyS3 日志完整参数litellm/integrations/s3.pyPrometheus 指标实现litellm/integrations/prometheus_services.py护栏基类与钩子枚举litellm/integrations/custom_guardrail.py、litellm/types/guardrails.py贡献新集成CONTRIBUTING.md【免费下载链接】litellmThe fastest, litest AI Gateway. Rust core with Python SDK. Call 100 LLM APIs in OpenAI (or native) format with cost tracking, guardrails, load balancing, and logging [Bedrock, Azure, OpenAI, Anthropic, OpenAI, VertexAI, vLLM, Nvidia NIM]项目地址: https://gitcode.com/GitHub_Trending/li/litellm创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考