Python SDK Reference
Complete guide to the LLMGovernor Python SDK for AI agent cost tracking and budget enforcement.
Installation
pip install llmgovernor
Configuration
Environment Variables
Set your LLMGovernor API key:
export LLMGOVERNOR_API_KEY="llmg_sk_xxxxx"
Optional Configuration
from blaze_agents import configure
configure(
api_key="your-api-key", # Override env var
base_url="https://api.blazeforagents.com", # Custom endpoint
timeout=30, # Request timeout in seconds
debug=True # Enable debug logging
)
Core Decorator: @fleet.agent
The @fleet.agent decorator is the primary interface for tracking AI agent costs.
Basic Usage
from blaze_agents import fleet
@fleet.agent()
def simple_agent():
# Your agent logic here
pass
Parameters
| Parameter | Type | Description | Default |
|---|---|---|---|
budget |
float | Maximum spend in USD | None (no limit) |
user_id |
str | User/session identifier | None |
agent_name |
str | Custom agent name | Function name |
tags |
dict | Custom metadata tags | {} |
alert_threshold |
float | Alert at % of budget (0.0-1.0) | 0.8 |
enforcement |
str | "block" or "warn" | "block" |
Examples
Budget Enforcement
@fleet.agent(budget=25.00)
def research_agent(query: str):
"""Agent with $25 budget limit."""
response = openai.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": query}]
)
return response.choices[0].message.content
User-Specific Tracking
@fleet.agent(budget=10.00, user_id="user_123")
def personal_assistant(user_input: str, user_id: str):
"""Track costs per user."""
# Agent logic
pass
Custom Tags and Metadata
@fleet.agent(
budget=100.00,
tags={
"department": "marketing",
"campaign": "holiday_2024",
"model_type": "gpt-4"
},
agent_name="HolidayMarketingAgent"
)
def marketing_agent(content: str):
"""Marketing agent with detailed tracking."""
# Agent logic
pass
Warning Mode
@fleet.agent(
budget=50.00,
enforcement="warn", # Don't block, just warn
alert_threshold=0.5 # Alert at 50% of budget
)
def experimental_agent(query: str):
"""Agent that warns but doesn't block on budget."""
# Agent logic
pass
Supported Providers
LLMGovernor automatically detects and tracks costs for these providers:
OpenAI
import openai
@fleet.agent(budget=20.00)
def openai_agent(prompt: str):
response = openai.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
Anthropic
import anthropic
@fleet.agent(budget=15.00)
def claude_agent(prompt: str):
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-3-sonnet-20240229",
max_tokens=1000,
messages=[{"role": "user", "content": prompt}]
)
return response.content[0].text
Google (Gemini/PaLM)
import google.generativeai as genai
@fleet.agent(budget=10.00)
def gemini_agent(prompt: str):
model = genai.GenerativeModel('gemini-pro')
response = model.generate_content(prompt)
return response.text
Azure OpenAI
from openai import AzureOpenAI
@fleet.agent(budget=30.00)
def azure_agent(prompt: str):
client = AzureOpenAI(
azure_endpoint="https://your-resource.openai.azure.com",
api_key="your-api-key",
api_version="2024-02-15-preview"
)
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
Exception Handling
BudgetExceededError
Raised when an agent exceeds its budget:
from blaze_agents.exceptions import BudgetExceededError
try:
result = expensive_agent("Complex analysis")
except BudgetExceededError as e:
print(f"Budget exceeded: {e.message}")
print(f"Current spend: ${e.current_spend}")
print(f"Budget limit: ${e.budget_limit}")
# Fallback to cheaper alternative
result = cheap_agent("Simple analysis")
Rate Limit Errors
from blaze_agents.exceptions import RateLimitError
try:
result = my_agent("Query")
except RateLimitError as e:
print(f"Rate limited by provider: {e}")
# Implement retry logic
Manual Tracking
For non-standard providers or custom tracking:
from blaze_agents import track_cost
@fleet.agent(budget=20.00)
def custom_agent(prompt: str):
# Your custom LLM call
response = custom_llm_api(prompt)
# Manually track the cost
track_cost(
amount=0.15, # Cost in USD
provider="custom_llm",
model="custom-model-v1",
tokens_input=len(prompt.split()),
tokens_output=len(response.split()),
metadata={"custom_field": "value"}
)
return response
Async Support
Full async/await support:
import asyncio
import openai
@fleet.agent(budget=25.00)
async def async_agent(prompt: str):
client = openai.AsyncOpenAI()
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
# Usage
async def main():
result = await async_agent("What is machine learning?")
print(result)
asyncio.run(main())
Context Managers
Temporary budget and tag overrides:
from blaze_agents import budget_context, tag_context
@fleet.agent()
def flexible_agent(query: str, complexity: str):
if complexity == "high":
with budget_context(50.00):
# Temporarily increase budget for complex queries
return expensive_analysis(query)
else:
with tag_context({"complexity": "low"}):
return simple_analysis(query)
Logging and Debug
Enable detailed logging:
import logging
from blaze_agents import configure
# Enable debug logging
configure(debug=True)
logging.getLogger("blaze_agents").setLevel(logging.DEBUG)
@fleet.agent(budget=10.00)
def debug_agent(prompt: str):
# Will log detailed cost information
return openai_call(prompt)
Best Practices
1. Set Appropriate Budgets
# Good: Reasonable budget for expected usage
@fleet.agent(budget=5.00)
def simple_qa_agent(question: str):
pass
# Bad: Budget too low for model capabilities
@fleet.agent(budget=0.01) # GPT-4 call costs ~$0.03+
def complex_analysis_agent(data: str):
pass
2. Use Meaningful Tags
@fleet.agent(
budget=20.00,
tags={
"version": "v2.1",
"environment": "production",
"feature": "customer_support"
}
)
def support_agent(ticket: str):
"""Well-tagged agent for cost analysis."""
pass
3. Implement Graceful Fallbacks
@fleet.agent(budget=30.00)
def smart_agent_with_fallback(query: str):
try:
# Try expensive model first
return gpt4_analysis(query)
except BudgetExceededError:
# Fall back to cheaper alternative
return gpt35_analysis(query)
4. Monitor and Alert
@fleet.agent(
budget=100.00,
alert_threshold=0.7, # Alert at 70%
tags={"critical": "true"}
)
def production_agent(data: str):
"""Critical production agent with early warnings."""
pass