Database Migration

While your pipeline logic is dynamically discovered by the code, the Admin Dashboard requires the pipeline to be registered in the database to display it in the user interface.

Quick Summary

After generating your pipeline with make new-pipeline, simply run:

make migrate

That's it! The migration automatically:

• Registers your pipeline across all organizations

• Creates necessary components and presets

• Sets up default configuration fields

• Prepares everything for the Admin Dashboard

No modification needed unless you want to add custom configuration fields (which also requires creating a PresetConfig class).

---

Overview

When you use the make new-pipeline CLI tool, a migration file is automatically created for you in migration/versions/ with fully functional upgrade() and downgrade() functions. In most cases, you don't need to modify the generated migration file - it's ready to use as-is.

The generated migration handles:

• Pipeline registration across all organizations (multi-tenancy support)

• Component creation and linking

• Preset creation with default configuration

• Configuration fields and values setup

• Catalog creation for prompt builders and language models

• Chatbot application creation (for non-global pipelines)

Migration File Structure

The CLI tool generates a migration file with the following structure:

migration/versions/v<version>_add_<pipeline_name>_pipeline.py

The file includes:

• Proper Alembic revision tracking with 12-character hex revision ID

• Auto-detected down_revision from the latest migration

• Fully implemented upgrade() and downgrade() functions - ready to use!

• Comprehensive pipeline setup for all organizations

• Built-in multi-tenancy support

• Default configuration fields with customization examples

Applying the Migration

The generated migration file is ready to use. Simply apply it:

make migrate

This will:

1. Create the pipeline for all organizations in your system

2. Set up all required components and presets

3. Configure default settings

4. Create catalogs and chatbot applications

The migration is ready to use as-is! You don't need to modify it unless you want to add custom configuration fields.

What does the generated migration look like?

The CLI tool generates a comprehensive migration file that handles all aspects of pipeline registration:

"""Add my-custom-pipeline pipeline.

Authors:
    John Doe ([e-mail])

Revision ID: c122cbd4e5ab
Create Date: 2024-01-15
"""

from collections.abc import Sequence

from alembic import op
from sqlalchemy.sql import select

from migration.pipeline_util import (
    TABLES,
    add_config_fields,
    add_config_values,
    create_catalogs,
    create_chatbot,
    create_pipeline_and_components,
    create_preset,
    delete_chatbots_and_related_items,
    delete_config_fields_and_components,
    get_all_pipeline_ids,
)

# Optional imports for customizing config fields:
# from migration.pipeline_util import ConfigCategory, get_config_category_id

# revision identifiers, used by Alembic.
revision: str = "c122cbd4e5ab"
down_revision: str | None = "b5b4fcfe43a7"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

