Workflow Task

task task

The Workflow Task runs a workflow. Allows creating workflows of workflows.

Example

The following shows a simple example using this task as part of a workflow.

  1. from txtai.workflow import WorkflowTask, Workflow
  3. workflow = Workflow([WorkflowTask(otherworkflow)])
  4. workflow(["input data"])

Methods

Python documentation for the task.

Source code in txtai/workflow/task/base.py

  1. def __init__(
  2.     self,
  3.     action=None,
  4.     select=None,
  5.     unpack=True,
  6.     column=None,
  7.     merge="hstack",
  8.     initialize=None,
  9.     finalize=None,
  10.     concurrency=None,
  11.     onetomany=True,
  12.     **kwargs,
  13. ):
  14.     """
  15.     Creates a new task. A task defines two methods, type of data it accepts and the action to execute
  16.     for each data element. Action is a callable function or list of callable functions.
  18.     Args:
  19.         action: action(s) to execute on each data element
  20.         select: filter(s) used to select data to process
  21.         unpack: if data elements should be unpacked or unwrapped from (id, data, tag) tuples
  22.         column: column index to select if element is a tuple, defaults to all
  23.         merge: merge mode for joining multi-action outputs, defaults to hstack
  24.         initialize: action to execute before processing
  25.         finalize: action to execute after processing
  26.         concurrency: sets concurrency method when execute instance available
  27.                      valid values: "thread" for thread-based concurrency, "process" for process-based concurrency
  28.         onetomany: if one-to-many data transformations should be enabled, defaults to True
  29.         kwargs: additional keyword arguments
  30.     """
  32.     # Standardize into list of actions
  33.     if not action:
  34.         action = []
  35.     elif not isinstance(action, list):
  36.         action = [action]
  38.     self.action = action
  39.     self.select = select
  40.     self.unpack = unpack
  41.     self.column = column
  42.     self.merge = merge
  43.     self.initialize = initialize
  44.     self.finalize = finalize
  45.     self.concurrency = concurrency
  46.     self.onetomany = onetomany
  48.     # Check for custom registration. Adds additional instance members and validates required dependencies available.
  49.     if hasattr(self, "register"):
  50.         self.register(**kwargs)
  51.     elif kwargs:
  52.         # Raise error if additional keyword arguments passed in without register method
  53.         kwargs = ", ".join(f"'{kw}'" for kw in kwargs)
  54.         raise TypeError(f"__init__() got unexpected keyword arguments: {kwargs}")