Early Exit Workflows with Conditional Routing

Comprehensive guide for implementing early exit workflows with conditional routing in Bast AI Agno.

Reference: Agno Early Stopping Documentation

---

Overview

Early exit workflows use conditional routing to determine whether to continue with the main workflow or exit early to specialized agents (greeting, end conversation, etc.). This pattern is useful for:

• Greeting Detection: Routing simple greetings to a greeting agent
• Conversation Ending: Detecting ending phrases and routing to end conversation agent
• Main Workflow: Continuing with the main workflow for substantive questions

---

Architecture

Workflow Structure

Input → Early Exit Agent → Routing Function → Conditional Wrappers → Agents
                                ↓
                    ┌───────────┴───────────┐
                    ↓                       ↓
            Routing Decision        Conditional Execution
                    ↓                       ↓
            {route: "GREETING"}    → greeting_agent
            {route: "END_CONVERSATION"} → end_conversation_agent
            {route: "KEEP_GOING"}  → main_workflow_agent

Components

1. Early Exit Agent: Analyzes input and returns EarlyExitDecision
2. Routing Function: Processes decision and returns routing information
3. Conditional Wrappers: Check routing decision and execute appropriate agent
4. Agents: Specialized agents for different routes (greeting, end_conversation, main_workflow)

---

Implementation Guide

Step 1: Create Agents

from bast_ai_agno.bp import AgnoWorkflowFactory
from bast_ai_agno.dto.agents.core_agent_configs import (
    early_exit_agent_config,
    greeting_agent_config,
    end_conversation_agent_config,
    basic_agent_config,
)

factory = AgnoWorkflowFactory(catname="your_catname")

# Create early exit agent
early_exit_agent = factory._agent_factory.create_openai_chat_agent(
    name="early_exit_agent",
    config_raw={
        **early_exit_agent_config,
        "model": "gpt-4o-mini",  # Fast model for quick decisions
    }
)

# Create greeting agent
greeting_agent = factory._agent_factory.create_openai_chat_agent(
    name="greeting_agent",
    config_raw={
        **greeting_agent_config,
        "model": "gpt-4o-mini",
    }
)

# Create end conversation agent
end_conversation_agent = factory._agent_factory.create_openai_chat_agent(
    name="end_conversation_agent",
    config_raw={
        **end_conversation_agent_config,
        "model": "gpt-4o-mini",
    }
)

# Create main workflow agent
main_workflow_agent = factory._agent_factory.create_openai_chat_agent(
    name="main_workflow_agent",
    config_raw={
        **basic_agent_config,
        "instructions": [
            "You are a helpful assistant.",
            "Answer the user's question clearly and helpfully."
        ],
        "model": "gpt-4o-mini",
    }
)

Step 2: Create Conditional Wrappers

from bast_ai_agno.dto.workflows.conditional_agent_routing import (
    make_conditional_greeting_wrapper,
    make_conditional_end_conversation_wrapper,
    make_conditional_main_workflow_wrapper,
)

# Create conditional wrappers
conditional_greeting = make_conditional_greeting_wrapper(greeting_agent)
conditional_end_conversation = make_conditional_end_conversation_wrapper(end_conversation_agent)
conditional_main_workflow = make_conditional_main_workflow_wrapper(main_workflow_agent)

Step 3: Create Workflow

from bast_ai_agno.dto.workflows import route_by_early_exit_decision

# Create workflow with conditional routing
workflow = factory.create_workflow(
    name="early_exit_workflow",
    steps=[
        early_exit_agent,  # Step 1: Analyze input
        route_by_early_exit_decision,  # Step 2: Route based on decision
        conditional_greeting,  # Step 3: Execute if route is GREETING
        conditional_end_conversation,  # Step 4: Execute if route is END_CONVERSATION
        conditional_main_workflow,  # Step 5: Execute if route is KEEP_GOING
    ]
)

Step 4: Run Workflow

# Run workflow
response = workflow.run(
    input="Hello",
    stream=False
)

print(response.content)
# Output: "Hi there, how can I help you today?"

