Pathway LLM-App

This template demonstrates how to build a multimodal Retrieval-Augmented Generation (RAG) application using the Pathway Live Data Framework. It is specifically designed to process unstructured financial documents (PDFs) containing visually formatted elements like tables, charts, and images.

Key features include:

• Vision-Language Integration: Uses GPT-4o to parse and understand tables and images within documents, outperforming traditional text-only RAG.
• Dynamic Indexing: The Pathway framework maintains a scalable in-memory index that automatically updates as documents in the source directory are modified or added.
• Continuous Sync: The application runs in permanent connection with your data sources (local folders, Google Drive, SharePoint, etc.).

The Realtime Document Indexing pipeline is a service powered by the Pathway Live Data Framework that creates a vector store with always-up-to-date knowledge. It supports real-time indexing from Microsoft 365 SharePoint, Google Drive, or local directories.

Key capabilities include:

• Similarity search: Query documents using natural language.
• Metadata filtering: Filter results using JMESPath format.
• Health monitoring: Access basic statistics about the indexer's status.

The pipeline works by reading data via Pathway connectors, parsing and chunking it with the unstructured library, embedding chunks via an embedding model (e.g., OpenAI), and indexing them using Pathway's machine-learning library.

This project implements a two-part AI pipeline designed to extract and structure data from unstructured sources (like PDFs) on the fly and allow querying via natural language.

Pipeline Components

1. Proactive Data Pipeline: A live pipeline that tracks file changes in a directory, reads documents (e.g., financial PDFs), structures them using the OpenAI API, and writes the results to a PostgreSQL instance.
2. Query Answering Pipeline: A REST API endpoint that accepts natural language queries, converts them into SQL queries using the OpenAI API, and executes them against the PostgreSQL data.

Extracted Data Schema

The pipeline extracts the following fields from financial PDF documents:

• company_symbol (str)
• year (int)
• quarter (str)
• revenue_md (float) - in millions of dollars
• eps (float) - in dollars
• net_income_md (float) - in millions of dollars

This template demonstrates how to build a real-time Retrieval-Augmented Generation (RAG) application using the Pathway Live Data Framework. Unlike traditional RAG setups that require separate ETL pipelines to sync data to a VectorDB, this application uses a streaming approach where changes in data sources (like local directories, Google Drive, or SharePoint) are automatically parsed, embedded, and indexed in real-time.

Key Capabilities:

• Real-time Indexing: Automatically processes new or updated documents from configured sources.
• Document Store: Uses a Pathway Document Store to manage knowledge without manual ETL.
• LLM Integration: Connects to LLMs (defaulting to OpenAI) for question-answering and summarization.
• Flexible Querying: Supports asking questions about specific files, folders, or the entire document set via filtering.

Pathway AI Pipelines are ready-to-deploy LLM application templates designed for high-accuracy RAG (Retrieval-Augmented Generation) and enterprise search at scale.

Key features include:

• Live Data Synchronization: Automatically syncs additions, deletions, and updates from data sources like File Systems, Google Drive, SharePoint, S3, Kafka, PostgreSQL, and real-time APIs.
• Built-in Indexing: Includes in-memory vector search, hybrid search, and full-text search without requiring external infrastructure like Pinecone, Weaviate, or Redis.
• Deployment Flexibility: Can be run as Docker containers, tested locally, or deployed to cloud providers (GCP, AWS, Azure, Render) and on-premises.
• Unified Stack: Combines backend logic, embedding, retrieval, and LLM management into a single framework using the Pathway Live Data Framework.

This template demonstrates how to build a fully private Retrieval-Augmented Generation (RAG) pipeline using the Pathway Live Data Framework. It is designed to ensure data security by keeping all processing local, utilizing Mistral 7B (via Ollama) and open-source embedding models.

Key components include:

• Pathway Live Data Framework: Handles real-time data synchronization and provides a built-in vector store.
• AdaptiveRAGQuestionAnswerer: An extension of BaseRAGQuestionAnswerer that implements adaptive retrieval, automatically adjusting the number of context chunks used until an answer is found.
• Local LLM: Deployed via Ollama to ensure no data leaves the local machine.

