跳至主要内容

路由器 - 负载均衡

LiteLLM 管理：

  • 跨多个部署（例如 Azure/OpenAI）进行负载均衡
  • 优先处理重要请求，以确保它们不会失败（即排队机制）
  • 基础可靠性逻辑 - 跨多个部署/提供商的冷却机制、回退、超时和重试（固定 + 指数退避）。

在生产环境中，litellm 支持使用 Redis 来跟踪冷却服务器和使用情况（管理 TPM/RPM 限制）。

信息

如果您希望通过服务器在不同的 LLM API 之间进行负载均衡，请使用我们的 LiteLLM 代理服务器

负载均衡​

(感谢 @paulpierre 和 sweep proxy 对此实现所做的贡献) 查看代码

快速入门​

跨多个 azure/bedrock/提供商 部署进行负载均衡。如果调用失败，LiteLLM 将处理不同区域的重试。

from litellm import Router

model_list = [{ # list of model deployments 
	"model_name": "gpt-3.5-turbo", # model alias -> loadbalance between models with same `model_name`
	"litellm_params": { # params for litellm completion/embedding call 
		"model": "azure/chatgpt-v-2", # actual model name
		"api_key": os.getenv("AZURE_API_KEY"),
		"api_version": os.getenv("AZURE_API_VERSION"),
		"api_base": os.getenv("AZURE_API_BASE")
	}
}, {
    "model_name": "gpt-3.5-turbo", 
	"litellm_params": { # params for litellm completion/embedding call 
		"model": "azure/chatgpt-functioncalling", 
		"api_key": os.getenv("AZURE_API_KEY"),
		"api_version": os.getenv("AZURE_API_VERSION"),
		"api_base": os.getenv("AZURE_API_BASE")
	}
}, {
    "model_name": "gpt-3.5-turbo", 
	"litellm_params": { # params for litellm completion/embedding call 
		"model": "gpt-3.5-turbo", 
		"api_key": os.getenv("OPENAI_API_KEY"),
	}
}, {
    "model_name": "gpt-4", 
	"litellm_params": { # params for litellm completion/embedding call 
		"model": "azure/gpt-4", 
		"api_key": os.getenv("AZURE_API_KEY"),
		"api_base": os.getenv("AZURE_API_BASE"),
		"api_version": os.getenv("AZURE_API_VERSION"),
	}
}, {
    "model_name": "gpt-4", 
	"litellm_params": { # params for litellm completion/embedding call 
		"model": "gpt-4", 
		"api_key": os.getenv("OPENAI_API_KEY"),
	}
},

]

router = Router(model_list=model_list)

# openai.ChatCompletion.create replacement
# requests with model="gpt-3.5-turbo" will pick a deployment where model_name="gpt-3.5-turbo"
response = await router.acompletion(model="gpt-3.5-turbo", 
				messages=[{"role": "user", "content": "Hey, how's it going?"}])

print(response)

# openai.ChatCompletion.create replacement
# requests with model="gpt-4" will pick a deployment where model_name="gpt-4"
response = await router.acompletion(model="gpt-4", 
				messages=[{"role": "user", "content": "Hey, how's it going?"}])

print(response)

可用端点​

  • router.completion() - 用于调用 100 多个 LLM 的聊天补全端点
  • router.acompletion() - 异步聊天补全调用
  • router.embedding() - 用于 Azure、OpenAI、Huggingface 端点的嵌入端点
  • router.aembedding() - 异步嵌入调用
  • router.text_completion() - 以旧版 OpenAI /v1/completions 端点格式进行的补全调用
  • router.atext_completion() - 异步文本补全调用
  • router.image_generation() - 以 OpenAI /v1/images/generations 端点格式进行的补全调用
  • router.aimage_generation() - 异步图像生成调用

高级 - 路由策略 ⭐️​

路由策略 - 加权选择、速率限制感知、最少繁忙、基于延迟、基于成本​

路由器提供了多种策略来跨多个部署路由您的调用。我们建议在生产环境中使用 simple-shuffle（默认）以获得最佳性能。

默认且推荐用于生产环境 - 以最小的延迟开销提供最佳性能。

根据提供的每分钟请求数 (rpm) 或每分钟令牌数 (tpm) 选择部署

如果未提供 rpm 或 tpm，则随机选择一个部署

您还可以设置 weight 参数，以指定何时应选择哪个模型。

