Fun AI Lab

Building autonomous AI agents sounds simple until your agent makes an expensive blunder because it acted on stale data. An AI agent deciding whether to block a fraudulent transaction, adjust dynamic pricing, or respond to an urgent customer message needs instant context. Giving an agent data that is five minutes old—or even five seconds old—is a recipe for failure.

To solve this, data teams traditionally build an unmaintainable multi-layer stack: Debezium for Change Data Capture (CDC), Kafka for message queueing, Apache Flink for continuous state processing, and PostgreSQL or Redis to serve the computed results.

RisingWave changes this dynamic completely. It eliminates the complex streaming Frankenstein by combining ingestion, stream processing, serving, and storage into a single SQL-native platform.

---

Why Agentic AI Demands Instant Context

Autonomous agents operate in continuous feedback loops. They inspect environment state, reason over choices, execute actions, and evaluate results. If the environment state lags behind reality, the agent’s reasoning breaks down.

+-----------------------------------------------------------------------+
|                       Traditional Distributed Stack                   |
|                                                                       |
|  [CDC Source] --> [Debezium] --> [Kafka] --> [Flink] --> [Postgres]  |
|                                                                 |     |
|                                                         (Latency Delay)
|                                                                 v     |
|                                                            [AI Agent] |
+-----------------------------------------------------------------------+
                                   vs.
+-----------------------------------------------------------------------+
|                          RisingWave Approach                          |
|                                                                       |
|  [CDC / Webhooks / Streams] -------> [ RisingWave ] -------> [AI Agent]|
|                                      (Sub-100ms)                      |
+-----------------------------------------------------------------------+

When you chain four or five separate distributed systems together:
1. Latency accumulates at every hop: Network serialization, message queueing, and stream computations introduce incremental delays.
2. State synchronization breaks: Cache invalidation, TTL management, and out-of-order event arrivals create race conditions.
3. Operational costs explode: Maintaining custom deployments for Kafka, Flink clusters, and dedicated cache layers demands heavy engineering bandwidth.

AI agents need clean, aggregated, fresh state available via simple SQL queries with millisecond-level response times. That is exact problem domain where RisingWave operates.

---

The Main Idea: Unified Stream Ingestion, Compute, and Serving

RisingWave takes a simple, powerful approach: treat streams as tables and continuous queries as materialized views.

Instead of executing a query from scratch every time a user or AI agent requests data, RisingWave uses incremental computation. When new records flow in from webhooks, database changes, or message brokers, RisingWave recomputes only the exact results affected by the incoming delta. End-to-end data freshness drops below 100 milliseconds, while direct point queries return in 10 to 20 milliseconds (P99 latency).

graph TD
    A[Webhooks & HTTP Events] -->|Direct Ingestion| RW[RisingWave Unified Engine]
    B[Database Logs / CDC] -->|Postgres / MySQL| RW
    C[Event Queues] -->|Kafka / Pulsar / Kinesis| RW
    D[Historical Storage] -->|S3 / Data Warehouses| RW

    subgraph RisingWave Core
        RW -->|Incremental Engine| MV[Materialized Views]
        MV -->|Row Store| SERVE[Low-Latency SQL Engine]
        MV -->|Auto-Compaction| ICEBERG[Apache Iceberg Catalog]
    end

    SERVE -->|10-20ms SQL Query| AGENT[Agentic AI Engines]
    SERVE -->|Live Stream| DASH[Real-Time Dashboards]
    ICEBERG -->|DataFusion Engine| ANALYTICS[Spark / Trino / DuckDB]

---

The Core Architectural Pillars

RisingWave simplifies data streaming by structuring its platform around four core duties: Ingest, Process, Serve, and Store.

1. Ingest Across Any Source

RisingWave removes the need for separate message bridge connectors. It natively ingests data directly across diverse channels:
* Webhooks: Direct HTTP ingestion from SaaS systems (GitHub, Stripe, custom API callbacks).
* Database CDC: Native change log ingestion from PostgreSQL, MySQL, and other databases.
* Event Streams: Direct connection to Kafka, Apache Pulsar, AWS Kinesis, and Redpanda.
* Historical Data: Batch ingestion directly from Amazon S3, data warehouses, or object stores.

All these sources are unified under standard PostgreSQL-compatible SQL syntax. You can join a live Kafka stream with a historical S3 table and a MySQL CDC topic in a single query.