The application exposes a REST endpoint at /v2/answer which accepts queries via a pw_ai_queries table.

This template provides a Model Context Protocol (MCP) server that exposes a real-time document indexing pipeline. It allows LLM applications to interact with indexed documents via standardized tools.

Key Capabilities:

• Real-time indexing: Automatically tracks changes in Microsoft 365 SharePoint, Google Drive, or local directories.
• Similarity Search: Perform vector searches based on user queries.
• Metadata Filtering: Filter results using JMESPath syntax.
• Standardized Interface: Accessible via any MCP-compliant client.

Available MCP Tools:

• retrieve_query: Performs similarity search on indexed documents.
• statistics_query: Returns health statistics for the indexer.
• inputs_query: Retrieves metadata for all files currently being processed.

The behavior and components of the RAG pipeline are configured using an app.yaml file. You can modify or replace the following sections:

• Input Connectors: Define where your data comes from (Local, Google Drive, SharePoint, etc.).
• LLM: Choose between local models (e.g., via Ollama) or API-based models.
• Embedder: Select the embedding model used for vectorization.
• Index: Configure the vector store settings.
• Webserver: Set the host and port for the API.
• Cache: Enable persistence_mode and persistence_backend to avoid repeated API calls.

Adaptive RAG (Retrieval-Augmented Generation) is a technique implemented using the Pathway Live Data Framework to optimize token usage without sacrificing accuracy.

This implementation extends the BaseRAGQuestionAnswerer class. The core logic resides in the answer function, which processes requests from the /v2/answer endpoint. It utilizes the answer_with_geometric_rag_strategy_from_index function from pathway.xpacks.llm.question_answering. This function iteratively queries the LLM with an increasing number of context documents retrieved from an index until an optimal answer is found.

The template is organized into the following files and directories:

• app.py: The core Python application logic using the Pathway Live Data Framework.
• app.yaml: Configuration file for the pipeline (defines LLM models, data sources, and server settings).
• requirements.txt: Python dependencies for the pipeline.
• Dockerfile: Configuration for containerizing the application.
• .env: Environment variables (e.g., OPENAI_API_KEY).
• data/: Directory containing sample files for testing.
• ui/: A Streamlit-based user interface for interacting with the RAG app.

This pipeline uses the Pathway Live Data Framework to index video content by combining two TwelveLabs models:

1. Pegasus (Video Understanding): Used via the TwelveLabsVideoParser. It converts video bytes into rich text descriptions (covering actions, people, settings, and on-screen text). This text is then indexed by Pathway.
2. Marengo (Multimodal Embedding): Used via the MarengoEmbedder. It produces 512-dimensional vectors in a shared space for text, image, audio, and video, serving as the retriever embedder.

Data Flow Architecture: video files $\rightarrow$ pw.io.fs.read $\rightarrow$ TwelveLabsVideoParser (Pegasus) $\rightarrow$ TokenCountSplitter $\rightarrow$ DocumentStore + UsearchKnnFactory (with MarengoEmbedder) $\rightarrow$ BaseRAGQuestionAnswerer (OpenAI LLM) $\rightarrow$ REST API.

The RAG pipeline follows a streaming architecture consisting of five main stages:

1. Data Ingestion: Sources (e.g., local data/ folder, Google Drive, SharePoint) are defined in app.yaml. The system polls these sources and automatically detects changes.
2. Parsing & Splitting: Documents are parsed using DoclingParser and chunked into smaller segments using TokenCountSplitter.
3. Embedding: Chunks are converted into vector embeddings using an embedder (e.g., OpenAIEmbedder).
4. Indexing: Embeddings are stored in a vector index via USearchKnnFactory. This process is incremental and streaming.
5. Serving: A SummaryQuestionAnswerer handles the logic, which is then exposed via a QASummaryRestServer for external API access.