Redis 数据摄取管道¶

本教程演示如何在数据摄取管道中同时使用 Redis 作为向量存储、缓存和文档存储。

依赖项¶

安装并启动 Redis，配置 OpenAI API 密钥

In [ ]:

%pip install llama-index-storage-docstore-redis
%pip install llama-index-vector-stores-redis
%pip install llama-index-embeddings-huggingface

%pip install llama-index-storage-docstore-redis %pip install llama-index-vector-stores-redis %pip install llama-index-embeddings-huggingface

In [ ]:

!docker run -d --name redis-stack -p 6379:6379 -p 8001:8001 redis/redis-stack:latest

!docker run -d --name redis-stack -p 6379:6379 -p 8001:8001 redis/redis-stack:latest

338c889086e8649aa80dfb79ebff4fffc98d72fc6d988ac158c6662e9e0cf04b

In [ ]:

import os

os.environ["OPENAI_API_KEY"] = "sk-..."
os.environ["TOKENIZERS_PARALLELISM"] = "false"

import os os.environ["OPENAI_API_KEY"] = "sk-..." os.environ["TOKENIZERS_PARALLELISM"] = "false"

创建种子数据¶

In [ ]:

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

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

In [ ]:

from llama_index.core import SimpleDirectoryReader

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

from llama_index.core import SimpleDirectoryReader # load documents with deterministic IDs documents = SimpleDirectoryReader( "./test_redis_data", filename_as_id=True ).load_data()

运行基于 Redis 的摄取管道¶

当关联向量存储后，该管道将负责将数据更新插入（upsert）至您的向量存储中。

若您仅需处理重复数据，可将策略更改为 DUPLICATES_ONLY。

In [ ]:

from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.core.ingestion import (
    DocstoreStrategy,
    IngestionPipeline,
    IngestionCache,
)
from llama_index.storage.kvstore.redis import RedisKVStore as RedisCache
from llama_index.storage.docstore.redis import RedisDocumentStore
from llama_index.core.node_parser import SentenceSplitter
from llama_index.vector_stores.redis import RedisVectorStore

from redisvl.schema import IndexSchema


embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")

custom_schema = IndexSchema.from_dict(
    {
        "index": {"name": "redis_vector_store", "prefix": "doc"},
        # customize fields that are indexed
        "fields": [
            # required fields for llamaindex
            {"type": "tag", "name": "id"},
            {"type": "tag", "name": "doc_id"},
            {"type": "text", "name": "text"},
            # custom vector field for bge-small-en-v1.5 embeddings
            {
                "type": "vector",
                "name": "vector",
                "attrs": {
                    "dims": 384,
                    "algorithm": "hnsw",
                    "distance_metric": "cosine",
                },
            },
        ],
    }
)

from llama_index.embeddings.huggingface import HuggingFaceEmbedding from llama_index.core.ingestion import ( DocstoreStrategy, IngestionPipeline, IngestionCache, ) from llama_index.storage.kvstore.redis import RedisKVStore as RedisCache from llama_index.storage.docstore.redis import RedisDocumentStore from llama_index.core.node_parser import SentenceSplitter from llama_index.vector_stores.redis import RedisVectorStore from redisvl.schema import IndexSchema embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5") custom_schema = IndexSchema.from_dict( { "index": {"name": "redis_vector_store", "prefix": "doc"}, # customize fields that are indexed "fields": [ # required fields for llamaindex {"type": "tag", "name": "id"}, {"type": "tag", "name": "doc_id"}, {"type": "text", "name": "text"}, # custom vector field for bge-small-en-v1.5 embeddings { "type": "vector", "name": "vector", "attrs": { "dims": 384, "algorithm": "hnsw", "distance_metric": "cosine", }, }, ], } )

In [ ]:

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(),
        embed_model,
    ],
    docstore=RedisDocumentStore.from_host_and_port(
        "localhost", 6379, namespace="document_store"
    ),
    vector_store=RedisVectorStore(
        schema=custom_schema,
        redis_url="redis://localhost:6379",
    ),
    cache=IngestionCache(
        cache=RedisCache.from_host_and_port("localhost", 6379),
        collection="redis_cache",
    ),
    docstore_strategy=DocstoreStrategy.UPSERTS,
)

pipeline = IngestionPipeline( transformations=[ SentenceSplitter(), embed_model, ], docstore=RedisDocumentStore.from_host_and_port( "localhost", 6379, namespace="document_store" ), vector_store=RedisVectorStore( schema=custom_schema, redis_url="redis://localhost:6379", ), cache=IngestionCache( cache=RedisCache.from_host_and_port("localhost", 6379), collection="redis_cache", ), docstore_strategy=DocstoreStrategy.UPSERTS, )

In [ ]:

nodes = pipeline.run(documents=documents)
print(f"Ingested {len(nodes)} Nodes")

nodes = pipeline.run(documents=documents) print(f"Ingested {len(nodes)} Nodes")

Ingested 2 Nodes

确认文档已成功导入¶

我们可以使用向量存储创建向量索引，并快速查询哪些文档已被识别。

In [ ]:

from llama_index.core import VectorStoreIndex

index = VectorStoreIndex.from_vector_store(
    pipeline.vector_store, embed_model=embed_model
)

from llama_index.core import VectorStoreIndex index = VectorStoreIndex.from_vector_store( pipeline.vector_store, embed_model=embed_model )

In [ ]:

print(
    index.as_query_engine(similarity_top_k=10).query(
        "What documents do you see?"
    )
)

print( index.as_query_engine(similarity_top_k=10).query( "What documents do you see?" ) )

I see two documents.

添加数据与导入¶

在这里，我们既可以更新现有文件，也能添加新文件！

In [ ]:

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

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

In [ ]:

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

nodes = pipeline.run(documents=documents)

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

documents = SimpleDirectoryReader( "./test_redis_data", filename_as_id=True ).load_data() nodes = pipeline.run(documents=documents) print(f"Ingested {len(nodes)} Nodes")

13:32:07 redisvl.index.index INFO   Index already exists, not overwriting.
Ingested 2 Nodes

In [ ]:

index = VectorStoreIndex.from_vector_store(
    pipeline.vector_store, embed_model=embed_model
)

response = index.as_query_engine(similarity_top_k=10).query(
    "What documents do you see?"
)

print(response)

for node in response.source_nodes:
    print(node.get_text())

index = VectorStoreIndex.from_vector_store( pipeline.vector_store, embed_model=embed_model ) response = index.as_query_engine(similarity_top_k=10).query( "What documents do you see?" ) print(response) for node in response.source_nodes: print(node.get_text())

You see three documents: test3.txt, test1.txt, and test2.txt.
This is a test file: three!
This is a NEW test file: one!
This is a test file: two!

如我们所见，数据已正确完成去重和更新插入操作！尽管完整流程运行了两次，但索引中仅存在三个节点。