core.pipeline

Processing Pipeline Module - Document Processing Workflow Management

Provides the Pipeline class for combining multiple processors and executing document processing workflows in sequence.

Classes

Pipeline([name])

Processing Pipeline for document workflow management.

class core.pipeline.Pipeline(name='DefaultPipeline')[source]

Bases: object

Processing Pipeline for document workflow management.

This class coordinates processors in a specific sequence for document processing. It provides methods for adding/removing processors, reordering them, and processing documents.

This is a core system component implementing the Pipeline Pattern, connecting a series of processing steps to form a complete processing workflow.

__init__(name='DefaultPipeline')[source]

Initialize Pipeline object.

Parameters:: name (str) – Pipeline name, defaults to “DefaultPipeline”

add_processor(processor)[source]

Add a processor to the pipeline.

Parameters:: processor (IProcessor) – IProcessor object to add

remove_processor(processor_name)[source]

从流水线中移除指定名称的处理器。

Parameters:: processor_name (str) – 要移除的处理器名称
Returns:: 如果成功移除处理器则返回True，否则返回False
Return type:: bool

get_processors()[source]

获取流水线中的所有处理器。

Returns:: 处理器列表
Return type:: List[IProcessor]

clear_processors()[source]

清空流水线中的所有处理器。

reorder_processors(processor_names)[source]

根据提供的名称列表重新排序处理器。

Parameters:: processor_names (List[str]) – 处理器名称列表，按期望的顺序排列
Returns:: 如果成功重排序则返回True，否则返回False
Return type:: bool

async process_document(document, start_from=None)[source]

异步处理文档，按顺序执行流水线中的处理器。

Parameters:

document (Document) – 要处理的Document对象
start_from (str | None) – 起始处理器名称，如果指定，则从该处理器开始处理，默认为None

Returns:

表示处理结果的ProcessResult对象

Return type:

ProcessResult

async process_documents(documents, max_concurrent=5)[source]

异步批量处理多个文档，支持并发处理以提高性能。

Parameters:

documents (List[Document]) – 要处理的Document对象列表
max_concurrent (int) – 最大并发处理数量，默认为5

Returns:

字典，键为文档ID，值为对应的ProcessResult对象

Return type:

Dict[str, ProcessResult]

process_document_sync(document, start_from=None)[source]

同步处理文档（保持向后兼容性）。

Parameters:

document (Document) – 要处理的Document对象
start_from (str | None) – 起始处理器名称，如果指定，则从该处理器开始处理，默认为None

Returns:

表示处理结果的ProcessResult对象

Return type:

ProcessResult

process_documents_sync(documents)[source]

同步批量处理多个文档（保持向后兼容性）。

Parameters:: documents (List[Document]) – 要处理的Document对象列表
Returns:: 字典，键为文档ID，值为对应的ProcessResult对象
Return type:: Dict[str, ProcessResult]