Pathway LLM App Templates Administrator & Developer Guide

pathwaycom/llm-app is a curated catalogue of ten reference applications — eight deployable templates plus two Jupyter cookbooks — each built on the Pathway streaming-dataflow engine and its pathway.xpacks.llm toolkit. This guide documents the architecture, deployment model, configuration surface, security posture, and developer internals for the catalogue. For end-user task walkthroughs, see the Pathway LLM App Templates User Guide.

The repository is small (126 files, ~62 KB of analyzed source). Most of the substance lives upstream in the Pathway library; the templates are configuration recipes and Pydantic glue that wire the library's RAG, indexing, and MCP primitives into deployable services. Read this guide together with the Pathway core docs at https://pathway.com/developers/ when you need to understand a class or method that the templates merely reference.

1. Architecture Overview

What the catalogue actually is

Each template under templates/ is a self-contained Pathway program with the following on-disk shape:

templates/<template_name>/
  app.py              ← the Pydantic App shell + entrypoint
  app.yaml            ← declarative pipeline config (App-class templates only)
  Dockerfile          ← image build: pathwaycom/pathway base + requirements
  docker-compose.yml  ← multi-container stack (app, ui, optional services)
  requirements.txt    ← typically just python-dotenv (pathway is in the base image)
  .env.example        ← env-var template (some templates only)
  README.md           ← per-template operational doc
  data/               ← seed documents (or files-for-indexing/)
  ui/                 ← Streamlit UI (some templates only)

Modern templates use a Pydantic App(BaseModel) class whose fields are parsed from app.yaml via pw.load_yaml. Legacy templates (drive_alert, unstructured_to_sql_on_the_fly) instead expose a procedural run(*, ...) function with os.environ.get defaults.

High-level system diagram

                         ┌────────────────────────────────────────────────┐
                         │      pathwaycom/llm-app TEMPLATE CATALOGUE     │
                         │  (this repository — Pydantic shells + YAML)    │
                         └──────────────────────┬─────────────────────────┘
                                                │
                                  ┌─────────────┴───────────────┐
                                  ▼                             ▼
        ┌──────────────────────────────────┐    ┌─────────────────────────────┐
        │ App(BaseModel) Pydantic shell    │    │ Legacy procedural run()     │
        │ extra="forbid"                   │    │ os.environ.get defaults     │
        │  ├── question_answerer | doc_    │    │  ├── drive_alert            │
        │  │   store | mcp_http (one of)   │    │  └── unstructured_to_sql_   │
        │  ├── host, port                  │    │      on_the_fly             │
        │  ├── persistence_mode/_backend   │    └─────────────────────────────┘
        │  ├── terminate_on_error          │                 │
        │  └── run() → pw.run(...)         │                 │
        └────────┬─────────────────────────┘                 │
                 │                                           │
                 └──────────────────┬────────────────────────┘
                                    ▼
        ┌────────────────────────────────────────────────────────────────┐
        │       pathway.xpacks.llm  (UPSTREAM LIBRARY — pathway[all])    │
        │                                                                │
        │  llms.OpenAIChat / LiteLLMChat                                 │
        │  embedders.OpenAIEmbedder / SentenceTransformerEmbedder        │
        │  parsers.DoclingParser / SlideParser / UnstructuredParser      │
        │  splitters.TokenCountSplitter                                  │
        │  document_store.DocumentStore / SlidesDocumentStore            │
        │  question_answering.{Base,Adaptive}RAGQuestionAnswerer         │
        │  question_answering.SummaryQuestionAnswerer                    │
        │  servers.{QASummary,DocumentStore}RestServer                   │
        │  mcp_server.PathwayMcp                                         │
        └────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
        ┌────────────────────────────────────────────────────────────────┐
        │     pathway core (pathway[all]) — Python+Rust dataflow         │
        │                                                                │
        │  pw.io.{fs,gdrive,postgres,slack,...}                          │
        │  pw.xpacks.connectors.sharepoint                               │
        │  pw.persistence.{Backend,Config}, pw.PersistenceMode           │
        │  pw.indexing.{UsearchKnnFactory,USearchMetricKind}             │
        │  pw.run(persistence_config=…, terminate_on_error=…)            │
        └────────────────────────────────────────────────────────────────┘

Note

The templates own almost no logic. The Pydantic App is ~50 lines per template; everything substantive — the streaming dataflow, the connectors, the LLM orchestration, the persistence layer — lives in the Pathway library. Audit and hardening work on the templates is therefore mostly about (a) configuration choices, (b) container/network posture, and (c) the legacy procedural code in drive_alert and unstructured_to_sql_on_the_fly.

Component inventory

Technology stack

Data flow — canonical RAG (question_answering_rag)

