Tutorial: A Customer-Support Agent with Live Context in 120 Lines of SQL

A customer-support agent with live context needs about 120 lines of SQL and 40 lines of Python when the context is maintained as materialized views in a streaming database. This tutorial builds one on RisingWave, a PostgreSQL-compatible streaming database: Kafka orders and clickstream, PostgreSQL customers and tickets via CDC, help articles embedded in-stream with an HNSW index, agent memory written back, and an agent loop that reads it all with two queries. The full architecture discussion, the alternatives table and the design decisions live in the companion guide, Real-Time Context for AI Agents. This post is the hands-on walkthrough.

What we are building

An online shop runs a support agent. When a customer asks "where is my order and can I still change the address?", the agent must know the order's current state, what the customer was just doing on the site, which help article applies, and what an earlier conversation may have promised. Each of those facts changes by the second, and each comes from a different system.

The usual answer is to give the agent six tools and let it look everything up. That is slow, expensive in tokens, and inconsistent, because each tool call sees a different moment. Instead, we precompute each context object as a materialized view and let the agent read one row.

The pieces:

Step 1: Start RisingWave

docker run -it --pull=always -p 4566:4566 -p 5691:5691 \
  risingwavelabs/risingwave:latest single_node

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

Port 4566 speaks the PostgreSQL wire protocol, so every driver and every agent framework that can talk to Postgres can talk to RisingWave.

Step 2: Connect the streams

Kafka topics become append-only sources. The operational tables come in through one shared CDC source, so an update to a customer's tier or a ticket's status arrives as a change rather than a new row.

CREATE SOURCE orders (
  order_id BIGINT, customer_id BIGINT, amount NUMERIC,
  status VARCHAR, order_ts TIMESTAMPTZ,
  WATERMARK FOR order_ts AS order_ts - INTERVAL '30 seconds'
) WITH (connector = 'kafka', topic = 'orders',
        properties.bootstrap.server = 'broker:9092')
FORMAT PLAIN ENCODE JSON;

CREATE SOURCE clickstream (
  customer_id BIGINT, event_type VARCHAR, page VARCHAR, event_ts TIMESTAMPTZ,
  WATERMARK FOR event_ts AS event_ts - INTERVAL '10 seconds'
) WITH (connector = 'kafka', topic = 'clickstream',
        properties.bootstrap.server = 'broker:9092')
FORMAT PLAIN ENCODE JSON;

CREATE SOURCE help_articles (
  article_id BIGINT, version INT, title VARCHAR, body VARCHAR, published_at TIMESTAMPTZ
) WITH (connector = 'kafka', topic = 'help_articles',
        properties.bootstrap.server = 'broker:9092')
FORMAT PLAIN ENCODE JSON;

CREATE SOURCE shop_pg WITH (
  connector = 'postgres-cdc', hostname = 'pg', port = '5432',
  username = 'rw', password = '...', database.name = 'shop', schema.name = 'public'
);

CREATE TABLE customers (customer_id BIGINT PRIMARY KEY, name VARCHAR, tier VARCHAR, region VARCHAR)
  FROM shop_pg TABLE 'public.customers';

CREATE TABLE tickets (ticket_id BIGINT PRIMARY KEY, customer_id BIGINT, status VARCHAR,
                      subject VARCHAR, opened_at TIMESTAMPTZ)
  FROM shop_pg TABLE 'public.tickets';

Step 3: Define the context as views

Each view is one context object. RisingWave maintains every one of them incrementally: when an order event arrives, only that customer's row in customer_360 changes.

CREATE MATERIALIZED VIEW customer_360 AS
SELECT c.customer_id, c.name, c.tier, c.region,
       COUNT(o.order_id) FILTER (WHERE o.status <> 'refunded')        AS orders_total,
       COALESCE(SUM(o.amount) FILTER (WHERE o.status = 'paid'), 0)  AS lifetime_value,
       MAX(o.order_ts)                                               AS last_order_at,
       COUNT(t.ticket_id) FILTER (WHERE t.status <> 'resolved')      AS open_tickets
FROM customers c
LEFT JOIN orders  o ON o.customer_id = c.customer_id
LEFT JOIN tickets t ON t.customer_id = c.customer_id
GROUP BY c.customer_id, c.name, c.tier, c.region;

CREATE MATERIALIZED VIEW session_activity AS
SELECT customer_id, window_start, window_end,
       COUNT(*) FILTER (WHERE event_type = 'page_view')      AS page_views,
       COUNT(*) FILTER (WHERE event_type = 'add_to_cart')    AS cart_adds,
       COUNT(*) FILTER (WHERE event_type = 'checkout_error') AS checkout_errors,
       ARRAY_AGG(page ORDER BY event_ts DESC)                AS recent_pages
FROM HOP(clickstream, event_ts, INTERVAL '1 minute', INTERVAL '15 minutes')
GROUP BY customer_id, window_start, window_end;

CREATE MATERIALIZED VIEW order_status_live AS
SELECT order_id, customer_id, amount, status, order_ts AS updated_at
FROM (SELECT *, ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY order_ts DESC) AS rn
      FROM orders) t
WHERE rn = 1 AND status NOT IN ('delivered', 'refunded');

Embeddings are a column in a view, not a separate pipeline. The openai_embedding function (RisingWave 2.5 and later) is wrapped in a SQL function so the model and key are defined once; the same function embeds the query at read time.

CREATE FUNCTION embed(VARCHAR) RETURNS VECTOR(1536) LANGUAGE SQL AS $$
  SELECT openai_embedding(
    '{"model": "text-embedding-3-small", "api_key": "sk-..."}'::jsonb, $1
  )::vector(1536)