LiteLLM 代理 Config.yaml​
model_list:
	- model_name: gpt-3.5-turbo
	  litellm_params:
	  	model: azure/chatgpt-v-2
		api_key: os.environ/AZURE_API_KEY
		api_version: os.environ/AZURE_API_VERSION
		api_base: os.environ/AZURE_API_BASE
		rpm: 900 
	- model_name: gpt-3.5-turbo
	  litellm_params:
	  	model: azure/chatgpt-functioncalling
		api_key: os.environ/AZURE_API_KEY
		api_version: os.environ/AZURE_API_VERSION
		api_base: os.environ/AZURE_API_BASE
		rpm: 10
Python SDK​
from litellm import Router 
import asyncio

model_list = [{ # list of model deployments 
	"model_name": "gpt-3.5-turbo", # model alias 
	"litellm_params": { # params for litellm completion/embedding call 
		"model": "azure/chatgpt-v-2", # actual model name
		"api_key": os.getenv("AZURE_API_KEY"),
		"api_version": os.getenv("AZURE_API_VERSION"),
		"api_base": os.getenv("AZURE_API_BASE"),
		"rpm": 900,			# requests per minute for this API
	}
}, {
    "model_name": "gpt-3.5-turbo", 
	"litellm_params": { # params for litellm completion/embedding call 
		"model": "azure/chatgpt-functioncalling", 
		"api_key": os.getenv("AZURE_API_KEY"),
		"api_version": os.getenv("AZURE_API_VERSION"),
		"api_base": os.getenv("AZURE_API_BASE"),
		"rpm": 10,
	}
},]

# init router
router = Router(model_list=model_list, routing_strategy="simple-shuffle")
async def router_acompletion():
	response = await router.acompletion(
		model="gpt-3.5-turbo", 
		messages=[{"role": "user", "content": "Hey, how's it going?"}]
	)
	print(response)
	return response

asyncio.run(router_acompletion())

路由组 - 按模型策略​

对同一路由器中的不同模型应用不同的路由策略。路由组将一系列 model_name 绑定到一个策略和（可选的）策略参数。未被任何组声明的模型将回退到路由器的顶级 routing_strategy。

提示

您还可以从仪表板创建、编辑和删除路由组。请参阅 通过 UI 管理路由组。

使用场景：您希望对 gpt-4o 使用基于延迟的路由，但对更便宜的模型使用简单的加权选择——而无需启动第二个路由器。

规则​

  • 每个 model_name 最多属于一个组。重叠会在初始化时引发 ValueError。
  • 不在任何组中的模型使用顶级 routing_strategy / routing_strategy_args（一个隐式的 "default" 组）。名称 "default" 是保留名称。
  • 每个组都可以覆盖 routing_strategy_args（例如延迟窗口 TTL、TPM 上限）。
  • 该组是基于预路由钩子之后的 model 名称按请求解析的。
model_list:
  - model_name: gpt-4o
    litellm_params:
      model: openai/gpt-4o
      api_key: os.environ/OPENAI_API_KEY
  - model_name: gpt-4o
    litellm_params:
      model: azure/gpt-4o
      api_base: os.environ/AZURE_API_BASE
      api_key: os.environ/AZURE_API_KEY
      api_version: "2024-08-01-preview"
  - model_name: cheap-model
    litellm_params:
      model: openai/gpt-4o-mini
      api_key: os.environ/OPENAI_API_KEY

router_settings:
  # fallback strategy for models not in any explicit group
  routing_strategy: simple-shuffle

  routing_groups:
    - group_name: latency-sensitive
      models: [gpt-4o]
      routing_strategy: latency-based-routing
      routing_strategy_args:
        ttl: 3600

行为

  • gpt-4o → 在 OpenAI + Azure 部署之间进行基于延迟的路由。
  • cheap-model → simple-shuffle（默认组）。

多个组​

两个组可以使用相同的策略但带有不同的参数；每个组都获得一个独立的状态实例。

router_settings:
  routing_strategy: simple-shuffle
  routing_groups:
    - group_name: hot-path
      models: [gpt-4o, claude-sonnet]
      routing_strategy: latency-based-routing
      routing_strategy_args:
        ttl: 60          # short window — react quickly to latency changes
    - group_name: batch
      models: [gpt-4o-mini, llama-70b]
      routing_strategy: usage-based-routing-v2
      routing_strategy_args:
        rpm: 10000

在运行时更新​

路由组可以通过 Router.update_settings(routing_groups=[...]) 或代理的 /config/update 端点进行更新。每个组的状态会在更新时重建。

流量镜像 / 无声实验​

