Skip to content

Workflow Performance Optimization Guide

This guide provides strategies to optimize Agno workflow performance and reduce response times.

Performance Comparison

Original Implementation

  • Response Time: ~16-18 seconds per execution
  • Issue: Workflow and agents recreated for each test/execution

Optimized Implementation

  • Response Time: ~6-8 seconds per execution
  • Improvement: ~50-60% faster by reusing workflow/agents

Key Optimization Strategies

1. Agent/Workflow Reuse (CRITICAL) ⚠️

NEVER create agents/workflows in loops or per request.

# ❌ WRONG - Recreates workflow every time (significant overhead)
def process_request(input_text):
    factory = AgnoWorkflowFactory(catname="testcat")
    workflow = factory.create_workflow(...)  # DON'T DO THIS
    return workflow.run(input_text)

# ✅ CORRECT - Create once, reuse
_workflow_cache = None

def get_or_create_workflow():
    global _workflow_cache
    if _workflow_cache is None:
        factory = AgnoWorkflowFactory(catname="testcat")
        _workflow_cache = factory.create_workflow(...)
    return _workflow_cache

def process_request(input_text):
    workflow = get_or_create_workflow()  # Reuse existing workflow
    return workflow.run(input_text)

Impact: This is the most critical optimization - can reduce overhead by 50-80%.

2. Model Selection

Use faster models for quick decisions:

# For early exit decisions - use fast, cheap model
early_exit_agent_config = {
    **early_exit_agent_config,
    "model": "gpt-4o-mini",  # Fast and cost-effective
}

# For simple responses - use fast model
greeting_agent_config = {
    **greeting_agent_config,
    "model": "gpt-4o-mini",  # Fast model
}

# For complex tasks - use more capable model
main_workflow_agent_config = {
    **basic_agent_config,
    "model": "gpt-4o",  # More capable for complex tasks
}

Model Speed Comparison: - gpt-4o-mini: Fastest, cheapest, good for simple tasks - gpt-4o: Slower but more capable, use for complex tasks - gpt-3.5-turbo: Very fast but less capable

3. Disable Unnecessary Features

For early exit decisions, disable features that add overhead:

early_exit_agent_config = {
    # Memory (disabled) - early exit decisions don't need memory
    "enable_agentic_memory": False,
    "enable_user_memories": False,

    # Storage & Sessions (minimal history)
    "add_history_to_context": True,
    "num_history_runs": 1,  # Minimal history

    # Knowledge (disabled) - no need for knowledge base
    "search_knowledge": False,

    # Fast model
    "model": "gpt-4o-mini",
}

Features to disable when not needed: - Memory (agentic/user memories) - adds overhead - Knowledge search - adds significant overhead - Extensive history - keep minimal (1-2 turns) - Tools - only if necessary

4. Early Stopping Pattern

Use early stopping to avoid unnecessary steps:

from agno.workflow import StepInput, StepOutput

def route_by_early_exit_decision(step_input: StepInput) -> StepOutput:
    # Parse decision
    decision = parse_decision(step_input.previous_step_content)

    # Stop early when needed
    if decision.action == 'HARD_EXIT':
        return StepOutput(
            content="STOP_EARLY",
            stop=True  # Stops workflow immediately
        )

    # Continue when appropriate
    return StepOutput(
        content="CONTINUE",
        stop=False
    )

Impact: Prevents unnecessary agent executions when decision is clear.

5. Workflow Structure Optimization

Structure workflow to fail fast:

# Optimized workflow structure
workflow = factory.create_workflow(
    name="optimized_workflow",
    steps=[
        # Step 1: Fast decision (early exit agent)
        early_exit_agent,  # Fast model, minimal config

        # Step 2: Routing (function - very fast)
        Step(name="route", executor=route_function),

        # Step 3: Action agents (only reached if needed)
        greeting_agent,  # Only if routed
        end_conversation_agent,  # Only if routed
        main_workflow_agent,  # Only if KEEP_GOING
    ]
)

