IngestionPipeline 使用一组 Transformations 作用于输入数据。这些 Transformations 应用到你的输入数据后，得到的节点会被直接返回，或（在提供时）插入向量数据库。每个「节点 + 变换」的组合都会被缓存，因此后续运行（前提是缓存被持久化）遇到同样的组合时可直接复用缓存结果，为你省下时间。

使用方式

最简单的用法是像这样实例化一个 IngestionPipeline：

from llama_index.core import Document
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.extractors import TitleExtractor
from llama_index.core.ingestion import IngestionPipeline, IngestionCache

# create the pipeline with transformations
pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=25, chunk_overlap=0),
        TitleExtractor(),
        OpenAIEmbedding(),
    ]
)

# run the pipeline
nodes = pipeline.run(documents=[Document.example()])

注意在真实场景下，你的文档应来自 SimpleDirectoryReader 或 Llama Hub 上的其他 reader。

连接向量数据库

运行摄取流水线时，你还可以选择把结果节点自动写入远端的向量库。

随后你可以从这个向量库构建索引。

from llama_index.core import Document
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.extractors import TitleExtractor
from llama_index.core.ingestion import IngestionPipeline
from llama_index.vector_stores.qdrant import QdrantVectorStore

import qdrant_client

client = qdrant_client.QdrantClient(location=":memory:")
vector_store = QdrantVectorStore(client=client, collection_name="test_store")

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=25, chunk_overlap=0),
        TitleExtractor(),
        OpenAIEmbedding(),
    ],
    vector_store=vector_store,
)

# Ingest directly into a vector db
pipeline.run(documents=[Document.example()])

# Create your index
from llama_index.core import VectorStoreIndex

index = VectorStoreIndex.from_vector_store(vector_store)

在流水线中计算嵌入

请注意，上面例子里的嵌入是作为流水线的一部分计算的。如果你把流水线连到向量库，嵌入就必须是流水线中的一个环节，否则之后实例化索引时会失败。

如果不连向量库、只是产出一个节点列表，那么流水线里可以不包含嵌入这一步。

缓存

在 IngestionPipeline 中，每个「节点 + 变换」组合都会被哈希并缓存。这在后续使用相同数据的运行中能省下时间。

下面几节介绍缓存相关的基本用法。

本地缓存管理

有了流水线之后，你可能想保存并加载缓存。

# save
pipeline.persist("./pipeline_storage")

# load and restore state
new_pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=25, chunk_overlap=0),
        TitleExtractor(),
    ],
)
new_pipeline.load("./pipeline_storage")

# will run instantly due to the cache
nodes = pipeline.run(documents=[Document.example()])

如果缓存变得太大，你可以清空它

# delete all context of the cache
cache.clear()

远端缓存管理

我们为缓存支持多种远端存储后端

• RedisCache
• MongoDBCache
• FirestoreCache

这里以 RedisCache 为例：

from llama_index.core import Document
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.extractors import TitleExtractor
from llama_index.core.ingestion import IngestionPipeline, IngestionCache
from llama_index.storage.kvstore.redis import RedisKVStore as RedisCache

ingest_cache = IngestionCache(
    cache=RedisCache.from_host_and_port(host="127.0.0.1", port=6379),
    collection="my_test_cache",
)

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=25, chunk_overlap=0),
        TitleExtractor(),
        OpenAIEmbedding(),
    ],
    cache=ingest_cache,
)

# Ingest directly into a vector db
nodes = pipeline.run(documents=[Document.example()])

这种情况下不需要持久化步骤，因为一切都边运行边缓存进了指定的远端集合。

异步支持

IngestionPipeline 也支持异步操作

nodes = await pipeline.arun(documents=documents)

文档管理

给摄取流水线挂上 docstore 即可启用文档管理。

以 document.doc_id 或 node.ref_doc_id 作为锚点，摄取流水线会主动查找重复文档。

它的工作方式是：

• 保存一份 doc_id -> document_hash 的映射

• 如果附加了向量库：

  • 若检测到重复 doc_id 且哈希已变化，则重新处理该文档并 upsert
  • 若检测到重复 doc_id 但哈希未变，则跳过该节点

• 如果没有附加向量库：

  • 对每个节点检查所有已存在的哈希
  • 若发现重复，则跳过该节点
  • 否则处理该节点

**注意：**如果不附加向量库，我们只能检查并去除重复的输入。

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.storage.docstore import SimpleDocumentStore

pipeline = IngestionPipeline(
    transformations=[...], docstore=SimpleDocumentStore()
)

完整演示见我们的示例 notebook。

也可以看看另一篇用 Redis 承担整个摄取技术栈的指南。

并行处理

IngestionPipeline 的 run 方法可以用多进程执行。它借助 multiprocessing.Pool，把一批批节点分配到各个处理器上。

要并行执行，只需把 num_workers 设为你想使用的进程数：

from llama_index.core.ingestion import IngestionPipeline

pipeline = IngestionPipeline(
    transformations=[...],
)
pipeline.run(documents=[...], num_workers=4)

模块

• 变换指南
• 进阶摄取流水线
• 异步摄取流水线
• 文档管理流水线
• Redis 摄取流水线
• Google Drive 摄取流水线
• 并行执行流水线