流量镜像允许您将生产流量“模仿”到辅助（静默）模型进行评估目的。静默模型的响应在后台收集，不会影响主请求的延迟或结果。

查看有关 A/B 测试 - 流量镜像的详细指南请点击此处

基本可靠性​

部署排序（优先级）​

在 litellm_params 中设置 order 以确定部署优先级。值越小，优先级越高。当多个部署共享相同的 order 时，路由策略会在它们之间进行选择。

当对 order=1 部署的请求失败时（连接错误、404、429 等），路由器会自动尝试 order=2 的部署，然后是 order=3，以此类推。每个订单级别在升级到下一个级别之前都有自己的一组重试机制。如果所有级别都耗尽，路由器将回退到任何已配置的 回退。

from litellm import Router

model_list = [
    {
        "model_name": "gpt-4",
        "litellm_params": {
            "model": "azure/gpt-4-primary",
            "api_key": os.getenv("AZURE_API_KEY"),
            "order": 1,  # 👈 Highest priority
        },
    },
    {
        "model_name": "gpt-4",
        "litellm_params": {
            "model": "azure/gpt-4-fallback",
            "api_key": os.getenv("AZURE_API_KEY_2"),
            "order": 2,  # 👈 Tried when order=1 fails
        },
    },
]

router = Router(model_list=model_list)

加权部署​

在部署上设置 weight，以便比其他部署更频繁地选择该部署。

这适用于 simple-shuffle 路由策略（这是默认设置，如果没有选择路由策略）。

from litellm import Router 

model_list = [
	{
		"model_name": "o1",
		"litellm_params": {
			"model": "o1-preview", 
			"api_key": os.getenv("OPENAI_API_KEY"), 
			"weight": 1
		},
	},
	{
		"model_name": "o1",
		"litellm_params": {
			"model": "o1-preview", 
			"api_key": os.getenv("OPENAI_API_KEY"), 
			"weight": 2 # 👈 PICK THIS DEPLOYMENT 2x MORE OFTEN THAN o1-preview
		},
	},
]

router = Router(model_list=model_list, routing_strategy="cost-based-routing")

response = await router.acompletion(
	model="gpt-3.5-turbo", 
	messages=[{"role": "user", "content": "Hey, how's it going?"}]
)
print(response)

加权故障转移​

默认情况下，当模型组中的部署失败时，路由器会移动到 fallbacks 中的下一个条目（不同的模型组）。通过 enable_weighted_failover，路由器首先在同一模型组内重试，通过使用现有权重重新选择不同的部署，仅在组内每个部署都尝试过后才升级到跨组回退。

当您有同一个模型的多个区域副本（例如 Azure eastus2 + swedencentral）并希望失败的区域回退到具有相同 model_name 的健康对等方时，这非常有用，而不是立即切换到不同的模型。

行为

  • 仅在 routing_strategy="simple-shuffle"（默认）时激活。
  • 在可重试的故障上，失败的部署 ID 被排除，并从同一模型组中的剩余对等方中选择一个新的部署，同时尊重 weight / rpm / tpm。
  • 排除项会在跳转之间累积：每次重试都会将之前的故障添加到排除集中，因此在同一个请求链中，刚刚失败的部署永远不会再次被选中。
  • 受 max_fallbacks（默认 5）限制。
  • 不针对 ContextWindowExceededError 或 ContentPolicyViolationError 触发 - 那些保持其专用的回退路径。
  • 仅异步：由 router.acompletion() 和其他异步入口点支持。同步 router.completion() 路径回退到常规回退。
  • 冷却仍然适用：超过 allowed_fails 的部署将独立于加权故障转移进行冷却。

顺序与权重

如果同一组也使用 order，则顺序过滤器会在加权选择之前运行。因此，加权故障转移仅在当前的最小订单层中的部署之间重新选择。晋升到下一个订单层是通过现有的基于顺序的回退路径进行的。

配置

from litellm import Router

model_list = [
    {
        "model_name": "gpt-4.1-mini",
        "litellm_params": {
            "model": "azure/gpt-4.1-mini",
            "api_base": "https://eastus2.example.azure.com",
            "api_key": os.getenv("AZURE_EASTUS2_KEY"),
            "weight": 1,
        },
    },
    {
        "model_name": "gpt-4.1-mini",
        "litellm_params": {
            "model": "azure/gpt-4.1-mini",
            "api_base": "https://swedencentral.example.azure.com",
            "api_key": os.getenv("AZURE_SWEDEN_KEY"),
            "weight": 1,
        },
    },
]

