Nested Budgets¶
Independent spend caps per workflow stage, with automatic rollup — a child stage can never exceed its own cap or what the parent has left.
Overview¶
Nested budgets let you enforce independent spend limits at every stage of a multi-step workflow, with automatic rollup to a parent cap. You get both control and visibility:
- Control costs per stage — Cap each phase independently
- Track attribution — See exactly where money was spent
- Enforce hierarchy — Parent budgets contain child budgets
- Auto-cap safely — Children can't exceed parent's remaining budget
Quick Example¶
from shekel import budget
with budget(max_usd=10.00, name="workflow") as workflow:
# Research stage: $2 budget
with budget(max_usd=2.00, name="research"):
research_results = run_research_agent()
# Analysis stage: $5 budget
with budget(max_usd=5.00, name="analysis"):
insights = run_analysis_agent()
# Parent can spend directly too
generate_final_report()
print(f"Total cost: ${workflow.spent:.2f}")
Core Concepts¶
1. Automatic Spend Propagation¶
When a child budget exits, its spend automatically propagates to the parent:
with budget(max_usd=10.00, name="parent") as parent:
with budget(max_usd=2.00, name="child") as child:
work() # Spends $1.50
# child exits → parent.spent += $1.50
print(parent.spent) # $1.50
print(parent.spent_by_children) # $1.50
print(parent.spent_direct) # $0.00
2. Auto-Capping¶
Children are automatically capped to the parent's remaining budget:
with budget(max_usd=10.00, name="parent") as parent:
parent_work() # Spends $7.00
# Child wants $5, but only $3 left
with budget(max_usd=5.00, name="child") as child:
print(child.limit) # $3.00 (auto-capped!)
work()
Why this matters: Prevents children from exceeding parent's limit, even if their requested budget is higher.
3. Parent Locking¶
Parents cannot spend while a child is active:
with budget(max_usd=10.00, name="parent") as parent:
with budget(max_usd=2.00, name="child"):
work()
# This would raise RuntimeError:
# parent_work() # ❌ Cannot spend on parent while child active
# Child exited, parent can spend again
parent_work() # ✅ OK
Why this matters: Ensures sequential execution and clear cost attribution.
4. Named Budgets¶
Names are required when nesting budgets:
# ✅ Correct
with budget(max_usd=10, name="parent"):
with budget(max_usd=5, name="child"):
work()
# ❌ Raises ValueError
with budget(max_usd=10): # Missing name!
with budget(max_usd=5):
work()
Why this matters: Names are essential for debugging, logging, and cost attribution in complex workflows.
Advanced Features¶
Hierarchical Names¶
Use full_name to get the complete path:
with budget(max_usd=20, name="pipeline") as root:
with budget(max_usd=10, name="processing") as proc:
with budget(max_usd=5, name="validation") as val:
print(val.full_name) # "pipeline.processing.validation"
Cost Attribution¶
Track where money was spent:
with budget(max_usd=20.00, name="workflow") as w:
with budget(max_usd=5.00, name="stage1"):
work1() # $3.00
with budget(max_usd=8.00, name="stage2"):
work2() # $6.00
work_parent() # $2.00
print(f"Total: ${w.spent:.2f}") # $11.00
print(f"Direct: ${w.spent_direct:.2f}") # $2.00
print(f"Children: ${w.spent_by_children:.2f}") # $9.00
Budget Tree Visualization¶
Use tree() to visualize the hierarchy:
with budget(max_usd=50.00, name="pipeline") as p:
with budget(max_usd=10.00, name="ingestion") as i:
ingest()
with budget(max_usd=20.00, name="processing") as proc:
with budget(max_usd=8.00, name="validation") as v:
validate()
with budget(max_usd=12.00, name="transform") as t:
transform()
generate()
print(p.tree())
# pipeline: $35.50 / $50.00 (direct: $2.00)
# ingestion: $8.50 / $10.00 (direct: $8.50)
# processing: $25.00 / $20.00 (direct: $0.00)
# validation: $12.00 / $8.00 (direct: $12.00)
# transform: $13.00 / $12.00 (direct: $13.00)
Track-Only Children¶
Create unlimited child budgets for tracking:
with budget(max_usd=20.00, name="workflow") as w:
# Track exploration without limits
with budget(max_usd=None, name="exploration"):
explore_options() # No limit!
# But finalization is capped
with budget(max_usd=5.00, name="finalization"):
finalize()
print(f"Exploration cost: ${w.children[0].spent:.2f}")
Real-World Examples¶
Multi-Agent System¶
def research_agent(topic: str, budget_usd: float = 10.0):
"""Research agent with per-stage budgets."""
with budget(max_usd=budget_usd, name="research_agent") as agent:
# Stage 1: Web search
with budget(max_usd=2.00, name="web_search"):
results = search_web(topic)
# Stage 2: Content analysis
with budget(max_usd=5.00, name="analysis"):
insights = analyze(results)
# Stage 3: Report generation
with budget(max_usd=3.00, name="report"):
report = generate_report(insights)
print(agent.tree())
return report
LangGraph Integration¶
from langgraph.graph import StateGraph
from shekel import budget
def create_workflow():
"""LangGraph workflow with nested budgets."""
workflow = StateGraph(AgentState)
# Wrap entire workflow
with budget(max_usd=20.00, name="langgraph_workflow") as main:
# Research node
def research_node(state):
with budget(max_usd=5.00, name="research"):
return {"research": run_research(state)}
# Analysis node
def analysis_node(state):
with budget(max_usd=10.00, name="analysis"):
return {"analysis": run_analysis(state)}
workflow.add_node("research", research_node)
workflow.add_node("analysis", analysis_node)
workflow.add_edge("research", "analysis")
app = workflow.compile()
result = app.invoke({"query": "..."})
print(f"Total cost: ${main.spent:.2f}")
print(main.tree())
CrewAI Per-Agent Budgets¶
from crewai import Crew, Agent
from shekel import budget
def run_crew():
"""CrewAI with per-agent budgets."""
with budget(max_usd=15.00, name="crew") as crew_budget:
# Researcher agent
with budget(max_usd=5.00, name="researcher"):
researcher = Agent(
role="Research Analyst",
goal="Find relevant information"
)
researcher.execute_task(research_task)
# Writer agent
with budget(max_usd=8.00, name="writer"):
writer = Agent(
role="Content Writer",
goal="Write comprehensive report"
)
writer.execute_task(writing_task)
print(crew_budget.tree())
Best Practices¶
1. Always Name Nested Budgets¶
# ❌ Bad: No names
with budget(max_usd=10):
with budget(max_usd=5):
work()
# ✅ Good: Clear names
with budget(max_usd=10, name="workflow"):
with budget(max_usd=5, name="processing"):
work()
2. Budget for Each Major Stage¶
with budget(max_usd=20, name="pipeline") as p:
with budget(max_usd=5, name="ingestion"):
ingest()
with budget(max_usd=10, name="processing"):
process()
with budget(max_usd=5, name="output"):
output()
3. Use tree() for Debugging¶
try:
with budget(max_usd=10, name="workflow") as w:
run_workflow()
except BudgetExceededError:
print("Budget exceeded! Cost breakdown:")
print(w.tree())
4. Check Auto-Capping¶
with budget(max_usd=10, name="parent") as parent:
expensive_operation() # Might use most of budget
with budget(max_usd=5, name="child") as child:
if child.limit < 5.00:
print(f"⚠️ Child was auto-capped to ${child.limit:.2f}")
# Maybe use a cheaper model?
Limitations¶
Maximum Nesting Depth¶
Maximum depth is 5 levels (root at depth 0):
with budget(max_usd=100, name="L0"): # Depth 0 ✅
with budget(max_usd=50, name="L1"): # Depth 1 ✅
with budget(max_usd=25, name="L2"): # Depth 2 ✅
with budget(max_usd=12, name="L3"): # Depth 3 ✅
with budget(max_usd=6, name="L4"): # Depth 4 ✅
work() # ✅ OK
# Depth 5 would raise ValueError
# with budget(max_usd=3, name="L5"): # ❌
Async Nesting Not Supported¶
# ❌ Async nesting raises RuntimeError
async with budget(max_usd=10, name="parent"):
async with budget(max_usd=5, name="child"): # ❌ Error!
await work()
# ✅ Use sync nested budgets instead
with budget(max_usd=10, name="parent"):
with budget(max_usd=5, name="child"):
await work() # ✅ OK
Unique Sibling Names¶
Children under the same parent must have unique names:
with budget(max_usd=10, name="parent"):
with budget(max_usd=3, name="stage1"): # ✅
work()
with budget(max_usd=3, name="stage1"): # ❌ Duplicate!
work()
Migration Notes¶
Accumulating Budgets¶
Budget variables always accumulate across uses. Create a new budget() instance if you need a fresh budget:
# Accumulates across uses
b = budget(max_usd=10.00)
with b: work1() # Spends $2
with b: work2() # Spends $2 more — b.spent == $4
# Fresh budget each time
with budget(max_usd=10.00): work1() # $2 (separate budget)
with budget(max_usd=10.00): work2() # $2 (separate budget)
API Reference¶
See API Reference for complete details:
budget.parent— Parent budget referencebudget.children— List of child budgetsbudget.active_child— Currently active childbudget.full_name— Hierarchical path namebudget.spent_direct— Direct spend (excluding children)budget.spent_by_children— Sum of child spendbudget.tree()— Visual hierarchy
Examples¶
See examples/ for runnable demos:
nested_research_agent.py— Multi-stage research workflownested_content_pipeline.py— Real OpenAI API integration