跳至主要内容

预算管理器

不想在调用 LLM API 或者用户调用这些 API 时产生巨额账单?使用这个功能即可。

信息

如果您需要一个服务器来管理用户密钥、预算等,请使用我们的 LiteLLM 代理服务器

LiteLLM 提供

  • litellm.max_budget:一个全局变量,可用于设置所有 litellm 调用中的最大预算(以美元为单位)。如果超过此预算,将引发 BudgetExceededError 异常。
  • BudgetManager:一个用于帮助设置每个用户预算的类。BudgetManager 会创建一个字典来管理用户预算,其中 key 是用户,对象是其当前成本 + 模型特定的成本。
  • LiteLLM 代理服务器:一个使用与 OpenAI 兼容的端点来调用 100 多个 LLM 的服务器。它负责管理用户预算、支出跟踪、负载均衡等。

快速入门

import litellm, os 
from litellm import completion

# set env variable
os.environ["OPENAI_API_KEY"] = "your-api-key"

litellm.max_budget = 0.001 # sets a max budget of $0.001

messages = [{"role": "user", "content": "Hey, how's it going"}]
completion(model="gpt-4", messages=messages)
print(litellm._current_cost)
completion(model="gpt-4", messages=messages)

基于用户的速率限制

Open In Colab
from litellm import BudgetManager, completion 

budget_manager = BudgetManager(project_name="test_project")

user = "1234"

# create a budget if new user user
if not budget_manager.is_valid_user(user):
budget_manager.create_budget(total_budget=10, user=user)

# check if a given call can be made
if budget_manager.get_current_cost(user=user) <= budget_manager.get_total_budget(user):
response = completion(model="gpt-3.5-turbo", messages=[{"role": "user", "content": "Hey, how's it going?"}])
budget_manager.update_cost(completion_obj=response, user=user)
else:
response = "Sorry - no budget!"

实现代码

与文本输入/输出结合使用

只需传入文本输入/输出和模型名称即可更新成本。

from litellm import BudgetManager

budget_manager = BudgetManager(project_name="test_project")
user = "12345"
budget_manager.create_budget(total_budget=10, user=user, duration="daily")

input_text = "hello world"
output_text = "it's a sunny day in san francisco"
model = "gpt-3.5-turbo"

budget_manager.update_cost(user=user, model=model, input_text=input_text, output_text=output_text) # 👈
print(budget_manager.get_current_cost(user))

进阶用法

在生产环境中,我们需要

  • 将用户预算存储在数据库中
  • 根据设定的持续时间重置用户预算

LiteLLM API

LiteLLM API 两者兼备。它将用户对象存储在托管数据库中,并每天运行一次 cron 任务,根据设定的持续时间(例如,每天/每周/每月等重置预算)来重置用户预算。

用法

budget_manager = BudgetManager(project_name="<my-unique-project>", client_type="hosted")

完整代码

from litellm import BudgetManager, completion 

budget_manager = BudgetManager(project_name="<my-unique-project>", client_type="hosted")

user = "1234"

# create a budget if new user user
if not budget_manager.is_valid_user(user):
budget_manager.create_budget(total_budget=10, user=user, duration="monthly") # 👈 duration = 'daily'/'weekly'/'monthly'/'yearly'

# check if a given call can be made
if budget_manager.get_current_cost(user=user) <= budget_manager.get_total_budget(user):
response = completion(model="gpt-3.5-turbo", messages=[{"role": "user", "content": "Hey, how's it going?"}])
budget_manager.update_cost(completion_obj=response, user=user)
else:
response = "Sorry - no budget!"

自托管

要使用您自己的数据库,请将 BudgetManager 客户端类型设置为 hosted 设置 api_base。

您的 API 需要公开 /get_budget/set_budget 端点。查看代码详情

用法

budget_manager = BudgetManager(project_name="<my-unique-project>", client_type="hosted", api_base="your_custom_api")

完整代码

from litellm import BudgetManager, completion 

budget_manager = BudgetManager(project_name="<my-unique-project>", client_type="hosted", api_base="your_custom_api")

user = "1234"

# create a budget if new user user
if not budget_manager.is_valid_user(user):
budget_manager.create_budget(total_budget=10, user=user, duration="monthly") # 👈 duration = 'daily'/'weekly'/'monthly'/'yearly'

# check if a given call can be made
if budget_manager.get_current_cost(user=user) <= budget_manager.get_total_budget(user):
response = completion(model="gpt-3.5-turbo", messages=[{"role": "user", "content": "Hey, how's it going?"}])
budget_manager.update_cost(completion_obj=response, user=user)
else:
response = "Sorry - no budget!"

预算管理器类

BudgetManager 类用于管理不同用户的预算。它提供了各种函数来创建、更新和检索预算信息。

以下是预算管理器类公开的公共函数及其输入/输出列表。

init

def __init__(self, project_name: str, client_type: str = "local", api_base: Optional[str] = None)
  • project_name (str):项目名称。
  • client_type (str):客户端类型("local" 或 "hosted")。默认为 "local"。
  • api_base (Optional[str]):API 的基础 URL。默认为 None。

create_budget

def create_budget(self, total_budget: float, user: str, duration: Literal["daily", "weekly", "monthly", "yearly"], created_at: float = time.time())

为用户创建预算。

  • total_budget (float):用户的总预算。
  • user (str):用户 ID。
  • duration (Literal["daily", "weekly", "monthly", "yearly"]):预算持续时间。
  • created_at (float):创建时间。默认为当前时间。

projected_cost

def projected_cost(self, model: str, messages: list, user: str)

计算会话的预计成本。

  • model (str):模型名称。
  • messages (list):消息列表。
  • user (str):用户 ID。

get_total_budget

def get_total_budget(self, user: str)

返回用户的总预算。

  • user (str):用户 ID。

update_cost

def update_cost(self, completion_obj: ModelResponse, user: str)

更新用户的成本。

  • completion_obj (ModelResponse):从模型接收到的完成对象。
  • user (str):用户 ID。

get_current_cost

def get_current_cost(self, user: str)

返回用户的当前成本。

  • user (str):用户 ID。

get_model_cost

def get_model_cost(self, user: str)

返回用户的模型成本。

  • user (str):用户 ID。

is_valid_user

def is_valid_user(self, user: str) -> bool

检查用户是否有效。

  • user (str):用户 ID。

get_users

def get_users(self)

返回所有用户的列表。

reset_cost

def reset_cost(self, user: str)

重置用户成本。

  • user (str):用户 ID。

reset_on_duration

def reset_on_duration(self, user: str)

根据持续时间重置用户成本。

  • user (str):用户 ID。

update_budget_all_users

def update_budget_all_users(self)

更新所有用户的预算。

save_data

def save_data(self)

存储用户字典。