router = Router(
    model_list=model_list,
    routing_strategy="simple-shuffle",
    enable_weighted_failover=True,  # 👈 retry within the same model group on failure
)

response = await router.acompletion(
    model="gpt-4.1-mini",
    messages=[{"role": "user", "content": "Hey"}],
)

演练

使用上述配置和对 gpt-4.1-mini 的请求

  1. simple-shuffle 使用 weight 从两个部署中选择一个。
  2. 如果所选部署引发提供商错误（例如 RateLimitError、InternalServerError），其部署 ID 会被添加到 metadata._failover_excluded_ids。
  3. 路由器在排除失败的部署的情况下重新进入 simple-shuffle，并将权重重新归一化到剩余的选项上。
  4. 步骤 2–3 重复进行，直到部署成功、每个对等方都被排除或达到 max_fallbacks。
  5. 只有在所有对等方都耗尽后，路由器才会回退到为该组配置的任何 fallbacks。

有关该标志，请参阅路由器设置参考中的 enable_weighted_failover。

最大并行请求 (异步)​

用于路由器上异步请求的信号量。限制对部署发出的最大并发调用数。在高流量场景下很有用。

如果设置了 tpm/rpm，并且没有给出最大并行请求限制，我们使用 RPM 或计算出的 RPM (tpm/1000/6) 作为最大并行请求限制。

from litellm import Router 

model_list = [{
	"model_name": "gpt-4",
	"litellm_params": {
		"model": "azure/gpt-4",
		...
		"max_parallel_requests": 10 # 👈 SET PER DEPLOYMENT
	}
}]

### OR ### 

router = Router(model_list=model_list, default_max_parallel_requests=20) # 👈 SET DEFAULT MAX PARALLEL REQUESTS 


# deployment max parallel requests > default max parallel requests

查看代码

冷却​

设置模型在冷却一分钟之前，一分钟内允许失败的调用次数限制。

from litellm import Router

model_list = [{...}]

router = Router(model_list=model_list, 
                allowed_fails=1,      # cooldown model if it fails > 1 call in a minute. 
				cooldown_time=100    # cooldown the deployment for 100 seconds if it num_fails > allowed_fails
		)

user_message = "Hello, whats the weather in San Francisco??"
messages = [{"content": user_message, "role": "user"}]

# normal call 
response = router.completion(model="gpt-3.5-turbo", messages=messages)

print(f"response: {response}")

预期响应

No deployments available for selected model, Try again in 60 seconds. Passed model=claude-3-5-sonnet. pre-call-checks=False, allowed_model_region=n/a.

禁用冷却​

from litellm import Router 


router = Router(..., disable_cooldowns=True)

冷却的工作原理​

冷却适用于单个部署，而不是整个模型组。路由器将故障隔离到特定部署，同时保持健康的替代方案可用。

什么是部署？​

部署是 config.yaml 模型列表中的单个条目。每个部署代表一个独特的配置，并带有其自己的 litellm_params。

LiteLLM 通过创建所有 litellm_params 的确定性哈希，为每个部署生成一个唯一的 model_id。这使得路由器可以独立跟踪和管理每个部署。

示例：同一模型的多个部署

负载均衡 config.yaml
model_list:
  - model_name: sonnet-4              # Deployment 1
    litellm_params:
      model: anthropic/claude-sonnet-4-20250514
      api_key: <our-real-key>
      
  - model_name: byok-sonnet-4         # Deployment 2  
    litellm_params:
      model: anthropic/claude-sonnet-4-20250514
      api_key: <customer-managed-key>
      api_base: https://proxy.litellm.ai/api.anthropic.com
      
  - model_name: sonnet-4              # Deployment 3
    litellm_params:
      model: vertex_ai/claude-sonnet-4-20250514
      vertex_project: my-project

每个部署都会获得一个唯一的 model_id（例如，1234567890、9129922、4982929292），路由器使用它来跟踪健康状况和冷却状态。

部署何时会被冷却？​

路由器会根据以下条件自动冷却部署

条件触发器冷却持续时间
速率限制 (429)收到 429 响应时立即冷却5 秒 (默认)
高故障率当前分钟内 >50% 的失败5 秒 (默认)
不可重试错误401 (认证), 404 (未找到), 408 (超时)5 秒 (默认)

在冷却期间，特定的部署会被暂时从可用池中删除，而其他健康的部署则继续处理请求。

冷却恢复​

