ContextQMD
Back to Pathway

Installing Pathway with Python 3.10+

examples/notebooks/showcases/live_vector_indexing_pipeline.ipynb

0.30.112.7 KB
Original Source

<a href="https://colab.research.google.com/github/pathwaycom/pathway/blob/main/examples/notebooks/showcases/live_vector_indexing_pipeline.ipynb" target="_parent"></a>

Installing Pathway with Python 3.10+

In the cell below, we install Pathway into a Python 3.10+ Linux runtime.

If you are running in Google Colab, please run the colab notebook (Ctrl+F9), disregarding the 'not authored by Google' warning.

The installation and loading time is less than 1 minute.

python
%%capture --no-display
!pip install --prefer-binary pathway
python
import logging

logging.basicConfig(level=logging.CRITICAL)

Always Up-to-date Data Indexing pipeline

This showcase demonstrates how to use Pathway to deploy a live data indexing pipeline that can be queried similarly to a typical vector store. Unlike traditional approaches, Pathway updates the index with every data change, ensuring consistently up-to-date answers.

<!-- canva link: https://www.canva.com/design/DAF1cxQW5Vg/LcFdDrPApBrgwM5kyirY6w/edit -->

::article-img

src: '/assets/content/showcases/vectorstore/vectorstore_doc.png' alt: 'Pathway data indexing pipeline' class: 'mx-auto' zoomable: true

::

Pathway Vectorstore enables building a document index on top of your documents without the complexity of ETL pipelines, managing different containers for storing, embedding, and serving. It allows for easy to manage, always up-to-date, LLM pipelines accessible using a RESTful API and with integrations to popular LLM toolkits such as Langchain and LlamaIndex.

In this article, we will use a simple document processing pipeline that:

  1. Monitors several data sources (files, S3 folders, cloud storages) for data changes.
  2. Parses, splits and embeds the documents.
  3. Builds a vector index for the data.

However, If you prefer not to create the pipeline from the ground up and would like to check out the functionality, take a look at our managed pipelines in action.

We will connect to the index using a VectorStore client, which allows retrieval of semantically similar documents.

Prerequisites

Install the pathway package. You can also install the unstructured package to use the most powerful unstructured.io-based parser.

Then download sample data.

python
!pip install "pathway[xpack-llm,xpack-llm-docs]"
 !pip install unstructured[all-docs]
!mkdir -p sample_documents
![ -f sample_documents/repo_readme.md ] || wget 'https://gist.githubusercontent.com/janchorowski/dd22a293f3d99d1b726eedc7d46d2fc0/raw/pathway_readme.md' -O 'sample_documents/repo_readme.md'
python
import logging
import sys
import time

logging.basicConfig(stream=sys.stderr, level=logging.WARN, force=True)

Building the data pipeline

First, make sure you have an API key with an LLM provider such as OpenAI.

python
import getpass
import os

if "OPENAI_API_KEY" not in os.environ:
    os.environ["OPENAI_API_KEY"] = getpass.getpass("OpenAI API Key:")

We will now assemble the data vectorization pipeline, using a simple UTF8 file parser, a character splitter and an embedder from the Pathway LLM xpack.

First, we define the data sources. We use the files-based one for simplicity, but any supported pathway connector, such as s3 or Google Drive will also work.

Then, we define the embedder and splitter.

Last, we assemble the data pipeline. We will start it running in a background thread to be able to query it immediately from the demonstration. Please note that in a production deployment, the server will run in another process, possibly on another machine. For the quick-start, we keep the server and client as different threads of the same Python process.

python
import pathway as pw

# To use advanced features with Pathway Scale, get your free license key from
# https://pathway.com/features and paste it below.
# To use Pathway Community, comment out the line below.
pw.set_license_key("demo-license-key-with-telemetry")


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

# This creates a connector that tracks files in Google Drive.
# Please follow the instructions at /developers/user-guide/connect/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))
python
# We now build the VectorStore pipeline

from pathway.xpacks.llm.embedders import OpenAIEmbedder
from pathway.xpacks.llm.splitters import TokenCountSplitter
from pathway.xpacks.llm.vector_store import VectorStoreClient, VectorStoreServer

PATHWAY_PORT = 8765

# Choose document transformers
text_splitter = TokenCountSplitter()
embedder = OpenAIEmbedder(api_key=os.environ["OPENAI_API_KEY"])

# The `PathwayVectorServer` is a wrapper over `pathway.xpacks.llm.vector_store` to accept LangChain transformers.
# Fell free to fork it to develop bespoke document processing pipelines.
vector_server = VectorStoreServer(
    *data_sources,
    embedder=embedder,
    splitter=text_splitter,
)
vector_server.run_server(
    host="127.0.0.1", port=PATHWAY_PORT, threaded=True, with_cache=False
)
time.sleep(
    30
)  # Workaround for Colab - messages from threads are not visible unless a cell is running

We now instantiate and configure the client

python
client = VectorStoreClient(
    host="127.0.0.1",
    port=PATHWAY_PORT,
)

And we can start asking queries

python
query = "What is Pathway?"
docs = client(query)
docs

Your turn! Now make a change to the source documents or make a fresh one and retry the query!

Integrations

Langchain

