跳至主要内容

预算管理器

不想在调用 LLM API 或用户调用它们时收到天价账单？使用这个。

信息

如果你想有一个服务器来管理用户密钥、预算等，请使用我们的 LiteLLM 代理服务器

LiteLLM 提供

  • litellm.max_budget: 您可以使用此全局变量设置所有 litellm 调用中的最大预算（美元）。如果超出此预算，将引发 BudgetExceededError。
  • BudgetManager: 一个用于帮助设置每个用户预算的类。BudgetManager 创建一个字典来管理用户预算，其中键是用户，对象是他们的当前成本 + 模型特定成本。
  • LiteLLM 代理服务器: 一个使用 openai 兼容端点调用 100+ LLMs 的服务器。管理用户预算、费用跟踪、负载均衡等。

快速入门​

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!"

Budget Manager 类​

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

以下是 Budget Manager 类公开的公共函数及其输入/输出列表。

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 (可选[str]): API 的 Base 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)

存储用户字典。