Skip to main content

预算管理器

不想因为调用LLM API（无论是在你调用时还是用户调用时）而收到疯狂的账单吗？使用这个。

info

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

LiteLLM提供了以下功能：

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

快速开始​

import litellm, os 
from litellm import completion

# 设置环境变量 
os.environ["OPENAI_API_KEY"] = "你的API密钥"

litellm.max_budget = 0.001 # 设置最大预算为$0.001

messages = [{"role": "user", "content": "嘿，最近怎么样"}]
completion(model="gpt-4", messages=messages)
print(litellm._current_cost)
completion(model="gpt-4", messages=messages)

基于用户的速率限制​

在Colab中打开
from litellm import BudgetManager, completion 

budget_manager = BudgetManager(project_name="测试项目")

user = "1234"

# 如果是新用户，创建预算
if not budget_manager.is_valid_user(user):
    budget_manager.create_budget(total_budget=10, user=user)

# 检查是否可以进行给定的调用
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": "嘿，最近怎么样？"}])
    budget_manager.update_cost(completion_obj=response, user=user)
else:
    response = "抱歉 - 没有预算了！"

实现代码

与文本输入/输出一起使用​

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

from litellm import BudgetManager

budget_manager = BudgetManager(project_name="测试项目")
user = "12345"
budget_manager.create_budget(total_budget=10, user=user, duration="每日")

input_text = "你好，世界"
output_text = "旧金山的阳光明媚"
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="<我的唯一项目>", client_type="hosted")

完整代码

from litellm import BudgetManager, completion 

budget_manager = BudgetManager(project_name="<我的唯一项目>", client_type="hosted")

user = "1234"

# 如果是新用户，创建预算
if not budget_manager.is_valid_user(user):
    budget_manager.create_budget(total_budget=10, user=user, duration="每月") # 👈 持续时间 = '每日'/'每周'/'每月'/'每年'

# 检查是否可以进行给定的调用
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": "嘿，最近怎么样？"}])
    budget_manager.update_cost(completion_obj=response, user=user)
else:
    response = "抱歉 - 没有预算了！"

自托管​

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

你的API应暴露/get_budget和/set_budget端点。查看代码了解详情

使用方法

budget_manager = BudgetManager(project_name="<我的唯一项目>", client_type="hosted", api_base="你的自定义API")

完整代码

from litellm import BudgetManager, completion 

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

user = "1234"

# 如果用户是新用户，则创建一个预算
if not budget_manager.is_valid_user(user):
    budget_manager.create_budget(total_budget=10, user=user, duration="monthly") # 👈 duration = 'daily'/'weekly'/'monthly'/'yearly'

# 检查是否可以进行特定调用
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 类用于管理不同用户的预算。它提供了创建、更新和检索预算信息的各种功能。

以下是 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)

存储用户字典。

优云智算