You can use a Pathway Vector Store in LangChain pipelines with PathwayVectorClient and configure a VectorStoreServer using LangChain components. For more information see our article or LangChain documentation.

python
!pip install langchain
!pip install langchain-openai
!pip install langchain-community
python
from langchain_community.vectorstores import PathwayVectorClient

# PathwayVectorClient implements regular VectorStore API of LangChain
client = PathwayVectorClient(host="127.0.0.1", port=PATHWAY_PORT)
docs = client.similarity_search("What is Pathway?")
python
# Here we show how to configure a server that uses LangChain document processing components

from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter

# Choose proper LangChain document transformers
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
embeddings_model = OpenAIEmbeddings(openai_api_key=os.environ["OPENAI_API_KEY"])

# Use VectorStoreServer.from_langchain_components to create a vector server using LangChain
# document processors
vector_server = VectorStoreServer.from_langchain_components(
    *data_sources,
    embedder=embeddings_model,
    splitter=text_splitter,
)
vector_server.run_server(
    host="127.0.0.1", port=PATHWAY_PORT + 1, threaded=True, with_cache=False
)
time.sleep(30)  # colab workaround
python
# You can connect to the Pathway+LangChain server using any client - Pathway's, Langchain's or LlamaIndex's!
client = VectorStoreClient(
    host="127.0.0.1",
    port=PATHWAY_PORT + 1,
)

client.query("pathway")

LlamaIndex

Pathway is fully integrated with LlamaIndex! We show below how to instantiate a Llama-Index retriever that queries the Pathway VectorStoreServer and how to configure a server using LlamaIndex components.

For more information see Pathway Retriever cookbook.

python
!pip install llama-index llama-index-retrievers-pathway llama-index-embeddings-openai
python
# You can connect to the PathwayVectorStore using a llama-index compatible retriever
from llama_index.retrievers.pathway import PathwayRetriever

# PathwayRetriever implements the Retriever interface
pr = PathwayRetriever(host="127.0.0.1", port=PATHWAY_PORT)
pr.retrieve(str_or_query_bundle="What is Pathway?")
python
# Here we show how to configure a server that uses LlamaIndex document processing components

from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.node_parser import TokenTextSplitter

# Choose proper LlamaIndex document transformers
embed_model = OpenAIEmbedding(embed_batch_size=10)

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

# Use VectorStoreServer.from_llamaindex_components to create a vector server using LlamaIndex
# document processors
vector_server = VectorStoreServer.from_llamaindex_components(
    *data_sources,
    transformations=transformations_example,
)
vector_server.run_server(
    host="127.0.0.1", port=PATHWAY_PORT + 2, threaded=True, with_cache=False
)
time.sleep(30)  # colab workaround
python
# You can connect to the Pathway+LlamaIndex server using any client - Pathway's, Langchain's or LlamaIndex's!
client = VectorStoreClient(
    host="127.0.0.1",
    port=PATHWAY_PORT + 2,
)

client.query("pathway")

Advanced topics

Getting information on indexed files

PathwayVectorClient.get_vectorstore_statistics() gives essential statistics on the state of the vector store, like the number of indexed files and the timestamp of the last updated one. You can use it in your chains to tell the user how fresh your knowledge base is.

python
client.get_vectorstore_statistics()

You can also use PathwayVectorClient.get_input_files() to get the list of indexed files along with the associated metadata.

python
client.get_input_files()

Filtering based on file metadata

We support document filtering using jmespath expressions, for instance:

python
# take into account only sources modified later than unix timestamp
docs = client(query, metadata_filter="modified_at >= `1702672093`")

# take into account only sources modified later than unix timestamp
docs = client(query, metadata_filter="owner == `james`")

# take into account only sources with path containing 'repo_readme'
docs = client(query, metadata_filter="contains(path, 'repo_readme')")

# and of two conditions
docs = client(query, metadata_filter="owner == `james` && modified_at >= `1702672093`")

# or of two conditions
docs = client(query, metadata_filter="owner == `james` || modified_at >= `1702672093`")

Configuring the parser

The vectorization pipeline supports pluggable parsers. If not provided, defaults to UTF-8 parser. You can find available parsers here. An example parser that can read PDFs, Word documents and other formats is provided with parsers.UnstructuredParser:

python
from pathway.xpacks.llm import parsers

vector_server = VectorStoreServer(
    *data_sources,
    parser=parsers.UnstructuredParser(),
    embedder=embeddings_model,
    splitter=text_splitter,
)

Configuring the cache

The Pathway vectorizing pipeline comes with an embeddings cache:

python
vector_server.run_server(..., with_cache=True)

The default cache configuration is the locally hosted disk cache, stored in the ./Cache directory. However, it can be customized by explicitly specifying the caching backend chosen among several persistent backend options.

Running in production

A production deployment will typically run the server in a separate process. We recommend running the Pathway data indexing pipeline in a container-based deployment environment like Docker or Kubernetes. For more info, see Pathway's deployment guide.

::shoutout-banner

href: "https://discord.gg/pathway" icon: "ic:baseline-discord"

#title Discuss tricks & tips for RAG #description Join our Discord community and dive into discussions on tricks and tips for mastering Retrieval Augmented Generation ::