
pipeline 中的最常见位置	在索引管道中，位于 Converter 之后，CSVDocumentCleaner 之前
强制运行变量	"documents": 包含 CSV 格式内容的文档列表
输出变量	"documents": 文档列表，每个文档包含从原始 CSV 文件中提取的一个子表
API 参考	PreProcessors (预处理器)
GitHub 链接	https://github.com/deepset-ai/haystack/blob/main/haystack/components/preprocessors/csv_document_splitter.py

概述

CSVDocumentSplitter 期望接收包含 CSV 格式内容的文档列表，并返回一个新的Document 对象列表，每个对象代表从原始文档中提取的一个子表。

拆分器有两种操作模式：

threshold (默认)：识别超过给定阈值的空行或空列，并相应地拆分文档。
row-wise：将每一行拆分成一个单独的文档，并将每一行视为一个独立的子表。

拆分过程遵循以下规则：

基于行的拆分：如果设置了row_split_threshold，则等于或超过此阈值的连续空行将触发拆分。
基于列的拆分：如果设置了column_split_threshold，则等于或超过此阈值的连续空列将触发拆分。
递归拆分：如果同时提供了两个阈值，则CSVDocumentSplitter 将首先按行拆分，然后按列拆分。如果检测到更多空行，则会再次调用拆分过程。这确保了子表的完全分离。

每个提取的子表都保留了原始文档的元数据，并包含其他字段：

source_id：原始文档的 ID。
row_idx_start：子表在原始文档中的起始行索引。
col_idx_start：子表在原始文档中的起始列索引。
split_id：文档中拆分的顺序 ID。

此组件对于需要高效提取和存储结构化数据的文档处理管道尤其有用。

支持的文档存储：

CSVDocumentSplitter 与以下文档存储兼容：

用法

单独使用

您可以使用在管道外使用 CSVDocumentSplitter 直接处理 CSV 文档。

from haystack import Document
from haystack.components.preprocessors import CSVDocumentSplitter

splitter = CSVDocumentSplitter(row_split_threshold=1, column_split_threshold=1)

doc = Document(
    content="""ID,LeftVal,,,RightVal,Extra
1,Hello,,,World,Joined
2,StillLeft,,,StillRight,Bridge
,,,,,
A,B,,,C,D
E,F,,,G,H
"""
)
split_result = splitter.run([doc])
print(split_result["documents"])  # List of split tables as Documents

在 pipeline 中

以下是如何集成CSVDocumentSplitter 到 Haystack 索引管道中。

from haystack import Pipeline, Document
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.converters.csv import CSVToDocument
from haystack.components.preprocessors import CSVDocumentSplitter
from haystack.components.preprocessors import CSVDocumentCleaner
from haystack.components.writers import DocumentWriter

# Initialize components
document_store = InMemoryDocumentStore()
p = Pipeline()
p.add_component(instance=CSVToDocument(), name="csv_file_converter")
p.add_component(instance=CSVDocumentSplitter(), name="splitter")
p.add_component(instance=CSVDocumentCleaner(), name="cleaner")
p.add_component(instance=DocumentWriter(document_store=document_store), name="writer")

# Connect components
p.connect("csv_file_converter.documents", "splitter.documents")
p.connect("splitter.documents", "cleaner.documents")
p.connect("cleaner.documents", "writer.documents")

# Run pipeline
p.run({"csv_file_converter": {"sources": ["path/to/your/file.csv"]}})

此管道将提取 CSV 内容，将其拆分为结构化子表，通过删除空行和空列来清理 CSV 文档，并将生成的文档存储在 Document Store 中以便进一步检索和处理。