尧图网站建设 
litellm 回调系统一次 LLM 调用埋下四道钩子日志与监控一次接齐【免费下载链接】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 的回调系统在每次 LLM 调用经过它时触发一组钩子函数你只需要订阅其中关心的节点就能把日志、监控、告警接进自己的体系。如果不想写代码内置的 Slack 告警和 Datadog 适配器各配一行即可启用如果要定制业务逻辑继承一个基类、覆写一两个钩子方法就能上线。一次 LLM 调用的事件链四个钩子按时间顺序触发一次 litellm 调用的生命周期固定经过四个节点每个节点对应一个钩子方法按时间排开如下请求发出前pre_api_call收到 model、messages 和全部请求参数适合做审计记录或参数校验响应到达后post_api_call收到请求参数、响应对象和起止时间戳是记录耗时与成本的主要位置流式响应中stream_event每收到一块chunk就触发一次需要逐块消费时必须实现它调用收尾成功走success_event失败走failure_event二者都携带完整参数用来区分成败做统计。这五个方法在基类 custom_logger.py 中都有定义全部为空实现——你实现哪个就订阅哪个节点。每个钩子还有async_前缀的异步版本比如async_log_post_api_call日志落库或外发建议走异步版本避免拖慢主请求。监控集成不改代码接入 Slack 告警与 Datadoglitellm 监控集成走内置适配器路线每个平台是一个现成类配好凭据后交给 litellm 自动调度。最小组合方式启用 Slack 告警Slack 告警由 slack_alerting.py 中的SlackAlerting类提供最小接入只需两个参数import litellm from litellm.integrations.SlackAlerting import SlackAlerting handler SlackAlerting( alert_to_webhook_url{slow_callbacks: https://hooks.slack.com/services/xxx}, # 告警接收的 webhook 地址 alerting_threshold120, # 超过该秒数判定为慢调用并告警 ) litellm.success_callback [handler] # 全局注册之后所有调用都会走它告警内容会先聚合再定时批量发出避免高流量下刷屏预算超支、团队告警等更多事件类型在 SlackAlerting 的说明文档 中有列出。最小组合方式接入 DatadogDatadog 方向更省事走配置文件即可在 proxy 的litellm_settings中加一行litellm_settings: success_callback: [datadog]再在进程环境导出DD_API_KEY和DD_SITE例如us5.datadoghq.com或改用LITELLM_DD_AGENT_HOST指向本地 Datadog Agent。日志上报由 datadog.py 中的DataDogLogger完成批量大小、标签维度都有默认值不用手动循环发送。内置适配器产出的数据最终长成这样——调用链、token、成本逐条可查除了 Slack 与 DatadogLangSmith、Arize 也是同类的内置适配器配置方式一致这里不再展开。从零写一个 litellm 自定义回调继承、覆写、注册、验证基类叫CustomLogger它是所有回调处理器的公共父类。下面以医疗聊天机器人为例需要记录每次调用的模型与耗时并把用户消息里的手机号脱敏后再落日志。四步走完。第一步继承基类。只需继承并覆写关心的钩子其余保持默认空实现import re from litellm.integrations.custom_logger import CustomLogger PHONE re.compile(r1[3-9]\d{9}) class MedicalAuditLogger(CustomLogger): 记录医疗对话审计日志先脱敏再落盘。 def log_post_api_call(self, kwargs, response_obj, start_time, end_time): # 脱敏用户消息中的手机号 for msg in kwargs.get(messages) or []: if msg.get(role) user and isinstance(msg.get(content), str): msg[content] PHONE.sub(1**********, msg[content]) with open(audit.log, a) as f: f.write(f{kwargs.get(model)}|{(end_time - start_time).total_seconds():.2f}s\n)第二步注册。按单次调用注册就把处理器实例放进callbackslitellm.completion( modelgpt-4o-mini, messages[{role: user, content: 我手机 13812345678帮我记录}], callbacks[MedicalAuditLogger()], # 本次调用触发该处理器 )也可放进litellm_settings全局生效。第三步验证。发起请求后确认audit.log多出一行且内容里的手机号已被替换。流式场景要单独覆写log_stream_event才能拿到逐块数据。回调上报到平台后的审计视图大致如下生产环境调优清单与踩坑四条清单按优先级排列异步优先落库、发 HTTP 等动作一律改用async_log_post_api_call等异步钩子别让日志 IO 阻塞主请求链路脱敏配置CustomLogger.__init__支持turn_off_message_loggingTrue开启后消息内容从标准日志负载中抹去合规要求高的场景直接用它批量聚合高流量下逐个事件上报会打爆下游改用批量基类 custom_batch_logger.py 中的CustomBatchLogger它是内置适配器的公共父类天然带定时聚合失败隔离日志回调内部异常被 litellm 捕获并记录不会反向击穿业务请求但要在自己的钩子外补一层监控防止静默无日志。常见坑按「现象 → 原因 → 解决」排查现象流式请求一条日志都没有。→原因只实现了log_post_api_call流式路径根本不走它。→解决另覆写log_stream_event处理逐块数据。现象日志体积爆炸磁盘先于业务告警。→原因默认把完整 messages 写进日志。→解决开启turn_off_message_loggingTrue或自己只在钩子里提取 model、耗时等字段。现象Slack 告警总是晚几分钟才到。→原因批量发送机制攒够一批或到定时窗口才发出。→解决属设计行为别逐条排查网络确需即时就到可调整聚合窗口。资源钩子基类litellm/integrations/custom_logger.pySlack 告警litellm/integrations/SlackAlerting/slack_alerting.pyDatadog 适配器litellm/integrations/datadog/datadog.py批量上报基类litellm/integrations/custom_batch_logger.py适配器目录说明litellm/integrations/Readme.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 回调系统：一次 LLM 调用埋下四道钩子，日志与监控一次接齐
litellm 回调系统一次 LLM 调用埋下四道钩子日志与监控一次接齐【免费下载链接】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 的回调系统在每次 LLM 调用经过它时触发一组钩子函数你只需要订阅其中关心的节点就能把日志、监控、告警接进自己的体系。如果不想写代码内置的 Slack 告警和 Datadog 适配器各配一行即可启用如果要定制业务逻辑继承一个基类、覆写一两个钩子方法就能上线。一次 LLM 调用的事件链四个钩子按时间顺序触发一次 litellm 调用的生命周期固定经过四个节点每个节点对应一个钩子方法按时间排开如下请求发出前pre_api_call收到 model、messages 和全部请求参数适合做审计记录或参数校验响应到达后post_api_call收到请求参数、响应对象和起止时间戳是记录耗时与成本的主要位置流式响应中stream_event每收到一块chunk就触发一次需要逐块消费时必须实现它调用收尾成功走success_event失败走failure_event二者都携带完整参数用来区分成败做统计。这五个方法在基类 custom_logger.py 中都有定义全部为空实现——你实现哪个就订阅哪个节点。每个钩子还有async_前缀的异步版本比如async_log_post_api_call日志落库或外发建议走异步版本避免拖慢主请求。监控集成不改代码接入 Slack 告警与 Datadoglitellm 监控集成走内置适配器路线每个平台是一个现成类配好凭据后交给 litellm 自动调度。最小组合方式启用 Slack 告警Slack 告警由 slack_alerting.py 中的SlackAlerting类提供最小接入只需两个参数import litellm from litellm.integrations.SlackAlerting import SlackAlerting handler SlackAlerting( alert_to_webhook_url{slow_callbacks: https://hooks.slack.com/services/xxx}, # 告警接收的 webhook 地址 alerting_threshold120, # 超过该秒数判定为慢调用并告警 ) litellm.success_callback [handler] # 全局注册之后所有调用都会走它告警内容会先聚合再定时批量发出避免高流量下刷屏预算超支、团队告警等更多事件类型在 SlackAlerting 的说明文档 中有列出。最小组合方式接入 DatadogDatadog 方向更省事走配置文件即可在 proxy 的litellm_settings中加一行litellm_settings: success_callback: [datadog]再在进程环境导出DD_API_KEY和DD_SITE例如us5.datadoghq.com或改用LITELLM_DD_AGENT_HOST指向本地 Datadog Agent。日志上报由 datadog.py 中的DataDogLogger完成批量大小、标签维度都有默认值不用手动循环发送。内置适配器产出的数据最终长成这样——调用链、token、成本逐条可查除了 Slack 与 DatadogLangSmith、Arize 也是同类的内置适配器配置方式一致这里不再展开。从零写一个 litellm 自定义回调继承、覆写、注册、验证基类叫CustomLogger它是所有回调处理器的公共父类。下面以医疗聊天机器人为例需要记录每次调用的模型与耗时并把用户消息里的手机号脱敏后再落日志。四步走完。第一步继承基类。只需继承并覆写关心的钩子其余保持默认空实现import re from litellm.integrations.custom_logger import CustomLogger PHONE re.compile(r1[3-9]\d{9}) class MedicalAuditLogger(CustomLogger): 记录医疗对话审计日志先脱敏再落盘。 def log_post_api_call(self, kwargs, response_obj, start_time, end_time): # 脱敏用户消息中的手机号 for msg in kwargs.get(messages) or []: if msg.get(role) user and isinstance(msg.get(content), str): msg[content] PHONE.sub(1**********, msg[content]) with open(audit.log, a) as f: f.write(f{kwargs.get(model)}|{(end_time - start_time).total_seconds():.2f}s\n)第二步注册。按单次调用注册就把处理器实例放进callbackslitellm.completion( modelgpt-4o-mini, messages[{role: user, content: 我手机 13812345678帮我记录}], callbacks[MedicalAuditLogger()], # 本次调用触发该处理器 )也可放进litellm_settings全局生效。第三步验证。发起请求后确认audit.log多出一行且内容里的手机号已被替换。流式场景要单独覆写log_stream_event才能拿到逐块数据。回调上报到平台后的审计视图大致如下生产环境调优清单与踩坑四条清单按优先级排列异步优先落库、发 HTTP 等动作一律改用async_log_post_api_call等异步钩子别让日志 IO 阻塞主请求链路脱敏配置CustomLogger.__init__支持turn_off_message_loggingTrue开启后消息内容从标准日志负载中抹去合规要求高的场景直接用它批量聚合高流量下逐个事件上报会打爆下游改用批量基类 custom_batch_logger.py 中的CustomBatchLogger它是内置适配器的公共父类天然带定时聚合失败隔离日志回调内部异常被 litellm 捕获并记录不会反向击穿业务请求但要在自己的钩子外补一层监控防止静默无日志。常见坑按「现象 → 原因 → 解决」排查现象流式请求一条日志都没有。→原因只实现了log_post_api_call流式路径根本不走它。→解决另覆写log_stream_event处理逐块数据。现象日志体积爆炸磁盘先于业务告警。→原因默认把完整 messages 写进日志。→解决开启turn_off_message_loggingTrue或自己只在钩子里提取 model、耗时等字段。现象Slack 告警总是晚几分钟才到。→原因批量发送机制攒够一批或到定时窗口才发出。→解决属设计行为别逐条排查网络确需即时就到可调整聚合窗口。资源钩子基类litellm/integrations/custom_logger.pySlack 告警litellm/integrations/SlackAlerting/slack_alerting.pyDatadog 适配器litellm/integrations/datadog/datadog.py批量上报基类litellm/integrations/custom_batch_logger.py适配器目录说明litellm/integrations/Readme.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），仅供参考