部署在冷却期结束后会自动恢复。路由器将

  1. 监控每个部署的冷却计时器
  2. 自动重新启用冷却结束后的部署
  3. 逐步重新引入冷却过的部署回到循环中
  4. 重置失败计数器，一旦部署再次健康

实际案例​

考虑这种具有多个提供商的高可用性设置

负载均衡 config.yaml
model_list:
  - model_name: sonnet-4              # Primary: Anthropic Direct
    litellm_params:
      model: anthropic/claude-sonnet-4-20250514
      api_key: <anthropic-key>
      
  - model_name: byok-sonnet-4         # BYOK: Customer-managed keys
    litellm_params:
      model: anthropic/claude-sonnet-4-20250514
      api_key: <customer-managed-key>
      api_base: https://proxy.litellm.ai/api.anthropic.com
      
  - model_name: sonnet-4              # Fallback: Vertex AI
    litellm_params:
      model: vertex_ai/claude-sonnet-4-20250514
      vertex_project: my-project

故障场景

重试​

对于异步和同步函数，我们都支持重试失败的请求。

对于 RateLimitError，我们实现了指数退避

对于通用错误，我们立即重试

这里简要展示了如何设置 num_retries = 3

from litellm import Router

model_list = [{...}]

router = Router(model_list=model_list,  
                num_retries=3)

user_message = "Hello, whats the weather in San Francisco??"
messages = [{"content": user_message, "role": "user"}]

# normal call 
response = router.completion(model="gpt-3.5-turbo", messages=messages)

print(f"response: {response}")

我们还支持在重试失败请求之前设置最短等待时间。这是通过 retry_after 参数实现的。

from litellm import Router

model_list = [{...}]

router = Router(model_list=model_list,  
                num_retries=3, retry_after=5) # waits min 5s before retrying request

user_message = "Hello, whats the weather in San Francisco??"
messages = [{"content": user_message, "role": "user"}]

# normal call 
response = router.completion(model="gpt-3.5-turbo", messages=messages)

print(f"response: {response}")

[高级]：自定义重试、基于错误类型的冷却​

  • 如果您想根据收到的异常设置 num_retries，请使用 RetryPolicy
  • 使用 AllowedFailsPolicy 设置在冷却部署之前每分钟允许的 allowed_fails 自定义数量

查看所有异常类型

示例

retry_policy = RetryPolicy(
    ContentPolicyViolationErrorRetries=3, 		  # run 3 retries for ContentPolicyViolationErrors
    AuthenticationErrorRetries=0,         		  # run 0 retries for AuthenticationErrorRetries
)

allowed_fails_policy = AllowedFailsPolicy(
	ContentPolicyViolationErrorAllowedFails=1000, # Allow 1000 ContentPolicyViolationError before cooling down a deployment
	RateLimitErrorAllowedFails=100,               # Allow 100 RateLimitErrors before cooling down a deployment
)

示例用法

from litellm.router import RetryPolicy, AllowedFailsPolicy

retry_policy = RetryPolicy(
	ContentPolicyViolationErrorRetries=3,         # run 3 retries for ContentPolicyViolationErrors
	AuthenticationErrorRetries=0,		          # run 0 retries for AuthenticationErrorRetries
	BadRequestErrorRetries=1,
	TimeoutErrorRetries=2,
	RateLimitErrorRetries=3,
)

allowed_fails_policy = AllowedFailsPolicy(
	ContentPolicyViolationErrorAllowedFails=1000, # Allow 1000 ContentPolicyViolationError before cooling down a deployment
	RateLimitErrorAllowedFails=100,               # Allow 100 RateLimitErrors before cooling down a deployment
)

router = litellm.Router(
	model_list=[
		{
			"model_name": "gpt-3.5-turbo",  # openai model name
			"litellm_params": {  # params for litellm completion/embedding call
				"model": "azure/chatgpt-v-2",
				"api_key": os.getenv("AZURE_API_KEY"),
				"api_version": os.getenv("AZURE_API_VERSION"),
				"api_base": os.getenv("AZURE_API_BASE"),
			},
		},
		{
			"model_name": "bad-model",  # openai model name
			"litellm_params": {  # params for litellm completion/embedding call
				"model": "azure/chatgpt-v-2",
				"api_key": "bad-key",
				"api_version": os.getenv("AZURE_API_VERSION"),
				"api_base": os.getenv("AZURE_API_BASE"),
			},
		},
	],
	retry_policy=retry_policy,
	allowed_fails_policy=allowed_fails_policy,
)

response = await router.acompletion(
	model=model,
	messages=messages,
)

