`OSSS.ai.workflows.definition`¶

`OSSS.ai.workflows.definition` ¶

Declarative workflow definition schema for OSSS DAG workflows.

This module provides the core data models for defining sophisticated DAG workflows with advanced node types, conditional routing, and ecosystem-ready metadata. Supports workflow versioning, attribution, and plugin architecture foundation.

Uses Pydantic for robust validation, automatic serialization, and schema generation.

`NodeCategory` ¶

Bases: str, Enum

Node category taxonomy for clear distinction between node types.

`AdvancedNodeType` ¶

Bases: str, Enum

Advanced node types for sophisticated workflow orchestration.

`BaseNodeType` ¶

Bases: str, Enum

Base node types for standard agent execution.

`EdgeDefinition` ¶

Bases: BaseModel

Definition of an edge between workflow nodes with sophisticated conditional routing.

Supports metadata-aware routing, success/failure paths, and classification-based conditional logic for Airflow/Prefect-level expressiveness.

`to_dict()` ¶

Convert to dictionary representation for serialization.

`FlowDefinition` ¶

Bases: BaseModel

Sophisticated flow definition with conditional routing and metadata awareness.

Supports advanced routing logic, success/failure paths, and metadata-driven conditional execution for production-grade workflow orchestration.

`to_dict()` ¶

Convert to dictionary representation for serialization.

`WorkflowNodeConfiguration` ¶

Bases: BaseModel

Node configuration with clear taxonomy and classification filters.

Provides explicit BASE vs ADVANCED categorization, type-safe node types, and metadata-aware routing support for the plugin architecture foundation.

`to_dict()` ¶

Convert to dictionary representation for serialization.

`ExecutionConfiguration` ¶

Bases: BaseModel

Configuration for workflow execution settings.

`OutputConfiguration` ¶

Bases: BaseModel

Configuration for workflow output formatting.

`QualityGates` ¶

Bases: BaseModel

Quality gates and validation criteria for workflow execution.

`ResourceLimits` ¶

Bases: BaseModel

Resource limits for workflow execution.

`WorkflowDefinition` ¶

Bases: BaseModel

Ecosystem-ready workflow definition with versioning, attribution, and sharing contracts.

This is the core schema for declarative DAG workflows, designed for the "Kubernetes of intelligent DAG workflows" ecosystem with plugin architecture foundation and reproducible workflow sharing capabilities.

Supports both simple workflows and rich configuration options including: - Execution configuration (checkpoints, parallelization) - Output formatting (format, sections, metadata) - Quality gates (confidence thresholds, time limits) - Resource limits (timeouts, LLM calls, context size)

`create(name, version, created_by, nodes, flow, description=None, tags=None, metadata=None)` `classmethod` ¶

Create a new workflow definition with auto-generated ID and timestamp.

Parameters¶

name : str Human-readable workflow name version : str Workflow version (semantic versioning recommended) created_by : str Creator attribution nodes : List[NodeConfiguration] Workflow node configurations flow : FlowDefinition Workflow flow definition description : Optional[str] Human-readable description tags : Optional[List[str]] Categorization tags metadata : Optional[Dict[str, Any]] Additional metadata

Returns¶

WorkflowDefinition New workflow definition instance

`to_json_snapshot()` ¶

Serialization contract for ecosystem sharing and reproducibility.

Returns¶

Dict[str, Any] Complete serialized workflow definition

`export(format='json')` ¶

Multi-format export for workflow sharing and storage.

Parameters¶

format : str Export format ("json", "yaml", future: "mermaid")

Returns¶

str Exported workflow definition

Raises¶

ValueError If export format is not supported

`validated_by()` ¶

Validation attribution for ecosystem trust.

Returns¶

str Validation attribution string

`from_dict(data)` `classmethod` ¶

Create workflow definition from dictionary data.

`from_json_snapshot(data)` `classmethod` ¶

Deserialize workflow definition from JSON snapshot.

Supports both new format (with flow structure) and legacy format (with edges array).

Parameters¶

data : Dict[str, Any] JSON snapshot data

Returns¶

WorkflowDefinition Deserialized workflow definition

`from_yaml_file(file_path)` `classmethod` ¶

Load workflow definition from YAML file.

Parameters¶

file_path : str Path to YAML workflow definition file

Returns¶

WorkflowDefinition Loaded workflow definition

`from_json_file(file_path)` `classmethod` ¶

Load workflow definition from JSON file.

Parameters¶

file_path : str Path to JSON workflow definition file

Returns¶

WorkflowDefinition Loaded workflow definition

`save_to_file(file_path, format=None)` ¶

Save workflow definition to file.

Parameters¶

file_path : str Output file path format : Optional[str] Export format (auto-detected from file extension if None)

OSSS.ai.workflows.definition¶