def upgrade() -> None:
    """Add my-custom-pipeline pipeline configuration.

    Performs the database migration to add the my-custom-pipeline pipeline type with all its
    associated components, presets, and configuration fields.

    This includes:
    1. Creating the pipeline and components for each organization
    2. Creating a default preset for each pipeline
    3. Setting up configuration fields and values
    """
    conn = op.get_bind()

    # Get all organization IDs from organizations table
    org_result = conn.execute(select(TABLES["organizations"].c.id)).fetchall()
    organization_ids = [row[0] for row in org_result]
    organization_ids.append(None)

    # Create pipelines for remaining organizations (reusing the same components)
    for org_id in organization_ids:
        # Create pipeline without creating new components
        (
            pipeline_id,
            my_custom_pipeline_component_id,
            rago_global_component_id,
        ) = create_pipeline_and_components(
            conn=conn,
            name="my-custom-pipeline",
            display_name="My Custom Pipeline",
            description="My Custom Pipeline Description",
            organization_id=org_id,
            pipelines=TABLES["pipelines"],
            components=TABLES["components"],
            pipeline_components=TABLES["pipeline_components"],
        )

        # Create preset for this pipeline
        preset_id = create_preset(conn, pipeline_id, TABLES["presets"], organization_id=org_id)

        # Add config fields - uses default RAG-specific config fields
        # To customize config fields, uncomment and modify the example below:
        #
        # from migration.pipeline_util import get_config_category_id, ConfigCategory
        #
        # retrieval_config_cat_id = get_config_category_id(
        #     conn, ConfigCategory.RETRIEVAL_KNOWLEDGE_CONFIGURATION, org_id
        # )
        # search_behavior_cat_id = get_config_category_id(
        #     conn, ConfigCategory.SEARCH_RETRIEVAL_BEHAVIOR, org_id
        # )
        #
        # custom_config_fields = {
        #     "my_custom_field": {
        #         "type": "string",
        #         "default_value": "custom_value",
        #         "ui_type": "text_field",
        #         "level": "PRESET_CHATBOT",
        #         "category_id": retrieval_config_cat_id,
        #     },
        #     "my_bool_field": {
        #         "type": "bool",
        #         "default_value": "true",
        #         "ui_type": "toggle",
        #         "level": "PRESET_CHATBOT",
        #         "category_id": search_behavior_cat_id,
        #     },
        #     # Add more custom fields as needed
        # }
        #
        # add_config_fields(
        #     conn, my_custom_pipeline_component_id, TABLES, org_id, custom_config_fields
        # )

        add_config_fields(conn, my_custom_pipeline_component_id, TABLES, organization_id=org_id)

        # Add config values for this pipeline's preset
        add_config_values(
            conn,
            preset_id,
            my_custom_pipeline_component_id,
            rago_global_component_id,
            TABLES,
            organization_id=org_id,
        )

        if org_id is None:
            create_catalogs(
                conn=conn,
                pipeline_id=pipeline_id,
                pipeline_name="my-custom-pipeline",
                chatbot_prompt_builder_catalog_id=None,
                chatbot_lmrp_catalog_id=None,
                organization_id=org_id,
                tables=TABLES,
            )
            continue

        # Check if chatbot already exists to avoid duplicate key errors
        existing_chatbot = conn.execute(
            select(TABLES["chatbots"].c.prompt_builder_catalog_id, TABLES["chatbots"].c.lmrp_catalog_id)
            .where(TABLES["chatbots"].c.chatbot_id == "my-custom-pipeline-application")
            .where(TABLES["chatbots"].c.organization_id == org_id)
        ).first()

        if existing_chatbot:
            chatbot_prompt_builder_catalog_id = existing_chatbot[0]
            chatbot_lmrp_catalog_id = existing_chatbot[1]
        else:
            (chatbot_prompt_builder_catalog_id, chatbot_lmrp_catalog_id) = create_chatbot(
                conn=conn,
                preset_id=preset_id,
                tables=TABLES,
                chatbot_id_value="my-custom-pipeline-application",
                chatbot_name="My Custom Pipeline Application",
                chatbot_description="My Custom Pipeline Application",
                organization_id=org_id,
            )

        create_catalogs(
            conn=conn,
            pipeline_id=pipeline_id,
            pipeline_name="my-custom-pipeline",
            chatbot_prompt_builder_catalog_id=chatbot_prompt_builder_catalog_id,
            chatbot_lmrp_catalog_id=chatbot_lmrp_catalog_id,
            organization_id=org_id,
            tables=TABLES,
        )


def downgrade() -> None:
    """Remove my-custom-pipeline pipeline configuration.

    Performs the database migration to remove the my-custom-pipeline pipeline type with all its
    associated components, presets, and configuration fields.

    This includes:
    1. Deleting all related items for all pipeline instances
    2. Deleting configuration fields and components
    3. Deleting all pipeline instances (across all organizations)
    """
    conn = op.get_bind()

    # Get all pipeline IDs for this pipeline name (across all organizations)
    pipeline_ids = get_all_pipeline_ids(conn, TABLES["pipelines"], "my-custom-pipeline", "RAGO")
    if pipeline_ids:
        delete_chatbots_and_related_items(conn, pipeline_ids, TABLES)

    # Delete config fields and components
    delete_config_fields_and_components(conn, TABLES, "my-custom-pipeline-component")

