Building with Open AI

import logging
import sys
import click
import pathway as pw
import yaml
from dotenv import load_dotenv
from pathway.udfs import DiskCache, ExponentialBackoffRetryStrategy
from pathway.xpacks.llm import embedders, llms, parsers, splitters
from pathway.xpacks.llm.question_answering import BaseRAGQuestionAnswerer
from pathway.xpacks.llm.vector_store import VectorStoreServer

# Set your Pathway license key here to use advanced features.
# You can obtain a free license key from: https://pathway.com/get-license
# If you're using the Community version, you can comment this out.
pw.set_license_key("demo-license-key-with-telemetry")

# Set up basic logging to capture key events and errors.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(name)s %(levelname)s %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

# Load environment variables (e.g., API keys) from the .env file.
load_dotenv()

# Function to handle data sources. The example uses a local folder first,
# but you can easily connect to Google Drive or SharePoint as alternative data sources.
def data_sources(source_configs) -> list[pw.Table]:
    sources = []
    for source_config in source_configs:
        if source_config["kind"] == "local":
            # Reading data from a local directory (default is the 'data' folder).
            # This is the first option used in this example, but it's flexible for other sources.
            source = pw.io.fs.read(
                **source_config["config"],
                format="binary",
                with_metadata=True,
            )
            sources.append(source)

        elif source_config["kind"] == "gdrive":
            # Reading data from a Google Drive folder.
            # Requires OAuth credentials specified in the config.yaml file.
            source = pw.io.gdrive.read(
                **source_config["config"],
                with_metadata=True,
            )
            sources.append(source)

        elif source_config["kind"] == "sharepoint":
            try:
                # Import the SharePoint connector for reading data from SharePoint.
                import pathway.xpacks.connectors.sharepoint as io_sp
                
                # Reading data from a SharePoint folder.
                # Note: The SharePoint connector is part of Pathway's commercial offering.
                source = io_sp.read(**source_config["config"], with_metadata=True)
                sources.append(source)
            except ImportError:
                # If SharePoint is configured but the connector isn't available, exit the program.
                print(
                    "The Pathway Sharepoint connector is part of the commercial offering. "
                    "Please contact us for a commercial license."
                )
                sys.exit(1)

    return sources

# Command-line interface (CLI) function to run the app with a specified config file.
@click.command()
@click.option("--config_file", default="config.yaml", help="Config file to be used.")
def run(config_file: str = "config.yaml"):
    # Load the configuration from the YAML file.
    with open(config_file) as config_f:
        configuration = yaml.safe_load(config_f)

    # Fetch the GPT model from the YAML configuration.
    GPT_MODEL = configuration["llm_config"]["model"]

    # Initialize the OpenAI Embedder to handle embeddings with caching enabled.
    embedder = embedders.OpenAIEmbedder(
        model="text-embedding-ada-002",
        cache_strategy=DiskCache(),
    )

    # Set up OpenAI's GPT model for answering questions with retry and caching strategies.
    chat = llms.OpenAIChat(
        model=GPT_MODEL,
        retry_strategy=ExponentialBackoffRetryStrategy(max_retries=6),
        cache_strategy=DiskCache(),
        temperature=0.05,  # Low temperature for less random responses.
    )

    # Host and port configuration for running the server.
    host_config = configuration["host_config"]
    host, port = host_config["host"], host_config["port"]

    # Initialize the vector store for storing document embeddings in memory.
    # This vector store updates the index dynamically whenever the data source changes
    # and can scale to handle over a million documents.
    doc_store = VectorStoreServer(
        *data_sources(configuration["sources"]),
        embedder=embedder,
        splitter=splitters.TokenCountSplitter(max_tokens=400),  # Split documents by token count.
        parser=parsers.ParseUnstructured(),  # Parse unstructured data for better handling.
    )

    # Create a RAG (Retrieve and Generate) question-answering application.
    rag_app = BaseRAGQuestionAnswerer(llm=chat, indexer=doc_store)

    # Build the server to handle requests at the specified host and port.
    rag_app.build_server(host=host, port=port)

    # Run the server with caching enabled, and handle errors without shutting down.
    rag_app.run_server(with_cache=True, terminate_on_error=False)

# Entry point to execute the app if the script is run directly.
if __name__ == "__main__":
    run()