Workflow Configuration Documentation

Comprehensive documentation for workflow configurations in Bast AI Agno.

Reference: Agno Workflow Documentation

---

Overview

Agno Workflows enable deterministic, controlled agentic flows by orchestrating agents, teams, and functions through a series of defined steps. This provides structured automation with predictable execution patterns, making them ideal for production systems that require reliable, repeatable processes.

---

Quick Start

Basic Usage

from bast_ai_agno.bp import AgnoWorkflowFactory
from bast_ai_agno.dto.workflows import simple_workflow_config

# Create factory
factory = AgnoWorkflowFactory(catname="your_catname")

# Create workflow from config
workflow = factory.create_workflow_from_json({
    "name": "my_workflow",
    **simple_workflow_config
})

# Run workflow
response = workflow.run(
    input="What are the main benefits of using AI in healthcare?",
    stream=False
)

print(response.content)

Running a Workflow

Based on Agno Workflow Documentation:

from bast_ai_agno.bp import AgnoWorkflowFactory
from bast_ai_agno.dto.workflows import simple_workflow_config
from agno.run.workflow import WorkflowRunOutput

factory = AgnoWorkflowFactory(catname="your_catname")

# Create workflow
workflow = factory.create_workflow_from_json({
    "name": "content_workflow",
    **simple_workflow_config
})

# Run workflow synchronously
response: WorkflowRunOutput = workflow.run(
    input="AI trends in 2024",
    stream=False
)

# Access response content
print(response.content)

Async Execution

import asyncio

async def main():
    factory = AgnoWorkflowFactory(catname="your_catname")
    workflow = factory.create_workflow_from_json({
        "name": "async_workflow",
        **simple_workflow_config
    })

    # Run workflow asynchronously
    response: WorkflowRunOutput = await workflow.arun(
        input="Recent breakthroughs in quantum computing"
    )

    print(response.content)

if __name__ == "__main__":
    asyncio.run(main())

---

Module Structure

Business Process Modules

• bp/agno_workflows.py - AgnoWorkflowFactory class for creating workflows from configuration/templates
• bp/mongodb_resources.py - MongoDB storage resources with separate workflow database

Configuration Modules

• dto/workflows/simple_workflow_config.py - Simple workflow configuration template
• dto/workflows/__init__.py - Workflow configuration exports

Integration Tests

• integration/workflows/simple_workflow_check.py - Integration tests for workflows

---

Workflow Factory

AgnoWorkflowFactory

The AgnoWorkflowFactory class provides methods to create workflows from JSON configurations and templates.

Methods

1. create_workflow_from_json(workflow_json)
2. Creates a workflow from JSON configuration
3. Handles nested config structures

4. Returns a Workflow object

5. create_workflow(name, steps, db, config_raw)

6. Creates a workflow with direct parameters
7. Supports steps as agents, teams, or functions

8. Automatically configures workflow storage

9. check_and_define_step(step_data)

10. Analyzes step data and creates appropriate step type
11. Supports agents, teams, and functions

12. Handles nested configurations

13. create_workflow_from_template(template_name, **kwargs)

14. Creates workflow from predefined template (ready for extension)
15. Template system for common workflow patterns

Example Usage

from bast_ai_agno.bp import AgnoWorkflowFactory

factory = AgnoWorkflowFactory(catname="your_catname")

# Create from JSON
workflow = factory.create_workflow_from_json({
    "name": "my_workflow",
    "steps": [
        {"agent": {"name": "researcher", "instructions": ["Research topics"]}},
        {"agent": {"name": "writer", "instructions": ["Write content"]}}
    ]
})

# Create directly
workflow = factory.create_workflow(
    name="my_workflow",
    steps=[agent1, team1, custom_function]
)

---

Workflow Storage

Separate MongoDB Database

Workflows use a separate MongoDB database dedicated to workflow storage, distinct from agent storage. This ensures workflow execution data is stored independently, facilitating better organization, scalability, and maintenance.

Database Structure