---

Routing Logic

EarlyExitDecision

The early exit agent returns an EarlyExitDecision with:

• action: HARD_EXIT, SOFT_EXIT, or KEEP_GOING
• reason: GREETING_MSG, HAS_ENDING_PHRASE, EMPTY_INPUT, NO_EXIT, etc.

Routing Function

The routing function (route_by_early_exit_decision) processes the decision and returns:

# For GREETING
{"route": "GREETING", "action": "HARD_EXIT", "reason": "GREETING_MSG", "input": "Hello"}

# For END_CONVERSATION
{"route": "END_CONVERSATION", "action": "HARD_EXIT", "reason": "HAS_ENDING_PHRASE", "input": "Thank you, goodbye"}

# For KEEP_GOING
{"route": "KEEP_GOING", "action": "KEEP_GOING", "reason": "NO_EXIT", "input": "What is the weather?"}

Conditional Wrappers

Each conditional wrapper: 1. Receives the routing decision from the previous step 2. Parses the routing decision (handles dict, string, or Pydantic model) 3. Checks if the route matches its condition 4. Executes the agent if the route matches, otherwise passes through

---

Common Issues and Solutions

Issue 1: Step Objects Not Recognized

Problem: Step objects from agno.workflow are not recognized by AgnoWorkflowFactory.

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

# ❌ WRONG - Step objects are not recognized
from agno.workflow import Step

workflow = factory.create_workflow(
    name="my_workflow",
    steps=[
        Step(name="route", executor=route_function),  # Won't work!
    ]
)

# ✅ CORRECT - Use functions directly
workflow = factory.create_workflow(
    name="my_workflow",
    steps=[
        route_function,  # Works!
    ]
)

Why: The factory's check_and_define_step method only recognizes Agent, Team, or callable objects. Step objects are not recognized, so they're skipped.

Issue 2: Input Not Available in Conditional Wrappers

Problem: Original input is not accessible in conditional wrappers.

Solution: Pass the original input through the routing decision.

