run-llama--llama_index/docs/examples/data_connectors/PathwayReaderDemo.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a href=\"https://colab.research.google.com/github/run-llama/llama_index/blob/main/docs/examples/data_connectors/PathwayReaderDemo.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Pathway Reader"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> [Pathway](https://pathway.com/) 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.\n",
    "\n",
    "This notebook demonstrates how to set up a live data indexing pipeline. You can query the results of this pipeline from your LLM application in the same manner as you would a regular reader. However, under the hood, Pathway updates the index on each data change giving you always up-to-date answers.\n",
    "\n",
    "In this notebook, we will first connect the `llama_index.readers.pathway.PathwayReader` reader to a [public demo document processing pipeline](https://pathway.com/solutions/ai-pipelines#try-it-out) that:\n",
    "\n",
    "1. Monitors several cloud data sources for data changes.\n",
    "2. Builds a vector index for the data.\n",
    "\n",
    "To have your own document processing pipeline check the [hosted offering](https://pathway.com/solutions/ai-pipelines) or [build your own](https://pathway.com/developers/user-guide/llm-xpack/vectorstore_pipeline/) by following this notebook. \n",
    "\n",
    "The basic pipeline described in this document allows to effortlessly build a simple index of files stored in a cloud location. However, Pathway provides everything needed to build realtime data pipelines and apps, including SQL-like able operations such as groupby-reductions and joins between disparate data sources, time-based grouping and windowing of data, and a wide array of connectors. \n",
    "\n",
    "For more details about Pathway data ingestion pipeline and vector store, visit [vector store pipeline](https://pathway.com/developers/user-guide/llm-xpack/vectorstore_pipeline/)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Prerequisites\n",
    "\n",
    "Install the `llama-index-readers-pathway` integration"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%pip install llama-index-readers-pathway"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Configure logging"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "import sys\n",
    "\n",
    "logging.basicConfig(stream=sys.stdout, level=logging.ERROR)\n",
    "logging.getLogger().addHandler(logging.StreamHandler(stream=sys.stdout))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Set up your OpenAI API key."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import getpass\n",
    "import os\n",
    "\n",
    "# omit if embedder of choice is not OpenAI\n",
    "if \"OPENAI_API_KEY\" not in os.environ:\n",
    "    os.environ[\"OPENAI_API_KEY\"] = getpass.getpass(\"OpenAI API Key:\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create the reader and connect to a public pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "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](https://pathway.com/solutions/ai-pipelines#try-it-out), which REST API you can access at `https://demo-document-indexing.pathway.stream`. This demo ingests documents from [Google Drive](https://drive.google.com/drive/u/0/folders/1cULDv2OaViJBmOfG5WB0oWcgayNrGtVs) and [Sharepoint](https://navalgo.sharepoint.com/sites/ConnectorSandbox/Shared%20Documents/Forms/AllItems.aspx?id=%2Fsites%2FConnectorSandbox%2FShared%20Documents%2FIndexerSandbox&p=true&ga=1) and maintains an index for retrieving documents."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from llama_index.readers.pathway import PathwayReader\n",
    "\n",
    "reader = PathwayReader(url=\"https://demo-document-indexing.pathway.stream\")\n",
    "# let us search with some text\n",
    "reader.load_data(query_text=\"What is Pathway\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create a summary index with llama-index"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "docs = reader.load_data(query_text=\"What is Pathway\", k=2)\n",
    "from llama_index.core import SummaryIndex\n",
    "\n",
    "index = SummaryIndex.from_documents(docs)\n",
    "query_engine = index.as_query_engine()\n",
    "response = query_engine.query(\"What does Pathway do?\")\n",
    "print(response)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Building your own data processing pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Prerequisites\n",
    "\n",
    "Install `pathway` package. Then download sample data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%pip install pathway\n",
    "%pip install llama-index-embeddings-openai"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!mkdir -p 'data/'\n",
    "!wget 'https://gist.githubusercontent.com/janchorowski/dd22a293f3d99d1b726eedc7d46d2fc0/raw/pathway_readme.md' -O 'data/pathway_readme.md'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define data sources tracked by Pathway"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Pathway can listen to many sources simultaneously, such as local files, S3 folders, cloud storage and any data stream for data changes.\n",
    "\n",
    "See [pathway-io](https://pathway.com/developers/api-docs/pathway-io) for more information."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pathway as pw\n",
    "\n",
    "data_sources = []\n",
    "data_sources.append(\n",
    "    pw.io.fs.read(\n",
    "        \"./data\",\n",
    "        format=\"binary\",\n",
    "        mode=\"streaming\",\n",
    "        with_metadata=True,\n",
    "    )  # This creates a `pathway` connector that tracks\n",
    "    # all the files in the ./data directory\n",
    ")\n",
    "\n",
    "# This creates a connector that tracks files in Google drive.\n",
    "# please follow the instructions at https://pathway.com/developers/tutorials/connectors/gdrive-connector/ to get credentials\n",
    "# data_sources.append(\n",
    "#     pw.io.gdrive.read(object_id=\"17H4YpBOAKQzEJ93xmC2z170l0bP2npMy\", service_user_credentials_file=\"credentials.json\", with_metadata=True))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create the document indexing pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let us create the document indexing pipeline. The `transformations` should be a list of `TransformComponent`s ending with an `Embedding` transformation.\n",
    "\n",
    "In this example, let's first split the text first using `TokenTextSplitter`, then embed with `OpenAIEmbedding`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pathway.xpacks.llm.vector_store import VectorStoreServer\n",
    "from llama_index.embeddings.openai import OpenAIEmbedding\n",
    "from llama_index.core.node_parser import TokenTextSplitter\n",
    "\n",
    "embed_model = OpenAIEmbedding(embed_batch_size=10)\n",
    "\n",
    "transformations_example = [\n",
    "    TokenTextSplitter(\n",
    "        chunk_size=150,\n",
    "        chunk_overlap=10,\n",
    "        separator=\" \",\n",
    "    ),\n",
    "    embed_model,\n",
    "]\n",
    "\n",
    "processing_pipeline = VectorStoreServer.from_llamaindex_components(\n",
    "    *data_sources,\n",
    "    transformations=transformations_example,\n",
    ")\n",
    "\n",
    "# Define the Host and port that Pathway will be on\n",
    "PATHWAY_HOST = \"127.0.0.1\"\n",
    "PATHWAY_PORT = 8754\n",
    "\n",
    "# `threaded` runs pathway in detached mode, we have to set it to False when running from terminal or container\n",
    "# for more information on `with_cache` check out https://pathway.com/developers/api-docs/persistence-api\n",
    "processing_pipeline.run_server(\n",
    "    host=PATHWAY_HOST, port=PATHWAY_PORT, with_cache=False, threaded=True\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Connect the reader to the custom pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from llama_index.readers.pathway import PathwayReader\n",
    "\n",
    "reader = PathwayReader(host=PATHWAY_HOST, port=PATHWAY_PORT)\n",
    "# let us search with some text\n",
    "reader.load_data(query_text=\"What is Pathway\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}