检索任务

检索任务连接到一个URL并将内容下载到本地。当处理需要本地可用数据的操作时，此任务非常有用。

示例

以下展示了一个简单示例，将此任务作为工作流的一部分使用。

from txtai.workflow import RetrieveTask, Workflow

workflow = Workflow([RetrieveTask(directory="/tmp")])
workflow(["https://file.to.download", "/local/file/to/copy"])

配置驱动示例

此任务也可以通过工作流配置创建。

workflow:
  tasks:
    - task: retrieve
      directory: /tmp

方法

任务的Python文档。

__init__(action=None, select=None, unpack=True, column=None, merge='hstack', initialize=None, finalize=None, concurrency=None, onetomany=True, **kwargs)

Creates a new task. A task defines two methods, type of data it accepts and the action to execute for each data element. Action is a callable function or list of callable functions.

Parameters:

Name	Type	Description	Default
action		action(s) to execute on each data element	None
select		filter(s) used to select data to process	None
unpack		if data elements should be unpacked or unwrapped from (id, data, tag) tuples	True
column		column index to select if element is a tuple, defaults to all	None
merge		merge mode for joining multi-action outputs, defaults to hstack	'hstack'
initialize		action to execute before processing	None
finalize		action to execute after processing	None
concurrency		sets concurrency method when execute instance available valid values: "thread" for thread-based concurrency, "process" for process-based concurrency	None
onetomany		if one-to-many data transformations should be enabled, defaults to True	True
kwargs		additional keyword arguments	{}

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

def __init__(
    self,
    action=None,
    select=None,
    unpack=True,
    column=None,
    merge="hstack",
    initialize=None,
    finalize=None,
    concurrency=None,
    onetomany=True,
    **kwargs,
):
    """
    Creates a new task. A task defines two methods, type of data it accepts and the action to execute
    for each data element. Action is a callable function or list of callable functions.

    Args:
        action: action(s) to execute on each data element
        select: filter(s) used to select data to process
        unpack: if data elements should be unpacked or unwrapped from (id, data, tag) tuples
        column: column index to select if element is a tuple, defaults to all
        merge: merge mode for joining multi-action outputs, defaults to hstack
        initialize: action to execute before processing
        finalize: action to execute after processing
        concurrency: sets concurrency method when execute instance available
                     valid values: "thread" for thread-based concurrency, "process" for process-based concurrency
        onetomany: if one-to-many data transformations should be enabled, defaults to True
        kwargs: additional keyword arguments
    """

    # Standardize into list of actions
    if not action:
        action = []
    elif not isinstance(action, list):
        action = [action]

    self.action = action
    self.select = select
    self.unpack = unpack
    self.column = column
    self.merge = merge
    self.initialize = initialize
    self.finalize = finalize
    self.concurrency = concurrency
    self.onetomany = onetomany

    # Check for custom registration. Adds additional instance members and validates required dependencies available.
    if hasattr(self, "register"):
        self.register(**kwargs)
    elif kwargs:
        # Raise error if additional keyword arguments passed in without register method
        kwargs = ", ".join(f"'{kw}'" for kw in kwargs)
        raise TypeError(f"__init__() got unexpected keyword arguments: {kwargs}")

register(directory=None, flatten=True)

Adds retrieve parameters to task.

Parameters:

Name	Type	Description	Default
directory		local directory used to store retrieved files	None
flatten		flatten input directory structure, defaults to True	True

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

def register(self, directory=None, flatten=True):
    """
    Adds retrieve parameters to task.

    Args:
        directory: local directory used to store retrieved files
        flatten: flatten input directory structure, defaults to True
    """

    # pylint: disable=W0201
    # Create default temporary directory if not specified
    if not directory:
        # Save tempdir to prevent content from being deleted until this task is out of scope
        # pylint: disable=R1732
        self.tempdir = tempfile.TemporaryDirectory()
        directory = self.tempdir.name

    # Create output directory if necessary
    os.makedirs(directory, exist_ok=True)

    self.directory = directory
    self.flatten = flatten