Best Practices: - Put fast decisions first - Use function steps for routing (no LLM overhead) - Structure so common paths are shorter

6. Streaming (Optional)

For better perceived performance, use streaming:

# For async/await patterns
response = await workflow.arun(input_text, stream=True)

# For sync patterns
response = workflow.run(input_text, stream=True)

Note: Streaming improves perceived performance but doesn't reduce total time.

7. Caching Strategy

Cache workflow instances and agents:

# Global cache
_workflow_cache = {}
_agent_cache = {}

def get_cached_workflow(name: str):
    if name not in _workflow_cache:
        factory = AgnoWorkflowFactory(catname="testcat")
        _workflow_cache[name] = factory.create_workflow(...)
    return _workflow_cache[name]

def get_cached_agent(name: str, config: dict):
    cache_key = f"{name}_{hash(str(config))}"
    if cache_key not in _agent_cache:
        factory = AgnoWorkflowFactory(catname="testcat")
        _agent_cache[cache_key] = factory._agent_factory.create_openai_chat_agent(...)
    return _agent_cache[cache_key]

Performance Benchmarks

Early Exit Workflow

Original (Recreating Workflow): - Setup: ~0.8s per test - Execution: ~16-18s per test - Total: ~17-19s per test

Optimized (Reusing Workflow): - Setup: ~0.8s (one-time) - Execution: ~6-8s per test - Total: ~6-8s per test (after first setup)

Improvement: ~50-60% faster

Expected Performance

Simple Tasks (Greeting, Early Exit): - Target: 2-5 seconds - Model: gpt-4o-mini - Features: Minimal config

Complex Tasks (Main Workflow): - Target: 5-10 seconds - Model: gpt-4o or gpt-4o-mini - Features: Full config as needed

Additional Tips

  1. Monitor Performance: Use Stopwatch to measure execution times
  2. Profile Steps: Identify which steps are slowest
  3. Parallel Execution: For independent steps, consider parallel execution
  4. Reduce Context: Keep context windows small when possible
  5. Use Appropriate Models: Match model capability to task complexity

Common Mistakes

  1. ❌ Creating agents/workflows in loops
  2. ❌ Using heavy models for simple tasks
  3. ❌ Enabling unnecessary features (memory, knowledge, tools)
  4. ❌ Not using early stopping when appropriate
  5. ❌ Recreating workflows per request
  6. ❌ Wrapping functions in Step objects (factory doesn't recognize them)
  7. ❌ Not passing original input through routing decisions

Lessons Learned from Early Exit Workflow Implementation

Issue: Step Objects Not Recognized

When implementing conditional routing in workflows, we discovered that Step objects from agno.workflow are not recognized by AgnoWorkflowFactory.create_workflow(). The factory only recognizes Agent, Team, or callable objects.

Solution: Use functions directly instead of wrapping them in Step objects.

# ❌ WRONG
from agno.workflow import Step
workflow = factory.create_workflow(
    steps=[Step(name="route", executor=route_function)]  # Won't work!
)

# ✅ CORRECT
workflow = factory.create_workflow(
    steps=[route_function]  # Works!
)

Issue: Input Not Available in Conditional Wrappers

When implementing conditional wrappers, we found that the original input is not directly accessible in workflow steps.

Solution: Pass the original input through the routing decision as JSON.

# In routing function
routing_dict = {
    "route": "GREETING",
    "input": original_input  # Include original input
}
return StepOutput(
    content=json.dumps(routing_dict),  # Serialize as JSON
    stop=False
)

# In conditional wrapper
routing_decision = json.loads(step_input.previous_step_content)
user_input = routing_decision.get('input', '')

Performance Impact

Implementing conditional routing with reusable workflows: - Setup Time: ~0.8s (one-time) - Execution Time: ~6-8s per test (after first setup) - Improvement: ~50-60% faster than recreating workflows per request

See EARLY_EXIT_WORKFLOWS.md for complete implementation details.

References