The migration uses utility functions from migration.pipeline_util to handle complex database operations consistently across all pipelines.

When to Modify the Migration

In most cases, you don't need to modify the migration! The generated migration handles all standard pipeline setup automatically.

You only need to modify it if you want to add custom configuration fields that appear in the Admin Dashboard UI.

📘 Note: Adding custom fields requires two steps:

1. Add fields to the migration (this guide)

2. Create a PresetConfig class (Advanced Configuration Guide)

Adding Custom Configuration Fields

⚠️ Important: If you add custom configuration fields in the migration, you MUST also create a PresetConfig class in your pipeline. See the Advanced Configuration Guide for complete instructions.

The generated migration includes commented examples showing how to add custom config fields. To add custom fields:

1. Add fields to the migration file: Uncomment and modify the example code

2. Create a PresetConfig class: Define a Pydantic model with matching fields (see Advanced Configuration)

3. Run the migration: make migrate

Both steps are required - the migration creates fields in the database, and PresetConfig provides type safety and validation in your code.

Example: Adding Custom Fields to Migration

# Inside the upgrade() function in your migration file:

from migration.pipeline_util import get_config_category_id, ConfigCategory

retrieval_config_cat_id = get_config_category_id(
    conn, ConfigCategory.RETRIEVAL_KNOWLEDGE_CONFIGURATION, org_id
)
search_behavior_cat_id = get_config_category_id(
    conn, ConfigCategory.SEARCH_RETRIEVAL_BEHAVIOR, org_id
)

custom_config_fields = {
    "custom_threshold": {
        "type": "float",
        "default_value": "0.7",
        "ui_type": "number",
        "level": "PRESET_CHATBOT",
        "category_id": retrieval_config_cat_id,
    },
    "enable_special_mode": {
        "type": "bool",
        "default_value": "false",
        "ui_type": "toggle",
        "level": "PRESET_CHATBOT",
        "category_id": search_behavior_cat_id,
    },
}

add_config_fields(
    conn, my_custom_pipeline_component_id, TABLES, org_id, custom_config_fields
)

After adding these fields to the migration, you must create a matching PresetConfig class. See the Advanced Configuration Guide - Preset Configuration section for complete examples of creating and registering PresetConfig classes.

Configuration Field Options

Field	Description	Example Values
type	Data type of the field (must match PresetConfig type)	"string", "bool", "int", "float"
default_value	Default value as string (must match PresetConfig default)	"true", "0.7", "custom_value"
ui_type	UI component in Admin Dashboard	"text_field", "toggle", "number", "dropdown"
level	Configuration level	"PRESET_CHATBOT", "CHATBOT"
category_id	UI grouping category (optional)	Retrieved via get_config_category_id() or None

Important: Field names, types, and defaults in the migration must exactly match your PresetConfig class definition.

Configuration Categories (UI Grouping)

Configuration categories are used to group related fields in the Admin Dashboard UI. Fields with the same category_id will be displayed together in a collapsible section.

Available ConfigCategory options:

• RETRIEVAL_KNOWLEDGE_CONFIGURATION: Groups retrieval and knowledge base settings

• SEARCH_RETRIEVAL_BEHAVIOR: Groups search behavior and filtering options

• RESPONSE_GENERATION: Groups response generation settings

• LANGUAGE_SETTINGS: Groups language and localization settings

If category_id is None: The field will be placed in an "Others" group in the Admin Dashboard.

Example:

# Fields with category - will be grouped together
custom_config_fields = {
    "retrieval_threshold": {
        "type": "float",
        "default_value": "0.7",
        "ui_type": "number",
        "level": "PRESET_CHATBOT",
        "category_id": retrieval_config_cat_id,  # Grouped in "Retrieval Knowledge Configuration"
    },
    "search_limit": {
        "type": "int",
        "default_value": "10",
        "ui_type": "number",
        "level": "PRESET_CHATBOT",
        "category_id": retrieval_config_cat_id,  # Grouped with retrieval_threshold
    },
    "custom_setting": {
        "type": "string",
        "default_value": "default",
        "ui_type": "text_field",
        "level": "PRESET_CHATBOT",
        "category_id": None,  # Will appear in "Others" group
    },
}

Note: Custom fields will appear in the Admin Dashboard UI grouped by their category, allowing users to configure them per chatbot or preset.

Accessing Custom Fields in Pipeline Code

Once you've defined custom configuration fields in the migration and created the corresponding PresetConfig class (see Advanced Configuration), you can access them in your pipeline code through the pipeline_config parameter:

In the build() function:

async def build(
    pipeline_config: dict[str, Any],
    prompt_builder_catalogs: dict[str, BaseCatalog[Any]] | None = None,
    lmrp_catalogs: dict[str, BaseCatalog[Any]] | None = None,
) -> Pipeline:
    """Constructs the pipeline instance."""
    # Access custom fields from pipeline_config
    custom_threshold = pipeline_config.get("custom_threshold", 0.7)
    enable_special_mode = pipeline_config.get("enable_special_mode", False)
    
    # Use the custom config values in your pipeline logic
    response_synthesizer = ResponseSynthesizer.stuff_preset(
        model_id="openai/gpt-4-mini",
        credentials=os.getenv("OPENAI_API_KEY"),
        temperature=custom_threshold,  # Use custom config
    )
    
    # ... rest of pipeline setup

In the build_initial_state() function:

def build_initial_state(
    request: dict[str, Any],
    pipeline_config: dict[str, Any],
    previous_state: dict[str, Any] | None = None,
    **kwargs: Any,
) -> dict[str, Any]:
    """Creates the starting state for a pipeline execution."""
    # Access custom fields from pipeline_config
    enable_special_mode = pipeline_config.get("enable_special_mode", False)
    
    return {
        "user_query": previous_state.get("user_query"),
        "event_emitter": previous_state.get("event_emitter"),
        "history": previous_state.get("history", []),
        "special_mode_enabled": enable_special_mode,  # Pass to pipeline state
    }

The pipeline_config dictionary contains all configuration values set by users in the Admin Dashboard for the specific chatbot instance.

Troubleshooting

Multiple Heads Error

If you get a "Multiple heads" error when running migrations:

Solution: The CLI tool automatically detects the correct down_revision, but if you manually modified the migration, verify that the down_revision matches the latest migration's revision field in migration/versions/.

Migration Fails During Upgrade

If the migration fails during make migrate:

1. Check the error message: It usually indicates which step failed

2. Verify database state: Ensure your database is in a consistent state

3. Check for existing data: If the pipeline already exists, the migration may fail

4. Rollback if needed: Use make downgrade1 to rollback one migration step

Rollback command:

make downgrade1

This will downgrade the most recent migration, executing its downgrade() function to remove the pipeline and related data.

Pipeline Not Visible in Admin Dashboard

If your pipeline doesn't appear after running the migration:

1. Verify migration ran successfully: Check the output of make migrate

2. Check organization: Ensure you're viewing the correct organization

3. Restart the application: Sometimes a restart is needed to reload pipelines

4. Check database: Verify the pipeline exists in the pipelines table

Custom Config Fields Not Appearing

If your custom configuration fields don't show up in the Admin Dashboard:

1. Verify field definitions: Check that field types and UI types are valid

2. Check category assignment: Ensure the category ID is correct

3. Restart application: Config fields are cached on startup

4. Check database: Verify fields exist in config_fields table

References

• Alembic Documentation