• Database Name: {prefix}_agno_workflow
• Collections:
• {workflow_name}_sessions - Workflow session data
• {workflow_name}_memory - Workflow memory
• {workflow_name}_metrics - Workflow metrics
• {workflow_name}_eval - Evaluation data
• {workflow_name}_kb - Knowledge base data

Usage

from bast_ai_agno.bp import AgnoMongoDBResources

resources = AgnoMongoDBResources(catname="your_catname")

# Get workflow storage
workflow_storage = resources.get_workflow_storage("my_workflow")

# Use in workflow creation
workflow = factory.create_workflow(
    name="my_workflow",
    db=workflow_storage,
    steps=[...]
)

---

Workflow Configurations

Simple Workflow Configuration

The simple_workflow_config provides a basic workflow template with two sequential agents:

from bast_ai_agno.dto.workflows import simple_workflow_config

# Simple workflow with researcher and writer agents
simple_workflow_config = {
    "steps": [
        {
            "agent": {
                "name": "researcher",
                **basic_agent_config,
                "instructions": [
                    "You are a research assistant.",
                    "Gather information and provide comprehensive summaries."
                ],
                "model": "gpt-4o-mini",
            }
        },
        {
            "agent": {
                "name": "writer",
                **basic_agent_config,
                "instructions": [
                    "You are a content writer.",
                    "Create well-structured content based on the research provided."
                ],
                "model": "gpt-4o-mini",
            }
        }
    ]
}

Custom Workflow Configuration

You can create custom workflow configurations:

from bast_ai_agno.dto.agents.agent_configs import basic_agent_config

custom_workflow_config = {
    "steps": [
        # Step 1: Research agent
        {
            "agent": {
                "name": "researcher",
                **basic_agent_config,
                "instructions": ["Research the topic"],
                "model": "gpt-4o",
            }
        },
        # Step 2: Analysis team
        {
            "team": {
                "name": "analysis_team",
                "members": [
                    {"name": "analyst1"},
                    {"name": "analyst2"}
                ],
                "model": "gpt-4o",
            }
        },
        # Step 3: Custom function
        my_custom_processing_function
    ]
}

---

Workflow Steps

Workflow steps can be:

1. Agents - Individual AI executors
2. Teams - Coordinated groups of agents
3. Functions - Custom Python functions (async functions)

Step Definition

Steps can be defined in multiple ways:

As Dictionary

{
    "agent": {
        "name": "agent_name",
        "instructions": ["..."],
        "model": "gpt-4o-mini"
    }
}

As Object

from bast_ai_agno.bp import AgnoAgentFactory

agent_factory = AgnoAgentFactory(catname="your_catname")
agent = agent_factory.create_openai_chat_agent(
    name="my_agent",
    config_raw=basic_agent_config
)

# Use directly in workflow
workflow = factory.create_workflow(
    name="my_workflow",
    steps=[agent]  # Use agent object directly
)

As Function

async def custom_processing(step_input):
    # Process input
    result = await some_processing(step_input.previous_step_content)
    return result

# Use in workflow
workflow = factory.create_workflow(
    name="my_workflow",
    steps=[custom_processing]  # Use function directly
)

---

Integration Tests

Running Workflow Tests

# Run workflow integration test
pants run src/python/agno/integration:simple_workflow_check

Test Coverage

The integration test (simple_workflow_check.py) covers:

1. Factory Initialization - Verifies factory can be created
2. Workflow Creation from Config - Tests JSON-based workflow creation
3. Direct Workflow Creation - Tests direct workflow creation
4. Step Definition - Validates workflow steps are correctly defined
5. Storage Configuration - Verifies workflow database is configured
6. Workflow Execution - Tests actual workflow execution with workflow.run()
7. Error Handling - Tests error handling with invalid inputs

---

DTO Folder Reorganization

The dto folder has been reorganized for better structure:

New Structure

