检索任务

检索任务连接到一个 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)

创建一个新任务。任务定义了两种方法：它接受的数据类型以及对每个数据元素执行的操作。操作是一个可调用的函数或可调用函数的列表。

参数

名称	类型	描述	默认值
action		对每个数据元素执行的操作	None
select		用于选择要处理的数据的过滤器	None
unpack		是否应从 (id, data, tag) 元组中解包或展开数据元素	True
column		如果元素是元组，则选择的列索引，默认为全部	None
merge		用于连接多操作输出的合并模式，默认为 hstack	'hstack'
initialize		处理前执行的操作	None
finalize		处理后执行的操作	None
concurrency		设置可执行实例可用时的并发方法有效值："thread" 用于基于线程的并发，"process" 用于基于进程的并发	None
onetomany		是否启用一对多数据转换，默认为 True	True
kwargs		附加关键字参数	{}

源代码位于 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)

将检索参数添加到任务。

参数

名称	类型	描述	默认值
directory		用于存储检索到的文件的本地目录	None
flatten		展平输入目录结构，默认为 True	True

源代码位于 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