缓存​

在生产环境中，我们建议使用 Redis 缓存。对于在本地快速测试，我们也支持简单的内存缓存。

内存缓存

router = Router(model_list=model_list, 
                cache_responses=True)

print(response)

Redis 缓存

router = Router(model_list=model_list, 
                redis_host=os.getenv("REDIS_HOST"), 
                redis_password=os.getenv("REDIS_PASSWORD"), 
                redis_port=os.getenv("REDIS_PORT"),
                cache_responses=True)

print(response)

传入 Redis URL，其他 kwargs

router = Router(model_list: Optional[list] = None,
                 ## CACHING ## 
                 redis_url=os.getenv("REDIS_URL")",
				 cache_kwargs= {}, # additional kwargs to pass to RedisCache (see caching.py)
				 cache_responses=True)
信息

在路由器设置中配置 Redis 缓存时，请使用 cache_kwargs 传入其他 Redis 参数，特别是对于通过 REDIS_* 环境变量设置可能会失败的非字符串值。

预调用检查（上下文窗口、欧盟区域）​

启用预调用检查以过滤掉

  1. 上下文窗口限制 < 调用所需消息量的部署。
  2. 位于欧盟区域之外的部署

1. 启用预调用检查

from litellm import Router 
# ...
router = Router(model_list=model_list, enable_pre_call_checks=True) # 👈 Set to True

2. 设置模型列表

对于 Azure 部署的上下文窗口检查，设置基础模型。从此列表中选择基础模型，所有 azure 模型都以 azure/ 开头。

对于“欧盟区域”过滤，设置部署的“region_name”。

注意：我们根据您的 litellm 参数自动推断 Vertex AI、Bedrock 和 IBM WatsonxAI 的 region_name。对于 Azure，请设置 litellm.enable_preview = True。

查看代码

model_list = [
            {
                "model_name": "gpt-3.5-turbo", # model group name
                "litellm_params": {  # params for litellm completion/embedding call
                    "model": "azure/chatgpt-v-2",
                    "api_key": os.getenv("AZURE_API_KEY"),
                    "api_version": os.getenv("AZURE_API_VERSION"),
                    "api_base": os.getenv("AZURE_API_BASE"),
					"region_name": "eu" # 👈 SET 'EU' REGION NAME
					"base_model": "azure/gpt-35-turbo", # 👈 (Azure-only) SET BASE MODEL
                },
            },
            {
                "model_name": "gpt-3.5-turbo", # model group name
                "litellm_params": {  # params for litellm completion/embedding call
                    "model": "gpt-3.5-turbo-1106",
                    "api_key": os.getenv("OPENAI_API_KEY"),
                },
            },
			{
				"model_name": "gemini-pro",
				"litellm_params: {
					"model": "vertex_ai/gemini-pro-1.5", 
					"vertex_project": "adroit-crow-1234",
					"vertex_location": "us-east1" # 👈 AUTOMATICALLY INFERS 'region_name'
				}
			}
        ]

router = Router(model_list=model_list, enable_pre_call_checks=True)

3. 测试它！

"""
- Give a gpt-3.5-turbo model group with different context windows (4k vs. 16k)
- Send a 5k prompt
- Assert it works
"""
from litellm import Router
import os

model_list = [
	{
		"model_name": "gpt-3.5-turbo",  # model group name
		"litellm_params": {  # params for litellm completion/embedding call
			"model": "azure/chatgpt-v-2",
			"api_key": os.getenv("AZURE_API_KEY"),
			"api_version": os.getenv("AZURE_API_VERSION"),
			"api_base": os.getenv("AZURE_API_BASE"),
			"base_model": "azure/gpt-35-turbo",
		},
		"model_info": {
			"base_model": "azure/gpt-35-turbo", 
		}
	},
	{
		"model_name": "gpt-3.5-turbo",  # model group name
		"litellm_params": {  # params for litellm completion/embedding call
			"model": "gpt-3.5-turbo-1106",
			"api_key": os.getenv("OPENAI_API_KEY"),
		},
	},
]

router = Router(model_list=model_list, enable_pre_call_checks=True) 

text = "What is the meaning of 42?" * 5000

response = router.completion(
	model="gpt-3.5-turbo",
	messages=[
		{"role": "system", "content": text},
		{"role": "user", "content": "Who was Alexander?"},
	],
)

print(f"response: {response}")

跨模型组缓存​

如果您想跨 2 个不同的模型组（例如 azure 部署和 openai）进行缓存，请使用缓存组。

