预算管理器
不想在调用 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)
基于用户的速率限制
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)
存储用户字典。