Pathway 是一个开放的数据处理框架。它让你可以轻松开发可处理实时数据源和变化数据的数据转换流水线与机器学习应用。

本 notebook 演示如何将实时数据索引流水线与 LlamaIndex 配合使用。你可以使用提供的 PathwayRetriever 从 LLM 应用中查询此流水线的结果。然而，在底层，Pathway 会在每次数据变更时更新索引，从而为你提供始终最新的答案。

在本 notebook 中，我们将使用一个公开演示文档处理流水线，它会：

1. 监控多个云数据源的数据变更。
2. 为数据构建向量索引。

要拥有你自己的文档处理流水线，请查看托管服务，或按照本 notebook 自行构建。

我们将使用 llama_index.retrievers.pathway.PathwayRetriever 检索器连接到索引，它实现了 retrieve 接口。

本文档描述的基本流水线让你能够轻松地为存储在云位置的若干文件构建简单索引。然而，Pathway 提供了构建实时数据流水线和应用所需的一切，包括类似 SQL 的操作（如 groupby 归约和跨不同数据源的连接）、基于时间的数据分组和窗口化，以及大量连接器。

有关 Pathway 数据摄取流水线和向量存储的更多细节，请访问向量存储流水线。

前置条件

要使用 PathwayRetrievier，你必须安装 llama-index-retrievers-pathway 包。

!pip install llama-index-retrievers-pathway

为 llama-index 创建检索器

要实例化并配置 PathwayRetriever，你需要提供文档索引流水线的 url，或者 host 和 port。在下面的代码中，我们使用一个公开可用的演示流水线，其 REST API 可通过 https://demo-document-indexing.pathway.stream 访问。该演示从 Google Drive 和 Sharepoint 摄取文档，并维护一个用于检索文档的索引。

from llama_index.retrievers.pathway import PathwayRetriever

retriever = PathwayRetriever(
    url="https://demo-document-indexing.pathway.stream"
)
retriever.retrieve(str_or_query_bundle="what is pathway")

轮到你了！ 获取你的流水线，或向演示流水线上传新文档，然后重试查询！

在查询引擎中使用

from llama_index.core.query_engine import RetrieverQueryEngine

query_engine = RetrieverQueryEngine.from_args(
    retriever,
)

response = query_engine.query("Tell me about Pathway")
print(str(response))

构建你自己的数据处理流水线

前提条件

安装 pathway 包。然后下载示例数据。

%pip install pathway
%pip install llama-index-embeddings-openai

!mkdir -p 'data/'
!wget 'https://gist.githubusercontent.com/janchorowski/dd22a293f3d99d1b726eedc7d46d2fc0/raw/pathway_readme.md' -O 'data/pathway_readme.md'

定义由 Pathway 跟踪的数据源

Pathway 可以同时监听多种来源，例如本地文件、S3 文件夹、云存储以及任何用于数据变更的数据流。

更多信息请参见 pathway-io。

import pathway as pw

data_sources = []
data_sources.append(
    pw.io.fs.read(
        "./data",
        format="binary",
        mode="streaming",
        with_metadata=True,
    )  # This creates a `pathway` connector that tracks
    # all the files in the ./data directory
)

# This creates a connector that tracks files in Google drive.
# please follow the instructions at https://pathway.com/developers/tutorials/connectors/gdrive-connector/ to get credentials
# data_sources.append(
#     pw.io.gdrive.read(object_id="17H4YpBOAKQzEJ93xmC2z170l0bP2npMy", service_user_credentials_file="credentials.json", with_metadata=True))

创建文档索引流水线

让我们创建文档索引流水线。transformations 应该是一个 TransformComponent 列表，并以 Embedding 转换结尾。

在本示例中，我们先使用 TokenTextSplitter 对文本进行切分，然后使用 OpenAIEmbedding 进行嵌入。

from pathway.xpacks.llm.vector_store import VectorStoreServer
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.node_parser import TokenTextSplitter

embed_model = OpenAIEmbedding(embed_batch_size=10)

transformations_example = [
    TokenTextSplitter(
        chunk_size=150,
        chunk_overlap=10,
        separator=" ",
    ),
    embed_model,
]

processing_pipeline = VectorStoreServer.from_llamaindex_components(
    *data_sources,
    transformations=transformations_example,
)

# Define the Host and port that Pathway will be on
PATHWAY_HOST = "127.0.0.1"
PATHWAY_PORT = 8754

# `threaded` runs pathway in detached mode, we have to set it to False when running from terminal or container
# for more information on `with_cache` check out https://pathway.com/developers/api-docs/persistence-api
processing_pipeline.run_server(
    host=PATHWAY_HOST, port=PATHWAY_PORT, with_cache=False, threaded=True
)

将检索器连接到自定义流水线

from llama_index.retrievers.pathway import PathwayRetriever

retriever = PathwayRetriever(host=PATHWAY_HOST, port=PATHWAY_PORT)
retriever.retrieve(str_or_query_bundle="what is pathway")