import litellm, asyncio, time
from litellm import Router 

# set os env
os.environ["OPENAI_API_KEY"] = ""
os.environ["AZURE_API_KEY"] = ""
os.environ["AZURE_API_BASE"] = ""
os.environ["AZURE_API_VERSION"] = ""

async def test_acompletion_caching_on_router_caching_groups(): 
	# tests acompletion + caching on router 
	try:
		litellm.set_verbose = True
		model_list = [
			{
				"model_name": "openai-gpt-3.5-turbo",
				"litellm_params": {
					"model": "gpt-3.5-turbo-0613",
					"api_key": os.getenv("OPENAI_API_KEY"),
				},
			},
			{
				"model_name": "azure-gpt-3.5-turbo",
				"litellm_params": {
					"model": "azure/chatgpt-v-2",
					"api_key": os.getenv("AZURE_API_KEY"),
					"api_base": os.getenv("AZURE_API_BASE"),
					"api_version": os.getenv("AZURE_API_VERSION")
				},
			}
		]

		messages = [
			{"role": "user", "content": f"write a one sentence poem {time.time()}?"}
		]
		start_time = time.time()
		router = Router(model_list=model_list, 
				cache_responses=True, 
				caching_groups=[("openai-gpt-3.5-turbo", "azure-gpt-3.5-turbo")])
		response1 = await router.acompletion(model="openai-gpt-3.5-turbo", messages=messages, temperature=1)
		print(f"response1: {response1}")
		await asyncio.sleep(1) # add cache is async, async sleep for cache to get set
		response2 = await router.acompletion(model="azure-gpt-3.5-turbo", messages=messages, temperature=1)
		assert response1.id == response2.id
		assert len(response1.choices[0].message.content) > 0
		assert response1.choices[0].message.content == response2.choices[0].message.content
	except Exception as e:
		traceback.print_exc()

asyncio.run(test_acompletion_caching_on_router_caching_groups())

告警 🚨​

针对以下事件向 Slack / 您的 webhook URL 发送告警

  • LLM API 异常
  • LLM 响应缓慢

从 https://api.slack.com/messaging/webhooks 获取 Slack Webhook URL

用法​

初始化一个 AlertingConfig 并将其传递给 litellm.Router。以下代码将触发告警，因为 api_key=bad-key 是无效的

import litellm
from litellm.router import Router
from litellm.types.router import AlertingConfig
import os
import asyncio

router = Router(
	model_list=[
		{
			"model_name": "gpt-3.5-turbo",
			"litellm_params": {
				"model": "gpt-3.5-turbo",
				"api_key": "bad_key",
			},
		}
	],
	alerting_config= AlertingConfig(
		alerting_threshold=10,
		webhook_url= "https:/..."
	),
)

async def main():
	print(f"\n=== Configuration ===")
	print(f"Slack logger exists: {router.slack_alerting_logger is not None}")
	
	try:
		await router.acompletion(
			model="gpt-3.5-turbo",
			messages=[{"role": "user", "content": "Hey, how's it going?"}],
		)
	except Exception as e:
		print(f"\n=== Exception caught ===")
		print(f"Waiting 10 seconds for alerts to be sent via periodic flush...")
		await asyncio.sleep(10)
		print(f"\n=== After waiting ===")
		print(f"Alert should have been sent to Slack!")

asyncio.run(main())

跟踪 Azure 部署成本​

问题：Azure 在使用 azure/gpt-4-1106-preview 时在响应中返回 gpt-4。这导致成本跟踪不准确

解决方案 ✅ ：在路由器初始化时设置 model_info["base_model"]，以便 litellm 使用正确的模型来计算 azure 成本

步骤 1. 路由器设置

from litellm import Router

model_list = [
	{ # list of model deployments 
		"model_name": "gpt-4-preview", # model alias 
		"litellm_params": { # params for litellm completion/embedding call 
			"model": "azure/chatgpt-v-2", # actual model name
			"api_key": os.getenv("AZURE_API_KEY"),
			"api_version": os.getenv("AZURE_API_VERSION"),
			"api_base": os.getenv("AZURE_API_BASE")
		},
		"model_info": {
			"base_model": "azure/gpt-4-1106-preview" # azure/gpt-4-1106-preview will be used for cost tracking, ensure this exists in litellm model_prices_and_context_window.json
		}
	}, 
	{
		"model_name": "gpt-4-32k", 
		"litellm_params": { # params for litellm completion/embedding call 
			"model": "azure/chatgpt-functioncalling", 
			"api_key": os.getenv("AZURE_API_KEY"),
			"api_version": os.getenv("AZURE_API_VERSION"),
			"api_base": os.getenv("AZURE_API_BASE")
		},
		"model_info": {
			"base_model": "azure/gpt-4-32k" # azure/gpt-4-32k will be used for cost tracking, ensure this exists in litellm model_prices_and_context_window.json
		}
	}
]

