Implementation:BerriAI Litellm Budget Manager
| Attribute | Value |
|---|---|
| Sources | litellm/budget_manager.py
|
| Domains | Budget Management, Cost Tracking, User Management |
| Last Updated | 2026-02-15 16:00 GMT |
Overview
BudgetManager is a client-side budget management class that tracks per-user LLM spending against configurable budgets with support for periodic reset cycles and both local file and hosted API storage.
Description
The BudgetManager class provides comprehensive per-user budget tracking for LiteLLM API calls. It supports two storage backends: "local" (persists to a user_cost.json file) and "hosted" (syncs with the LiteLLM API at api.litellm.ai). Key features include:
- Budget creation with optional duration-based reset cycles (
"daily","weekly","monthly","yearly") - Cost tracking via
update_cost, which computes costs from either aModelResponseobject or raw input/output text usinglitellm.completion_costandlitellm.cost_per_token - Projected cost estimation before making API calls via
projected_cost - Per-model cost breakdown tracked in
model_costwithin each user's record - Automatic budget reset when the configured duration elapses via
reset_on_durationandupdate_budget_all_users - Non-blocking persistence using background threads for save operations
Note: This is the client-side budget manager, not the proxy budget manager (which is in proxy_server.py).
Usage
Import and instantiate BudgetManager when you need client-side per-user budget enforcement. Use it to create budgets, check spending before API calls, and update costs after completions.
Code Reference
Source Location
Signature
class BudgetManager:
def __init__(self, project_name: str, client_type: str = "local",
api_base: Optional[str] = None, headers: Optional[dict] = None)
def load_data(self)
def create_budget(self, total_budget: float, user: str,
duration: Optional[Literal["daily", "weekly", "monthly", "yearly"]] = None,
created_at: float = time.time())
def projected_cost(self, model: str, messages: list, user: str) -> float
def get_total_budget(self, user: str) -> float
def update_cost(self, user: str, completion_obj: Optional[ModelResponse] = None,
model: Optional[str] = None, input_text: Optional[str] = None,
output_text: Optional[str] = None) -> dict
def get_current_cost(self, user: str) -> float
def get_model_cost(self, user: str)
def is_valid_user(self, user: str) -> bool
def get_users(self) -> list
def reset_cost(self, user: str) -> dict
def reset_on_duration(self, user: str)
def update_budget_all_users(self)
def save_data(self)
Import
from litellm.budget_manager import BudgetManager
I/O Contract
Inputs
| Parameter | Type | Description |
|---|---|---|
project_name |
str |
Name of the project for hosted storage. |
client_type |
str |
Storage backend: "local" or "hosted". Defaults to "local".
|
total_budget |
float |
Maximum budget in USD for a user. |
user |
str |
User identifier. |
duration |
Optional[Literal["daily", "weekly", "monthly", "yearly"]] |
Budget reset cycle. |
completion_obj |
Optional[ModelResponse] |
A completion response to extract cost from. |
model |
str |
Model name for cost calculation. |
input_text / output_text |
Optional[str] |
Raw text for manual cost calculation. |
Outputs
| Method | Return Type | Description |
|---|---|---|
create_budget |
dict |
The user's budget record. |
projected_cost |
float |
Estimated total cost after the next API call. |
get_total_budget |
float |
The user's total budget limit. |
update_cost |
dict |
The updated user record including current_cost and model_cost.
|
get_current_cost |
float |
The user's current accumulated cost. |
is_valid_user |
bool |
Whether the user exists in the budget system. |
get_users |
list |
List of all tracked user identifiers. |
Usage Examples
from litellm.budget_manager import BudgetManager
import litellm
budget_manager = BudgetManager(project_name="my-project")
# Create a monthly budget of $50 for a user
budget_manager.create_budget(total_budget=50.0, user="user-123", duration="monthly")
# Check projected cost before making a call
messages = [{"role": "user", "content": "Write a long essay about AI"}]
projected = budget_manager.projected_cost(model="gpt-4", messages=messages, user="user-123")
if projected < budget_manager.get_total_budget("user-123"):
# Make the API call
response = litellm.completion(model="gpt-4", messages=messages)
# Update the cost
budget_manager.update_cost(user="user-123", completion_obj=response)
# Check current spending
current = budget_manager.get_current_cost("user-123")
print(f"Current cost: ${current:.4f}")
# Reset budgets for all users if duration has elapsed
budget_manager.update_budget_all_users()
Related Pages
- BerriAI_Litellm_Batch_Completion - Batch completion that may benefit from budget checks