fabricatio_core.models.action

Module that contains the classes for defining and executing task workflows.

This module provides the Action and WorkFlow classes for creating structured task execution pipelines. Actions represent atomic operations, while WorkFlows orchestrate sequences of actions with shared context and error handling.

Classes:

Action: Base class for defining executable actions with context management. WorkFlow: Manages action sequences, context propagation, and task lifecycle.

Attributes

OUTPUT_KEY
INPUT_KEY

Classes

Action	Class that represents an action to be executed in a workflow.
WorkFlow	Manages sequences of actions to fulfill tasks.

Module Contents

fabricatio_core.models.action.OUTPUT_KEY = 'task_output'

fabricatio_core.models.action.INPUT_KEY = 'task_input'

class fabricatio_core.models.action.Action(/, **data: Any)

Bases: fabricatio_core.models.generic.WithBriefing, abc.ABC

Class that represents an action to be executed in a workflow.

Actions are the atomic units of work in a workflow. Each action performs a specific operation and can modify the shared context data.

ctx_override: ClassVar[bool] = False

Whether to override the instance attr by the context variable.

name: str = None

The name of the action.

description: str = None

The description of the action.

output_key: str = None

The key used to store this action’s output in the context dictionary.

model_post_init(__context: Any) → None

Initialize the action by setting default name and description if not provided.

Parameters:

__context – The context to be used for initialization.

async act(cxt: Dict[str, Any]) → Dict[str, Any]

Execute action and update context.

Parameters:

cxt (Dict[str, Any]) – Shared context dictionary.

Returns:

Updated context dictionary with new/modified entries.

to_task_output(to: str | WorkFlow = OUTPUT_KEY) → Self

Set the output key to OUTPUT_KEY and return the action instance.

class fabricatio_core.models.action.WorkFlow(/, **data: Any)

Bases: fabricatio_core.models.generic.WithBriefing

Manages sequences of actions to fulfill tasks.

Handles context propagation between actions, error handling, and task lifecycle events like cancellation and completion.

name: str = 'WorkFlow'

The name of the workflow, which is used to identify and describe the workflow.

description: str = ''

The description of the workflow, which describes the workflow’s purpose and requirements.

steps: Sequence[Type[Action] | Action] = None

The sequence of actions to be executed, can be action classes or instances.

task_input_key: ClassVar[str] = 'task_input'

Key used to store the input task in the context dictionary.

task_output_key: ClassVar[str] = 'task_output'

Key used to extract the final result from the context dictionary.

extra_init_context: Dict[str, Any] = None

Additional initial context values to be included at workflow start.

classmethod set_task_input_key(input_key: str) → Type[Self]

Set the task input key for the workflow.

classmethod set_task_output_key(output_key: str) → Type[Self]

Set the task output key for the workflow.

model_post_init(__context: Any) → None

Initialize the workflow by instantiating any action classes.

Parameters:

__context – The context to be used for initialization.

iter_actions() → Generator[Action, None, None]

Iterate over action instances.

override_action_variable(action: Action, ctx: Dict[str, Any]) → Self

Override action variable with context values.

async serve(task: fabricatio_core.models.task.Task) → None

Execute workflow to complete given task.

Parameters:

task (Task) – Task instance to be processed.

Steps:

1. Initialize context with task instance and extra data

2. Execute each action sequentially

3. Handle task cancellation and exceptions

4. Extract final result from context

update_init_context(/, **kwargs) → Self

Update the initial context with additional key-value pairs.

Parameters:

**kwargs – Key-value pairs to add to the initial context.

Returns:

The workflow instance for method chaining.

Return type:

Self