[ source files (filesystem / GDrive / SharePoint) ]
                        │
                        ▼ (Pathway streaming connector watches changes)
              pw.io.fs.read(path=data, format=binary, with_metadata=true)
                        │
                        ▼  (table of binary files)
                DoclingParser  (LLM-aware table parsing)
                        │
                        ▼  (table of text strings)
              TokenCountSplitter(max_tokens=400)
                        │
                        ▼  (table of chunks)
              OpenAIEmbedder(text-embedding-3-small)
                        │
                        ▼  (table of (chunk, vector))
              UsearchKnnFactory(metric=COS) → DocumentStore
                        │
                        ▼  (live, always-current vector index)
   ┌────────────────────┴─────────────────────────┐
   │                                              │
   ▼                                              ▼
BaseRAGQuestionAnswerer.answer()           DocumentStore.list_documents()
(top-k retrieval → prompt build →           (returns metadata for UI)
OpenAIChat → response)
   │
   ▼
QASummaryRestServer (HTTP, port 8000)
   │
   ▼
[ /v2/answer | /v1/retrieve | /v1/statistics | /v2/list_documents | /v2/summarize ]

Data flow — MCP server (document_store_mcp_server)

[ files-for-indexing/ ] ──► pw.io.fs.read ──► DoclingParser ──► TokenCountSplitter
        ──► SentenceTransformerEmbedder(mxbai-embed-large-v1) ──► UsearchKnn
        ──► DocumentStore ──► PathwayMcp(streamable-http, port 8068)
                              │
                              ▼
              [ MCP-aware client: Claude Desktop, agent, etc. ]

Data flow — drive_alert (legacy)

[ Google Drive folder ]
        │ pw.io.gdrive.read(refresh_interval=30s)
        ▼
   UnstructuredParser ──► TokenCountSplitter(40-120 tokens)
        │
        ▼
   OpenAIEmbedder(text-embedding-ada-002) ──► KNNIndex
        │
        │     ┌──────────────── HTTP POST / on port 8080 (REST connector) ──┐
        │     ▼                                                             │
        │  QueryInputSchema(query, user)                                    │
        │     │                                                             │
        │     ▼  prompt: "Is this an alerting query?" → OpenAIChat          │
        │     ▼  yields (alert_enabled: bool, query: str)                   │
        ▼     ▼
     KNNIndex.get_nearest_items(k=3)
        │
        ▼  build_prompt(documents_list, query) → OpenAIChat → response
        │
        ▼  HTTP response_writer(output)
        │
        ▼  filter(alert_enabled) → pw.stateful.deduplicate(acceptor=LLM)
        │
        ▼  pw.io.slack.send_alerts(message, channel_id, token)

Data flow — unstructured_to_sql_on_the_fly (legacy)

