litellm 请求拦截实战：一次请求从发出到返回，被哪些钩子处理过-编程实验室

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

litellm 是一个 AI 网关（AI Gateway），它让你用 OpenAI 的格式去调用 Bedrock、Azure、Anthropic 等 100 多家 LLM 服务，同时自带成本统计、负载均衡和日志能力。对大多数开发者来说，真正的价值藏在它的钩子（Hook）机制里：在请求发给模型之前、在响应返回给你的之后，你可以插入自定义逻辑做拦截、过滤和记录，而不用改动任何业务代码。

这篇文章不按照"先讲预处理、再讲后处理"的顺序来，而是跟着一次真实请求走完整条链路——它从你的服务发出，经过网关的几道检查，到达模型，再把结果带回来。你会看到每个环节 litellm 预留了哪些"拦截点"，以及现成的钩子能解决什么问题。

为什么模型调用需要一个"检查站"

假设你的应用已经跑起来了，某天突然遇到三个问题：某个被禁用的用户还在发请求；有用户在提示词里粘贴了生产环境的 API 密钥；运营要求"回答里不能出现竞品名字"。

如果应用是直连模型 API 的，这三个问题都得改业务代码。但把调用流量收拢到 litellm 网关之后，它们就变成同一类问题：在请求经过的通道上设检查站。你可以把钩子机制想象成高速公路上的收费站——车（请求）必须依次通过，每个站可以查身份、查货、收过路费，而司机（你的业务代码）对此一无所知，也不需要改变驾驶方式。

litellm 把这些检查站分成了两类：请求发出前触发的前置钩子（pre-call hook），和响应回来后触发的后置钩子（post-call hook）。下面跟着请求走一遍。

第一关：请求还没出门，先验身份

当你的请求到达代理时，第一件被检查的事是"你是谁"。enterprise/enterprise_hooks/目录下的blocked_user_list.py提供了一个企业级钩子，它的async_pre_call_hook方法会在调用模型之前查询当前的最终用户是否在你的阻止名单里，命中则直接拒绝，请求根本到不了模型。

身份没问题之后，第二件事是"你带了什么"。这是数据泄露的高发环节——用户完全可能把密钥、令牌当作上下文粘贴进 prompt。enterprise/litellm_enterprise/enterprise_callbacks/中的secret_detection.py就干这件事：它在发送前扫描请求文本，内置了几十种检测器（AWS 密钥、Stripe 令牌、JWT、私钥等），发现敏感信息就将其打码或拦截，防止你的密钥被模型"记住"或出现在日志里。

第三道门是内容合规。同一目录里的banned_keywords.py钩子支持配置一个禁词列表（可以是 Python 列表，也可以是一个文件路径），前置钩子阶段会检查输入，命中禁词就返回 400。类似地，google_text_moderation.py和openai_moderation.py则借助第三方内容审核服务对输入做更细粒度的毒性、暴力等分类检测。

这一关的核心价值是：不合规的流量在产生 token 费用之前就被挡掉了。

第二关：响应回来后，再翻一遍"行李"

模型不是万能的，输出同样需要把关。

banned_keywords.py里的钩子不止检查输入。它的async_post_call_success_hook会在拿到完整响应后再次扫描文本内容，而async_post_call_streaming_hook则负责流式场景——当响应以分块（streaming chunk）一段段返回时，逐块检查，发现违禁内容可以中断流。这意味着你可以用同一套禁词配置，同时约束"进"和"出"。

如果你的合规要求更严格，还可以在这一层接入 Aporia 等第三方护栏服务（见enterprise/enterprise_hooks/aporia_ai.py），把整个安全判定外包给专业平台，网关只负责触发时机和结果处理。

自己加一道检查站：自定义钩子的 4 个步骤

现成钩子不够用时，照着这个流程写一个自己的即可：

新建一个文件（比如custom_hooks.py），让你的类继承 litellm 提供的钩子基类；
实现你需要的生命周期方法：想拦请求就实现async_pre_call_hook（请求发出前），想处理结果就实现async_post_call_success_hook（响应成功后），流式场景再补一个async_post_call_streaming_hook；
在代理的 yaml 配置里注册它，通常通过general_settings或litellm_settings相关字段挂载钩子；
重启 litellm 服务让配置生效。

几个常见的自定义场景：按请求特征做智能路由到更便宜的模型（前置钩子里改参数即可）；对重复请求做缓存拦截（前置钩子查缓存命中后直接返回）；把每次请求的关键字段落库，形成完整审计链（前后钩子各记一次）。

验证效果：三种方式确认钩子在起作用

钩子这类"隐式"逻辑最怕静默失效，确认它生效有三条路：

看面板：litellm 代理自带的统计面板能看到请求量、响应时间分布和当前 RPS，钩子拦截掉的请求也会体现在状态码分布里；
接 Langfuse 等追踪工具：litellm/integrations/langfuse/目录下的实现会把每次调用的输入输出、耗时、token 用量和成本画成一条完整轨迹，某个请求被哪个钩子拦下、在哪一步花掉时间，一目了然：

查审计日志：网关管理界面保留了操作与请求的审计记录，配合日志检索可以做合规取证：

另外提醒一句：如果你用限流类能力（TPM/RPM 路由）来配合钩子做资源保护，记得在压测里专门验证限流触发的边界行为，别只测正常路径。

回顾与上手

把一次请求的生命周期摊开看，litellm 的钩子机制就是在每个关键节点留了口子：发出前验身份（blocked_user_list）、查密钥（secret_detection）、过禁词（banned_keywords）；返回后同样过禁词、接第三方护栏。你的业务代码不需要知道这些细节，只需要把流量指到网关。

想要动手试试，可以先克隆仓库看一遍现成钩子的写法：git clone https://gitcode.com/GitHub_Trending/li/litellm，然后从enterprise/enterprise_hooks/里最简单的banned_keywords.py读起——不到 120 行，是理解整个钩子机制最快的一条路径。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考