dto/
├── agents/          # All agent configurations
│   ├── agent_configs.py
│   ├── core_agent_configs.py
│   ├── memory_configs.py
│   ├── knowledge_configs.py
│   └── ...
├── utils/           # All utility functions
│   ├── agent_config_utils.py
│   ├── agent_tool_utils.py
│   ├── agent_reasoning_utils.py
│   └── ...
├── workflows/       # Workflow configurations
│   ├── simple_workflow_config.py
│   └── __init__.py
└── output/          # Output schemas
    └── early_exit_output.py

Backward Compatibility

All old import paths still work:

# Old imports still work
from bast_ai_agno.dto.agent_configs import basic_agent_config
from bast_ai_agno.dto.agent_config_utils import add_tools_to_config
from bast_ai_agno.dto.core_agent_configs import greeting_agent_config

# New imports (preferred)
from bast_ai_agno.dto.agents.agent_configs import basic_agent_config
from bast_ai_agno.dto.utils.agent_config_utils import add_tools_to_config
from bast_ai_agno.dto.agents.core_agent_configs import greeting_agent_config

---

Examples

Example 1: Simple Research Workflow

from bast_ai_agno.bp import AgnoWorkflowFactory
from bast_ai_agno.dto.workflows import simple_workflow_config

factory = AgnoWorkflowFactory(catname="your_catname")

workflow = factory.create_workflow_from_json({
    "name": "research_workflow",
    **simple_workflow_config
})

response = workflow.run(
    input="What are the latest trends in AI?",
    stream=False
)

print(response.content)

Example 2: Custom Workflow with Teams

from bast_ai_agno.bp import AgnoWorkflowFactory, AgnoTeamFactory

workflow_factory = AgnoWorkflowFactory(catname="your_catname")
team_factory = AgnoTeamFactory(catname="your_catname")

# Create team
research_team = team_factory.create_team_from_json({
    "name": "research_team",
    "members": [
        {"name": "researcher1"},
        {"name": "researcher2"}
    ]
})

# Create workflow with team
workflow = workflow_factory.create_workflow(
    name="team_workflow",
    steps=[research_team]
)

response = workflow.run(input="Research quantum computing")

Example 3: Workflow with Custom Function

async def preprocess_input(step_input):
    """Preprocess input before agent processing"""
    processed = step_input.input.lower().strip()
    return {"content": processed}

async def postprocess_output(step_input):
    """Postprocess output after agent processing"""
    output = step_input.previous_step_content
    return {"content": f"[PROCESSED] {output}"}

workflow = factory.create_workflow(
    name="processing_workflow",
    steps=[
        preprocess_input,
        {"agent": {"name": "processor", "instructions": ["Process the input"]}},
        postprocess_output
    ]
)

---

Best Practices

1. Reuse Workflows - Create workflows once and reuse them (similar to agents)
2. Use Configurations - Leverage workflow configurations for consistency
3. Separate Storage - Use separate workflow databases for better organization
4. Error Handling - Wrap workflow execution in try-except blocks
5. Logging - Enable workflow event storage for debugging and auditing
6. Testing - Use integration tests to verify workflow functionality

---

Early Exit Workflows

For workflows with conditional routing based on early exit decisions, see:

• EARLY_EXIT_WORKFLOWS.md - Complete guide for early exit workflows with conditional routing
• Includes patterns for greeting detection, conversation ending, and main workflow routing
• Covers common issues and solutions (Step objects vs functions, input passing, etc.)

Quick Example

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,
)

# 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 with conditional routing
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,
    ]
)

Important Notes: - Use functions directly, not Step objects (factory doesn't recognize Step objects) - Pass original input through routing decision for conditional wrappers - Reuse workflows and agents for performance (see WORKFLOW_PERFORMANCE_OPTIMIZATION.md)

---

External Resources

• Agno Workflows Overview
• Building Workflows
• Running Workflows
• Workflow Patterns
• Agno Early Stopping

---

Getting Help

1. Check the Agno Workflow Documentation
2. Review integration tests in integration/workflows/
3. See workflow configurations in dto/workflows/
4. Refer to bp/agno_workflows.py for factory implementation
5. For early exit workflows, see EARLY_EXIT_WORKFLOWS.md