litellm 回调系统实操：用 5 个钩子把日志、告警与监控挂上去

litellm 回调系统实操用 5 个钩子把日志、告警与监控挂上去【免费下载链接】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把 litellm 部署成统一 LLM 网关之后总会被问两个问题这次请求为什么慢了这周某个模型烧了多少钱答案通常不是去改业务代码而是挂上 litellm 的回调callback系统——网关在请求生命周期的每个阶段都预留了钩子hook你的日志、监控逻辑插进去就能自动拿到数据。这份实操按一条请求走过的路来讲触发点在哪、内置适配器怎么一行启用、自定义 Logger 怎么写、批量上报和采样怎么配。先数清楚一条请求会被拦截几次打开litellm/integrations/custom_logger.pyCustomLogger基类里那一排空方法就是全部可插的位置。按一次 chat completion 的时间线排开大致是发请求之前log_pre_api_call/async_log_pre_api_call还能拿到 model 和 messages 原始入参如果想在发出前改写参数用专门的async_pre_request_hook源码注释里明确区分了记录和转换两类钩子。拿到完整响应后log_post_api_call此时 kwargs 里有完整请求上下文response_obj 里有模型返回。流式响应的每个分片log_stream_eventchunk 到达时逐次触发。收尾判定log_success_event/log_failure_event同步和异步async_log_*两套签名异步版本适合做 IO 密集的上报。隐藏彩蛋走 fallback 时还有log_success_fallback_event/log_failure_fallback_event也就是说原模型挂了、切到备用模型这件事本身也是可观测的。每个钩子都是空实现你覆写哪几个网关就只打哪几个点。这也是它比中间件方案省心的地方不用自己维护请求状态机start_time、end_time、kwargs 都由网关递到手上。第一行配置把内置监控先跑起来litellm 的litellm/integrations/目录下放着十几个现成适配器Slack 告警、Datadog、Prometheus、Langfuse、LangSmith、Arize、Datadog 之外还有 S3、GCS 等落盘类完整清单看 litellm/integrations/Readme.md。以 Langfuse 为例SDK 里一行字符串就够了import litellm # 字符串名走回调注册表SDK 会按同名环境变量自动构造实例 litellm.callbacks [langfuse]挂上之后一次completion调用在 Langfuse 里就变成一条带耗时、token 数、成本的 trace长这样Slack 告警稍微多一点配置但仍然是实例化 注册两步from litellm.integrations.SlackAlerting import SlackAlerting litellm.callbacks [ SlackAlerting( alert_types[timeout, rate_limit], alert_to_webhook_url{ timeout: https://hooks.slack.com/services/xxx, }, ) ]这里有两个源码里可以直接验证的细节alerting_threshold默认 300 秒超过就视为慢请求告警见litellm/integrations/SlackAlerting/slack_alerting.pySlackAlerting继承自CustomBatchLogger告警不是逐条 POST而是攒够一批或每 10 秒发一轮高流量下不会把 webhook 打爆见同目录batching_handler.py和 Readme。提示代理模式下不用写 Python在 proxy 配置的callbacks里登记名字即可各回调对应哪些环境变量可以调CustomLogger.get_callback_env_vars(回调名)反查这个方法就是为了防止配了名字却没给 key而存在的。流式响应里日志怎么记这是最容易漏的点如果你只覆写了log_post_api_call非流式请求没问题但streamTrue时响应是个生成器post 钩子里看不到中间分片。想观测流式过程要覆写log_stream_event或异步版async_log_stream_event每个 chunk 到达都会带着response_obj进来。这里有个坑流式场景下逐 chunk 落盘等于写放大正确姿势是在 stream 钩子里做增量聚合比如只累计 token 数把真正的持久化留给log_success_event这个收尾钩子。CustomLogger里还有一组async_post_call_streaming_*钩子粒度更细具体行为参考源码确认。自定义 Logger 的最小骨架内置适配器覆盖不了的场景合规审计、内部风控系统继承CustomLogger即可。一个只记录哪个模型、花了多久的最小实现from litellm.integrations.custom_logger import CustomLogger class AuditLogger(CustomLogger): def __init__(self, **kwargs): # 标准日志 payload 中的消息内容将被抹掉只留元数据 super().__init__(turn_off_message_loggingTrue, **kwargs) async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): record { model: response_obj.model, latency: end_time - start_time, } await audit_store.write(record) litellm.callbacks [AuditLogger()]三个实用开关都是基类现成的能力turn_off_message_loggingTrue消息体不进标准日志 payload做审计时默认带上比较稳妥覆写truncate_standard_logging_payload_content给长文本字段设截断上限防止一次多模态请求写出 MB 级日志handle_callback_failure你的上报抛异常时网关会走这个钩子做降级处理而不是让业务请求陪葬。失败路径记得也写一份async_log_failure_event——只记成功的审计日志复盘时等于瞎了一半。批量上报与采样别让日志吃掉网关性能逐条上报在大流量下有两个问题网络往返多、下游 API 被限流。CustomBatchLoggerlitellm/integrations/custom_batch_logger.py把入队—攒批—定时冲刷这套逻辑做成了基类你只需实现async_send_batchimport asyncio from litellm.integrations.custom_batch_logger import CustomBatchLogger class BufferedAuditLogger(CustomBatchLogger): def __init__(self, **kwargs): self.flush_lock asyncio.Lock() super().__init__(batch_size50, flush_interval10, **kwargs) async def async_send_batch(self): events, self.log_queue self.log_queue, [] await audit_store.write_many(events)队列是内存log_queue默认上限 5 万条DEFAULT_MAX_QUEUE_SIZE冲刷失败时事件会保留到下轮重试超过上限才丢最老的——也就是说下游短暂不可用不会让你丢日志但下游长期不可用会先丢历史。高并发下还有一招是采样并非所有请求都值得全量上报。部分适配器内置了sampling_rate参数比如litellm/integrations/argilla.py里就是随机数小于采样率才记录的标准写法自己实现时照这个模式抄即可。怎么选告警、指标、trace 各归其位三类内置工具解决的问题不同混用会既贵又吵即时响应类Slack 告警盯现在有事超时、限流、预算告警秒级推到值班群趋势指标类Datadog、Prometheus盯长期健康延迟分位数、成本曲线、错误分布数据在litellm/integrations/datadog/和litellm/integrations/prometheus_services.py里各有一套实现调用追踪类Langfuse、LangSmith、Arize盯单次为什么一条请求的完整入参、分片、耗时拆开看开发调试阶段最有用。实践上三件套可以同时挂litellm.callbacks是列表互不干扰。顺序上建议先接追踪类把数据流跑通肉眼可见、反馈快再叠加指标和告警——顺序反了告警响了却查不到根因等于白配。延伸阅读集成模块总览litellm/integrations/Readme.md可观测性实践 notebookLangfuse、Langtrace 等接入示例cookbook/logging_observability/自定义回调的类型定义与标准 payload 结构litellm/types/utils.py中的StandardLoggingPayload【免费下载链接】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），仅供参考