用 LiteLLM 把十几家大模型收进一个网关:统一 key、限流与记账

一个团队做 AI 应用,半年下来手里往往攥着七八把 API key:OpenAI 的、Anthropic 的、DeepSeek 的、智谱的、还有本地 llama.cpp 的。每接一家,代码里就多一套字段名、一套错误码、一套计费口径。LiteLLM 要解决的就是这个:它对外只暴露 OpenAI 兼容协议,对内把各家模型翻译成同一套调用格式。 它到底统一了什么 一套 SDK:代码里只写 client.chat.completions.create(...),模型名写成 deepseek/deepseek-chat、gemini/gemini-2.5-pro 这样的前缀,剩下的转换由网关做; 一个 key 出口:应用只认网关地址,各家真实密钥存在服务端,前端拿不到; 统一记账:网关天然知道每次调用花了多少 token、哪家模型、哪个用户,可以直接出账单。 起一个网关 最省事的是用它的 proxy 模式:写一个 config.yaml 列出模型与对应环境变量里的密钥,然后 litellm --config config.yaml --port 4000。应用把 base_url 指向 http://localhost:4000,就像在调一个 OpenAI。 再往上走,可以开它的后台 UI,按团队、按 key 分组限额——市场部的 key 每天最多花 50 元,研发的 key 不限速但要记账。 为什么这件事值得做 Manus 2.0 把 Token 成本砍掉两成多,靠的是框架优化。但如果你连「谁在花、花在哪」都看不见,优化就是盲人摸象。一个统一网关是把账单从黑盒变成仪表盘的前提。等哪天你要在 GPT-6 和 DeepSeek 之间按价格动态切换模型,这个网关就是路由层——没有它,多模型容灾只是纸上谈兵。 补一个排障细节:网关层最容易出问题的是超时。长任务(Agent 跑十几步)很容易超过默认的短 HTTP 超时,导致前端报错但后端其实还在跑。把网关和客户端的超时都调到五分钟以上,并开启流式响应,长任务的体验才顺。另外,key 分组限额要配告警,别等账单来了才发现某个测试 key 被循环调用。 把网关立起来之后,你会发现「换模型」「看账单」「做限流」从改代码变成了改配置——这层抽象一旦到位,后面的多模型策略、价格路由才有地方生长,不至于散落在业务代码里。 去论坛讨论 这篇文章里的观点,欢迎到硅基AGI论坛讨论。按标题搜索:LiteLLM网关 💬 更多交流内容,请点击本站论坛交流区:silicon-agi.com ...

2026-10-08 · 1 min · 71 words · 王波
LLM API网关设计:限流、缓存、故障转移的工程实践

LLM API网关设计:限流、缓存、故障转移的工程实践

为什么需要LLM API网关 当企业级应用调用LLM API时,直接暴露上游接口会面临三个核心问题:成本失控、延迟波动和服务不可用。一个设计良好的API网关能在这三者之间取得平衡,同时提供统一的鉴权、计费和可观测能力。 本文将从工程实践角度,拆解LLM API网关的三大核心模块。 限流策略设计 多维度限流模型 与传统API不同,LLM调用具有token维度的特性。一个简单的QPS限流无法覆盖真实场景,需要多维度组合: 维度 限流键 典型阈值 适用场景 QPS API Key + 路径 50 req/s 防恶意刷量 TPM API Key + 模型 100K tokens/min 控制成本 并发数 用户ID 5 concurrent 防资源独占 日总量 租户ID 10M tokens/day 预算控制 令牌桶+滑动窗口的混合实现 以下是基于Redis的混合限流实现: import asyncio import time import redis.asyncio as redis class TokenBucketRateLimiter: def __init__(self, redis_client: redis.Redis, capacity: int, refill_rate: float): self.redis = redis_client self.capacity = capacity self.refill_rate = refill_rate # tokens per second async def acquire(self, key: str, tokens: int = 1) -> bool: lua_script = """ local key = KEYS[1] local capacity = tonumber(ARGV[1]) local refill_rate = tonumber(ARGV[2]) local requested = tonumber(ARGV[3]) local now = tonumber(ARGV[4]) local bucket = redis.call('HMGET', key, 'tokens', 'timestamp') local current_tokens = tonumber(bucket[1]) or capacity local last_timestamp = tonumber(bucket[2]) or now -- 补充令牌 local elapsed = now - last_timestamp current_tokens = math.min(capacity, current_tokens + elapsed * refill_rate) if current_tokens < requested then return 0 end current_tokens = current_tokens - requested redis.call('HMSET', key, 'tokens', current_tokens, 'timestamp', now) redis.call('EXPIRE', key, 3600) return 1 """ now = time.time() result = await self.redis.eval( lua_script, 1, f"ratelimit:{key}", self.capacity, self.refill_rate, tokens, now ) return bool(result) 关键设计点:使用Lua脚本保证原子性,避免竞态条件。对于TPM限流,需要在请求完成后根据实际token消耗做补充扣减。 ...

2026-07-29 · 3 min · 447 words · 硅基 AGI 探索者
🔗 硅基 AGI 论坛 · silicon-agi.com | 📡 RSS 订阅
鲁ICP备2026018361号
📮 联系信箱:13789892969@139.com