Skip to content

Action Framework

The action framework provides the foundational classes for building reusable workflow components that follow GitHub Actions patterns.

Core Classes

causaliq_workflow.action

Action

Base class for all workflow actions.

Methods:

  • run –

    Execute action with validated inputs, return outputs.

  • validate_inputs –

    Validate input values against input specifications.

run abstractmethod 

run(
    inputs: Dict[str, Any], mode: str = "run", context: Optional[WorkflowContext] = None
) -> Dict[str, Any]

Execute action with validated inputs, return outputs.

Parameters:

  • inputs
    (Dict[str, Any]) –

    Dictionary of input values keyed by input name

  • mode
    (str, default: 'run' ) –

    Execution mode ('dry-run', 'run', 'compare')

  • context
    (Optional[WorkflowContext], default: None ) –

    Workflow context for optimization and intelligence

Returns:

  • Dict[str, Any] –

    Dictionary of output values keyed by output name

Raises:

validate_inputs 

validate_inputs(inputs: Dict[str, Any]) -> bool

Validate input values against input specifications.

Parameters:

  • inputs
    (Dict[str, Any]) –

    Dictionary of input values to validate

Returns:

  • bool –

    True if all inputs are valid

Raises:

ActionInput dataclass 

ActionInput(
    name: str,
    description: str,
    required: bool = False,
    default: Any = None,
    type_hint: str = "Any",
)

Define action input specification.

ActionOutput dataclass 

ActionOutput(name: str, description: str, value: Any)

Define action output specification.

Exception Handling

ActionExecutionError

Raised when action execution fails.

ActionValidationError

Raised when action input validation fails.

Quick Example

from causaliq_workflow.action import Action, ActionExecutionError
from typing import Any, Dict

class MyStructureLearnerAction(Action):
    """Custom structure learning action."""

    name = "my-structure-learner"
    version = "1.0.0"
    description = "Custom causal structure learning implementation"

    def run(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        """Execute the structure learning algorithm."""
        try:
            # Your implementation here
            return {
                "graph_path": "/path/to/output.graphml",
                "node_count": 5,
                "edge_count": 8,
            }
        except Exception as e:
            raise ActionExecutionError(f"Structure learning failed: {e}") from e

Design Patterns

Action Implementation Guidelines

  1. Inherit from Action base class - Provides standardized interface
  2. Define comprehensive inputs - Use ActionInput for type safety
  3. Document outputs clearly - Help users understand action results
  4. Handle errors gracefully - Use ActionExecutionError and ActionValidationError
  5. Follow semantic versioning - Enable workflow compatibility tracking
  6. Create GraphML output - Use standardized format for causal graphs

Testing Your Actions

import pytest
from pathlib import Path
from causaliq_workflow.action import ActionExecutionError

def test_my_action_success():
    """Test successful action execution."""
    action = MyStructureLearnerAction()
    inputs = {
        "data_path": "/path/to/test_data.csv",
        "output_dir": "/path/to/output",
        "alpha": 0.05,
    }

    result = action.run(inputs)

    assert "graph_path" in result
    assert "node_count" in result
    assert "edge_count" in result
    assert Path(result["graph_path"]).exists()

def test_my_action_missing_file():
    """Test action fails gracefully with missing input."""
    action = MyStructureLearnerAction()
    inputs = {
        "data_path": "/nonexistent/file.csv",
        "output_dir": "/path/to/output",
    }

    with pytest.raises(ActionExecutionError):
        action.run(inputs)

← Back to API Overview | Next: Action Registry →