路径读取器

Pathway is an open data processing framework. It allows you to easily develop data transformation pipelines and Machine Learning applications that work with live data sources and changing data.

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

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

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

To have your own document processing pipeline check the hosted offering or build your own by following this notebook.

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

For more details about Pathway data ingestion pipeline and vector store, visit vector store pipeline.

前提条件

安装 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:")

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

To instantiate and configure PathwayReader you need to provide either the url or the host and port of your document indexing pipeline. In the code below we use a publicly available demo pipeline, which REST API you can access at https://demo-document-indexing.pathway.stream. This demo ingests documents from Google Drive and Sharepoint and maintains an index for retrieving documents.

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)

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

前提条件

Install pathway package. Then download sample data.

%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文件夹、云存储以及任何数据流中的数据变更。

See pathway-io for more information.

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))

创建文档索引流水线

Let us create the document indexing pipeline. The transformations should be a list of TransformComponents ending with an Embedding transformation.

In this example, let’s first split the text first using TokenTextSplitter, then embed with 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")