CascadeAgent (Python)

CascadeAgent (Python)

The main agent class for intelligent AI model cascading with automatic cost optimization.

Class: CascadeAgent

from cascadeflow import CascadeAgent, ModelConfig

Constructor

CascadeAgent(
    models: List[ModelConfig],
    quality_config: Optional[QualityConfig] = None,
    cascade: Optional[CascadeConfig] = None
)

Parameters:

• models (List[ModelConfig], required): List of models to cascade through, automatically sorted by cost
• quality_config (QualityConfig, optional): Quality validation configuration
• cascade (CascadeConfig, optional): Advanced cascade settings

Example:

from cascadeflow import CascadeAgent, ModelConfig

agent = CascadeAgent(models=[
    ModelConfig(name="gpt-4o-mini", provider="openai", cost=0.00015),
    ModelConfig(name="gpt-4o", provider="openai", cost=0.00625)
])

Methods

run()

Execute a query with automatic cascading.

async def run(
    prompt: str | List[Message],
    max_tokens: Optional[int] = None,
    temperature: Optional[float] = None,
    system_prompt: Optional[str] = None,
    tools: Optional[List[Dict]] = None,
    force_direct: bool = False
) -> CascadeResult

Parameters:

• prompt (str or List[Message]): Query text or message list
• max_tokens (int, optional): Maximum tokens to generate
• temperature (float, optional): Temperature (0-2), default: 0.7
• system_prompt (str, optional): System prompt override
• tools (List[Dict], optional): Tools/functions available for calling
• force_direct (bool, optional): Skip cascade, use best model directly

Returns: CascadeResult - Result object with content, costs, and metrics

Example:

# Basic usage
result = await agent.run("What is Python?")
print(result.content)
print(f"Cost: ${result.total_cost:.6f}")

# With options
result = await agent.run(
    "Explain quantum computing",
    max_tokens=500,
    temperature=0.3,
    system_prompt="You are a physics expert"
)

stream()

Stream responses with real-time events.

async def stream(
    prompt: str | List[Message],
    max_tokens: Optional[int] = None,
    temperature: Optional[float] = None,
    system_prompt: Optional[str] = None,
    tools: Optional[List[Dict]] = None
) -> AsyncIterator[StreamEvent]

Parameters: Same as run()

Yields: StreamEvent - Stream events with incremental content

Example:

async for event in agent.stream("Tell me a story"):
    if event.type == StreamEventType.CONTENT_DELTA:
        print(event.content, end="", flush=True)
    elif event.type == StreamEventType.COMPLETE:
        print(f"\nCost: ${event.total_cost:.6f}")

Presets

Create an agent from a production preset.

Recommended API: auto_agent(preset=...)

Example:

from cascadeflow import auto_agent

# Balanced performance
agent = auto_agent(preset="balanced")

# Optimized for speed
agent_fast = auto_agent(preset="speed_optimized")

Complete Examples

Basic Query

from cascadeflow import CascadeAgent, ModelConfig

agent = CascadeAgent(models=[
    ModelConfig(name="gpt-4o-mini", provider="openai", cost=0.00015),
    ModelConfig(name="gpt-4o", provider="openai", cost=0.00625)
])

result = await agent.run("What is the capital of France?")

print(f"Answer: {result.content}")
print(f"Model used: {result.model_used}")
print(f"Cost: ${result.total_cost:.6f}")
print(f"Savings: {result.cost_saved_percentage:.1f}%")

With Quality Configuration

from cascadeflow import CascadeAgent, ModelConfig, QualityConfig

agent = CascadeAgent(
    models=[
        ModelConfig(name="gpt-4o-mini", provider="openai", cost=0.00015),
        ModelConfig(name="gpt-4o", provider="openai", cost=0.00625)
    ],
    quality_config=QualityConfig(
        threshold=0.8,  # Stricter quality requirements
        require_minimum_tokens=20
    )
)

result = await agent.run("Explain machine learning")

Multi-Provider Setup

agent = CascadeAgent(models=[
    # Groq (fastest)
    ModelConfig(name="llama-3.1-8b", provider="groq", cost=0.00005),
    # OpenAI (balanced)
    ModelConfig(name="gpt-4o-mini", provider="openai", cost=0.000375),
    # Anthropic (quality)
    ModelConfig(name="claude-sonnet-4-5", provider="anthropic", cost=0.003)
])

Tool Calling

tools = [
    {
        "name": "get_weather",
        "description": "Get current weather for a location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City name"}
            },
            "required": ["location"]
        }
    }
]

result = await agent.run(
    "What's the weather in San Francisco?",
    tools=tools
)

if result.has_tool_calls:
    for call in result.tool_calls:
        print(f"Tool: {call['name']}")
        print(f"Args: {call['arguments']}")

Streaming with Events

async for event in agent.stream("Write a short poem about AI"):
    match event.type:
        case StreamEventType.MODEL_SELECTED:
            print(f"Using model: {event.model}")

        case StreamEventType.CONTENT_DELTA:
            print(event.content, end="", flush=True)

        case StreamEventType.CASCADE_TRIGGERED:
            print("\n[Escalating to better model...]")

        case StreamEventType.COMPLETE:
            print(f"\nTotal cost: ${event.total_cost:.6f}")

See Also

• ModelConfig - Model configuration options (includes QualityConfig)
• CascadeResult - Result object documentation
• Streaming Guide - Streaming responses
• Tools Guide - Function calling
• Presets Guide - Built-in preset configurations