[ ./data/quarterly_earnings/*.pdf ]
        │ pw.io.fs.read(format=binary)
        ▼
   UnstructuredParser ──► strip_metadata
        │
        ▼  build_prompt_structure (insert FinancialStatementSchema typehints)
        ▼  OpenAIChat → JSON response
        ▼  parse_str_to_list → unpack_col → typed columns
        ▼  pw.io.postgres.write(structured_table, settings, table=quarterly_earnings)
        ▼  pw.io.csv.write(structured_table, "./data/quarterly_earnings.csv")

──── alongside ────

[ HTTP POST / on port 8080 ] ──► NLQuerySchema(query, user)
        │
        ▼  build_prompt_query(table, query) → OpenAIChat → SQL string
        ▼  psycopg cursor.execute(sql_query) → fetchall()
        ▼  pw.make_tuple(sql_query, result) → response_writer

Note

Both legacy templates use pw.io.http.rest_connector (port 8080), not the modern QASummaryRestServer. They predate the App-class refactor and therefore expose a different REST shape (single POST / rather than the /v2/answer-family). Keep this in mind when fronting them with a reverse proxy.

2. System Requirements

Runtime requirements

Native (non-Docker) runtime

External services

Pathway license

Templates ship with pw.set_license_key("demo-license-key-with-telemetry"). Operational implications:

Important

The templates do not read PATHWAY_LICENSE_KEY from the environment by default — the license string is hard-coded in app.py. To use a personal Scale key, edit app.py to call pw.set_license_key(os.environ["PATHWAY_LICENSE_KEY"]) and put the key in .env. This is a small but important hardening step for production deployments.

3. Configuration Reference

Configuration files (per template)

Note

The repo root has pyproject.toml and setup.cfg, but they declare package-mode = false and only configure linters. There is no PyPI-installable llm-app package. Each template is its own freestanding application; cloning the repo is the supported install method.

App Pydantic model — the shared shape

Every modern template's App(BaseModel) exposes a near-identical surface (the variable field is the domain object: question_answerer, document_store, or mcp_http):

model_config = ConfigDict(extra="forbid", arbitrary_types_allowed=True) is set on every App. Forbid is the load-bearing setting: it turns YAML typos into Pydantic validation errors at startup rather than silently ignored fields.

run() resolves persistence as follows:

1. If persistence_mode is None and the deprecated with_cache is True: warn, use UDF_CACHING.
2. If persistence_mode is None and with_cache is not True: persistence disabled.
3. Else: use the explicit persistence_mode.
4. If persistence is active and persistence_backend is None: default to Backend.filesystem("./Cache").
5. Build pw.persistence.Config and pass it to pw.run.
6. Always call pw.run(persistence_config=…, terminate_on_error=…, monitoring_level=pw.MonitoringLevel.NONE).

Slides AI Search App — extra fields

templates/slides_ai_search/app.py's App has additional fields tied to its slide-specific pipeline:

Note that sources, llm, retriever_factory are top-level (read by App) in this template, whereas in the canonical RAG they are $-prefixed YAML anchors consumed by $document_store and question_answerer.

YAML configuration grammar (per-template differences)

BaseRAGQuestionAnswerer knobs (recognised under question_answerer:)

AdaptiveRAGQuestionAnswerer knobs

Connector configuration

Every template's app.yaml includes commented examples for SharePoint and Google Drive sources. Live examples in shipped configs:

Filesystem (default, every template)

- !pw.io.fs.read
  path: data                # or files-for-indexing for indexing/MCP templates
  format: binary
  with_metadata: true

Google Drive (commented in every template)

- !pw.io.gdrive.read
  object_id: $DRIVE_ID
  service_user_credentials_file: gdrive_indexer.json
  file_name_pattern:
    - "*.pdf"
    - "*.pptx"
  object_size_limit: null
  with_metadata: true
  refresh_interval: 30

SharePoint (commented in every template)

- !pw.xpacks.connectors.sharepoint.read
  url: $SHAREPOINT_URL
  tenant: $SHAREPOINT_TENANT
  client_id: $SHAREPOINT_CLIENT_ID
  cert_path: sharepointcert.pem
  thumbprint: $SHAREPOINT_THUMBPRINT
  root_path: $SHAREPOINT_ROOT
  with_metadata: true
  refresh_interval: 30

Important

SharePoint uses app-only authentication via certificate + thumbprint, not username/password. The cert file (sharepointcert.pem) must be next to app.yaml (or wherever cert_path: points) and the matching thumbprint registered with the Azure AD app registration. The app registration must have application (not delegated) permissions for Sites.Read.All.

Environment variable reference

Important

Several variables (SHAREPOINT_*, DRIVE_ID) are referenced only inside app.yaml via $VAR interpolation, so they will not appear in any AST scan of os.environ.get — they are still required at runtime when the corresponding connector block is uncommented. Keep this matrix as the canonical inventory; the Python source alone is incomplete.

Annotated walkthrough — production-hardened app.yaml

# Persistence: keep the cache between restarts so we don't re-embed on deploy
persistence_mode: !pw.PersistenceMode.PERSISTING
persistence_backend: !pw.persistence.Backend.filesystem
  path: "/var/lib/pathway/cache"

# Halt on errors so a partial pipeline doesn't silently serve bad answers
terminate_on_error: true

# Listen only on the in-cluster network; reverse-proxy will terminate TLS
host: "0.0.0.0"
port: 8000

# Sources: SharePoint as primary, with file-system fallback (mounted volume)
$sources:
  - !pw.xpacks.connectors.sharepoint.read
    url: $SHAREPOINT_URL
    tenant: $SHAREPOINT_TENANT
    client_id: $SHAREPOINT_CLIENT_ID
    cert_path: /etc/secrets/sharepointcert.pem
    thumbprint: $SHAREPOINT_THUMBPRINT
    root_path: $SHAREPOINT_ROOT
    with_metadata: true
    refresh_interval: 60
  - !pw.io.fs.read
    path: /data/local
    format: binary
    with_metadata: true

# Production model + capacity tuned to OpenAI tier-3 limits
$llm: !pw.xpacks.llm.llms.OpenAIChat
  model: "gpt-4o"
  retry_strategy: !pw.udfs.ExponentialBackoffRetryStrategy
    max_retries: 8
  cache_strategy: !pw.udfs.DefaultCache {}
  temperature: 0
  capacity: 32
  async_mode: "fully_async"

$embedder: !pw.xpacks.llm.embedders.OpenAIEmbedder
  model: "text-embedding-3-small"
  cache_strategy: !pw.udfs.DefaultCache {}
  retry_strategy: !pw.udfs.ExponentialBackoffRetryStrategy {}

$splitter: !pw.xpacks.llm.splitters.TokenCountSplitter
  max_tokens: 400

$parser: !pw.xpacks.llm.parsers.DoclingParser
  table_parsing_strategy: "llm"
  async_mode: "fully_async"
  chunk: false
  cache_strategy: !pw.udfs.DefaultCache {}

$retriever_factory: !pw.indexing.UsearchKnnFactory
  reserved_space: 100000   # tune to expected doc count
  embedder: $embedder
  metric: !pw.indexing.USearchMetricKind.COS

$document_store: !pw.xpacks.llm.document_store.DocumentStore
  docs: $sources
  parser: $parser
  splitter: $splitter
  retriever_factory: $retriever_factory

question_answerer: !pw.xpacks.llm.question_answering.BaseRAGQuestionAnswerer
  llm: $llm
  indexer: $document_store
  search_topk: 8
  prompt_template: |
    You are a customer-support assistant for ACME Corp.
    Use ONLY the provided context. If the answer isn't in the context,
    say "I don't have that information."
    Context: {context}
    Question: {query}
    Answer:

4. Deployment & Scaling

Single-machine Docker Compose (the shipped path)

Each template ships a docker-compose.yml (most do — multimodal_rag, private_rag, adaptive_rag, slides_ai_search, question_answering_rag, document_indexing, document_store_mcp_server, unstructured_to_sql_on_the_fly, drive_alert all do). For the canonical RAG:

cd templates/question_answering_rag
docker compose up --build

Compose typically starts:

Slides adds an nginx service (8080); unstructured-to-SQL adds a postgres service.

Single Dockerfile (no UI)

cd templates/<template>
docker build -t myorg/llm-app-<template>:5fa97b7 .
docker run -d \
  -p 8000:8000 \
  -v "$PWD/data:/app/data" \
  -v "/var/lib/pathway/cache:/app/Cache" \
  --env-file .env \
  --restart unless-stopped \
  myorg/llm-app-<template>:5fa97b7

Kubernetes (manifest sketch — not bundled)

The repository does not ship Kubernetes manifests. A typical Deployment looks like:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: pathway-rag
spec:
  replicas: 1            # see "scaling" below before raising
  selector:
    matchLabels: {app: pathway-rag}
  template:
    metadata:
      labels: {app: pathway-rag}
    spec:
      containers:
        - name: app
          image: myorg/llm-app-question_answering_rag:5fa97b7
          ports: [{containerPort: 8000}]
          envFrom:
            - secretRef: {name: pathway-rag-secrets}
          volumeMounts:
            - {name: data, mountPath: /app/data}
            - {name: cache, mountPath: /app/Cache}
          resources:
            requests: {cpu: 1, memory: 4Gi}
            limits:   {cpu: 4, memory: 8Gi}
      volumes:
        - {name: data,  persistentVolumeClaim: {claimName: pathway-data}}
        - {name: cache, persistentVolumeClaim: {claimName: pathway-cache}}
---
apiVersion: v1
kind: Service
metadata: {name: pathway-rag}
spec:
  selector: {app: pathway-rag}
  ports: [{port: 80, targetPort: 8000}]

Multi-machine clustering (Pathway Scale)

Pathway supports multi-machine workers via the --addresses runtime flag. The catalogue's app.py files do not invoke --addresses themselves — you would either:

1. Modify app.py to call pw.run with the processes and nodes arguments (Pathway core API), or
2. Run multiple instances of the container with appropriate environment variables (Pathway's docs on multi-machine cover the exact incantation).

Both require PATHWAY_LICENSE_KEY (Scale tier) — without it pw.run raises a license error when clustering is requested.

Warning

Do not raise the Kubernetes replicas count above 1 unless you have wired up Pathway's multi-machine clustering. Two instances of the same App without coordination will both try to be authoritative for the same index and produce duplicated work, not horizontal scaling.

Persistence and state

pw.PersistenceMode controls what the cache holds:

pw.persistence.Backend.filesystem(path) is the only backend used in the shipped templates. Pathway Scale supports additional backends (S3, Azure Blob) — those require a Scale license and app.py changes.

Restart and reload behaviour

• Config change (app.yaml): edit, docker compose restart app. Pathway re-reads YAML at process start.
• Code change (app.py): rebuild image with docker compose up --build.
• New documents: Pathway picks them up live; no restart.
• Document deletion: Pathway notices the file disappear; the corresponding chunks leave the index.
• OPENAI_API_KEY rotation: restart; key is read once at process start.

5. Operations

Health checks

Every App-class template exposes the pw.io.http.PathwayWebserver-backed endpoints. A simple liveness probe:

curl -f -X POST http://localhost:8000/v1/statistics

Returns 200 with index stats when the pipeline is alive. For Kubernetes:

livenessProbe:
  httpGet: {path: /v1/statistics, port: 8000, scheme: HTTP}
  initialDelaySeconds: 60
  periodSeconds: 30
readinessProbe:
  httpGet: {path: /v1/statistics, port: 8000, scheme: HTTP}
  initialDelaySeconds: 10
  periodSeconds: 5

Note

Initial delay matters — first index build can take a minute on a large corpus. The pipeline will return 503 (or refuse connections) until then.

Logging

Every template configures Python's logging to INFO level with a fixed format:

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(name)s %(levelname)s %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

Pathway emits log lines like Indexed N documents, OpenAI request retried, and Document store updated. For production, redirect Docker logs to your log aggregator (the included compose files do not configure log drivers — add logging: blocks).

Streamlit UI logging

The UIs log structured events for analytics:

logger.info({"_type": "search_request_event", "query": question})
logger.info({"_type": "search_response_event", "query": question, "response": type(response)})
logger.info({"_type": "rate_event", "rating": event, "rank": idx, ...})

Slides UI also logs _type: create_adjacent_image_urls. These are JSON-shaped log lines — filter on the _type field.

Monitoring

Pathway emits internal monitoring at monitoring_level=pw.MonitoringLevel.NONE (the catalogue's default). Higher levels (IN_OUT, ALL) write to stdout; configure them by editing pw.run(monitoring_level=…) in app.py. For production, integrate with Prometheus by scraping the Pathway metrics endpoint (see Pathway core docs).

Scaling guidance

Backup and disaster recovery

Ops checklist for go-live

• [ ] Replace demo license key with personal key or remove entirely.
• [ ] Move OPENAI_API_KEY (and Slack/SharePoint secrets) to a secret manager (Kubernetes Secrets, AWS Secrets Manager).
• [ ] Set terminate_on_error: true so partial-pipeline answers don't reach customers.
• [ ] Mount /app/Cache to a persistent volume.
• [ ] Configure persistence_mode: !pw.PersistenceMode.PERSISTING if you need restart-resumption.
• [ ] Place the API behind a reverse proxy with TLS (Caddy, nginx, Traefik).
• [ ] Add an authentication layer in front (the templates do not authenticate REST calls).
• [ ] Set log shipping (the templates write to stdout — wire to your log aggregator).
• [ ] Add liveness/readiness probes.
• [ ] Pin OpenAI model versions explicitly (avoid floating gpt-4o-class names that update silently).
• [ ] Test failure modes: kill OpenAI access, kill Postgres (legacy), kill Drive access, confirm the pipeline degrades sensibly.

Tip

Keep one Streamlit UI deployed even if your end users hit the API directly — it's the fastest way to debug the index and confirm a deployment is healthy without writing curl commands.

6. Performance

Where the time and money go

For a typical RAG query on the canonical template:

Indexing is the heavy phase: parsing PDFs (especially with DoclingParser table_parsing_strategy: "llm"), chunking, and embedding. A 100-page PDF with table-LLM parsing can take 30-60 s and ~10 cents in OpenAI calls. Subsequent runs hit DefaultCache and are nearly free.

Tuning levers

Benchmarks (rough orders of magnitude)

These are not from a benchmark suite — they are typical orders of magnitude for the shipped configurations on commodity hardware.

7. Security

Threat model

Warning

The shipped templates have no authentication on the REST or MCP endpoints. They are designed for evaluation and internal deployments behind a reverse proxy. Putting a template directly on the public internet without a fronting authn/authz layer is unsafe — anyone can query your index, read list_documents, and incur LLM cost.

Secret handling

The templates rely on python-dotenv's load_dotenv() plus os.environ.get (legacy) or YAML $VAR interpolation (modern):

For production, replace .env with environment variables sourced from your secret manager. Do not bake secrets into images.

Auditing

Pathway logs every retry and connector poll. The Streamlit UIs log structured search_request_event / search_response_event lines that can feed audit trails. For full audit coverage of "who asked what":

1. Front the API with a proxy that logs (Caddy, Traefik, ALB).
2. Pass user identity through to the request body or a header that ends up in the prompt.
3. Capture the JSON-shaped UI log lines.

Caution

Pathway's pw.io.slack.send_alerts posts to Slack with the bot token in plaintext at request time. Anyone with read access to the running container's environment or to outgoing-traffic mirrors can recover the token. Treat the container as a security boundary equivalent to the Slack bot itself.

License compliance

The catalogue is MIT and you can use, modify, and redistribute the templates freely. The Pathway engine you depend on is under BSL 1.1 — for commercial production use of advanced features you need a Pathway Scale license (which also unlocks the demo telemetry-free mode). Read the BSL terms carefully if you are forking Pathway itself.

8. Developer Internals

Repository structure

pathwaycom/llm-app/
├── README.md                          ← marketing landing + selector table
├── CONTRIBUTING.md
├── CODE_OF_CONDUCT.md
├── LICENSE                            ← MIT
├── pyproject.toml                     ← linter config; package-mode = false
├── setup.cfg                          ← flake8/black config
├── .gitignore
├── assets/                            ← logos, marketing images
├── cookbooks/
│   └── self-rag-agents/
│       ├── pathway_langgraph_agentic_rag.ipynb
│       └── pathway_deploy_langgraph_agents.ipynb
└── templates/
    ├── adaptive_rag/
    │   ├── app.py                     ← App(BaseModel), question_answerer
    │   ├── app.yaml
    │   ├── Dockerfile
    │   ├── requirements.txt           ← python-dotenv~=1.0
    │   ├── .env.example
    │   ├── README.md
    │   └── data/                      ← seed PDF
    ├── document_indexing/              ← App(BaseModel), document_store
    │   └── … (similar shape)
    ├── document_store_mcp_server/      ← App(BaseModel), mcp_http (PathwayMcp)
    │   └── …
    ├── drive_alert/                    ← LEGACY — procedural run(*, ...)
    │   ├── app.py
    │   ├── __init__.py                 ← re-exports run
    │   ├── ui/server.py
    │   └── …
    ├── multimodal_rag/                 ← App(BaseModel), question_answerer + $parsing_llm
    │   └── …
    ├── private_rag/                    ← App(BaseModel), AdaptiveRAG + LiteLLM
    │   └── …
    ├── question_answering_rag/         ← canonical App(BaseModel), question_answerer
    │   ├── app.py
    │   ├── app.yaml
    │   ├── docker-compose.yml
    │   ├── ui/                         ← Streamlit RAGClient UI
    │   │   ├── ui.py
    │   │   ├── Dockerfile
    │   │   ├── requirements.txt        ← streamlit==1.37.0, load_dotenv
    │   │   └── static/pathway-logo-black.png
    │   └── data/
    ├── slides_ai_search/               ← App(BaseModel) + SlidesDocumentStore + nginx
    │   ├── app.py
    │   ├── app.yaml
    │   ├── nginx/                      ← file-server stack
    │   │   ├── Dockerfile
    │   │   └── nginx.conf
    │   ├── pathway_slides_ai_search/   ← per-template helper module
    │   │   └── __init__.py
    │   └── ui/                         ← Streamlit UI (categories, languages, slide cards)
    │       ├── ui.py
    │       └── requirements.txt        ← includes openai==1.35.10, beautifulsoup4
    └── unstructured_to_sql_on_the_fly/ ← LEGACY — procedural run(*, ...)
        ├── app.py
        ├── postgres/init-db.sql
        ├── ui/server.py
        └── data/quarterly_earnings/    ← seed PDFs

The shared App-class lifecycle (deep dive)

Every modern template's app.py follows this exact recipe:

import logging
from warnings import warn
import pathway as pw
from dotenv import load_dotenv
# template-specific xpack imports (one of):
from pathway.xpacks.llm.question_answering import SummaryQuestionAnswerer
from pathway.xpacks.llm.servers import QASummaryRestServer
# or DocumentStore + DocumentStoreServer
# or PathwayMcp
from pydantic import BaseModel, ConfigDict, InstanceOf

pw.set_license_key("demo-license-key-with-telemetry")

logging.basicConfig(level=logging.INFO, format="…", datefmt="…")
load_dotenv()


class App(BaseModel):
    # exactly one domain object: question_answerer / document_store / mcp_http
    question_answerer: InstanceOf[SummaryQuestionAnswerer]
    host: str = "0.0.0.0"
    port: int = 8000
    with_cache: bool | None = None  # deprecated
    persistence_backend: pw.persistence.Backend | None = None
    persistence_mode: pw.PersistenceMode | None = pw.PersistenceMode.UDF_CACHING
    terminate_on_error: bool = False

    def run(self) -> None:
        server = QASummaryRestServer(self.host, self.port, self.question_answerer)
        # … persistence resolution …
        pw.run(persistence_config=…, terminate_on_error=…,
               monitoring_level=pw.MonitoringLevel.NONE)

    model_config = ConfigDict(extra="forbid", arbitrary_types_allowed=True)


if __name__ == "__main__":
    with open("app.yaml") as f:
        config = pw.load_yaml(f)
    app = App(**config)
    app.run()

The five blocks that vary across modern templates:

1. The xpack import — which class is used as the domain object.
2. The single Pydantic field — question_answerer, document_store, or mcp_http.
3. The server instantiation in run() — which xpack server (or none, for MCP) is built.
4. Whether with_cache exists — the MCP template omits the deprecated field.
5. base_dir resolution — slides_ai_search reads the YAML relative to __file__ because its compose mounts a non-default working directory.

Legacy template internals

drive_alert/app.py

A 200-line procedural run(*, ...) that builds a four-stage Pathway dataflow:

1. Index build: pw.io.gdrive.read → UnstructuredParser → TokenCountSplitter(min_tokens=40, max_tokens=120) → OpenAIEmbedder → KNNIndex.
2. Query intake + intent detection: pw.io.http.rest_connector (port 8080, schema QueryInputSchema(query, user)), build_prompt_check_for_alert_request_and_extract_query UDF, OpenAIChat, split_answer UDF that yields (alert_enabled: bool, cleaned_query: str).
3. Query answer: for each query, fetch top-3 nearest chunks, build prompt, call OpenAIChat, send back via response_writer.
4. Alert loop: filter alert_enabled=True queries, run pw.stateful.deduplicate with an LLM-driven acceptor(new, old) -> bool that asks the LLM whether two answers materially differ. When the dedupe accepts a new answer, pw.io.slack.send_alerts posts a notification.

Key UDFs:

Schemas:

class DocumentInputSchema(pw.Schema):
    doc: str

class QueryInputSchema(pw.Schema):
    query: str
    user: str

Defaults that matter:

unstructured_to_sql_on_the_fly/app.py

A 240-line procedural run(*, ...) that builds two pipelines in one process:

Pipeline 1 — extraction:

files = pw.io.fs.read(data_dir, format="binary")
parser = UnstructuredParser()
unstructured_documents = files.select(texts=parser(pw.this.data)).select(
    texts=strip_metadata(pw.this.texts)
)
structured_table = structure_on_the_fly(
    unstructured_documents, api_key, model_locator, max_tokens, temperature
)
pw.io.postgres.write(structured_table, postgreSQL_settings, postresql_table)
pw.io.csv.write(structured_table, "./data/quarterly_earnings.csv")

structure_on_the_fly builds a token-aware prompt, calls OpenAIChat, parses JSON, unpacks columns, types them. Schema:

class FinancialStatementSchema(pw.Schema):
    company_symbol: str
    year: int
    quarter: str
    revenue_md: float
    eps: float
    net_income_md: float

Pipeline 2 — NL→SQL query:

query, response_writer = pw.io.http.rest_connector(
    host=host, port=port, schema=NLQuerySchema,
    autocommit_duration_ms=50, delete_completed_queries=True,
)
query += query.select(prompt=build_prompt_query(postresql_table, pw.this.query))
query += query.select(sql_query=model(prompt_chat_single_qa(pw.this.prompt)))

@pw.udf
def execute_sql_query(sql_query):
    cursor.execute(sql_query); answer = cursor.fetchall(); conn.commit(); return answer

query = query.select(query, sql_query, result=execute_sql_query(pw.this.sql_query))
answers = query.select(result=pw.make_tuple(pw.this.sql_query, pw.this.result))
response_writer(answers)

The prompt grammar pinned by build_prompt_query:

• SELECT clause specifies columns or aggregators (SUM, COUNT, MIN, MAX, AVG).
• WHERE uses standard binary operators with AND/OR.
• GROUP BY includes any WHERE columns to avoid psycopg2.errors.GroupingError.
• If no aggregator, always include company_symbol, year, quarter.

Caution

The prompt is the only SQL guard. An LLM can be tricked into producing malicious SQL with sufficient prompt injection. Run against a read-only Postgres role; never share a connection with privileged tables.

Streamlit UIs

Every UI is a single ui.py (or server.py) that:

1. Loads .env, reads PATHWAY_HOST and PATHWAY_PORT.
2. Constructs RAGClient(url=f"http://{PATHWAY_HOST}:{PATHWAY_PORT}").
3. Renders a Streamlit page with a query input, a sidebar listing indexed docs (from conn.list_documents(keys=[])), and a result region.
4. On submit, calls conn.answer(question, ...) (or for slides, a filtered variant) and renders.

question_answering_rag/ui/ui.py (5408 chars) is the simplest. slides_ai_search/ui/ui.py (11886 chars) adds:

• Image rendering via the nginx file server (FILE_SERVER_URL).
• Filename → slide-ID parsing (parse_slide_id_components, create_slide_url, get_adjacent_image_urls).
• Category and language multi-selects → JMESPath filter strings (combine_filters).
• BeautifulSoup scrape of the nginx index page to surface "Indexing in Progress" files.

drive_alert/ui/server.py and unstructured_to_sql_on_the_fly/ui/server.py are smaller (2047 and 2502 chars) — they use the legacy requests.post against port 8080 rather than RAGClient.

Module reference

Code patterns and conventions

Pydantic-with-extra-forbid is the hard contract. Every modern template sets extra="forbid". This shifts every config error from runtime mystery to startup-time validation error with a clear message. Keep this when forking.

YAML anchors over inheritance. $llm, $embedder, $splitter, $parser, $retriever_factory are conventionally $-prefixed; question_answerer, host, port, persistence_*, terminate_on_error are top-level. The $-prefix doesn't change Pathway behaviour — it's a project convention indicating "this is referenced elsewhere in the YAML, not consumed by App". Keep the convention.

pw.set_license_key at module-import time. Every app.py calls pw.set_license_key("demo-…") at the top, before App is even defined. Do not move this — Pathway needs the license set before the dataflow is described.

load_dotenv() once, early. Same pattern in every template.

monitoring_level=pw.MonitoringLevel.NONE. The default for the catalogue. Raise to IN_OUT or ALL for production.

Procedural legacy is not deprecated as a Pathway pattern. Pathway still supports building dataflows procedurally (in fact, app.py is a procedural Pathway program — it just happens to be pre-wired by Pydantic). The legacy templates are "legacy" only in the sense that they predate the App-class catalogue convention; the underlying API is current.

Testing strategy

The repository ships no test suite — there are no tests/ or pytest.ini files. The catalogue's correctness story is:

1. Each template has a working seed dataset (PDFs in data/ or files-for-indexing/).
2. docker compose up --build is the smoke test: it must produce indexed documents and answer a curl query.
3. Upstream Pathway has its own test suite that covers the xpacks the templates use.

For your fork, recommended tests:

• Build smoke test — docker compose up --build reaches INFO Indexed N documents.
• Endpoint smoke test — curl /v2/answer returns a non-empty JSON response.
• YAML schema test — python -c "import yaml; from app import App; cfg = yaml.safe_load(open('app.yaml')); App(**cfg)" passes Pydantic validation.
• Connector test — point at a tiny S3 bucket / SharePoint folder / Drive folder and confirm indexing.

Potential improvements (tech debt)

• Legacy templates lack the App-class shell. drive_alert and unstructured_to_sql_on_the_fly could be refactored to expose configuration via app.yaml like the modern templates. They predate the convention; refactoring is straightforward but breaks API surface.
• No central README with the selector table. The root README is marketing; users discover templates by ls'ing templates/. Adding a selector table to the README would close GitHub issue #124. (This guide's section 4.1 is the de-facto fix.)
• PATHWAY_LICENSE_KEY should be read from env by default. Hard-coding "demo-license-key-with-telemetry" in every app.py is a footgun for production deployments.
• Streaming response support (issue #46) — OpenAIChat doesn't expose stream=True through Pathway xpacks. Would require upstream library changes.
• Templates do not expose Pathway monitoring metrics. A metrics endpoint and a tiny Prometheus scrape would help operators.
• No authentication on REST/MCP endpoints. A simple API-key middleware is a 50-line fix, currently absent.
• No rate-limiting on REST/MCP endpoints. Any client can hammer the LLM and run up the OpenAI bill.
• Per-template pyproject.toml would help dependency pinning. Current requirements.txt files are sparse — they rely on the Pathway base image for everything substantive.
• Tests — see above.

Contributing

The repository is open-source under MIT (see LICENSE). CONTRIBUTING.md exists at the root with the contribution workflow. Pull requests typically:

• Add a new template (mirror the App-class shape — Pydantic shell + app.yaml + Dockerfile + compose + README).
• Update an existing template to a newer Pathway xpack version.
• Refactor a legacy template to App-class.
• Fix a connector configuration.

For substantial work (e.g., wiring multi-machine clustering into the templates), open an issue first.

9. Integration Points

Inbound integrations (data sources)

Outbound integrations (sinks and services)

Client-side integration

The Streamlit UIs use pathway.xpacks.llm.question_answering.RAGClient — a Python client pointed at the Pathway REST API. You can use the same client from any Python application:

from pathway.xpacks.llm.question_answering import RAGClient

conn = RAGClient(url="http://your-pathway-host:8000")
response = conn.answer("Your question here", return_context_docs=True)
print(response["response"])
print(response["context_docs"])

# Other methods used by the shipped UIs
docs = conn.list_documents(keys=[])  # for a sidebar/admin view

For non-Python clients, hit the REST endpoints directly (see section 4.3 of the User Guide).

MCP integration

document_store_mcp_server exposes the live Pathway document store to MCP-aware clients via streamable HTTP. Configuration in app.yaml:

mcp_http: !pw.xpacks.llm.mcp_server.PathwayMcp
  name: "Streamable MCP Server"
  transport: "streamable-http"
  host: "0.0.0.0"
  port: 8068
  serve:
    - $document_store

To consume from Claude Desktop or another MCP client, register http://your-host:8068 as an MCP server. The server exposes the document store as MCP tools (retrieve, list, etc.) — your client sees them as native tool calls.

LangGraph integration (cookbook only)

cookbooks/self-rag-agents/ demonstrates wiring Pathway's live DocumentStore into a LangGraph state machine. The notebooks pin LangGraph and pathway[all] but the cookbooks are not deployable — they are tutorials. To productionise, port the LangGraph state machine into a templates/<your_name>/app.py that spins up its own Pathway document store, exposes a REST endpoint, and runs the graph as a UDF over the query stream.

Tip

The cookbook pattern is the bridge for adding agentic behaviour to a Pathway-powered RAG: keep the live document store on Pathway, run the agent reasoning in LangGraph, expose the combined service as REST or MCP. The catalogue does not yet ship this combined shape as a deployable template — building it is a natural extension if you need multi-step agentic RAG.