2. Continuous Incremental Processing

At the core of RisingWave lies its continuous compute engine. When you write a CREATE MATERIALIZED VIEW statement, RisingWave maintains that view perpetually in memory and disk state.

Instead of scanning millions of rows on every trigger, the engine evaluates incoming stream deltas. If 10 new rows arrive in a stream of 100 million records, RisingWave updates only the state associated with those 10 records.

3. Native Low-Latency Serving

In traditional architectures, Flink streams data into an external database (like Redis, Cassandra, or PostgreSQL) so applications can query it.

RisingWave integrates an internal, high-performance row store directly into the platform. AI agents query this internal storage layer using standard SQL drivers over the PostgreSQL wire protocol. You query the stream processing state directly, without maintaining external caches, TTLs, or sync jobs.

4. Open Lakehouse Storage via Apache Iceberg

Real-time streaming state handles active operational needs, but historical context remains essential for machine learning model training and long-term analytics.

RisingWave writes cold and warm analytical data directly into Apache Iceberg™ format. It hosts an embedded Iceberg REST catalog and automates maintenance tasks like small-file compaction, snapshot pruning, and snapshot cleanup. Historical analysis runs seamlessly using vectorized query engines like Apache DataFusion, or external query engines like DuckDB, Spark, and Trino.

---

Stack Comparison: Legacy vs. RisingWave

To understand the architecture consolidation, let’s compare the operational requirements of a traditional streaming stack against RisingWave:

Architectural Layer	Traditional Distributed Stack	RisingWave Stream Platform
Change Ingestion (CDC)	Debezium + Connectors	Native SQL CDC Sources
Message Transport	Apache Kafka / Redpanda Cluster	Internal Streaming Pipelines
Stream Processing Engine	Apache Flink (Java/Scala)	Incremental SQL Compute Engine
Serving Database / Cache	Redis / PostgreSQL / Cassandra	Built-in Low-Latency Row Store
Long-Term Lakehouse Storage	Separate Spark / Iceberg maintenance scripts	Native Iceberg Sink + Built-in Catalog Maintenance
Query Interface	Java APIs, Flink SQL, Custom REST APIs	Standard PostgreSQL Wire Protocol

---

Hands-On Example: Real-Time AI Fraud Context Pipeline

Let me show you how straightforward it is to build a real-time event pipeline for an AI agent using RisingWave.

Imagine an autonomous AI agent responsible for identifying payment fraud. The agent requires an up-to-second summary of a user’s activity over the past 10 minutes: total spend, failed transaction count, and unique location count.

Step 1: Install and Launch RisingWave

You can spin up RisingWave locally in seconds using their single-line installer script or Docker.

# Quick local installation
curl -L https://risingwave.com/sh | sh

# Start RisingWave instance
./risingwave standalone

Alternatively, launch it with Docker:

docker run -d --name risingwave -p 4566:4566 -p 5691:5691 risingwavelabs/risingwave:latest standalone

Step 2: Connect to RisingWave

RisingWave uses the standard PostgreSQL wire protocol. Connect using psql or any standard Postgres library in Python, Node.js, or Go.

psql -h localhost -p 4566 -d dev -U root

Step 3: Define a Streaming Data Source

Define a live stream coming from a Kafka topic containing payment transactions:

CREATE SOURCE user_transactions (
    transaction_id VARCHAR,
    user_id VARCHAR,
    amount DECIMAL,
    location VARCHAR,
    status VARCHAR,
    transaction_time TIMESTAMP
) WITH (
    connector = 'kafka',
    topic = 'payment_events',
    properties.bootstrap.server = 'localhost:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

Step 4: Create a Real-Time Materialized View

Now create a materialized view that continuously computes user risk metrics. RisingWave incrementally updates this state as fast as transaction events hit Kafka.

CREATE MATERIALIZED VIEW user_risk_context AS
SELECT
    user_id,
    COUNT(transaction_id) AS tx_count_10m,
    SUM(CASE WHEN status = 'FAILED' THEN 1 ELSE 0 END) AS failed_tx_count_10m,
    SUM(amount) AS total_spent_10m,
    COUNT(DISTINCT location) AS distinct_locations_10m,
    MAX(transaction_time) AS last_seen
FROM
    user_transactions
WHERE
    transaction_time > NOW() - INTERVAL '10 minutes'
GROUP BY
    user_id;

Step 5: Query Fresh Context from Your AI Agent (Python)

Your AI agent can now execute a standard SQL query to grab instant user context in under 15 milliseconds:

import psycopg2

# Connect to RisingWave via standard Postgres driver
conn = psycopg2.connect("postgresql://root@localhost:4566/dev")
cursor = conn.cursor()

def get_agent_context(user_id: str):
    query = """
        SELECT tx_count_10m, failed_tx_count_10m, total_spent_10m, distinct_locations_10m
        FROM user_risk_context
        WHERE user_id = %s;
    """
    cursor.execute(query, (user_id,))
    result = cursor.fetchone()

    if result:
        return {
            "tx_count_10m": result[0],
            "failed_tx_count": result[1],
            "total_spent": float(result[2]),
            "distinct_locations": result[3]
        }
    return None

# Fetch real-time context for decision making
context = get_agent_context("usr_99218")
print(f"Agent Context Payload: {context}")

There are no cache updates to trigger, no Flink clusters to recompile, and no intermediate queues to sync. The query returns fresh stream computational data directly.

---

Real-World Use Cases for Real-Time Streaming

1. Autonomous Customer Service Agents

When an customer opens a chat session, an AI support agent needs immediate context: active cart items, recent failed checkout attempts, and current API status outages. RisingWave aggregates webhooks from Shopify, Stripe, and Zendesk into a single materialized view, providing the AI agent with complete context before it generates its first response word.

2. Algorithmic Trading & Dynamic Pricing

Financial agents and dynamic pricing engines must react instantly to market order books and inventory swings. Stale metric windows result in lost arbitrage opportunities or inventory mispricing. RisingWave executes complex temporal sliding windows and multi-stream joins with sub-100ms freshness.

3. Real-Time Vector Indexing & RAG Pipelines

Retrieval-Augmented Generation (RAG) applications depend on updated knowledge bases. When core system databases update, RisingWave tracks change streams and pushes modified records directly to vector database ingestion pipelines, keeping vector indices completely in sync with operational databases.

---

Common Myths and Operational Mistakes

Myth 1: “Streaming SQL is too slow for production low-latency serving.”

Fact: Traditional batch SQL engines like Hive or Spark SQL are slow because they scan cold files on disk. RisingWave is explicitly architected for continuous incremental stream evaluation. Query execution checks pre-computed memory and optimized row storage indexes, delivering 10–20ms query responses.

Myth 2: “You still need Redis in front of your stream processor.”

Fact: Engineers often put Redis after Flink because Flink isn’t designed as a direct query engine. RisingWave built an internal high-speed row store specifically to eliminate this extra layer. Agents query RisingWave directly.

Operational Pitfall to Avoid: Over-materializing Everything

Because creating materialized views in SQL is effortless, teams sometimes create dozens of complex continuous queries without indexing strategy planning. Always construct targeted materialized views tailored specifically to your active application or AI agent query patterns to maintain optimal memory efficiency.

---

Practical Action Plan to Get Started

If you want to simplify your real-time data infrastructure:

1. Audit your current latency gaps: Identify where your AI agents or downstream services suffer from stale context due to delayed batch ETLs or multi-hop messaging queues.
2. Start with a single stream source: Point RisingWave to an existing Kafka topic or database change log using standard SQL syntax.
3. Build targeted materialized views: Replace custom stream processing code or complex SQL batch scripts with continuous materialized views.
4. Query via standard Postgres tools: Connect your Python, TypeScript, or Go application directly to RisingWave using standard PostgreSQL libraries.
5. Dump cold historical state to Iceberg: Configure an Iceberg sink to store historical stream records safely on Amazon S3 without manual maintenance scripts.

---

Streamlining the Future of Agentic Infrastructure

As AI agents transition from simple single-prompt chatbots into real-time operational decision engines, streaming architecture becomes a primary system bottleneck. Systems reliant on batch pipelines or fragile, multi-component streaming stacks will struggle under operational complexity and latency lag.

RisingWave brings the clarity and simplicity of SQL to real-time event streaming. By unifying ingestion, incremental processing, low-latency querying, and storage into one engine, it frees developers to focus on building intelligent systems instead of maintaining server pipelines.

📂 Explore the open-source repository on GitHub: https://github.com/risingwavelabs/risingwave