Workflow
This section covers the workflow management components of the Kailash SDK, including the new Universal Hybrid Cyclic Graph Architecture introduced in v0.2.0.
Workflow Class
The core class for building and managing workflows.
- class kailash.workflow.Workflow(workflow_id: str, name: str, description: str = '', version: str = '1.0.0', author: str = '', metadata: dict[str, Any] | None = None)[source]
Bases:
objectRepresents a workflow DAG of nodes.
- Parameters:
- __init__(workflow_id: str, name: str, description: str = '', version: str = '1.0.0', author: str = '', metadata: dict[str, Any] | None = None)[source]
Initialize a workflow.
- Parameters:
- Raises:
WorkflowValidationError – If workflow initialization fails
- add_node(node_id: str, node_or_type: Any, **config) None[source]
Add a node to the workflow.
- Parameters:
- Raises:
WorkflowValidationError – If node is invalid
NodeConfigurationError – If node configuration fails
- Return type:
None
- connect(source_node: str, target_node: str, mapping: dict[str, str] | None = None, cycle: bool = False, max_iterations: int | None = None, convergence_check: str | None = None, cycle_id: str | None = None, timeout: float | None = None, memory_limit: int | None = None, condition: str | None = None, parent_cycle: str | None = None) None[source]
Connect two nodes in the workflow.
- Parameters:
source_node (str) – Source node ID
target_node (str) – Target node ID
mapping (dict[str, str] | None) – Dict mapping source outputs to target inputs
cycle (bool) – Whether this connection creates a cycle
max_iterations (int | None) – Maximum cycle iterations (required if cycle=True)
convergence_check (str | None) – Convergence condition expression
cycle_id (str | None) – Logical cycle group identifier
timeout (float | None) – Cycle timeout in seconds
memory_limit (int | None) – Memory limit in MB
condition (str | None) – Conditional cycle routing expression
parent_cycle (str | None) – Parent cycle for nested cycles
- Raises:
ConnectionError – If connection is invalid
WorkflowValidationError – If nodes don’t exist or cycle parameters invalid
- Return type:
None
- create_cycle(cycle_id: str | None = None)[source]
Create a new CycleBuilder for intuitive cycle configuration.
This method provides the entry point to the enhanced CycleBuilder API, which offers a fluent, chainable interface for creating cyclic workflow connections with better developer experience than the raw connect() method.
- Design Philosophy:
Replaces verbose parameter-heavy cycle creation with an intuitive builder pattern that guides developers through cycle configuration with IDE auto-completion and method chaining.
- Upstream Dependencies:
Requires source and target nodes to exist in workflow
Uses existing connection validation and cycle infrastructure
- Downstream Consumers:
CycleBuilder.build() calls back to workflow.connect() internally
CyclicWorkflowExecutor for execution of configured cycles
Cycle debugging and visualization tools
- Usage Patterns:
Simple cycles: create_cycle().connect().max_iterations().build()
Convergence-based: create_cycle().connect().converge_when().build()
Complex cycles: Full builder chain with timeouts and conditions
- Implementation Details:
Creates a CycleBuilder instance that accumulates configuration through method chaining, then applies it via workflow.connect() when build() is called. Maintains full backward compatibility.
- Error Handling:
WorkflowValidationError: If cycle_id conflicts with existing cycles
CycleConfigurationError: Raised by CycleBuilder for invalid config
- Side Effects:
Creates CycleBuilder instance but does not modify workflow until build() is called. No validation occurs until build() time.
- Parameters:
cycle_id (Optional[str]) – Optional identifier for the cycle group. If None, cycles are grouped by connection pattern. Used for nested cycles and debugging identification.
- Returns:
Fluent builder instance for configuring the cycle
- Return type:
- Raises:
ImportError – If CycleBuilder module cannot be imported
Example
>>> # Basic cycle with iteration limit >>> workflow.create_cycle("optimization") \ ... .connect("processor", "evaluator") \ ... .max_iterations(50) \ ... .build()
>>> # Convergence-based cycle with timeout >>> workflow.create_cycle("quality_improvement") \ ... .connect("cleaner", "validator", {"result": "data"}) \ ... .converge_when("quality > 0.95") \ ... .timeout(300) \ ... .build()
>>> # Nested cycle with memory limit >>> workflow.create_cycle("inner_optimization") \ ... .connect("fine_tuner", "evaluator") \ ... .max_iterations(10) \ ... .nested_in("outer_optimization") \ ... .memory_limit(1024) \ ... .build()
- separate_dag_and_cycle_edges() tuple[tuple[tuple, ...], tuple[tuple, ...]][source]
Separate DAG edges from cycle edges.
- get_cycle_groups() dict[str, list[tuple]][source]
Get cycle edges grouped by cycle_id with enhanced multi-node cycle detection.
For multi-node cycles like A → B → C → A where only C → A is marked as cycle, this method identifies all nodes (A, B, C) that are part of the same strongly connected component and groups them together.
- has_cycles() bool[source]
Check if the workflow contains any cycle connections.
- Returns:
True if workflow has cycle connections, False otherwise
- Return type:
- get_execution_order() tuple[str, ...] | list[str][source]
Get topological execution order for nodes, handling cycles gracefully.
- validate(runtime_parameters: dict[str, Any] | None = None) None[source]
Validate the workflow structure.
- run(task_manager: TaskManager | None = None, **overrides) tuple[dict[str, Any], str | None][source]
Execute the workflow.
- Parameters:
task_manager (TaskManager | None) – Optional task manager for tracking
**overrides – Parameter overrides
- Returns:
Tuple of (results dict, run_id)
- Raises:
WorkflowExecutionError – If workflow execution fails
WorkflowValidationError – If workflow is invalid
- Return type:
- execute(inputs: dict[str, Any] | None = None, task_manager: TaskManager | None = None) dict[str, Any][source]
Execute the workflow.
- export_to_kailash(output_path: str, format: str = 'yaml', **config) None[source]
Export workflow to Kailash-compatible format.
- save(path: str, format: str = 'json') None[source]
Save workflow to file.
- Parameters:
- Raises:
ValueError – If format is invalid
- Return type:
None
- create_state_wrapper(state_model: BaseModel) WorkflowStateWrapper[source]
Create a state manager wrapper for a workflow.
This wrapper provides convenient methods for updating state immutably, making it easier to manage state in workflow nodes.
- execute_with_state(state_model: BaseModel, wrap_state: bool = True, task_manager: TaskManager | None = None, **overrides) tuple[BaseModel, dict[str, Any]][source]
Execute the workflow with state management.
This method provides a simplified interface for executing workflows with automatic state management, making it easier to manage state transitions.
- Parameters:
state_model (BaseModel) – The initial state for workflow execution
wrap_state (bool) – Whether to wrap state in WorkflowStateWrapper
task_manager (TaskManager | None) – Optional task manager for tracking
**overrides – Additional parameter overrides
- Returns:
Tuple of (final state, all results)
- Raises:
WorkflowExecutionError – If execution fails
WorkflowValidationError – If workflow is invalid
- Return type:
- add_api_integration_pattern(auth_node: str = 'api_auth', data_fetcher: str = 'api_client', transformer: str = 'data_transformer', validator: str = 'response_validator', output: str = 'api_output') str
Add an API integration pattern to this workflow.
- add_batch_processing_cycle(processor_node: str, batch_size: int = 100, total_items: int | None = None, cycle_id: str | None = None) str
Add a batch processing cycle to this workflow.
- add_convergence_cycle(processor_node: str, tolerance: float = 0.001, max_iterations: int = 1000, cycle_id: str | None = None) str
Add a numerical convergence cycle to this workflow.
- add_data_processing_pipeline(data_reader: str = 'data_reader', cleaner: str = 'data_cleaner', enricher: str = 'data_enricher', aggregator: str = 'data_aggregator', writer: str = 'data_writer') str
Add a data processing pipeline to this workflow.
- add_data_quality_cycle(cleaner_node: str, validator_node: str, quality_threshold: float = 0.95, max_iterations: int = 10, cycle_id: str | None = None) str
Add a data quality improvement cycle to this workflow.
- add_document_ai_workflow(document_reader: str = 'pdf_reader', text_processor: str = 'ai_analyzer', extractor: str = 'data_extractor', output: str = 'structured_data') str
Add a document AI processing workflow to this workflow.
- add_investment_pipeline(data_source: str = 'market_data', processor: str = 'portfolio_analyzer', validator: str = 'risk_assessor', output: str = 'investment_report') str
Add an investment data processing pipeline to this workflow.
- add_learning_cycle(trainer_node: str, evaluator_node: str, target_accuracy: float = 0.95, max_epochs: int = 100, early_stopping_patience: int = 10, cycle_id: str | None = None) str
Add a machine learning training cycle to this workflow.
- add_optimization_cycle(processor_node: str, evaluator_node: str, convergence: str = 'quality > 0.9', max_iterations: int = 50, cycle_id: str | None = None) str
Add an optimization cycle pattern to this workflow.
- add_retry_cycle(target_node: str, max_retries: int = 3, backoff_strategy: str = 'exponential', success_condition: str = 'success == True', cycle_id: str | None = None) str
Add a retry cycle pattern to this workflow.
- classmethod from_brief(brief, **kwargs)
Realize a natural-language brief into a
WorkflowBuilder.See
kailash.workflow.from_brief.workflow_from_brief()for the full contract, accepted keyword arguments, and raised exceptions.The classmethod returns a
WorkflowBuilder(not aWorkflowinstance) so the caller can compose further with the standard builder API before calling.build():wf = Workflow.from_brief("a workflow that reads CSV and counts rows") runtime = LocalRuntime() results, run_id = runtime.execute(wf.build())
- save_mermaid_markdown(filepath: str, title: str | None = None) None
Save workflow as markdown with Mermaid diagram.
Basic Usage:
from kailash import Workflow
# Create a workflow
workflow = Workflow("data_pipeline")
# Add nodes
workflow.add_node("CSVReaderNode", "input", config={"file_path": "data.csv"})
workflow.add_node("DataFilter", "filter", config={"column": "active", "value": True})
workflow.add_node("CSVWriterNode", "output", config={"file_path": "filtered.csv"})
# Connect nodes
workflow.connect_sequential(["input", "filter", "output"])
# Execute
results = workflow.run()
WorkflowBuilder
Builder pattern for constructing workflows programmatically.
- class kailash.workflow.builder.WorkflowBuilder(edge_config: dict[str, Any] | None = None)[source]
Bases:
objectBuilder pattern for creating Workflow instances.
- validate_parameter_declarations(warn_on_issues: bool = True) list[ValidationIssue][source]
Validate parameter declarations for all nodes in the workflow.
This method detects common parameter declaration issues that lead to silent parameter dropping and debugging difficulties.
- add_node(*args, **kwargs) str[source]
Unified add_node method supporting multiple API patterns.
Supported patterns: 1. add_node(“NodeType”, “node_id”, {“param”: value}) # Current/Preferred 2. add_node(“node_id”, NodeClass, param=value) # Legacy fluent 3. add_node(NodeClass, “node_id”, param=value) # Alternative
- Parameters:
*args – Positional arguments (pattern-dependent)
**kwargs – Keyword arguments for configuration
- Returns:
Node ID (useful for method chaining)
- Raises:
WorkflowValidationError – If node_id is already used or invalid pattern
- Return type:
- add_node_instance(node_instance: Any, node_id: str | None = None, *, _internal: bool = False) str[source]
Add a node instance to the workflow.
This is a convenience method for adding pre-configured node instances.
- Parameters:
node_instance (Any) – Pre-configured node instance
node_id (str | None) – Unique identifier for this node (auto-generated if not provided)
_internal (bool) – Keyword-only internal flag — when True, suppresses the consumer-facing instance-API advisory. Set ONLY by SDK-internal registration paths (e.g. Nexus
@app.handler()); consumer code MUST NOT pass this. Keyword-only (after*) so a positional True cannot accidentally suppress the warning. Genuine consumer instance-API usage still warns.
- Returns:
Node ID
- Raises:
WorkflowValidationError – If node_id is already used or instance is invalid
- Return type:
- add_node_type(node_type: str, node_id: str | None = None, config: dict[str, Any] | None = None) str[source]
Add a node by type name to the workflow.
This is the original string-based method, provided for clarity and backward compatibility.
- add_connection(from_node: str, from_output: str, to_node: str, to_input: str) WorkflowBuilder[source]
Connect two nodes in the workflow.
- Parameters:
- Raises:
WorkflowValidationError – If nodes don’t exist
ConnectionError – If connection is invalid
- Return type:
- connect(from_node: str, to_node: str, mapping: dict | None = None, from_output: str | None = None, to_input: str | None = None) None[source]
Connect two nodes in the workflow with flexible parameter formats.
This method provides a more intuitive API for connecting nodes and supports both simple connections and complex mapping-based connections.
- Parameters:
- Return type:
None
Examples
# Simple connection workflow.connect(“node1”, “node2”, from_output=”data”, to_input=”input”)
# Mapping-based connection workflow.connect(“node1”, “node2”, mapping={“data”: “input”})
# Default data flow workflow.connect(“node1”, “node2”) # Uses “data” -> “data”
- set_metadata(**kwargs) WorkflowBuilder[source]
Set workflow metadata.
- Parameters:
**kwargs – Metadata key-value pairs
- Returns:
Self for chaining
- Return type:
- add_typed_connection(from_node: str, from_output: str, to_node: str, to_input: str, contract: str | ConnectionContract, validate_immediately: bool = False) WorkflowBuilder[source]
Add a typed connection with contract validation.
This is the new contract-based connection method that enforces validation contracts on data flowing between nodes.
- Parameters:
from_node (str) – Source node ID
from_output (str) – Output field from source
to_node (str) – Target node ID
to_input (str) – Input field on target
contract (str | ConnectionContract) – Contract name (string) or ConnectionContract instance
validate_immediately (bool) – Whether to validate contract definitions now
- Returns:
Self for chaining
- Raises:
WorkflowValidationError – If contract is invalid or nodes don’t exist
ConnectionError – If connection setup fails
- Return type:
Example:
# Using predefined contract workflow.add_typed_connection( "csv_reader", "data", "processor", "input_data", contract="string_data" ) # Using custom contract custom_contract = ConnectionContract( name="user_data_flow", source_schema={"type": "object", "properties": {"id": {"type": "string"}}}, target_schema={"type": "object", "properties": {"id": {"type": "string"}}}, security_policies=[SecurityPolicy.NO_PII] ) workflow.add_typed_connection( "user_source", "user", "user_processor", "user_data", contract=custom_contract )
- get_connection_contract(connection_id: str) ConnectionContract | None[source]
Get the contract for a specific connection.
- Parameters:
connection_id (str) – Connection identifier in format “from.output → to.input”
- Returns:
ConnectionContract if found, None otherwise
- Return type:
ConnectionContract | None
- validate_all_contracts() tuple[bool, list[str]][source]
Validate all connection contracts in the workflow.
- add_workflow_inputs(input_node_id: str, input_mappings: dict) WorkflowBuilder[source]
Map workflow-level inputs to a specific node’s parameters.
- Parameters:
- Returns:
Self for chaining
- Return type:
- update_node(node_id: str, config_updates: dict[str, Any]) WorkflowBuilder[source]
Update the configuration of an existing node.
This is essential for enterprise scenarios like: - Dynamic environment-specific configuration - Runtime parameter injection - Security context updates - A/B testing and feature flags
- Parameters:
- Returns:
Self for chaining
- Raises:
WorkflowValidationError – If node doesn’t exist
- Return type:
- build(workflow_id: str | None = None, **kwargs) Workflow[source]
Build and return a Workflow instance.
- set_workflow_parameters(**parameters) WorkflowBuilder[source]
Set default parameters that will be passed to all nodes.
- Parameters:
**parameters – Key-value pairs of workflow-level parameters
- Returns:
Self for chaining
- Return type:
- add_parameter_mapping(node_id: str, mappings: dict[str, str]) WorkflowBuilder[source]
Add parameter mappings for a specific node.
- Parameters:
- Returns:
Self for chaining
- Return type:
- add_input_connection(to_node: str, to_input: str, from_workflow_param: str) WorkflowBuilder[source]
Connect a workflow parameter directly to a node input.
- Parameters:
- Returns:
Self for chaining
- Return type:
- clear() WorkflowBuilder[source]
Clear builder state.
- Returns:
Self for chaining
- Return type:
- classmethod from_dict(config: dict[str, Any]) WorkflowBuilder[source]
Create builder from dictionary configuration.
Example Usage:
from kailash.workflow import WorkflowBuilder
# Using builder pattern
builder = WorkflowBuilder("etl_pipeline")
workflow = (builder
.add_node("CSVReaderNode", "extract", config={"file_path": "input.csv"})
.add_node("DataTransformer", "transform", config={
"operations": [
{"type": "rename", "old": "id", "new": "customer_id"},
{"type": "cast", "column": "amount", "dtype": "float"}
]
})
.add_node("SQLWriter", "load", config={
"connection_string": "postgresql://localhost/db",
"table_name": "customers"
})
.connect_sequential(["extract", "transform", "load"])
.build()
)
results = workflow.run()
Workflow Graph Management
Internal graph representation is handled by the Workflow class.
- class kailash.workflow.graph.Workflow(workflow_id: str, name: str, description: str = '', version: str = '1.0.0', author: str = '', metadata: dict[str, Any] | None = None)[source]
Bases:
objectRepresents a workflow DAG of nodes.
- Parameters:
- __init__(workflow_id: str, name: str, description: str = '', version: str = '1.0.0', author: str = '', metadata: dict[str, Any] | None = None)[source]
Initialize a workflow.
- Parameters:
- Raises:
WorkflowValidationError – If workflow initialization fails
- add_node(node_id: str, node_or_type: Any, **config) None[source]
Add a node to the workflow.
- Parameters:
- Raises:
WorkflowValidationError – If node is invalid
NodeConfigurationError – If node configuration fails
- Return type:
None
- connect(source_node: str, target_node: str, mapping: dict[str, str] | None = None, cycle: bool = False, max_iterations: int | None = None, convergence_check: str | None = None, cycle_id: str | None = None, timeout: float | None = None, memory_limit: int | None = None, condition: str | None = None, parent_cycle: str | None = None) None[source]
Connect two nodes in the workflow.
- Parameters:
source_node (str) – Source node ID
target_node (str) – Target node ID
mapping (dict[str, str] | None) – Dict mapping source outputs to target inputs
cycle (bool) – Whether this connection creates a cycle
max_iterations (int | None) – Maximum cycle iterations (required if cycle=True)
convergence_check (str | None) – Convergence condition expression
cycle_id (str | None) – Logical cycle group identifier
timeout (float | None) – Cycle timeout in seconds
memory_limit (int | None) – Memory limit in MB
condition (str | None) – Conditional cycle routing expression
parent_cycle (str | None) – Parent cycle for nested cycles
- Raises:
ConnectionError – If connection is invalid
WorkflowValidationError – If nodes don’t exist or cycle parameters invalid
- Return type:
None
- create_cycle(cycle_id: str | None = None)[source]
Create a new CycleBuilder for intuitive cycle configuration.
This method provides the entry point to the enhanced CycleBuilder API, which offers a fluent, chainable interface for creating cyclic workflow connections with better developer experience than the raw connect() method.
- Design Philosophy:
Replaces verbose parameter-heavy cycle creation with an intuitive builder pattern that guides developers through cycle configuration with IDE auto-completion and method chaining.
- Upstream Dependencies:
Requires source and target nodes to exist in workflow
Uses existing connection validation and cycle infrastructure
- Downstream Consumers:
CycleBuilder.build() calls back to workflow.connect() internally
CyclicWorkflowExecutor for execution of configured cycles
Cycle debugging and visualization tools
- Usage Patterns:
Simple cycles: create_cycle().connect().max_iterations().build()
Convergence-based: create_cycle().connect().converge_when().build()
Complex cycles: Full builder chain with timeouts and conditions
- Implementation Details:
Creates a CycleBuilder instance that accumulates configuration through method chaining, then applies it via workflow.connect() when build() is called. Maintains full backward compatibility.
- Error Handling:
WorkflowValidationError: If cycle_id conflicts with existing cycles
CycleConfigurationError: Raised by CycleBuilder for invalid config
- Side Effects:
Creates CycleBuilder instance but does not modify workflow until build() is called. No validation occurs until build() time.
- Parameters:
cycle_id (Optional[str]) – Optional identifier for the cycle group. If None, cycles are grouped by connection pattern. Used for nested cycles and debugging identification.
- Returns:
Fluent builder instance for configuring the cycle
- Return type:
- Raises:
ImportError – If CycleBuilder module cannot be imported
Example
>>> # Basic cycle with iteration limit >>> workflow.create_cycle("optimization") \ ... .connect("processor", "evaluator") \ ... .max_iterations(50) \ ... .build()
>>> # Convergence-based cycle with timeout >>> workflow.create_cycle("quality_improvement") \ ... .connect("cleaner", "validator", {"result": "data"}) \ ... .converge_when("quality > 0.95") \ ... .timeout(300) \ ... .build()
>>> # Nested cycle with memory limit >>> workflow.create_cycle("inner_optimization") \ ... .connect("fine_tuner", "evaluator") \ ... .max_iterations(10) \ ... .nested_in("outer_optimization") \ ... .memory_limit(1024) \ ... .build()
- separate_dag_and_cycle_edges() tuple[tuple[tuple, ...], tuple[tuple, ...]][source]
Separate DAG edges from cycle edges.
- get_cycle_groups() dict[str, list[tuple]][source]
Get cycle edges grouped by cycle_id with enhanced multi-node cycle detection.
For multi-node cycles like A → B → C → A where only C → A is marked as cycle, this method identifies all nodes (A, B, C) that are part of the same strongly connected component and groups them together.
- has_cycles() bool[source]
Check if the workflow contains any cycle connections.
- Returns:
True if workflow has cycle connections, False otherwise
- Return type:
- get_execution_order() tuple[str, ...] | list[str][source]
Get topological execution order for nodes, handling cycles gracefully.
- validate(runtime_parameters: dict[str, Any] | None = None) None[source]
Validate the workflow structure.
- run(task_manager: TaskManager | None = None, **overrides) tuple[dict[str, Any], str | None][source]
Execute the workflow.
- Parameters:
task_manager (TaskManager | None) – Optional task manager for tracking
**overrides – Parameter overrides
- Returns:
Tuple of (results dict, run_id)
- Raises:
WorkflowExecutionError – If workflow execution fails
WorkflowValidationError – If workflow is invalid
- Return type:
- execute(inputs: dict[str, Any] | None = None, task_manager: TaskManager | None = None) dict[str, Any][source]
Execute the workflow.
- export_to_kailash(output_path: str, format: str = 'yaml', **config) None[source]
Export workflow to Kailash-compatible format.
- save(path: str, format: str = 'json') None[source]
Save workflow to file.
- Parameters:
- Raises:
ValueError – If format is invalid
- Return type:
None
- create_state_wrapper(state_model: BaseModel) WorkflowStateWrapper[source]
Create a state manager wrapper for a workflow.
This wrapper provides convenient methods for updating state immutably, making it easier to manage state in workflow nodes.
- execute_with_state(state_model: BaseModel, wrap_state: bool = True, task_manager: TaskManager | None = None, **overrides) tuple[BaseModel, dict[str, Any]][source]
Execute the workflow with state management.
This method provides a simplified interface for executing workflows with automatic state management, making it easier to manage state transitions.
- Parameters:
state_model (BaseModel) – The initial state for workflow execution
wrap_state (bool) – Whether to wrap state in WorkflowStateWrapper
task_manager (TaskManager | None) – Optional task manager for tracking
**overrides – Additional parameter overrides
- Returns:
Tuple of (final state, all results)
- Raises:
WorkflowExecutionError – If execution fails
WorkflowValidationError – If workflow is invalid
- Return type:
- add_api_integration_pattern(auth_node: str = 'api_auth', data_fetcher: str = 'api_client', transformer: str = 'data_transformer', validator: str = 'response_validator', output: str = 'api_output') str
Add an API integration pattern to this workflow.
- add_batch_processing_cycle(processor_node: str, batch_size: int = 100, total_items: int | None = None, cycle_id: str | None = None) str
Add a batch processing cycle to this workflow.
- add_convergence_cycle(processor_node: str, tolerance: float = 0.001, max_iterations: int = 1000, cycle_id: str | None = None) str
Add a numerical convergence cycle to this workflow.
- add_data_processing_pipeline(data_reader: str = 'data_reader', cleaner: str = 'data_cleaner', enricher: str = 'data_enricher', aggregator: str = 'data_aggregator', writer: str = 'data_writer') str
Add a data processing pipeline to this workflow.
- add_data_quality_cycle(cleaner_node: str, validator_node: str, quality_threshold: float = 0.95, max_iterations: int = 10, cycle_id: str | None = None) str
Add a data quality improvement cycle to this workflow.
- add_document_ai_workflow(document_reader: str = 'pdf_reader', text_processor: str = 'ai_analyzer', extractor: str = 'data_extractor', output: str = 'structured_data') str
Add a document AI processing workflow to this workflow.
- add_investment_pipeline(data_source: str = 'market_data', processor: str = 'portfolio_analyzer', validator: str = 'risk_assessor', output: str = 'investment_report') str
Add an investment data processing pipeline to this workflow.
- add_learning_cycle(trainer_node: str, evaluator_node: str, target_accuracy: float = 0.95, max_epochs: int = 100, early_stopping_patience: int = 10, cycle_id: str | None = None) str
Add a machine learning training cycle to this workflow.
- add_optimization_cycle(processor_node: str, evaluator_node: str, convergence: str = 'quality > 0.9', max_iterations: int = 50, cycle_id: str | None = None) str
Add an optimization cycle pattern to this workflow.
- add_retry_cycle(target_node: str, max_retries: int = 3, backoff_strategy: str = 'exponential', success_condition: str = 'success == True', cycle_id: str | None = None) str
Add a retry cycle pattern to this workflow.
- classmethod from_brief(brief, **kwargs)
Realize a natural-language brief into a
WorkflowBuilder.See
kailash.workflow.from_brief.workflow_from_brief()for the full contract, accepted keyword arguments, and raised exceptions.The classmethod returns a
WorkflowBuilder(not aWorkflowinstance) so the caller can compose further with the standard builder API before calling.build():wf = Workflow.from_brief("a workflow that reads CSV and counts rows") runtime = LocalRuntime() results, run_id = runtime.execute(wf.build())
- save_mermaid_markdown(filepath: str, title: str | None = None) None
Save workflow as markdown with Mermaid diagram.
Key Methods:
add_node(node_id, node_instance): Add a node to the graphadd_edge(source, target, metadata): Connect two nodesconnect(source, target, mapping=None, condition=None, cycle=False, max_iterations=100, convergence_check=None): Connect nodes with optional cycle supportget_execution_order(): Get topological sort of nodes (DAG workflows only)validate(): Check for proper graph structure and cycle configuration
Cyclic Workflows (Enhanced in v0.2.0)
Kailash v0.2.0 introduces the Universal Hybrid Cyclic Graph Architecture with high-performance iterative processing, automatic convergence detection, and comprehensive developer tools.
Performance: 30,000+ iterations per second for typical workflows.
CycleBuilder API (New in v0.2.0)
- class kailash.workflow.CycleBuilder(workflow: Workflow, cycle_id: str | None = None)[source]
Bases:
objectFluent builder for creating cyclic workflow connections.
This class provides an intuitive, chainable API for configuring cyclic connections in workflows. It replaces the verbose parameter-heavy approach with a more discoverable and type-safe builder pattern.
Examples
Creating a basic cycle:
>>> workflow = Workflow("optimization", "Optimization Loop") >>> cycle = workflow.create_cycle("quality_improvement") >>> cycle.connect("processor", "evaluator") ... .max_iterations(50) ... .converge_when("quality > 0.9") ... .timeout(300) ... .build()
- connect(source_node: str, target_node: str, mapping: dict[str, str] | None = None) CycleBuilder[source]
Configure the source and target nodes for the cycle connection.
Establishes which nodes will be connected in a cyclic pattern. The mapping parameter defines how outputs from the source node map to inputs of the target node.
- Parameters:
source_node (str) – Node ID that produces output for the cycle.
target_node (str) – Node ID that receives input from the cycle.
mapping (dict[str, str] | None) – Output-to-input mapping. Keys are source output fields, values are target input fields. If None, attempts automatic mapping based on parameter names.
- Returns:
Self for method chaining.
- Raises:
WorkflowValidationError – If source or target node doesn’t exist.
CycleConfigurationError – If nodes are invalid for cyclic connection.
- Return type:
Examples
>>> cycle.connect("processor", "evaluator", {"result": "input_data"}) >>> # Or with automatic mapping >>> cycle.connect("node_a", "node_b")
- max_iterations(iterations: int) CycleBuilder[source]
Set the maximum number of cycle iterations for safety.
Provides a hard limit on cycle execution to prevent infinite loops. This is a critical safety mechanism for production workflows.
- Parameters:
iterations (int) – Maximum number of iterations allowed. Must be positive. Recommended range: 10-1000 depending on use case.
- Returns:
Self for method chaining.
- Raises:
CycleConfigurationError – If iterations is not positive.
- Return type:
Examples
>>> cycle.max_iterations(100) # Allow up to 100 iterations >>> cycle.max_iterations(10) # Quick convergence expected
- converge_when(condition: str) CycleBuilder[source]
Set the convergence condition to terminate the cycle early.
Defines an expression that, when true, will stop cycle execution before reaching max_iterations. This enables efficient early termination when the desired result is achieved.
- Parameters:
condition (str) – Python expression evaluated against node outputs. Can reference any output field from cycle nodes. Examples: “error < 0.01”, “quality > 0.9”, “improvement < 0.001”
- Returns:
Self for method chaining.
- Raises:
CycleConfigurationError – If condition syntax is invalid.
- Return type:
Examples
>>> cycle.converge_when("error < 0.01") # Numerical convergence >>> cycle.converge_when("quality > 0.95") # Quality threshold >>> cycle.converge_when("improvement < 0.001") # Minimal improvement
- timeout(seconds: float) CycleBuilder[source]
Set a timeout limit for cycle execution.
Provides time-based safety limit to prevent cycles from running indefinitely. Useful for cycles that might have unpredictable convergence times.
- Parameters:
seconds (float) – Maximum execution time in seconds. Must be positive. Recommended: 30-3600 seconds.
- Returns:
Self for method chaining.
- Raises:
CycleConfigurationError – If timeout is not positive.
- Return type:
Examples
>>> cycle.timeout(300) # 5 minutes maximum >>> cycle.timeout(30.5) # 30.5 seconds for quick cycles
- memory_limit(mb: int) CycleBuilder[source]
Set a memory usage limit for cycle execution.
Provides memory-based safety limit to prevent cycles from consuming excessive memory through data accumulation across iterations.
- Parameters:
mb (int) – Maximum memory usage in megabytes. Must be positive. Recommended: 100-10000 MB.
- Returns:
Self for method chaining.
- Raises:
CycleConfigurationError – If memory limit is not positive.
- Return type:
Examples
>>> cycle.memory_limit(1024) # 1GB limit >>> cycle.memory_limit(512) # 512MB for smaller workflows
- when(condition: str) CycleBuilder[source]
Set a conditional expression for cycle routing.
Enables conditional cycle execution where the cycle only runs when the specified condition is met. Useful for adaptive workflows.
- Parameters:
condition (str) – Python expression for conditional execution. Evaluated before each cycle iteration.
- Returns:
Self for method chaining.
- Raises:
CycleConfigurationError – If condition syntax is invalid.
- Return type:
Examples
>>> cycle.when("retry_count < 3") # Retry logic >>> cycle.when("needs_optimization") # Conditional optimization
- nested_in(parent_cycle_id: str) CycleBuilder[source]
Make this cycle nested within another cycle.
Enables hierarchical cycle structures where one cycle operates within the iterations of a parent cycle. Useful for multi-level optimization scenarios.
- Parameters:
parent_cycle_id (str) – Identifier of the parent cycle.
- Returns:
Self for method chaining.
- Raises:
CycleConfigurationError – If parent cycle ID is invalid.
- Return type:
Examples
>>> cycle.nested_in("outer_optimization") # This cycle runs inside outer_optimization
- build() None[source]
Build and add the configured cycle to the workflow.
Validates the cycle configuration and creates the actual cyclic connection in the workflow. This finalizes the cycle builder pattern and applies all configured settings.
- Raises:
CycleConfigurationError – If cycle configuration is incomplete or invalid.
WorkflowValidationError – If workflow connection fails.
- Return type:
None
Examples
>>> cycle.connect("node_a", "node_b") ... .max_iterations(50) ... .converge_when("quality > 0.9") ... .build() # Creates the cycle in the workflow
- classmethod from_config(workflow: Workflow, config: CycleConfig) CycleBuilder[source]
Create a CycleBuilder from a CycleConfig instance.
Provides an alternative constructor that initializes the builder with all configuration from a type-safe CycleConfig object. This enables configuration reuse, templating, and structured configuration management across multiple cycles.
- Parameters:
workflow (Workflow) – Target workflow for the cycle.
config (CycleConfig) – Pre-configured cycle parameters.
- Returns:
Builder instance initialized with config values.
- Raises:
CycleConfigurationError – If config is invalid.
ImportError – If CycleConfig module is not available.
- Return type:
Examples
Using a template:
>>> config = CycleTemplates.optimization_loop(max_iterations=50) >>> builder = CycleBuilder.from_config(workflow, config) >>> builder.connect("optimizer", "evaluator").build()
Using custom configuration:
>>> config = CycleConfig(max_iterations=100, timeout=300) >>> builder = CycleBuilder.from_config(workflow, config) >>> builder.connect("processor", "evaluator").build()
- apply_config(config: CycleConfig) CycleBuilder[source]
Apply configuration from a CycleConfig instance to this builder.
Merges configuration parameters from a CycleConfig object into the current builder state. This allows combining fluent builder calls with structured configuration objects for maximum flexibility.
- Parameters:
config (CycleConfig) – Configuration to apply to this builder.
- Returns:
Self for method chaining.
- Raises:
CycleConfigurationError – If config is invalid.
- Return type:
Examples
>>> builder = workflow.create_cycle("custom") ... .connect("a", "b") ... .apply_config(CycleTemplates.optimization_loop()) ... .timeout(120) # Override the template timeout ... .build()
Phase 5 API - Simple and Intuitive:
from kailash.workflow import CycleBuilder
from kailash.nodes import PythonCodeNode
# Create builder
builder = CycleBuilder("optimization")
# Add cycle node with automatic state management
optimizer_code = '''
# Access previous state with automatic defaults
try:
x = cycle_state["x"]
loss = cycle_state["loss"]
except:
x = 10.0 # Initial value
loss = float('inf')
# Gradient descent step
gradient = 2 * x # derivative of x^2
learning_rate = 0.1
new_x = x - learning_rate * gradient
new_loss = new_x ** 2
# Check convergence
converged = abs(new_loss - loss) < 0.001
result = {"x": new_x, "loss": new_loss, "converged": converged}
'''
builder.add_cycle_node(
"optimizer",
PythonCodeNode(name="optimizer", code=optimizer_code),
convergence_check="converged == True",
max_iterations=100
)
# Build and run
workflow = builder.build()
results = workflow.run()
Advanced Features:
# Multi-node cycles with CycleBuilder
builder = CycleBuilder("multi_stage")
# Add multiple nodes in cycle
builder.add_cycle_node("stage1", Stage1Node())
builder.add_node("stage2", Stage2Node()) # Regular node in cycle
builder.add_node("stage3", Stage3Node())
# Define cycle path
builder.connect("stage1", "stage2")
builder.connect("stage2", "stage3")
builder.close_cycle("stage3", "stage1",
convergence_check="converged == True")
workflow = builder.build()
Traditional API (Still Supported):
from kailash import Workflow
workflow = Workflow("iterative_process")
# Add a cycle-aware node
workflow.add_node("processor", MyProcessorNode())
# Connect node to itself to create a cycle
workflow.create_cycle("processing_cycle") \
.connect("processor", "processor", mapping={"output": "input"}) \
.max_iterations(100) \
.converge_when("done == True") \
.build()
CycleAnalyzer (New in v0.2.0)
- class kailash.workflow.CycleAnalyzer(analysis_level: str = 'standard', enable_profiling: bool = True, enable_debugging: bool = True, output_directory: str | None = None)[source]
Bases:
objectComprehensive analysis tool combining debugging and profiling capabilities.
This class provides a unified interface for cycle analysis, combining the detailed tracking capabilities of CycleDebugger with the performance insights of CycleProfiler to provide comprehensive cycle optimization guidance and health monitoring.
Examples
>>> analyzer = CycleAnalyzer(analysis_level="comprehensive") >>> # Start analysis >>> session = analyzer.start_analysis_session("optimization_study") >>> trace = analyzer.start_cycle_analysis("cycle_1", "workflow_1") >>> # During execution... >>> analyzer.track_iteration(trace, input_data, output_data) >>> # Complete analysis >>> analyzer.complete_cycle_analysis(trace, converged=True) >>> report = analyzer.generate_comprehensive_report(session)
- Parameters:
- __init__(analysis_level: str = 'standard', enable_profiling: bool = True, enable_debugging: bool = True, output_directory: str | None = None)[source]
Initialize cycle analyzer.
- start_analysis_session(session_id: str) str[source]
Start a new analysis session for grouping related cycles.
Analysis sessions allow grouping multiple cycle executions for comparative analysis, trend identification, and comprehensive reporting across related workflow executions.
- Parameters:
session_id (str) – Unique identifier for the analysis session.
- Returns:
Session ID for reference.
- Return type:
Examples
>>> session = analyzer.start_analysis_session("optimization_experiment_1")
- start_cycle_analysis(cycle_id: str, workflow_id: str, max_iterations: int | None = None, timeout: float | None = None, convergence_condition: str | None = None) CycleExecutionTrace | None[source]
Start analysis for a new cycle execution.
Begins comprehensive tracking for a cycle execution, including debugging and profiling as configured. Returns a trace object for tracking iteration progress.
- Parameters:
- Returns:
Trace object for tracking, or None if debugging disabled.
- Return type:
CycleExecutionTrace | None
Examples
>>> trace = analyzer.start_cycle_analysis("opt_cycle", "workflow_1", max_iterations=100)
- track_iteration(trace: CycleExecutionTrace, input_data: dict[str, Any], output_data: dict[str, Any], convergence_value: float | None = None, node_executions: list[str] | None = None)[source]
Track a single cycle iteration with input/output data.
Records detailed information about a cycle iteration including timing, resource usage, convergence metrics, and execution flow for comprehensive analysis.
- Parameters:
Examples
>>> analyzer.track_iteration(trace, input_data, output_data, convergence_value=0.05)
- complete_cycle_analysis(trace: CycleExecutionTrace, converged: bool, termination_reason: str, convergence_iteration: int | None = None)[source]
Complete cycle analysis and generate insights.
Finalizes cycle tracking and performs comprehensive analysis including performance metrics, optimization recommendations, and comparative insights if multiple cycles are available.
- Parameters:
Examples
>>> analyzer.complete_cycle_analysis(trace, converged=True, termination_reason="convergence")
- generate_cycle_report(trace: CycleExecutionTrace) dict[str, Any][source]
Generate comprehensive report for a single cycle.
Creates a detailed analysis report for a specific cycle execution including debugging information, performance metrics, and optimization recommendations.
- Parameters:
trace (CycleExecutionTrace) – Completed cycle trace.
- Returns:
Comprehensive cycle analysis report.
- Return type:
Examples
>>> report = analyzer.generate_cycle_report(trace) >>> print(f"Cycle efficiency: {report['performance']['efficiency_score']}")
- generate_session_report(session_id: str | None = None) dict[str, Any][source]
Generate comprehensive report for an analysis session.
Creates a detailed analysis report covering all cycles in a session, including comparative analysis, trend identification, and overall optimization recommendations.
- Parameters:
session_id (str | None) – Session to analyze, or current session if None.
- Returns:
Comprehensive session analysis report.
- Return type:
Examples
>>> report = analyzer.generate_session_report() >>> print(f"Best cycle: {report['comparative_analysis']['best_cycle']}")
- get_real_time_metrics(trace: CycleExecutionTrace) dict[str, Any][source]
Get real-time metrics for an active cycle.
Provides current performance metrics and health indicators for a cycle that is currently executing, enabling real-time monitoring and early intervention if issues are detected.
- Parameters:
trace (CycleExecutionTrace) – Active cycle trace.
- Returns:
Real-time metrics and health indicators
- Return type:
Dict[str, Any]
- Side Effects:
None - this is a pure analysis method
Example
>>> metrics = analyzer.get_real_time_metrics(trace) >>> if metrics['health_score'] < 0.5: ... print("Cycle performance issue detected!")
- export_analysis_data(filepath: str | None = None, format: str = 'json', include_traces: bool = True)[source]
Export comprehensive analysis data.
Exports all analysis data including traces, performance metrics, and reports for external analysis, archival, or sharing.
- Parameters:
- Side Effects:
Creates export file with analysis data
Example
>>> analyzer.export_analysis_data("cycle_analysis.json", include_traces=True)
Analyze Cycle Performance:
from kailash.workflow import CycleAnalyzer
analyzer = CycleAnalyzer(workflow)
# Analyze execution
report = analyzer.analyze_execution(results)
print(f"Total iterations: {report['iterations']}")
print(f"Performance: {report['iterations_per_second']:.0f} iter/sec")
print(f"Convergence rate: {report['convergence_rate']:.2%}")
# Generate detailed report
analyzer.generate_report("cycle_analysis.json")
# Visualize convergence
analyzer.plot_convergence("convergence.png")
CycleDebugger (New in v0.2.0)
- class kailash.workflow.CycleDebugger(debug_level: str = 'basic', enable_profiling: bool = False)[source]
Bases:
objectComprehensive debugging tool for cyclic workflow execution.
This class provides real-time debugging capabilities for cycles, including iteration tracking, performance monitoring, convergence analysis, and detailed execution tracing. It integrates with the cycle execution system to provide insights into cycle behavior and performance.
- Design Philosophy:
Provides non-intrusive debugging that doesn’t affect cycle performance in production. Offers multiple levels of debugging detail from basic tracking to comprehensive profiling with rich analytics.
- Upstream Dependencies:
Used by CyclicWorkflowExecutor when debug mode is enabled
Integrates with cycle configuration and execution systems
- Downstream Consumers:
Debug reports and analysis tools
Performance optimization recommendations
Cycle visualization and monitoring dashboards
- Usage Patterns:
Real-time debugging during development
Performance profiling for optimization
Production monitoring for cycle health
Post-execution analysis for troubleshooting
Example
>>> debugger = CycleDebugger(debug_level="detailed") >>> trace = debugger.start_cycle("optimization", "workflow_1") >>> >>> # During cycle execution >>> iteration = debugger.start_iteration(trace, input_data) >>> debugger.end_iteration(iteration, output_data) >>> >>> # After cycle completion >>> debugger.end_cycle(trace, converged=True, reason="convergence") >>> report = debugger.generate_report(trace)
- __init__(debug_level: str = 'basic', enable_profiling: bool = False)[source]
Initialize cycle debugger.
- Parameters:
- Side Effects:
Configures logging and profiling settings
- start_cycle(cycle_id: str, workflow_id: str, max_iterations: int | None = None, timeout: float | None = None, convergence_condition: str | None = None) CycleExecutionTrace[source]
Start debugging a new cycle execution.
Creates a new execution trace and begins tracking cycle execution with all configured debugging features enabled.
- Parameters:
- Returns:
New trace object for tracking execution
- Return type:
CycleExecutionTrace
- Side Effects:
Creates new trace and adds to active_traces Logs cycle start event
Example
>>> trace = debugger.start_cycle("opt_cycle", "workflow_1", max_iterations=100)
- start_iteration(trace: CycleExecutionTrace, input_data: dict[str, Any], iteration_number: int | None = None) CycleIteration[source]
Start debugging a new cycle iteration.
Creates a new iteration object and begins tracking execution time, resource usage, and other iteration-specific metrics.
- Parameters:
- Returns:
New iteration object for tracking
- Return type:
CycleIteration
- Side Effects:
Creates new iteration and adds to trace Begins resource monitoring if profiling enabled
Example
>>> iteration = debugger.start_iteration(trace, {"value": 10})
- end_iteration(trace: CycleExecutionTrace, iteration: CycleIteration, output_data: dict[str, Any], convergence_value: float | None = None, node_executions: list[str] | None = None)[source]
Complete iteration tracking with output data and metrics.
Finalizes iteration tracking by recording output data, convergence metrics, and final resource usage measurements.
- Parameters:
- Side Effects:
Completes iteration and adds to trace Updates peak resource usage in trace Logs iteration completion
Example
>>> debugger.end_iteration(trace, iteration, {"result": 20}, convergence_value=0.05)
- end_cycle(trace: CycleExecutionTrace, converged: bool, termination_reason: str, convergence_iteration: int | None = None)[source]
Complete cycle tracking with final results and analysis.
Finalizes cycle execution tracking and generates comprehensive statistics and analysis for the complete cycle execution.
- Parameters:
- Side Effects:
Completes trace and removes from active_traces Logs cycle completion with statistics
Example
>>> debugger.end_cycle(trace, converged=True, termination_reason="convergence", convergence_iteration=15)
- generate_report(trace: CycleExecutionTrace) dict[str, Any][source]
Generate comprehensive debugging report for a cycle execution.
Creates a detailed report including execution statistics, performance analysis, convergence trends, and optimization recommendations based on the complete cycle execution trace.
- Parameters:
trace (CycleExecutionTrace) – Completed cycle trace to analyze
- Returns:
Comprehensive debugging report
- Return type:
Dict[str, Any]
- Side Effects:
None - this is a pure analysis method
Example
>>> report = debugger.generate_report(trace) >>> print(f"Efficiency score: {report['performance']['efficiency_score']}")
Debug Cyclic Workflows:
from kailash.workflow import CycleDebugger
debugger = CycleDebugger(workflow)
# Enable debugging
debugger.enable_debugging()
# Set breakpoints
debugger.set_breakpoint("optimizer", iteration=10)
debugger.set_conditional_breakpoint(
"optimizer",
condition=lambda state: state.get("loss", 0) < 0.1
)
# Run with debugging
results = workflow.run()
# Get debug information
debug_info = debugger.get_debug_info()
print(f"State at iteration 10: {debug_info['breakpoints'][10]}")
# Trace execution
trace = debugger.get_execution_trace()
for step in trace:
print(f"Iteration {step['iteration']}: {step['state']}")
CycleProfiler (New in v0.2.0)
- class kailash.workflow.CycleProfiler(enable_advanced_metrics: bool = True)[source]
Bases:
objectAdvanced profiling and performance analysis for cyclic workflows.
This class provides comprehensive performance analysis capabilities for cycles, including statistical analysis, bottleneck identification, comparative analysis across multiple cycles, and detailed optimization recommendations based on execution patterns.
Examples
>>> profiler = CycleProfiler() >>> profiler.add_trace(execution_trace) >>> metrics = profiler.analyze_performance() >>> recommendations = profiler.get_optimization_recommendations()
- Parameters:
enable_advanced_metrics (bool)
- __init__(enable_advanced_metrics: bool = True)[source]
Initialize cycle profiler.
- Parameters:
enable_advanced_metrics (bool) – Whether to enable advanced statistical analysis.
- add_trace(trace: CycleExecutionTrace)[source]
Add a cycle execution trace for analysis.
- Parameters:
trace (CycleExecutionTrace) – Completed execution trace to analyze.
Examples
>>> profiler.add_trace(execution_trace)
- analyze_performance() PerformanceMetrics[source]
Perform comprehensive performance analysis on all traces.
Analyzes all collected traces to generate comprehensive performance metrics, identify bottlenecks, and provide optimization recommendations based on statistical analysis of execution patterns.
- Returns:
Comprehensive performance analysis results.
- Return type:
PerformanceMetrics
Examples
>>> metrics = profiler.analyze_performance() >>> print(f"Average cycle time: {metrics.avg_cycle_time:.3f}s")
- compare_cycles(cycle_ids: list[str]) dict[str, Any][source]
Compare performance across multiple specific cycles.
Provides detailed comparative analysis between specific cycles, highlighting performance differences, convergence patterns, and relative efficiency metrics.
- Parameters:
cycle_ids (List[str]) – List of cycle IDs to compare
- Returns:
Comparative analysis results
- Return type:
Dict[str, Any]
- Side Effects:
None - this is a pure analysis method
Example
>>> comparison = profiler.compare_cycles(["cycle_1", "cycle_2"]) >>> print(f"Best performing cycle: {comparison['best_cycle']}")
- get_optimization_recommendations(trace: CycleExecutionTrace | None = None) list[dict[str, Any]][source]
Generate detailed optimization recommendations.
Provides specific, actionable optimization recommendations based on performance analysis, including parameter tuning suggestions, algorithmic improvements, and resource optimization strategies.
- Parameters:
trace (Optional[CycleExecutionTrace]) – Specific trace to analyze, or None for overall recommendations
- Returns:
List of optimization recommendations with details
- Return type:
List[Dict[str, Any]]
- Side Effects:
None - this is a pure analysis method
Example
>>> recommendations = profiler.get_optimization_recommendations() >>> for rec in recommendations: ... print(f"{rec['priority']}: {rec['description']}")
- generate_performance_report() dict[str, Any][source]
Generate comprehensive performance report.
Creates a detailed performance report including metrics analysis, recommendations, trends, and comparative insights across all analyzed cycles.
- Returns:
Comprehensive performance report
- Return type:
Dict[str, Any]
- Side Effects:
None - this is a pure analysis method
Example
>>> report = profiler.generate_performance_report() >>> print(f"Overall score: {report['overall_score']}")
Profile Performance:
from kailash.workflow import CycleProfiler
profiler = CycleProfiler(workflow)
# Profile execution
profile = profiler.profile_execution(results)
print(f"Average iteration time: {profile['avg_iteration_time']:.4f}s")
print(f"Peak memory usage: {profile['peak_memory_mb']:.1f} MB")
print(f"Bottleneck node: {profile['bottleneck_node']}")
# Generate performance report
profiler.generate_performance_report("performance_report.html")
# Identify optimization opportunities
suggestions = profiler.get_optimization_suggestions()
for suggestion in suggestions:
print(f"- {suggestion}")
CycleBuilder API (v0.2.0+):
create_cycle(name): Creates a new cycle with a unique nameconnect(source, target, mapping=None): Connects nodes within the cyclemax_iterations(n): Maximum iterations before forced stop (default: 100)converge_when(condition): Python expression evaluated against node outputstimeout(seconds): Maximum time limit for cycle executionbuild(): Finalizes the cycle configuration
Important Notes:
Use
workflow.create_cycle("name").connect(...).build()for modern cyclesThe deprecated
cycle=Trueparameter is superseded by CycleBuilder APIWorkflows with cycles use the
CyclicRunnerautomaticallyUse
CycleAwareNodebase class for built-in cycle featuresState is managed automatically between iterations
Performance optimized for 30,000+ iterations/second
WorkflowRunner
Executes workflows with various runtime configurations.
- class kailash.workflow.runner.WorkflowRunner[source]
Bases:
objectManages execution across multiple connected workflows.
This class allows building complex processing pipelines by connecting multiple workflows together, with conditional branching based on state.
- add_workflow(workflow_id: str, workflow: Workflow) None[source]
Add a workflow to the runner.
- Parameters:
- Raises:
ValueError – If a workflow with the given ID already exists
- Return type:
None
- connect_workflows(source_workflow_id: str, target_workflow_id: str, condition: dict[str, Any] | None = None, state_mapping: dict[str, str] | None = None) None[source]
Connect two workflows.
- Parameters:
- Raises:
ValueError – If any workflow ID is invalid
- Return type:
None
- get_next_workflows(current_workflow_id: str, state: BaseModel) list[tuple[str, dict[str, Any]]][source]
Get the next workflows to execute based on current state.
- execute(entry_workflow_id: str, initial_state: BaseModel, task_manager: TaskManager | None = None, max_steps: int = 10) tuple[BaseModel, dict[str, dict[str, Any]]][source]
Execute a sequence of connected workflows.
- Parameters:
entry_workflow_id (str) – ID of the first workflow to execute
initial_state (BaseModel) – Initial state for workflow execution
task_manager (TaskManager | None) – Optional task manager for tracking
max_steps (int) – Maximum number of workflow steps to execute
- Returns:
Tuple of (final state, all results by workflow)
- Raises:
WorkflowExecutionError – If workflow execution fails
ValueError – If entry workflow is not found
- Return type:
Example Usage:
from kailash.workflow import WorkflowRunner
from kailash.tracking import TaskManager
# Create runner with task tracking
task_manager = TaskManager()
runner = WorkflowRunner(
workflow=workflow,
task_manager=task_manager
)
# Run with configuration
results = runner.run(
initial_data={"customers": customer_df},
config={
"max_workers": 4,
"timeout": 300
}
)
# Access execution metrics
run_id = results["run_id"]
metrics = task_manager.get_run_metrics(run_id)
State Management
Manages workflow execution state and data flow.
- class kailash.workflow.state.StateManager[source]
Bases:
objectManages immutable state operations for workflow execution.
This class provides utilities for updating state objects immutably, focusing on Pydantic models to ensure type safety and validation.
- static update_in(state_obj: BaseModel, path: list[str], value: Any) BaseModel[source]
Update a nested property in the state and return a new state object.
- Parameters:
- Returns:
A new state object with the update applied
- Raises:
- Return type:
- static batch_update(state_obj: BaseModel, updates: list[tuple[list[str], Any]]) BaseModel[source]
Apply multiple updates to the state atomically.
- Parameters:
- Returns:
A new state object with all updates applied
- Raises:
- Return type:
- static get_in(state_obj: BaseModel, path: list[str]) Any[source]
Get the value at a nested path.
- Parameters:
- Returns:
The value at the specified path
- Raises:
- Return type:
- class kailash.workflow.state.WorkflowStateWrapper(state: StateT)[source]
Bases:
Generic[StateT]Wraps a state object with convenient update methods for use in workflows.
This wrapper provides a clean interface for immutable state updates within workflow nodes, simplifying state management.
- Parameters:
state (StateT)
- __init__(state: StateT)[source]
Initialize the state wrapper.
- Parameters:
state (StateT) – The Pydantic model state object to wrap
- update_in(path: list[str], value: Any) WorkflowStateWrapper[StateT][source]
Update state at path and return new wrapper.
- Parameters:
- Returns:
A new state wrapper with the update applied
- Return type:
WorkflowStateWrapper[StateT]
- batch_update(updates: list[tuple[list[str], Any]]) WorkflowStateWrapper[StateT][source]
Apply multiple updates to the state atomically.
- merge(**updates) WorkflowStateWrapper[StateT][source]
Merge flat updates into state and return a new wrapper.
- Parameters:
**updates – Attribute updates to apply to the top level
- Returns:
A new state wrapper with the updates applied
- Return type:
WorkflowStateWrapper[StateT]
- get_state() StateT[source]
Get the wrapped state object.
- Returns:
The current state object
- Return type:
StateT
- classmethod __class_getitem__(params)
Parameterizes a generic class.
At least, parameterizing a generic class is the main thing this method does. For example, for some generic class Foo, this is called when we do Foo[int] - there, with cls=Foo and params=int.
However, note that this method is also called when defining generic classes in the first place with class Foo(Generic[T]): ….
State Management:
from kailash.nodes import Node
# Access during node execution
class MyNode(Node):
def execute(self, inputs):
# Get workflow state
state = self.workflow_state
# Store intermediate results
state.set_node_output("my_node", {"result": data})
# Access previous outputs
previous = state.get_node_output("previous_node")
return {"processed": data}
Workflow Patterns
Sequential Pipeline
Simple linear data flow:
workflow = Workflow("sequential")
# Add nodes
nodes = ["read", "validate", "transform", "save"]
for i, node_id in enumerate(nodes):
workflow.add_node(f"Node{i}", node_id, config={})
# Connect in sequence
workflow.connect_sequential(nodes)
Parallel Processing
Execute multiple branches concurrently:
workflow = Workflow("parallel")
# Source node
workflow.add_node("CSVReaderNode", "source", config={"file_path": "data.csv"})
# Parallel branches
workflow.add_node("DataFilter", "filter1", config={"column": "type", "value": "A"})
workflow.add_node("DataFilter", "filter2", config={"column": "type", "value": "B"})
workflow.add_node("DataFilter", "filter3", config={"column": "type", "value": "C"})
# Merge results
workflow.add_node("MergeNode", "combine", config={"strategy": "concat"})
# Connect
for filter_id in ["filter1", "filter2", "filter3"]:
workflow.add_edge("source", filter_id)
workflow.add_edge(filter_id, "combine")
Conditional Routing
Route data based on conditions:
workflow = Workflow("conditional")
# Add switch node
workflow.add_node("SwitchNode", "router", config={
"condition": "priority",
"routes": {
"high": "priority == 'high'",
"medium": "priority == 'medium'",
"low": "default"
}
})
# Add handlers for each route
workflow.add_node("PythonCodeNode", "high_handler", config={
"code": "return {'processed': 'high priority'}"
})
workflow.add_node("PythonCodeNode", "medium_handler", config={
"code": "return {'processed': 'medium priority'}"
})
workflow.add_node("PythonCodeNode", "low_handler", config={
"code": "return {'processed': 'low priority'}"
})
# Connect routes
workflow.add_edge("router", "high_handler", output_port="high")
workflow.add_edge("router", "medium_handler", output_port="medium")
workflow.add_edge("router", "low_handler", output_port="low")
Error Handling
Handle errors gracefully:
workflow = Workflow("error_handling")
# Main processing node
workflow.add_node("DataProcessor", "process", config={})
# Error handler
workflow.add_node("PythonCodeNode", "error_handler", config={
"code": '''
import logging
logging.error(f"Processing failed: {inputs.get('error')}")
# Send notification, save to error log, etc.
return {"handled": True}
'''
})
# Connect error output
workflow.add_edge("process", "error_handler", output_port="error")
Dynamic Workflows
Build workflows dynamically based on configuration:
def build_dynamic_workflow(config):
workflow = Workflow("dynamic")
# Add nodes based on config
for step in config["steps"]:
workflow.add_node(
step["type"],
step["id"],
config=step.get("config", {})
)
# Connect based on config
for connection in config["connections"]:
workflow.add_edge(
connection["from"],
connection["to"],
output_port=connection.get("port", "default")
)
return workflow
# Use configuration
config = {
"steps": [
{"type": "CSVReaderNode", "id": "input", "config": {"file_path": "data.csv"}},
{"type": "DataFilter", "id": "filter", "config": {"column": "active", "value": True}},
{"type": "CSVWriterNode", "id": "output", "config": {"file_path": "output.csv"}}
],
"connections": [
{"from": "input", "to": "filter"},
{"from": "filter", "to": "output"}
]
}
workflow = build_dynamic_workflow(config)
Workflow Visualization
MermaidVisualizer
- class kailash.workflow.mermaid_visualizer.MermaidVisualizer(workflow: Workflow, direction: str = 'TB', node_styles: dict[str, str] | None = None)[source]
Bases:
objectGenerate Mermaid diagrams for workflow visualization.
This class provides methods to convert Kailash workflows into Mermaid diagram syntax, which can be embedded in markdown files for better documentation and visualization.
- Variables:
workflow – The workflow to visualize
node_styles – Custom styles for different node types
direction – Graph direction (TB, LR, etc.)
- Parameters:
- __init__(workflow: Workflow, direction: str = 'TB', node_styles: dict[str, str] | None = None)[source]
Initialize the Mermaid visualizer.
- generate() str[source]
Generate the Mermaid diagram code.
- Returns:
Complete Mermaid diagram as a string
- Return type:
- generate_markdown(title: str | None = None) str[source]
Generate a complete markdown section with the Mermaid diagram.
Example Usage:
# Generate Mermaid diagram
mermaid_diagram = workflow.to_mermaid()
print(mermaid_diagram)
# Save as markdown file
workflow.save_mermaid_markdown("workflow_diagram.md")
# Custom styling
from kailash.workflow import MermaidVisualizer
visualizer = MermaidVisualizer(workflow)
diagram = visualizer.generate_mermaid(
direction="LR", # Left to right
include_data_nodes=True,
custom_styles={
"data": "fill:#e1f5fe",
"transform": "fill:#fff3e0",
"logic": "fill:#f3e5f5"
}
)
Workflow Serialization
Export Workflows
# Export to YAML
workflow.export("workflow.yaml", format="yaml")
# Export to JSON
workflow.export("workflow.json", format="json")
# Get export data
export_data = workflow.to_dict()
Import Workflows
from kailash import Workflow
# Load from YAML
workflow = Workflow.from_file("workflow.yaml")
# Load from JSON
workflow = Workflow.from_file("workflow.json")
# Load from dict
workflow = Workflow.from_dict(export_data)
Best Practices
Use Descriptive Node IDs
# Good
workflow.add_node("CSVReaderNode", "read_customer_data", ...)
workflow.add_node("DataFilter", "filter_active_customers", ...)
# Avoid
workflow.add_node("CSVReaderNode", "node1", ...)
workflow.add_node("DataFilter", "node2", ...)
Validate Early
# Validate workflow before execution
try:
workflow.validate()
except Exception as e:
print(f"Workflow validation failed: {e}")
# Fix issues before running
Use Task Tracking
# Always use task manager for production
task_manager = TaskManager()
runner = WorkflowRunner(workflow, task_manager)
# Monitor execution
results = runner.run()
print(f"Execution took: {results['duration']}s")
Handle Partial Results
try:
results = workflow.run()
except WorkflowExecutionError as e:
# Access partial results
completed = e.partial_results
print(f"Completed nodes: {list(completed.keys())}")
# Handle cleanup
cleanup_partial_results(completed)
Modular Workflows
# Create reusable sub-workflows
def create_data_validation_workflow():
w = Workflow("validation")
w.add_node("SchemaValidator", "validate_schema", ...)
w.add_node("DataQualityChecker", "check_quality", ...)
return w
# Compose larger workflows
main_workflow = Workflow("main")
validation_sub = create_data_validation_workflow()
main_workflow.add_subworkflow("validation", validation_sub)
Cycle Configuration (New in v0.2.0)
- class kailash.workflow.CycleConfig(max_iterations: int | None = None, convergence_check: str | Callable | None = None, timeout: float | None = None, memory_limit: int | None = None, iteration_safety_factor: float = 1.5, cycle_id: str | None = None, parent_cycle: str | None = None, description: str = '', condition: str | None = None, priority: int = 0, retry_policy: dict[str, ~typing.Any]=<factory>, metadata: dict[str, ~typing.Any]=<factory>)[source]
Bases:
objectType-safe configuration for cyclic workflow connections.
This dataclass provides a structured, type-safe way to configure cycle parameters with validation, default values, and comprehensive error checking. It replaces loose parameter passing with a validated configuration object that can be reused across multiple cycles.
- Design Philosophy:
Provides compile-time type safety and runtime validation for cycle configurations. Enables configuration reuse, templating, and standardization across workflows while maintaining flexibility.
- Upstream Dependencies:
Used by CycleBuilder.build() for configuration validation
Can be used directly with Workflow.connect() for type safety
Supports serialization for configuration persistence
- Downstream Consumers:
CyclicWorkflowExecutor for execution of configured cycles
Cycle debugging and profiling tools for configuration analysis
Configuration templates and presets for common patterns
- Configuration Categories:
Termination Conditions: max_iterations, timeout, convergence_check
Safety Limits: memory_limit, iteration_safety_factor
Cycle Metadata: cycle_id, parent_cycle, description
Execution Control: condition, priority, retry_policy
Example
>>> # Basic configuration >>> config = CycleConfig(max_iterations=100, convergence_check="error < 0.01") >>> workflow.connect("a", "b", cycle_config=config)
>>> # Advanced configuration with all features >>> config = CycleConfig( ... max_iterations=50, ... convergence_check="quality > 0.95", ... timeout=300.0, ... memory_limit=1024, ... cycle_id="optimization_loop", ... description="Quality optimization cycle", ... condition="needs_optimization == True" ... )
- Parameters:
- __post_init__()[source]
Validate configuration after initialization.
Performs comprehensive validation of all configuration parameters to ensure they are valid, compatible, and safe for cycle execution.
- Raises:
CycleConfigurationError – If configuration is invalid or unsafe
- Side Effects:
Logs validation warnings for suboptimal configurations Applies automatic fixes for minor configuration issues
- validate() None[source]
Validate the cycle configuration for correctness and safety.
Performs comprehensive validation of all configuration parameters, checking for required fields, valid ranges, unsafe expressions, and configuration conflicts. Provides actionable error messages for any validation failures.
- Raises:
CycleConfigurationError – If configuration is invalid
- Return type:
None
- Side Effects:
Logs warnings for suboptimal but valid configurations May modify configuration for automatic safety improvements
Example
>>> config = CycleConfig(max_iterations=-5) # Will raise error >>> config.validate() # CycleConfigurationError
- get_effective_max_iterations() int | None[source]
Get the effective maximum iterations with safety factor applied.
Calculates the actual maximum iterations that will be used during cycle execution, including the safety factor multiplier to prevent runaway cycles even when convergence conditions fail.
- Returns:
Effective maximum iterations, or None if not configured
- Return type:
Optional[int]
- Side Effects:
None - this is a pure calculation method
Example
>>> config = CycleConfig(max_iterations=100, iteration_safety_factor=1.5) >>> config.get_effective_max_iterations() 150
- to_dict() dict[str, Any][source]
Convert configuration to dictionary format.
Serializes the configuration to a dictionary suitable for JSON/YAML export, API transmission, or storage. Excludes None values and callable convergence checks for clean serialization.
- Returns:
Dictionary representation of configuration
- Return type:
Dict[str, Any]
- Side Effects:
None - this method is pure
Example
>>> config = CycleConfig(max_iterations=100) >>> config.to_dict() {'max_iterations': 100, 'iteration_safety_factor': 1.5, ...}
- classmethod from_dict(data: dict[str, Any]) CycleConfig[source]
Create configuration from dictionary data.
Deserializes a configuration from dictionary format, typically loaded from JSON/YAML files or API requests. Handles missing fields gracefully with default values.
- Parameters:
data (Dict[str, Any]) – Dictionary containing configuration data
- Returns:
New configuration instance
- Return type:
- Raises:
CycleConfigurationError – If data contains invalid values
- Side Effects:
Validates the resulting configuration automatically
Example
>>> data = {'max_iterations': 100, 'timeout': 60.0} >>> config = CycleConfig.from_dict(data)
- merge(other: CycleConfig) CycleConfig[source]
Merge this configuration with another, with other taking precedence.
Creates a new configuration by merging two configurations, where non-None values from the other configuration override values in this configuration. Useful for applying templates and overlays.
- Parameters:
other (CycleConfig) – Configuration to merge with (takes precedence)
- Returns:
New merged configuration instance
- Return type:
- Raises:
CycleConfigurationError – If merged configuration is invalid
- Side Effects:
Validates the resulting merged configuration
Example
>>> base = CycleConfig(max_iterations=100) >>> override = CycleConfig(timeout=60.0, cycle_id="custom") >>> merged = base.merge(override) >>> # Result has max_iterations=100, timeout=60.0, cycle_id="custom"
- create_template(template_name: str) dict[str, Any][source]
Create a reusable template from this configuration.
Exports the configuration as a named template that can be stored, shared, and reused across multiple workflows. Templates include metadata about their intended use case and recommended parameters.
- Parameters:
template_name (str) – Name for the template
- Returns:
Template data including metadata
- Return type:
Dict[str, Any]
- Side Effects:
None - this method is pure
Example
>>> config = CycleConfig(max_iterations=50, convergence_check="quality > 0.9") >>> template = config.create_template("quality_optimization")
- __repr__() str[source]
Return string representation of the configuration.
- Returns:
Human-readable representation showing key configuration values
- Return type:
Example
>>> config = CycleConfig(max_iterations=100, timeout=60.0) >>> str(config) 'CycleConfig(max_iterations=100, timeout=60.0, cycle_id=None)'
- __init__(max_iterations: int | None = None, convergence_check: str | Callable | None = None, timeout: float | None = None, memory_limit: int | None = None, iteration_safety_factor: float = 1.5, cycle_id: str | None = None, parent_cycle: str | None = None, description: str = '', condition: str | None = None, priority: int = 0, retry_policy: dict[str, ~typing.Any]=<factory>, metadata: dict[str, ~typing.Any]=<factory>) None
Type-Safe Configuration:
from kailash.workflow import CycleConfig
# Create configuration
config = CycleConfig(
max_iterations=1000,
convergence_check="loss < 0.001",
early_termination="gradient < 1e-6",
save_checkpoints=True,
checkpoint_interval=100
)
# Use in workflow with CycleBuilder API
workflow.create_cycle("optimization_cycle") \
.connect("node1", "node2") \
.max_iterations(config.max_iterations) \
.converge_when(config.convergence_check) \
.timeout(config.timeout) \
.build()
Migration Tools (New in v0.2.0)
Intelligent Migration System for DAG to Cyclic Workflow Conversion.
This module provides comprehensive tools to analyze existing DAG workflows and intelligently suggest or automatically convert them to use cyclic patterns where appropriate. It identifies optimization opportunities, provides detailed implementation guidance, and automates the conversion process.
- Design Philosophy:
Provides intelligent analysis of existing workflows to identify patterns that would benefit from cyclification, offering both automated conversion and detailed guidance for manual implementation. Focuses on preserving workflow semantics while optimizing for performance and maintainability.
- Key Features:
Pattern recognition for cyclification opportunities
Confidence scoring for conversion recommendations
Automated conversion with safety validation
Detailed implementation guidance with code examples
Risk assessment and migration planning
Template-based conversion for common patterns
- Analysis Capabilities:
Retry pattern detection in manual implementations
Iterative improvement pattern identification
Data validation and cleaning pattern recognition
Batch processing pattern analysis
Numerical convergence pattern detection
Performance anti-pattern identification
- Core Components:
CyclificationOpportunity: Identified conversion opportunity
CyclificationSuggestion: Detailed implementation guidance
DAGToCycleConverter: Main analysis and conversion engine
Pattern detection algorithms for common workflows
- Conversion Strategy:
Non-destructive analysis preserving original workflows
Confidence-based prioritization of opportunities
Template-based conversion for reliability
Comprehensive validation of converted workflows
Rollback capabilities for failed conversions
- Upstream Dependencies:
Existing workflow structures and node implementations
CycleTemplates for automated conversion patterns
Workflow validation and safety systems
- Downstream Consumers:
Workflow development tools and IDEs
Automated workflow optimization systems
Migration planning and execution tools
Performance optimization recommendations
Educational and training systems
Examples
Analyze workflow for opportunities:
>>> from kailash.workflow.migration import DAGToCycleConverter
>>> converter = DAGToCycleConverter(existing_workflow)
>>> opportunities = converter.analyze_cyclification_opportunities()
>>> for opp in opportunities:
... print(f"Found {opp.pattern_type}: {opp.description}")
... print(f"Confidence: {opp.confidence:.2f}")
... print(f"Expected benefit: {opp.estimated_benefit}")
Generate detailed migration guidance:
>>> suggestions = converter.generate_detailed_suggestions()
>>> for suggestion in suggestions:
... print(f"Found {suggestion.opportunity.pattern_type}")
... print(f"Implementation steps:")
... for step in suggestion.implementation_steps:
... print(f" {step}")
... print(f"Code example: {suggestion.code_example}")
... print(f"Expected outcome: {suggestion.expected_outcome}")
Automated conversion:
>>> # Convert specific nodes to cycle
>>> cycle_id = converter.convert_to_cycle(
... nodes=["processor", "evaluator"],
... convergence_strategy="quality_improvement",
... max_iterations=50
... )
>>> print(f"Created cycle: {cycle_id}")
Comprehensive migration report:
>>> report = converter.generate_migration_report()
>>> print(f"Total opportunities: {report['summary']['total_opportunities']}")
>>> print(f"High confidence: {report['summary']['high_confidence']}")
>>> # Implementation priority order
>>> for item in report['implementation_order']:
... print(f"{item['priority']}: {item['justification']}")
See also
kailash.workflow.templatesfor conversion patternskailash.workflow.validationfor workflow analysis
- class kailash.workflow.migration.CyclificationOpportunity(nodes: list[str], pattern_type: str, confidence: float, description: str, suggested_convergence: str | None = None, estimated_benefit: str = 'unknown', implementation_complexity: str = 'medium')[source]
Bases:
objectRepresents an opportunity to convert a DAG pattern to a cycle.
- Parameters:
- class kailash.workflow.migration.CyclificationSuggestion(opportunity: CyclificationOpportunity, implementation_steps: list[str], code_example: str, expected_outcome: str, risks: list[str])[source]
Bases:
objectDetailed suggestion for converting nodes to a cycle.
- Parameters:
- opportunity: CyclificationOpportunity
- class kailash.workflow.migration.DAGToCycleConverter(workflow: Workflow)[source]
Bases:
objectAnalyzer and converter for transforming DAG workflows into cyclic workflows.
This class helps identify patterns in existing workflows that could benefit from cyclic execution and provides tools to convert them.
- Parameters:
workflow (Workflow)
- __init__(workflow: Workflow)[source]
Initialize converter with target workflow.
- Parameters:
workflow (Workflow) – The workflow to analyze and potentially convert
- opportunities: list[CyclificationOpportunity]
- analyze_cyclification_opportunities() list[CyclificationOpportunity][source]
Analyze workflow for patterns that could benefit from cyclification.
- Returns:
List of identified cyclification opportunities
- Return type:
Example
>>> workflow = create_example_workflow() >>> converter = DAGToCycleConverter(workflow) >>> opportunities = converter.analyze_cyclification_opportunities() >>> for opp in opportunities: ... print(f"{opp.pattern_type}: {opp.description}")
- generate_detailed_suggestions() list[CyclificationSuggestion][source]
Generate detailed suggestions with implementation guidance.
- Returns:
List of detailed suggestions for cyclification
- Return type:
Example
>>> converter = DAGToCycleConverter(workflow) >>> converter.analyze_cyclification_opportunities() >>> suggestions = converter.generate_detailed_suggestions() >>> for suggestion in suggestions: ... print(suggestion.code_example)
- convert_to_cycle(nodes: list[str], convergence_strategy: str = 'error_reduction', cycle_type: str | None = None, **kwargs) str[source]
Convert specific nodes to a cycle using the specified strategy.
- Parameters:
- Returns:
The created cycle identifier
- Return type:
Example
>>> converter = DAGToCycleConverter(workflow) >>> cycle_id = converter.convert_to_cycle( ... nodes=["processor", "evaluator"], ... convergence_strategy="quality_improvement", ... max_iterations=50 ... )
- generate_migration_report() dict[str, Any][source]
Generate comprehensive migration report with analysis and recommendations.
Example
>>> converter = DAGToCycleConverter(workflow) >>> converter.analyze_cyclification_opportunities() >>> report = converter.generate_migration_report() >>> print(report['summary']['total_opportunities'])
Migrate Existing Workflows:
from kailash.workflow.migration import WorkflowMigrator
# Migrate to use CycleBuilder
migrator = WorkflowMigrator()
# Analyze workflow
analysis = migrator.analyze_workflow(old_workflow)
print(f"Found {len(analysis['cycles'])} cycles")
# Generate migration code
new_code = migrator.generate_migration_code(old_workflow)
print(new_code)
# Auto-migrate
new_workflow = migrator.migrate_workflow(old_workflow)