Skip to main content
Create a scoped run context manager that tracks metrics and optionally enforces constraints for a block of agent execution.

Signature

def run(
    budget: Optional[float] = None,
    max_tool_calls: Optional[int] = None,
    max_latency_ms: Optional[float] = None,
    max_energy: Optional[float] = None,
    compliance: Optional[str] = None,
    kpi_weights: Optional[dict[str, float]] = None,
    kpi_targets: Optional[dict[str, float]] = None,
) -> ContextManager[HarnessRunContext]

Parameters

ParameterTypeDefaultDescription
budgetfloat | NoneNoneMax USD for this run
max_tool_callsint | NoneNoneMax tool/function calls
max_latency_msfloat | NoneNoneMax wall-clock ms per call
max_energyfloat | NoneNoneMax energy units
compliancestr | NoneNone"gdpr", "hipaa", "pci", or "strict"
kpi_weightsdict | NoneNoneKPI dimension weights
kpi_targetsdict | NoneNoneKPI dimension targets

Returns

Context manager yielding HarnessRunContext. See HarnessRunContext.

Usage

Basic budget

with cascadeflow.run(budget=0.50) as session:
    result = await agent.run("Analyze this data")
    print(session.summary())

Full configuration

with cascadeflow.run(
    budget=1.00,
    max_tool_calls=10,
    max_energy=100.0,
    compliance="gdpr",
    kpi_weights={"quality": 0.6, "cost": 0.3, "latency": 0.1},
    kpi_targets={"quality": 0.9},
) as session:
    result = await agent.run("Process EU customer data")
    print(session.summary())
    for record in session.trace():
        print(f"Step {record['step']}: {record['action']}")

Nested runs

Runs can be nested. Inner runs inherit the parent’s remaining budget:
with cascadeflow.run(budget=1.00) as outer:
    with cascadeflow.run(budget=0.30) as inner:
        await agent.run("Sub-task")
    # outer.summary() includes inner costs

Notes

  • run() requires init() to have been called first
  • Parameters override the global config for the duration of the block
  • Use session.summary() for aggregate metrics
  • Use session.trace() for per-step decision records