src/workflows/README.md

Workflows Subpackage

This subpackage provides a framework for defining, validating, and executing workflows composed of interconnected model steps with dependency management.

Overview

The workflows subpackage enables the creation and execution of workflows where multiple model steps can be combined, with outputs from earlier steps feeding into inputs of later steps. The package handles dependency resolution, execution order, and error handling.

Components

structs.py

Contains the core data structures used throughout the workflow system:

• Field: Represents an input or output field with name and type information
• ModelStep: Represents a single step in a workflow with input fields, output fields, and model details
• Workflow: A collection of ModelSteps with their identifiers

utils.py

Provides utility functions for workflow operations:

• _create_variable_step_mapping: Maps variables to the steps that produce them
• create_dependency_graph: Builds a dependency graph representing the execution order constraints
• topological_sort: Sorts steps in execution order based on their dependencies

workflow_executor.py

Handles the execution of workflows:

• Processes inputs and outputs between steps
• Coordinates the execution of model steps in the correct order
• Integrates with external model providers (e.g., via litellm)

errors.py

Defines custom exceptions for workflow-related errors:

• WorkflowError: Base class for workflow errors
• CyclicDependencyError: Raised when detecting cycles in the workflow graph
• UnknownVariableError: Raised when a step requires a variable that's not provided or produced

Usage Example

from workflows.structs import Field, ModelStep, Workflow

# Define a workflow with two steps
step1 = ModelStep(
    input_fields=[Field(name="query", type="string")],
    output_fields=[Field(name="summary", type="string")],
    model="gpt-3.5-turbo",
    system_prompt="Summarize the following text"
)

step2 = ModelStep(
    input_fields=[Field(name="summary", type="string", variable="step1.summary")],
    output_fields=[Field(name="key_points", type="array")],
    model="gpt-4",
    system_prompt="Extract key points from the summary"
)

workflow = Workflow(steps={"step1": step1, "step2": step2})

# Execute the workflow
from workflows.workflow_executor import execute_workflow

result = execute_workflow(
    workflow=workflow,
    input_values={"query": "Long text to summarize..."}
)

# Access results
summary = result["step1.summary"]
key_points = result["step2.key_points"]

Error Handling

The workflows system provides robust error handling:

• Detects cyclic dependencies in workflow definitions
• Validates input/output variable references
• Ensures all required inputs are provided

Extending the Workflows System

To extend the workflows system:

1. Add new model step types by extending the ModelStep class
2. Create custom field types by extending validation in the execution logic
3. Implement additional error types in errors.py for specialized error handling