摄取管道 + 文档管理¶

将 docstore 附加到摄取管道将启用文档管理功能。

使用 document.doc_id 或 node.ref_doc_id 作为基准点，摄取管道将主动查找重复文档。

其工作原理如下：

• 存储 doc_id -> document_hash 的映射
• 如果检测到重复的 doc_id 且哈希值已更改，则文档将重新处理
• 如果哈希值未更改，则管道将跳过该文档

如果我们不附加向量存储，我们只能检查并移除重复的输入。

如果附加了向量存储，我们还可以处理 upsert（更新或插入）！我们有另一篇指南介绍 upsert 和向量存储。

创建种子数据¶

输入 [ ]

%pip install llama-index-storage-docstore-redis
%pip install llama-index-storage-docstore-mongodb
%pip install llama-index-embeddings-huggingface

%pip install llama-index-storage-docstore-redis %pip install llama-index-storage-docstore-mongodb %pip install llama-index-embeddings-huggingface

输入 [ ]

# Make some test data
!mkdir -p data
!echo "This is a test file: one!" > data/test1.txt
!echo "This is a test file: two!" > data/test2.txt

# 创建一些测试数据 !mkdir -p data !echo "This is a test file: one!" > data/test1.txt !echo "This is a test file: two!" > data/test2.txt

输入 [ ]

from llama_index.core import SimpleDirectoryReader

# load documents with deterministic IDs
documents = SimpleDirectoryReader("./data", filename_as_id=True).load_data()

from llama_index.core import SimpleDirectoryReader # 加载带有确定性 ID 的文档 documents = SimpleDirectoryReader("./data", filename_as_id=True).load_data()

/home/loganm/.cache/pypoetry/virtualenvs/llama-index-4a-wkI5X-py3.11/lib/python3.11/site-packages/deeplake/util/check_latest_version.py:32: UserWarning: A newer version of deeplake (3.8.9) is available. It's recommended that you update to the latest version using `pip install -U deeplake`.
  warnings.warn(

使用文档存储创建管道¶

输入 [ ]

from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.storage.docstore.redis import RedisDocumentStore
from llama_index.storage.docstore.mongodb import MongoDocumentStore
from llama_index.core.node_parser import SentenceSplitter


pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(),
        HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5"),
    ],
    docstore=SimpleDocumentStore(),
)

from llama_index.embeddings.huggingface import HuggingFaceEmbedding from llama_index.core.ingestion import IngestionPipeline from llama_index.core.storage.docstore import SimpleDocumentStore from llama_index.storage.docstore.redis import RedisDocumentStore from llama_index.storage.docstore.mongodb import MongoDocumentStore from llama_index.core.node_parser import SentenceSplitter pipeline = IngestionPipeline( transformations=[ SentenceSplitter(), HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5"), ], docstore=SimpleDocumentStore(), )

输入 [ ]

nodes = pipeline.run(documents=documents)

nodes = pipeline.run(documents=documents)

Docstore strategy set to upserts, but no vector store. Switching to duplicates_only strategy.

输入 [ ]

print(f"Ingested {len(nodes)} Nodes")

print(f"已摄取 {len(nodes)} 个节点")

Ingested 2 Nodes

[可选] 保存/加载管道¶

保存管道将保存内部缓存和文档存储。

注意：如果您使用远程缓存/文档存储，则不需要此步骤

输入 [ ]

pipeline.persist("./pipeline_storage")

pipeline.persist("./pipeline_storage")

输入 [ ]

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(),
        HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5"),
    ]
)

# restore the pipeline
pipeline.load("./pipeline_storage")

pipeline = IngestionPipeline( transformations=[ SentenceSplitter(), HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5"), ] ) # 恢复管道 pipeline.load("./pipeline_storage")

测试文档管理¶

在这里，我们可以创建一个新文档，以及编辑现有文档，以测试文档管理功能。

新文档和已编辑文档都将摄取，而未更改的文档将被跳过

输入 [ ]

!echo "This is a test file: three!" > data/test3.txt
!echo "This is a NEW test file: one!" > data/test1.txt

!echo "This is a test file: three!" > data/test3.txt !echo "This is a NEW test file: one!" > data/test1.txt

输入 [ ]

documents = SimpleDirectoryReader("./data", filename_as_id=True).load_data()

documents = SimpleDirectoryReader("./data", filename_as_id=True).load_data()

输入 [ ]

nodes = pipeline.run(documents=documents)

nodes = pipeline.run(documents=documents)

Docstore strategy set to upserts, but no vector store. Switching to duplicates_only strategy.

输入 [ ]

print(f"Ingested {len(nodes)} Nodes")

print(f"已摄取 {len(nodes)} 个节点")

Ingested 2 Nodes

让我们确认哪些节点被摄取了

输入 [ ]

for node in nodes:
    print(f"Node: {node.text}")

for node in nodes: print(f"节点: {node.text}")

Node: This is a NEW test file: one!
Node: This is a test file: three!

我们还可以验证文档存储只跟踪了三个文档

输入 [ ]

print(len(pipeline.docstore.docs))

print(len(pipeline.docstore.docs))

3