router = Router(model_list=model_list)

步骤 2. 在自定义回调中访问 response_cost，litellm 会为您计算响应成本

import litellm
from litellm.integrations.custom_logger import CustomLogger

class MyCustomHandler(CustomLogger):        
	def log_success_event(self, kwargs, response_obj, start_time, end_time): 
		print(f"On Success")
		response_cost = kwargs.get("response_cost")
		print("response_cost=", response_cost)

customHandler = MyCustomHandler()
litellm.callbacks = [customHandler]

# router completion call
response = router.completion(
	model="gpt-4-32k", 
	messages=[{ "role": "user", "content": "Hi who are you"}]
)

默认 litellm.completion/embedding 参数​

您还可以为 litellm 补全/嵌入调用设置默认参数。以下是如何操作

from litellm import Router

fallback_dict = {"gpt-3.5-turbo": "gpt-3.5-turbo-16k"}

router = Router(model_list=model_list, 
                default_litellm_params={"context_window_fallback_dict": fallback_dict})

user_message = "Hello, whats the weather in San Francisco??"
messages = [{"content": user_message, "role": "user"}]

# normal call 
response = router.completion(model="gpt-3.5-turbo", messages=messages)

print(f"response: {response}")

自定义回调 - 跟踪 API 密钥、API 端点、使用的模型​

如果您需要跟踪每次补全调用所使用的 api_key、api 端点、模型、custom_llm_provider，您可以设置一个 自定义回调

用法​

import litellm
from litellm.integrations.custom_logger import CustomLogger

class MyCustomHandler(CustomLogger):        
	def log_success_event(self, kwargs, response_obj, start_time, end_time): 
		print(f"On Success")
		print("kwargs=", kwargs)
		litellm_params= kwargs.get("litellm_params")
		api_key = litellm_params.get("api_key")
		api_base = litellm_params.get("api_base")
		custom_llm_provider= litellm_params.get("custom_llm_provider")
		response_cost = kwargs.get("response_cost")

		# print the values
		print("api_key=", api_key)
		print("api_base=", api_base)
		print("custom_llm_provider=", custom_llm_provider)
		print("response_cost=", response_cost)

	def log_failure_event(self, kwargs, response_obj, start_time, end_time): 
		print(f"On Failure")
		print("kwargs=")

customHandler = MyCustomHandler()

litellm.callbacks = [customHandler]

# Init Router
router = Router(model_list=model_list, routing_strategy="simple-shuffle")

# router completion call
response = router.completion(
	model="gpt-3.5-turbo", 
	messages=[{ "role": "user", "content": "Hi who are you"}]
)

部署路由器​

如果您希望通过服务器在不同的 LLM API 之间进行负载均衡，请使用我们的 LiteLLM 代理服务器

调试路由器​

基本调试​

设置 Router(set_verbose=True)

from litellm import Router

router = Router(
    model_list=model_list,
    set_verbose=True
)

详细调试​

设置 Router(set_verbose=True,debug_level="DEBUG")

from litellm import Router

router = Router(
    model_list=model_list,
    set_verbose=True,
    debug_level="DEBUG"  # defaults to INFO
)

非常详细的调试​

设置 litellm.set_verbose=True 和 Router(set_verbose=True,debug_level="DEBUG")

from litellm import Router
import litellm

litellm.set_verbose = True

router = Router(
    model_list=model_list,
    set_verbose=True,
    debug_level="DEBUG"  # defaults to INFO
)

路由器常规设置​

用法​

router = Router(model_list=..., router_general_settings=RouterGeneralSettings(async_only_mode=True))

规范​

class RouterGeneralSettings(BaseModel):
    async_only_mode: bool = Field(
        default=False
    )  # this will only initialize async clients. Good for memory utils
    pass_through_all_models: bool = Field(
        default=False
    )  # if passed a model not llm_router model list, pass through the request to litellm.acompletion/embedding
🚅
LiteLLM 企业版
SSO/SAML、审计日志、支出跟踪、多团队管理和护栏 —— 专为生产环境构建。
了解更多 →