路径读取器

路径是一个开源的数据处理框架。它允许您轻松开发与实时数据源和变化数据交互的数据转换流水线和机器学习应用程序。

本笔记本演示如何设置实时数据索引管道。您可以像使用常规读取器一样，从您的LLM应用程序中查询此管道的结果。然而，在底层，Pathway会在每次数据变更时更新索引，为您提供始终最新的答案。

在本笔记本中，我们将首先将 llama_index.readers.pathway.PathwayReader 读取器连接到一个公共演示文档处理流水线，该流水线：

监控多个云数据源以检测数据变化。
为数据构建向量索引。

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

本文档描述的基础流程能够轻松构建云存储文件的简单索引。然而，Pathway 提供了构建实时数据管道和应用程序所需的一切功能，包括类SQL表操作（例如分组聚合和异构数据源连接）、基于时间的数据分组与窗口化，以及丰富的连接器库。

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

前提条件

安装 llama-index-readers-pathway 集成

%pip install llama-index-readers-pathway

配置日志记录

import logging
import sys

logging.basicConfig(stream=sys.stdout, level=logging.ERROR)
logging.getLogger().addHandler(logging.StreamHandler(stream=sys.stdout))

设置您的OpenAI API密钥。

import getpass
import os

# omit if embedder of choice is not OpenAI
if "OPENAI_API_KEY" not in os.environ:
    os.environ["OPENAI_API_KEY"] = getpass.getpass("OpenAI API Key:")

创建读取器并连接到公共管道

要实例化和配置 PathwayReader，您需要提供文档索引管道的 url 或 host 和 port。在以下代码中，我们使用一个公开可用的演示管道，您可以通过 https://demo-document-indexing.pathway.stream 访问其 REST API。该演示从Google 云端硬盘和Sharepoint摄取文档，并维护用于检索文档的索引。

from llama_index.readers.pathway import PathwayReader

reader = PathwayReader(url="https://demo-document-indexing.pathway.stream")
# let us search with some text
reader.load_data(query_text="What is Pathway")

使用 llama-index 创建摘要索引

docs = reader.load_data(query_text="What is Pathway", k=2)
from llama_index.core import SummaryIndex

index = SummaryIndex.from_documents(docs)
query_engine = index.as_query_engine()
response = query_engine.query("What does Pathway do?")
print(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 应该是一个以 Embedding 转换结尾的 TransformComponent 列表。

在这个示例中，我们首先使用 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.readers.pathway import PathwayReader

reader = PathwayReader(host=PATHWAY_HOST, port=PATHWAY_PORT)
# let us search with some text
reader.load_data(query_text="What is Pathway")