LiteLLM 回调接入实战指南三行代码给 AI 应用装上监控和审计【免费下载链接】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/litellmLiteLLM 的回调callback体系本质上是一套标准化的 LiteLLM 插件机制在每次 LLM 调用前后触发你注册的钩子用它就能把成本统计、审计日志、告警等 LiteLLM 集成第三方服务的工作挂上去而不用改动任何业务代码。适合已经会基础 Python、正在用 LiteLLM 统一管理多家模型、但没接触过这套机制的读者全文约十分钟。1️⃣ 三行接入先把回调挂上再说看完这节能得到一个最小可运行的接入方式之后所有高级玩法都是它的变体。最小示例继承CustomLogger覆写请求成功这个钩子把每次调用的模型名和 token 消耗写进本地文件。基类位于litellm/integrations/custom_logger.py。import datetime from litellm import completion from litellm.integrations.custom_logger import CustomLogger class 本地审计插件(CustomLogger): async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): 请求成功后被调用把关键信息追加写入日志文件 模型名 kwargs.get(model) 总token getattr(response_obj, total_tokens, 0) with open(审计日志.log, a, encodingutf-8) as 文件: 文件.write(f{datetime.datetime.now()} | {模型名} | {总token}\n) completion( modelgpt-4o-mini, messages[{role: user, content: 你好}], callbacks[本地审计插件()], # 单行传入立即生效 )如果希望全局生效把实例放进全局列表即可import litellm litellm.callbacks [本地审计插件()]代理部署场景下也可以不写代码litellm/proxy/的配置文件和回调注册表支持按名字启用内置回调Datadog、Slack、Langfuse 等对应名称清单在litellm/__init__.py里。2️⃣ 它为什么能跑通三个触发时机看完这节能回答钩子到底何时被调用。核心逻辑在litellm/litellm_core_utils/litellm_logging.py的分发器里一次请求最多经过三类时机请求前置钩子async_pre_call_hook调用发出前执行。返回异常或字符串可拒绝请求——拦截器、风控都挂在这里成功 / 失败事件async_log_success_event与async_log_failure_event日志、成本核算的主战场流式事件async_log_stream_event流式响应逐块触发适合做实时观察。custom_logger.py里每个钩子都有默认空实现你覆写哪几个都行。目录litellm/integrations/下现成的实现——s3_v2.py的 S3 落盘、prometheus.py的指标上报、SlackAlerting/的告警——全是同一套机制的示范。接入有两条路径像上文那样传对象实例或传内置回调的名字字符串配合环境变量传密钥适合不想写代码的场景。3️⃣ 自己写一个按模型累计 token 并拦截超预算看完这节能独立开发一个自己的 LiteLLM 集成第三方工具之外的纯本地插件。下面的守卫会按模型累计消耗超过预算时抛出异常直接阻断本次调用——这就是前置钩子的实战用法。class 预算守卫(CustomLogger): 按模型累计 token超过预算就阻断后续调用 def __init__(self, 预算上限100_000): self.预算上限 预算上限 self.消耗 {} # 模型名 - 累计 token async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): 模型名 kwargs.get(model, 未知模型) self.消耗[模型名] self.消耗.get(模型名, 0) getattr(response_obj, total_tokens, 0) async def async_pre_call_hook(self, user_api_key_dict, cache, data, call_type): 已用 sum(self.消耗.values()) if 已用 self.预算上限: return Exception(f预算已用尽{已用}/{self.预算上限}请下月再来) return None # 返回 None 表示放行用法和前面一致callbacks[预算守卫()]或注册进litellm.callbacks。4️⃣ 四个高频坑先知道再动手看完这节能避开绝大多数调试时间。优先覆写异步钩子。异步请求走的是async_log_success_event这条路径只覆写同步版log_success_event的插件在 async 场景下可能根本不触发重 I/O 要批量写。每次请求都读写一次外部存储会拖慢主链路litellm/integrations/s3_v2.py的批量上传实现值得抄作业顺序即顺序。多个回调按列表顺序依次执行没有优先级字段把强依赖放前面升级前查签名。钩子参数随版本演进升级 LiteLLM 后跑一遍冒烟测试比翻 changelog 更快发现问题。想继续深入直接读litellm/integrations/里的内置实现和callback_configs.json可以看到每个回调支持的配置项。下一篇计划写LiteLLM 代理的回调配置与批量参数调优把代理侧的玩法补齐。【免费下载链接】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），仅供参考