Introduction
Your fraud detection agent flags a transaction as suspicious, but the customer already updated their shipping address ten minutes ago. Your support agent tells a customer that an item is in stock, but the last inventory sync ran at midnight. Your recommendation agent suggests a product the user just purchased, because the order event has not propagated yet.
These are not edge cases. They are the default behavior when AI agents operate on stale data. The observe-think-act loop that every agent follows is only as good as the observation step, and if that observation reflects yesterday's state of the world, the thinking and acting steps inherit that blindness.
The fix is not a faster batch job or a shorter cron interval. It is a fundamentally different data architecture: one that keeps agent context continuously fresh as events happen. This post explains why AI agents need real-time data, what a real-time data layer looks like in practice, and how to build one with RisingWave, a PostgreSQL-compatible streaming database, using nothing but SQL.
Why Do AI Agents Fail with Stale Data?
AI agents follow a loop: observe the current state, reason about it, and take an action. This observe-think-act cycle repeats continuously. The quality of every decision depends on the freshness and completeness of the observation.
Most agent architectures retrieve context at query time from a database, a vector store, or an API. The assumption is that those sources reflect reality. But in a typical enterprise, operational databases sync to analytics stores on hourly or daily schedules. Feature stores refresh in batch windows. Even CDC pipelines can lag by minutes when the system is under load.
The Freshness Mismatch Problem
Consider what happens when an agent enriches a live event with context that is hours old:
- Fraud detection agent: A customer changes their billing address, then makes a large purchase five minutes later. The agent checks the customer profile, but the address change has not propagated yet. The agent sees the old address, flags the transaction as anomalous, and blocks a legitimate purchase.
- Customer support agent: A user asks about order status. The agent queries an inventory table that refreshes every six hours. The item shipped two hours ago, but the agent reports it as "processing" because it cannot see the shipment event.
- Trading agent: A portfolio rebalancing agent checks positions to decide whether to sell. The positions table reflects the state from 30 minutes ago. The agent sells a position that was already sold by another system, creating an overdraft.
- Recommendation agent: A user just added three items to their cart. The recommendation model still suggests those same items because the cart events have not reached the feature store.
In each case, the agent's logic is correct. The data is wrong, or more precisely, the data is old. According to a 2025 Confluent report, 86% of IT leaders now identify investments in data streaming as a top strategic priority, driven largely by the demands of AI workloads that cannot tolerate stale context.
Why Batch Refreshes Cannot Keep Up
You might think the solution is to run batch refreshes more frequently. Move from daily to hourly. From hourly to every five minutes. But this approach hits a wall quickly:
- Cost scales linearly: Each refresh recomputes the entire result set, even if only a handful of rows changed.
- Latency has a floor: Even a five-minute refresh means your agent is working with data that could be up to five minutes old. For fraud detection or trading, that is an eternity.
- Cascading delays: If your agent context depends on joining data from three different sources, and each refreshes on a different schedule, the effective staleness is the worst of the three.
What agents actually need is a data layer that updates continuously, not periodically. That is what a streaming database provides.
What Does a Real-Time Data Layer for AI Agents Look Like?
A real-time data layer for AI agents has three properties: it ingests events as they happen, it transforms and joins those events into agent-ready context, and it serves that context with low latency when the agent queries it.
A streaming database like RisingWave delivers all three in a single system. The core abstraction is the materialized view: a SQL query whose results are incrementally maintained as new data arrives. Instead of running an expensive aggregation every time an agent needs context, the materialized view keeps the result pre-computed and fresh.
Materialized Views as Pre-Computed Agent Context
A materialized view is a SQL query result that the database keeps in sync with the underlying data automatically. When a new event arrives in the source stream, RisingWave propagates only the change (the delta) through the query plan and updates the result. The agent always reads a fresh, pre-computed answer.
This is fundamentally different from a traditional materialized view in PostgreSQL, which is a snapshot that requires manual REFRESH commands. RisingWave materialized views are continuously maintained with sub-second freshness.
Here is a concrete example. Suppose you are building a customer support agent that needs to answer questions about order status, recent activity, and account health. You can define the agent's context as a materialized view:
-- Tested against RisingWave v2.8.0
-- Source: order events from Kafka
CREATE SOURCE order_events (
order_id VARCHAR,
customer_id VARCHAR,
product_id VARCHAR,
quantity INT,
total_amount DECIMAL,
status VARCHAR,
event_time TIMESTAMP
) WITH (
connector = 'kafka',
topic = 'ecommerce.orders',
properties.bootstrap.server = 'kafka:9092'
) FORMAT PLAIN ENCODE JSON;
-- Source: customer profile updates via CDC
CREATE SOURCE customer_updates (
customer_id VARCHAR,
name VARCHAR,
email VARCHAR,
tier VARCHAR,
updated_at TIMESTAMP
) WITH (
connector = 'kafka',
topic = 'ecommerce.customers',
properties.bootstrap.server = 'kafka:9092'
) FORMAT PLAIN ENCODE JSON;
-- Materialized view: always-fresh context for the support agent
CREATE MATERIALIZED VIEW customer_support_context AS
SELECT
c.customer_id,
c.name,
c.email,
c.tier,
COUNT(o.order_id) AS total_orders_30d,
SUM(o.total_amount) AS total_spend_30d,
MAX(o.event_time) AS last_order_time,
ARRAY_AGG(DISTINCT o.status) AS recent_statuses
FROM customer_updates c
LEFT JOIN order_events o
ON c.customer_id = o.customer_id
AND o.event_time > NOW() - INTERVAL '30 days'
GROUP BY c.customer_id, c.name, c.email, c.tier;
The agent queries this materialized view with a simple SELECT:
SELECT * FROM customer_support_context
WHERE customer_id = 'cust_8291';
Expected output:
customer_id | name | email | tier | total_orders_30d | total_spend_30d | last_order_time | recent_statuses
-------------+-----------+----------------------+-------+------------------+-----------------+---------------------+---------------------
cust_8291 | Jane Park | jane.park@email.com | gold | 12 | 2847.50 | 2026-03-29 14:22:00 | {shipped,delivered}
No custom infrastructure. No batch jobs. No cache invalidation logic. The materialized view handles all of it. The agent gets a single-row response with everything it needs, and the data is fresh to within seconds.
Why SQL Matters for Agent Infrastructure
Using SQL as the interface between agents and their data layer is a deliberate architectural choice:
- LLMs already generate SQL well. Most large language models can write correct SQL for common query patterns, which means agents can dynamically construct queries without custom code.
- Existing tooling works. Any PostgreSQL client, ORM, or connection pool works with RisingWave. You do not need a proprietary SDK.
- Debugging is straightforward. When an agent makes a bad decision, you can inspect exactly what data it saw by running the same query manually.
As described in the proactive agent architecture pattern, streaming databases let the LLM describe the result it wants in SQL, while the database handles the heavy lifting of continuous computation.
How Does RisingWave Power AI Agents with Sub-Second Freshness?
RisingWave is an open-source, PostgreSQL-compatible streaming database purpose-built for real-time workloads. Three architectural properties make it well-suited as the real-time data layer for AI agents.
Incremental Computation, Not Full Recomputation
When a new event arrives, RisingWave does not re-execute the entire materialized view query. It computes only the delta: the change to the result caused by the new event. This is called incremental view maintenance.
For an aggregation like COUNT(*) grouped by customer_id, inserting one new order increments a single counter. The cost is proportional to the size of the change, not the size of the dataset. This is what enables sub-second freshness even at high event volumes, because processing one event is fast regardless of whether the underlying dataset has millions or billions of rows.
PostgreSQL Wire Protocol Compatibility
RisingWave speaks the PostgreSQL wire protocol. Your agent framework (whether it is LangChain, CrewAI, AutoGen, or a custom orchestrator) connects to RisingWave using a standard PostgreSQL driver:
# Python example: agent queries RisingWave for fresh context
import psycopg2
conn = psycopg2.connect(
host="localhost",
port=4566,
dbname="dev",
user="root"
)
def get_customer_context(customer_id: str) -> dict:
"""Retrieve fresh customer context for the support agent."""
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM customer_support_context WHERE customer_id = %s",
(customer_id,)
)
row = cursor.fetchone()
columns = [desc[0] for desc in cursor.description]
return dict(zip(columns, row))
# The agent calls this function as a tool
context = get_customer_context("cust_8291")
# Returns: {'customer_id': 'cust_8291', 'name': 'Jane Park', ...}
No proprietary client. No new API to learn. If your agent can query PostgreSQL, it can query RisingWave.
Exactly-Once State Consistency
AI agents need deterministic, consistent context. If a materialized view processes the same event twice, the agent could see inflated totals or duplicate records, leading to incorrect decisions.
RisingWave uses a barrier-based checkpoint mechanism that guarantees exactly-once semantics for internal state updates. Every materialized view reflects each source event exactly once, even after failures and recovery. For downstream sinks, the guarantee depends on the connector, but the internal state that agents query is always consistent.
End-to-End Example: Kafka to Agent Query
Here is a complete pipeline: ingest transaction events from Kafka, compute a real-time fraud risk score per customer, and serve it to a fraud detection agent.
-- Tested against RisingWave v2.8.0
-- Step 1: Ingest raw transaction events from Kafka
CREATE SOURCE transactions (
transaction_id VARCHAR,
customer_id VARCHAR,
merchant_id VARCHAR,
amount DECIMAL,
currency VARCHAR,
location VARCHAR,
event_time TIMESTAMP,
card_present BOOLEAN
) WITH (
connector = 'kafka',
topic = 'payments.transactions',
properties.bootstrap.server = 'kafka:9092'
) FORMAT PLAIN ENCODE JSON;
-- Step 2: Compute rolling fraud risk signals per customer
CREATE MATERIALIZED VIEW fraud_risk_signals AS
SELECT
customer_id,
COUNT(*) AS txn_count_1h,
COUNT(DISTINCT merchant_id) AS unique_merchants_1h,
SUM(amount) AS total_spend_1h,
MAX(amount) AS max_single_txn_1h,
COUNT(DISTINCT location) AS unique_locations_1h,
SUM(CASE WHEN card_present = false THEN 1 ELSE 0 END) AS card_not_present_count_1h
FROM transactions
WHERE event_time > NOW() - INTERVAL '1 hour'
GROUP BY customer_id;
-- Step 3: The fraud agent queries this view in real time
SELECT
customer_id,
txn_count_1h,
unique_merchants_1h,
total_spend_1h,
unique_locations_1h,
card_not_present_count_1h
FROM fraud_risk_signals
WHERE customer_id = 'cust_5502';
Expected output:
customer_id | txn_count_1h | unique_merchants_1h | total_spend_1h | unique_locations_1h | card_not_present_count_1h
-------------+--------------+---------------------+----------------+---------------------+---------------------------
cust_5502 | 7 | 5 | 1823.40 | 3 | 4
The fraud agent sees that customer cust_5502 has made 7 transactions in the last hour across 5 different merchants and 3 locations, with 4 of those transactions being card-not-present. This pattern is a strong fraud signal, and the agent has this information within seconds of the latest transaction, not after the next batch refresh.
Architecture Overview
The data flow from source to agent looks like this:
Kafka / CDC Sources → RisingWave → Materialized Views → AI Agent → Action / Response
Each component has a single responsibility:
- Kafka/CDC handles event capture and transport
- RisingWave handles continuous transformation and stateful computation
- Materialized views serve as the pre-computed context layer
- The agent focuses purely on reasoning and action
FAQ
What is a streaming database for AI agents?
A streaming database is a database that continuously ingests event streams and maintains query results in real time through incremental computation. For AI agents, it serves as a context layer that provides always-fresh data without batch refreshes. RisingWave is one such database, offering PostgreSQL compatibility so agents can query fresh context using standard SQL.
How is a streaming materialized view different from a cached query?
A cached query stores a static result that must be explicitly invalidated or refreshed on a schedule. A streaming materialized view in RisingWave is incrementally maintained: every new event automatically updates the result. There is no cache invalidation logic, no TTL to configure, and no stale window between refreshes. The view stays fresh as long as events keep flowing.
Can AI agents query RisingWave directly?
Yes. RisingWave implements the PostgreSQL wire protocol, so any agent framework that supports PostgreSQL connections (LangChain, CrewAI, AutoGen, or custom Python/Node.js code) can query RisingWave directly using standard database drivers like psycopg2, asyncpg, or node-postgres. No proprietary SDK or adapter is required.
How fresh is the data in a RisingWave materialized view?
RisingWave delivers end-to-end freshness under 100 milliseconds for most workloads, meaning a new event arriving in Kafka is reflected in the materialized view result within that window. The exact latency depends on the complexity of the query and the event throughput, but sub-second freshness is the norm for typical agent context queries.
Conclusion
AI agents are only as good as the data they observe. Stale context leads to wrong decisions, regardless of how sophisticated the reasoning model is. The core takeaways:
- Batch refreshes create a freshness gap that agents inherit as decision errors. Five-minute-old data is not "almost real-time" for fraud detection, trading, or customer support.
- Materialized views are the right abstraction for agent context. They pre-compute joined, aggregated, filtered results and keep them continuously fresh.
- SQL is the ideal interface between agents and their data layer. LLMs generate SQL well, existing tooling works, and debugging is straightforward.
- Incremental computation is what makes sub-second freshness possible at scale, processing only changes rather than recomputing entire datasets.
- PostgreSQL compatibility means zero integration overhead. If your agent can query Postgres, it can use RisingWave.
To learn more about building agents with streaming data, read the context engineering guide and the RisingWave getting started documentation.
Ready to try this yourself? Try RisingWave Cloud free, no credit card required. Sign up here.
Join our Slack community to ask questions and connect with other stream processing developers.
