Pathway: A Python Framework for Real-Time Data and AI

pathwaycom/pathway

As a software engineer, you'll find Pathway invaluable because it simplifies a lot of the complexities of stream processing. Instead of dealing with low-level details of distributed systems or complex data synchronization, you can focus on the business logic of your application.

Here's why it's a game-changer

Python Native
If you already know Python, you're good to go! Pathway lets you build powerful streaming applications using the familiar syntax and libraries you love.

Automatic Incremental Computation
This is the magic behind Pathway. When new data arrives, it automatically updates the results of your computations efficiently, only processing the new data instead of re-calculating everything from scratch. This makes your applications fast and resource-efficient.

Unified API for Batch and Streaming
You write your code once, and it works for both batch and streaming data. No need to rewrite your logic for different data sources. This simplifies development and maintenance.

LLM and RAG Integration
Pathway provides built-in tools for working with LLMs. You can easily build pipelines that ingest streaming data, process it, and then use it to augment LLM queries in real-time. This is perfect for building intelligent chatbots, recommendation systems, or real-time summarization tools.

Getting started with Pathway is straightforward. You'll need Python installed on your system.

Installation

You can install Pathway using pip

pip install pathway

Writing Your First Pipeline

A Pathway application is a Python script. You import the necessary modules, define your data sources, and then specify the computations you want to perform.

Here's a simple example of a pipeline that reads data from a streaming source, filters it, and then writes the results to an output file.

Imagine you have a stream of log events, and you want to filter out all events that are not of a "CRITICAL" level.

data_stream.csv (simulated streaming data)

Let's assume this file is constantly being appended with new log entries.

timestamp,level,message
2025-08-15T09:00:00Z,INFO,User logged in
2025-08-15T09:01:05Z,CRITICAL,Database connection failed
2025-08-15T09:02:10Z,WARNING,Low disk space

etl_pipeline.py

import pathway as pw

# 1. Define the input data source
# We're simulating a streaming CSV file that new lines are appended to.
log_data = pw.io.csv.read(
    "data_stream.csv",
    mode="stream",
    schema=pw.Schema(
        timestamp=str,
        level=str,
        message=str
    )
)

# 2. Transform the data
# This is where we apply our business logic.
# Here, we filter for only the 'CRITICAL' log entries.
critical_logs = log_data.filter(log_data.level == "CRITICAL")

# 3. Define the output sink
# We'll write the results to a new CSV file.
pw.io.csv.write(critical_logs, "critical_logs_output.csv")

# 4. Run the pipeline
# This starts the computation and keeps it running
# to process new data as it arrives.
pw.run()

When you run this script with python etl_pipeline.py, Pathway will

Continuously monitor data_stream.csv for new lines.

Whenever a new line is added, it will efficiently check if level is "CRITICAL".

If it is, it will append that line to critical_logs_output.csv.

The magic is that this entire process is managed for you in real-time without you needing to manually manage loops, threads, or complex state management. It's a clean, declarative way to build robust data pipelines.

Pathway really shines when you want to build more complex applications, like a real-time RAG system.

Here's a conceptual example of a pipeline that

Ingests a stream of documents (e.g., from a news feed).

Indexes these documents in a vector store.

Takes a user query as input.

Retrieves relevant documents from the vector store in real-time.

Uses the retrieved context to augment a prompt to an LLM.

import pathway as pw
from pathway.xpacks.llm import prompt_llm, ragger
from pathway.xpacks.vector_store import VectorStoreServer

# 1. Ingest a stream of documents (e.g., news articles)
documents = pw.io.csv.read("news_articles.csv", mode="stream")

# 2. Index the documents in a vector store
# This happens automatically and in real-time as new articles arrive.
document_store = VectorStoreServer(
    documents=documents.select(pw.this.text)
)

# 3. Ingest a stream of user queries
queries = pw.io.csv.read("user_queries.csv", mode="stream")

# 4. Use the RAG module to combine queries with retrieved context
# Pathway automatically joins the queries with the relevant documents.
prompted_queries = ragger(
    rag_context=document_store,
    query=queries.query,
    prompt_template="Answer the following question based on the provided context: {query}\n\nContext: {context}"
)

# 5. Send the augmented prompt to an LLM and get the response
llm_responses = prompt_llm(prompted_queries.prompt, model="gpt-4")

# 6. Write the final responses to an output file
pw.io.csv.write(llm_responses, "llm_answers.csv")

# 7. Run the pipeline!
pw.run()