agents.base.workflow¶

Workflow base class - Pure workflow orchestration without engine dependencies.

This module provides the abstract Workflow class for building pure orchestration workflows that handle routing, transformation, and coordination without requiring language model engines.

Classes:: Workflow: Abstract base class for pure workflow orchestration.

Example

Creating a simple data processing workflow:

from haive.agents.base.workflow import Workflow

class DataProcessor(Workflow):
    async def execute(self, data):
        # Pure processing logic, no LLM
        processed = transform_data(data)
        validated = validate_data(processed)
        return validated

processor = DataProcessor(name="data_processor")
result = await processor.execute(raw_data)

Classes¶

Workflow

Pure workflow orchestration without engine dependencies.

Module Contents¶

class agents.base.workflow.Workflow(/, **data)¶

Bases: pydantic.BaseModel, abc.ABC

Pure workflow orchestration without engine dependencies.

Workflow handles pure orchestration - routing, transformation, coordination - without requiring engines. This is the foundation for building lightweight workflow components.

Parameters:: data (Any)

name¶: Name of the workflow, auto-generated from class name if not provided.

verbose¶: Enable verbose logging for detailed execution information.

debug¶: Enable debug mode for additional diagnostics.

Examples

Data processing workflow:

class DataProcessor(Workflow):
    async def execute(self, data):
        # Pure processing, no LLM
        return processed_data

Routing workflow:

class Router(Workflow):
    async def execute(self, request):
        # Route to appropriate handler
        return route_decision

Note

This is an abstract base class. Subclasses must implement the execute() method to define the workflow logic.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

classmethod auto_generate_name(values)¶

Auto-generate workflow name from class name if not provided.

Converts CamelCase class names to space-separated names for better readability in logs and debugging.

Parameters:: values (dict[str, Any]) – Dictionary of field values before validation.
Returns:: Updated values dictionary with auto-generated name if needed.
Return type:: dict[str, Any]

abstractmethod execute(input_data)¶

Async:
Parameters:: input_data (Any)
Return type:: Any

Execute the workflow logic.

This method must be implemented by subclasses to define the specific workflow behavior.

Parameters:: input_data (Any) – Input data for the workflow. Type depends on the specific workflow implementation.
Returns:: Output data from the workflow. Type depends on the specific workflow implementation.
Raises:: NotImplementedError – If not implemented by subclass.
Return type:: Any