# In routing function
routing_dict = {
    "route": "GREETING",
    "action": action,
    "reason": reason,
    "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', '')

Issue 3: Routing Decision Not Parsed Correctly

Problem: Routing decision is passed as a string representation instead of a dict.

Solution: Handle multiple formats in conditional wrappers.

def conditional_wrapper(step_input: StepInput) -> StepOutput:
    routing_decision = step_input.previous_step_content or {}

    # Parse routing decision - handle different formats
    route = None
    if isinstance(routing_decision, dict):
        route = routing_decision.get('route')
    elif isinstance(routing_decision, str):
        import json
        try:
            parsed = json.loads(routing_decision)
            if isinstance(parsed, dict):
                route = parsed.get('route')
        except:
            # Fallback: parse string representation
            if "GREETING_MSG" in routing_decision:
                route = "GREETING"
            # ... handle other cases

---

Performance Optimization

Agent/Workflow Reuse

CRITICAL: Always reuse agents and workflows. Creating them in loops causes significant overhead.

# ❌ WRONG - Creates workflow every time
def process_request(input_text):
    factory = AgnoWorkflowFactory(catname="testcat")
    workflow = factory.create_workflow(...)
    return workflow.run(input_text)

# ✅ CORRECT - Reuse workflow
_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()
    return workflow.run(input_text)

Impact: Reusing workflows can reduce execution time by 50-80%.

Model Selection

Use fast models for quick decisions:

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

Minimal Configuration

Disable unnecessary features for early exit decisions:

early_exit_agent_config = {
    "enable_agentic_memory": False,
    "enable_user_memories": False,
    "add_history_to_context": True,
    "num_history_runs": 1,  # Minimal history
    "search_knowledge": False,
    "model": "gpt-4o-mini",
}

See WORKFLOW_PERFORMANCE_OPTIMIZATION.md for more details.

---

Complete Example

from bast_ai_agno.bp import AgnoWorkflowFactory
from bast_ai_agno.dto.workflows import route_by_early_exit_decision
from bast_ai_agno.dto.workflows.conditional_agent_routing import (
    make_conditional_greeting_wrapper,
    make_conditional_end_conversation_wrapper,
    make_conditional_main_workflow_wrapper,
)
from bast_ai_agno.dto.agents.core_agent_configs import (
    early_exit_agent_config,
    greeting_agent_config,
    end_conversation_agent_config,
    basic_agent_config,
)

# Create factory (reuse for all workflows)
factory = AgnoWorkflowFactory(catname="your_catname")

# Create agents (reuse for all workflows)
early_exit_agent = factory._agent_factory.create_openai_chat_agent(
    name="early_exit_agent",
    config_raw={**early_exit_agent_config, "model": "gpt-4o-mini"}
)

greeting_agent = factory._agent_factory.create_openai_chat_agent(
    name="greeting_agent",
    config_raw={**greeting_agent_config, "model": "gpt-4o-mini"}
)

end_conversation_agent = factory._agent_factory.create_openai_chat_agent(
    name="end_conversation_agent",
    config_raw={**end_conversation_agent_config, "model": "gpt-4o-mini"}
)

main_workflow_agent = factory._agent_factory.create_openai_chat_agent(
    name="main_workflow_agent",
    config_raw={
        **basic_agent_config,
        "instructions": ["You are a helpful assistant."],
        "model": "gpt-4o-mini"
    }
)

# Create conditional wrappers
conditional_greeting = make_conditional_greeting_wrapper(greeting_agent)
conditional_end_conversation = make_conditional_end_conversation_wrapper(end_conversation_agent)
conditional_main_workflow = make_conditional_main_workflow_wrapper(main_workflow_agent)

# Create workflow (create once, reuse)
workflow = factory.create_workflow(
    name="early_exit_workflow",
    steps=[
        early_exit_agent,
        route_by_early_exit_decision,
        conditional_greeting,
        conditional_end_conversation,
        conditional_main_workflow,
    ]
)

# Test different inputs
test_cases = [
    "Hello",  # → greeting_agent
    "Thank you, goodbye",  # → end_conversation_agent
    "What is the weather today?",  # → main_workflow_agent
]

for test_input in test_cases:
    response = workflow.run(input=test_input, stream=False)
    print(f"Input: {test_input}")
    print(f"Response: {response.content}\n")

---

Best Practices

1. Reuse Workflows: Create workflows once and reuse them (similar to agents)
2. Use Fast Models: Use gpt-4o-mini for quick decisions
3. Minimal Configuration: Disable unnecessary features for early exit agents
4. Handle Multiple Formats: Conditional wrappers should handle dict, string, and Pydantic models
5. Pass Input Through: Include original input in routing decision for conditional wrappers
6. Use Functions Directly: Don't wrap functions in Step objects - use functions directly
7. JSON Serialization: Serialize routing decisions as JSON strings for reliable passing

---

Testing

Integration Test

See integration/workflows/early_exit_workflow_check_optimized.py for a complete example.

# Run integration test
pants run src/python/agno/integration/workflows/early_exit_workflow_check_optimized.py

Test Cases

1. Greeting Input: "Hello" → Should route to greeting_agent
2. Ending Phrase: "Thank you, goodbye" → Should route to end_conversation_agent
3. Keep Going: "What is the weather today?" → Should route to main_workflow_agent

---

Troubleshooting

Workflow Only Has 1 Step

Cause: Functions wrapped in Step objects are not recognized.

Solution: Use functions directly in the steps list.

Routing Decision Not Parsed

Cause: Routing decision is passed as a string representation.

Solution: Handle multiple formats in conditional wrappers (dict, string, Pydantic model).

Input Not Available in Conditional Wrappers

Cause: Original input is not passed through the workflow.

Solution: Include original input in the routing decision JSON.

Performance Issues

Cause: Workflows/agents are recreated for each request.

Solution: Reuse workflows and agents (see Performance Optimization section).

---

References

• Agno Early Stopping
• Agno Workflows
• WORKFLOW_PERFORMANCE_OPTIMIZATION.md
• WORKFLOWS.md