$$;

CREATE MATERIALIZED VIEW article_embeddings AS
SELECT article_id, version, title, body, published_at,
       embed(title || E'\n' || body) AS embedding
FROM help_articles;

CREATE INDEX article_embeddings_hnsw ON article_embeddings
USING HNSW (embedding) INCLUDE (article_id, version, title, body)
WITH (distance_type = 'cosine', m = 16, ef_construction = 64);

Vector indexes are built on append-only inputs, which is why help articles arrive as versions on a topic rather than as an updatable CDC table. The read query picks the latest version per article.

Memory is just another stream. The agent inserts, a view summarises, the context query joins.

CREATE TABLE agent_memory (
  memory_id BIGINT PRIMARY KEY, customer_id BIGINT,
  kind VARCHAR, content VARCHAR, created_at TIMESTAMPTZ
);

CREATE MATERIALIZED VIEW customer_memory AS
SELECT customer_id,
       ARRAY_AGG(kind || ': ' || content ORDER BY created_at DESC) AS notes
FROM agent_memory
WHERE created_at > NOW() - INTERVAL '90 days'
GROUP BY customer_id;

Step 4: Two queries per turn

The first query returns the whole structured context as JSON from one consistent snapshot. The second retrieves the three most relevant articles.

-- context.sql
SELECT to_jsonb(c) AS profile,
       to_jsonb(s) AS last_15_min,
       (SELECT jsonb_agg(to_jsonb(o)) FROM order_status_live o
         WHERE o.customer_id = c.customer_id) AS open_orders,
       m.notes AS memory
FROM customer_360 c
LEFT JOIN session_activity s
       ON s.customer_id = c.customer_id
      AND s.window_end = (SELECT MAX(window_end) FROM session_activity
                           WHERE customer_id = c.customer_id)
LEFT JOIN customer_memory m ON m.customer_id = c.customer_id
WHERE c.customer_id = $1;

-- retrieve.sql
SELECT DISTINCT ON (article_id) article_id, title, body
FROM article_embeddings
ORDER BY article_id, version DESC, embedding <=> embed($2)
LIMIT 3;

Step 5: The agent loop

The Python is short because the hard part moved into the views. Any LLM client works; the context is plain JSON.

import json, psycopg, anthropic

rw = psycopg.connect("postgresql://agent_ro@localhost:4566/dev", autocommit=True)
llm = anthropic.Anthropic()
CONTEXT_SQL, RETRIEVE_SQL = open("context.sql").read(), open("retrieve.sql").read()

def build_context(customer_id, question):
    with rw.cursor() as cur:
        cur.execute(CONTEXT_SQL, (customer_id,))
        profile, last_15_min, open_orders, memory = cur.fetchone()
        cur.execute(RETRIEVE_SQL, (question,))
        articles = [{"id": a, "title": t, "body": b} for a, t, b in cur.fetchall()]
    return {"profile": profile, "last_15_min": last_15_min,
            "open_orders": open_orders or [], "memory": memory or [],
            "relevant_articles": articles}

def answer(customer_id, question):
    ctx = build_context(customer_id, question)
    resp = llm.messages.create(
        model="claude-sonnet-5-5", max_tokens=600,
        system="You are a support agent. Use only the facts in <context>. "
               "The context is at most a few seconds old; treat it as current.",
        messages=[{"role": "user",
                   "content": f"<context>{json.dumps(ctx, default=str)}</context>\n\n{question}"}])
    return resp.content[0].text

If the agent runs inside Claude Desktop, Cursor or another MCP host, skip the Python: point the RisingWave MCP server at the same read-only user and let the model run the two queries itself.

Five things we learned building it

Shape the view for the prompt, not for a dashboard. The first version of customer_360 returned forty columns. The agent did not need them, and they cost tokens on every turn. Counts, the latest few items, and JSON casting in SQL keep the whole context under about 1,500 tokens.

Consistency matters more than we expected. With tool-call assembly, the agent occasionally saw an order as "shipped" in one call and "refund pending" in another, and reasoned its way to a confident wrong answer. Reading everything from one barrier-aligned snapshot removed that class of error entirely.

In-stream embedding changes what you can promise. A policy page edited at 10:00 is retrievable at 10:00:05. That is a different product from "updated nightly", and it needed no scheduler.

Give the agent a read-only user. agent_ro has SELECT on the five context views and INSERT on agent_memory, nothing else. The same connection string goes to the MCP server, so its management tools cannot change the pipeline.

Memory next to live data is more useful than memory alone. A view that joins promises in agent_memory with order_status_live flags "we told the customer it would ship Tuesday and it has not" before the customer asks.

FAQ

Does this need Kafka? No. Any RisingWave source works: Kafka, Pulsar, Kinesis, Redpanda, PostgreSQL, MySQL, MongoDB or SQL Server CDC, or plain INSERTs. The pattern is the views, not the transport.

How fresh is the context? RisingWave commits a consistent snapshot every barrier, one second by default, so the context is typically one to two seconds behind the event. The read itself is a point lookup in milliseconds.

Can I keep my existing vector database? Yes. Keep retrieval in Pinecone, Weaviate or pgvector and use RisingWave for the structured, time-sensitive part of the context. Many teams start that way and consolidate later when the embedding pipeline becomes the thing that is always stale.

What about agents that need very high QPS? Keep the views in RisingWave and sink them to Redis or a similar store for serving. The views stay the single definition of the context; the cache is just a faster read path.

The complete code, the alternatives comparison and the six design decisions are in the guide: Real-Time Context for AI Agents: Architecture, SQL and a Reference Implementation.