Build Compound Flows

This guide explains how to create, test, and deploy compound flows that implement complex, multi-stage processing pipelines. Compound flows provide sophisticated capabilities for implementing business logic, custom processing stages, and integration with existing elemental flows. 🔄💡

Development Lifecycle

Creating a compound flow follows a systematic process:

Configuration: Define your compound flow using YAML

Testing: Validate the entire processing pipeline

Deployment: Make your flow available

Execution: Run your multi-stage flow in production

Updates: Iterate and improve

Let’s explore each step in detail.

Use OpenAI (gpt-4o) or Anthropic (claude-3.5-sonnet) models when building complex workflows to ensure optimal performance. Checkout Available Models

Step 1: YAML Configuration

# Flow specification version
version: "0.1.0"                                    # Flow specification version

metadata:
  flow_type: "compound"                             # Specifies this as a compound flow
  name: "your-flow-name"                            # Unique identifier
  description: "Describe what this compound flow accomplishes"
  author: "your-username"                           # Your Mira Flows username
  tags: [tag1, tag2, tag3]                         # Discovery keywords
  private: true                                     # Access control setting

inputs:
  prime_input_1:                                    # Primary input parameter
    type: string
    description: "What is this input used for"
    required: true
    example: "Example value"
  prime_input_2:                                    # Secondary input parameter
    type: string
    description: "Description of this input"
    required: false
    example: "Example"

workflow:
  # Elemental Flow stage - starts immediately
  first_flow:                                       # First processing stage
    type: "elemental"
    flow_name: "john_doe/first_elemental_flow"
    inputs:
      abc: ${inputs.prime_input_1}
      xyz: ${inputs.prime_input_2}

  # Custom processing stage - starts immediately (parallel to first_flow)
  second_flow:                                      # Parallel processing stage
    type: "custom"
    inputs:
      input1: ${inputs.prime_input_1}
    model:
      provider: "provider-name"                     # e.g., anthropic, openai, meta, etc.
      name: "model-name"                            # Specific model identifier
    prompt: |
      Your flow's primary instruction or role...
      You can use {input1} placeholders to reference inputs.

  # Waits for both first_flow and second_flow to complete
  third_flow:                                       
    type: "custom"
    depends_on: [first_flow, second_flow]  # Dependent on fist_flow and second_flow
    inputs:
      input1: ${first_flow.output} # uses output of first_flow as input
      input2: ${second_flow.output} # uses output of second_flow as input
    model:
      provider: "provider-name"
      name: "model-name"
    dataset:
      source: "author_name/dataset_name"
    prompt: |
      Your flow's primary instruction or role...
      You can use {input1} and {input2} placeholders to reference inputs.

output:
  value:                                           # Combine & customise outputs in order
    - ${first_flow.output}
    - ${second_flow.output}
    - ${third_flow.output}

readme: |
  This is a detailed explanation of what your flow does.

  ### Workflow Overview:
  1. **First Flow**: Executes `john_doe/first_elemental_flow` with the provided inputs
  2. **Second Flow**: Runs in parallel with first flow
  3. **Third Flow**: Depends on both previous flows, executes after they complete

  ### Outputs:
  - Combined outputs from all flows are returned in order

Step 2: Testing

Test your compound flow to ensure all stages work correctly:

Python

from mira_sdk import MiraClient, CompoundFlow
from mira_sdk.exceptions import FlowError

client = MiraClient(config={"API_KEY": "YOUR_API_KEY"})     # Initialize Mira Client
flow = CompoundFlow(source="/path/to/compound_flow.yaml")           # Load flow configuration

test_input = {                                              # Prepare test inputs
    "prime_input_1": "test data",
    "prime_input_2": "test parameters"
}

try:
    response = client.flow.test(flow, test_input)           # Test entire pipeline
    print("Test response:", response)
except FlowError as e:
    print("Test failed:", str(e))                           # Handle test failure

Step 3: Deployment

Deploy your compound flow to make it available:

Python

from mira_sdk import MiraClient, CompoundFlow
from mira_sdk.exceptions import FlowError

flow = CompoundFlow(source="/path/to/compound_flow.yaml")           # Load flow configuration

try:
    client.flow.deploy(flow)                               # Deploy to platform
    print("Compound flow deployed successfully!")          # Success message
except FlowError as e:
    print(f"Deployment error: {str(e)}")                   # Handle deployment error

Step 4: Execution

After deployment, execute your compound flow:

Python

from mira_sdk import MiraClient
from mira_sdk.exceptions import FlowError

flow_name = "your-username/your-flow-name"                 # Flow identifier
input_data = {                                             # Execution inputs
    "prime_input_1": "production data",
    "prime_input_2": "production parameters"
}

try:
    result = client.flow.execute(flow_name, input_data)    # Execute workflow
    print("Execution result:", result)                     # Display result
except FlowError as e:
    print("Execution error:", str(e))                      # Handle execution error

Step 5: Updates

When you need to modify your compound flow:

Python

from mira_sdk import MiraClient
from mira_sdk.exceptions import FlowError

flow = client.flow.get("your-username/your-flow-name")     # Retrieve existing flow
flow.save("/path/to/compound_flow.yaml")                   # Save for editing

try:
    client.flow.deploy(flow)                               # Deploy updated version
    print("Updated flow deployed successfully!")           # Success message
except FlowError as e:
    print(f"Update error: {str(e)}")                       # Handle update error

# In case you forget to bump the flow version, it will get bumped by default every time you deploy the same flow.

By following these steps, you can create, test, deploy, and manage your own custom compound flows on the Mira Flows platform. 🌟🔧

​Development Lifecycle

​Step 1: YAML Configuration

​Step 2: Testing

​Step 3: Deployment

​Step 4: Execution

​Step 5: Updates

Development Lifecycle

Step 1: YAML Configuration

Step 2: Testing

Step 3: Deployment

Step 4: Execution

Step 5: Updates