03 - TransformComponent 数据流水线模式

03 - TransformComponent 数据流水线模式

学习目标

本章将带你分析 LlamaIndex 的数据流水线设计模式。你将学到：

• TransformComponent 统一接口设计
• IngestionPipeline 的编排与转换链
• SHA256 缓存键策略
• Docstore 去重机制与三种策略
• 增量更新的实现方式

项目实践

TransformComponent 统一接口

LlamaIndex 中数据处理流水线的每一步都实现 TransformComponent 接口：

class TransformComponent(Protocol):
    def __call__(self, nodes: Sequence[BaseNode], **kwargs) -> Sequence[BaseNode]: ...

关键设计：__call__ 方法接收 Sequence[BaseNode] 并返回 Sequence[BaseNode]。这使得所有转换步骤可以串联：

Documents → [SentenceSplitter] → Nodes → [Embedding] → NodesWithEmbeddings → [VectorStore] → 持久化

每个转换步骤的输入和输出类型相同（都是 Node），这允许任意组合和替换。

IngestionPipeline 编排

IngestionPipeline 是转换链的编排器，负责按序应用 transformations：

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=512, chunk_overlap=20),
        Settings.embed_model,  # 自动为 nodes 计算 embedding
    ],
    vector_store=my_vector_store,
    docstore=my_docstore,
)

nodes = pipeline.run(documents=documents)

执行流程：

SHA256 缓存策略

IngestionPipeline 的缓存键基于 SHA256 哈希：

def get_transformation_hash(node: BaseNode, transformation: TransformComponent) -> str:
    # 哈希输入 = 节点内容 + 转换配置
    content = node.get_content()
    config = transformation.get_config_hash()  # 转换器的参数哈希
    return sha256(f"{content}:{config}".encode()).hexdigest()

缓存失效的条件：

• 节点内容变化（文档更新）
• 转换配置变化（chunk_size、模型名称等参数改变）

陷阱：修改分块参数后，旧缓存仍然命中。因为缓存键包含节点内容（而非配置），旧分块的缓存仍然有效，导致新参数不生效。需要清空缓存或删除 docstore 数据。

Docstore 去重策略

通过 docstore 存储文档 hash，IngestionPipeline 支持三种去重策略：

策略	行为	适用场景
UPSERTS	文档 hash 变化时重新处理并更新	文档可能更新
DUPLICATES_ONLY	只处理新文档，跳过已存在的	增量添加，不更新旧文档
UPSERTS_AND_DELETE	重新处理 + 删除已不存在的文档	同步文档源（删除旧文档）

UPSERTS_AND_DELETE 的工作流程：

1. 检查 docstore 中已有的 ref_doc_id
2. 如果当前运行中没有遇到某个 ref_doc_id → 删除它及其所有关联 nodes
3. 处理新/更新的文档

并行处理

IngestionPipeline 支持多进程并行：

nodes = pipeline.run(
    documents=documents,
    num_workers=4,  # 4 个进程并行处理
    show_progress=True,
)

使用 multiprocessing.Pool 实现，适合大量文档的初始索引。

问题与规避

陷阱	对策
修改分块参数后缓存仍然命中	清空缓存目录或使用 IngestionPipeline(cache=None) 禁用缓存
UPSERTS_AND_DELETE 意外删除文档	确认输入 documents 包含所有应该保留的文档
多进程处理时嵌入 API 超额调用	控制 num_workers，避免并发过高导致 API rate limit
Embedding 失败导致整个 pipeline 失败	确保 embed_model 正确配置；使用 num_workers=1 调试

设计取舍

TransformComponent vs 专用方法

维度	TransformComponent (__call__)	专用方法 (transform_documents)
可组合性	高（统一接口，任意串联）	低（接口不统一）
可读性	中（需要理解 __call__ 模式）	高（方法名自解释）
扩展性	高（新增转换只需实现接口）	低（需要修改 pipeline 代码）

为什么选择 TransformComponent：统一接口使得 IngestionPipeline 可以接受任意转换步骤，用户可以在 transformations 列表中插入任何实现了该接口的组件。代价是需要理解 __call__ 的鸭子类型模式。

IngestionPipeline vs 直接 from_documents()

维度	IngestionPipeline	index.from_documents()
缓存	支持（SHA256 去重）	不支持
增量更新	支持（docstore 策略）	全量重建
灵活性	高（自定义 transformations）	低（固定流程）
代码复杂度	高（需要配置 pipeline）	低（一行代码）

参考来源

• LlamaIndex IngestionPipeline 源码：llama-index-core/llama_index/core/ingestion/pipeline.py
• LlamaIndex TransformComponent 接口：llama-index-core/llama_index/core/schema.py