Kafka, Flink and a local LLM: a fraud loop that closes itself

I wanted to see what an event-driven AI system looks like when every part is real and running: a message bus, a stream processor, and a language model that makes a decision something else then has to obey. So I built a small one. It is a pretend bank with a fraud problem, and this is its dashboard, recorded live:

The idea fits in three lines:

Layer	The question it answers
Kafka	What is happening?
Flink	Which of it matters?
A local LLM	What does it mean, and what should we do about it?

The rest of this post walks through each part. I am not assuming you have used Kafka or Flink before.

The parts

The parts Five services, and Kafka between every pair

• A simulator plays the bank’s card network and its online banking. It has 20,000 made-up customers and produces about 50 card purchases a second. Every 12 seconds or so it also starts a scripted story: five kinds of fraud, and two kinds of innocent behavior that only look like fraud.
• Kafka carries every event between the parts.
• Flink runs two jobs. The first adds context to every purchase. The second watches for suspicious patterns and raises an alert when it sees one.
• An agent picks up each alert, gives it to a language model running on my own hardware, and lets the model look things up before it decides.
• An action service carries out the decision. The simulator obeys it: a blocked card really is declined the next time someone tries it.

That last step is why I call it a closed loop. The model’s answer is not a report for someone to read later. It changes what the source system does next, and that change shows up in the same stream of events.

Kafka, in two minutes

Kafka is a log that many programs can share. A program appends a message to a named topic. Other programs read that topic in order, each at its own pace. Reading does not remove anything, so any number of readers can follow the same topic without knowing about each other.

That one property is what holds this demo together. The alert topic, fraud-alerts, has two readers: the agent, which acts on each alert, and the dashboard, which only displays it. Neither knows the other exists. When I stop the agent, alerts keep piling up in the topic. When I start it again, it carries on from where it left off, because Kafka remembers how far each reader got.

There are eleven topics. Here they are in Kafka’s web console about ten minutes into a run:

One detail matters later. Every purchase is sent with the card number as its key, and Kafka keeps all messages with the same key in order. So everything that happens to one card arrives in the order it happened, which is exactly what pattern detection needs.

Flink, in two minutes

Flink is a stream processor. You describe a query, and instead of running once and stopping, it runs forever, updating its answer as each event arrives. Both jobs in this demo are written in SQL.

Job 1 adds context. A raw purchase says very little: a card, an amount, a merchant ID. So for each one, Flink looks up the customer and the merchant in PostgreSQL and attaches what the bank knows. This is the heart of it:

SELECT t.*, m.name AS merchant_name, m.risk_rating AS merchant_risk,
       c.home_country, c.avg_amount,
       ROUND(t.amount / c.avg_amount, 2) AS amount_ratio,
       t.country <> c.home_country        AS is_foreign
FROM transactions AS t
LEFT JOIN customers FOR SYSTEM_TIME AS OF t.proc AS c ON t.customer_id = c.customer_id
LEFT JOIN merchants FOR SYSTEM_TIME AS OF t.proc AS m ON t.merchant_id = m.merchant_id;

transactions here is not a database table. It is the Kafka topic, declared as a table. The odd-looking FOR SYSTEM_TIME AS OF means “look this customer up as of right now”. After this job, a purchase is no longer just “$220 at merchant M-WEB-016”. It is “$220, five times this customer’s average, at a high-risk gift-card site”.

Job 2 looks for patterns. It reads the enriched purchases and runs five detectors, each using a different Flink technique:

Alert	What trips it	How Flink does it
Amount spike	one purchase at least 8 times the customer’s average, and $500 or more	a plain filter
Velocity burst	8 or more purchases on one card in 60 seconds	a sliding time window
Impossible travel	the physical card used in two countries within 15 minutes	sequence matching
Card testing	three or more charges under $5, then one of $200 or more	sequence matching with a repeat count
Account takeover	a password reset or new device from abroad, then a large online purchase within 10 minutes	a join between two streams

The card-testing rule is a good one to read, because the SQL is almost the sentence above. Thieves often check that a stolen card works with a few tiny charges before the big one:

MATCH_RECOGNIZE (
    PARTITION BY card_id            -- look at each card separately
    ORDER BY ts                     -- in the order things happened
    ...
    PATTERN (S{3,} L) WITHIN INTERVAL '5' MINUTE
    DEFINE
        S AS S.amount < 5,          -- S: a small charge
        L AS L.amount >= 200        -- L: a large one
)

S{3,} L is the whole rule: three or more small ones, then a large one.

These rules are broad on purpose. They catch real attacks, and they also fire on a customer paying a tuition bill. That is fine, because Flink’s job here is to miss nothing. Telling the two apart is somebody else’s job.

Where the local model comes in

A language model would be a terrible way to look at 50 purchases a second. Each call takes seconds, not milliseconds. But it is good at something rules are bad at: weighing messy context. “This is 13 times her average” is a rule. “This is 13 times her average, but she has paid this same college almost this same amount on the 3rd of each of the last three months” is a judgment.

So the model only sees what Flink flags. In the run I recorded, that was 57 alerts out of about 32,000 purchases.

The model is Qwen 3.8, served on my own workstation by SGLang, the same setup I described in the Rubik’s Cube post. Nothing goes to a cloud API. The server speaks the same protocol as OpenAI’s, so the standard Python client works once it is pointed at a different address:

from openai import OpenAI

client = OpenAI(base_url="http://<my-workstation>:8000/v1", api_key="dummy")
resp = client.chat.completions.create(model="qwen38-27b", messages=messages, tools=TOOLS)

The agent is a loop

The model does not get one question and give one answer. It gets the alert and a list of tools: functions it is allowed to ask for. It replies with the lookups it wants, my code runs them and sends back the results, and the model continues. The tools are:

• get_customer_profile: home city, how long they have been a customer, average purchase, card status.
• get_recent_transactions: the card’s latest purchases, including declined ones.
• get_spending_baseline: 90 days of habits. Typical and largest amounts, categories, countries, earlier bursts.
• get_merchant_history: the merchant’s risk rating, and everything this card bought there before.
• get_recent_auth_events: logins, password resets and new devices in the last 30 minutes.
• get_prior_cases: earlier alerts for this customer and what was decided.
• submit_decision: returns nothing. This one ends the investigation.

Two things about this list are worth pointing out.

First, it mixes “what normally happens” with “what is happening right now”. Five of the lookups read PostgreSQL, the database that Flink’s first job keeps filling. The logins come from memory: the agent follows the auth-events topic itself and keeps the last few events for each customer.

Second, the final answer is also a tool call. submit_decision has fixed fields: a verdict, a risk score from 0 to 100, one of four actions, an explanation for a human reviewer, and the text message for the customer. Because the actions are a fixed list, my code never has to guess what a paragraph of prose meant. If the model sends something invalid, the agent says so and the model tries again.

The four actions:

Action	Meaning	What the simulator does
Block card	the card details are stolen	declines everything on that card
Freeze account	someone else controls the online banking	the same, reported as a frozen account
Step-up	the evidence is mixed, so ask the customer to confirm in the app	the fraudster fails, the real customer passes
Allow	this fits the customer’s own behavior	nothing

The instructions the model gets do not tell it which pattern deserves which answer. They describe the situation and the stakes. This paragraph is from the prompt, word for word:

Both kinds of mistake are costly. Letting fraud continue loses money every few seconds. Blocking a genuine customer strands them at a checkout and damages trust.

Follow one case

Here is the account takeover from the clip, with the times the demo logged.

One real case An account takeover, from purchase to frozen card in 6.0 seconds

↻ Replay

1. T − 5 s

   Simulated bank → auth-events

   Someone in Hanoi resets the password on a US customer's online banking and enrolls a new device.

2. T + 0

   Simulated bank → transactions

   That device buys $220 of gift cards at GiftCardHub. The card is active, so the purchase is approved.

3. a moment later

   Flink job 1 → transactions-enriched

   Adds what the bank knows: home country US, merchant risk HIGH, and that this is 5.09 times the customer's average purchase.

4. T + 0.2 s

   Flink job 2 → fraud-alerts

   The account-takeover rule matches: a password reset from abroad, then a large online purchase. The alert reaches the agent.

5. T + 1.4 s

   Agent + Qwen → agent-trace

   The model's first reply asks for 6 lookups at once: profile, logins, recent purchases, 90-day baseline, history with this merchant, earlier cases.

6. T + 6.0 s

   Agent + Qwen → agent-decisions

   Its second reply is the decision: fraud, risk 95 out of 100, freeze the account. 756 tokens written in 2 rounds.

7. T + 6.0 s

   Action service → card-actions

   The safety check accepts it. The card is marked FROZEN and the customer's text message goes out.

8. afterwards

   Simulated bank → transactions

   The fraudster's next 3 attempts ($660) are declined. 2 purchases ($440) got through before the freeze.

And here is the same case as the dashboard shows it. The paragraph in the middle is the model’s own explanation:

The bottom line of each card is the simulator’s ground truth: what was really going on. It travels on its own topic that neither Flink nor the agent reads, so it works as an honest answer key.

Same alert, two answers

This is the reason the model is in the loop at all. Both of these cards started as the same Flink alert, an amount spike about 13 times the customer’s average. Flink cannot tell them apart. That is by design.

The difference came from one lookup, get_merchant_history. A rule could do this too, if you wrote one for recurring bills. And then another for commuters who tap a transit card nine times in half a minute, and another for the next thing. The model was not told about either case. It was told to check whether the history shows “this merchant, amount, place or rhythm before”, and it did.

Code gets the last word

A model’s output should not act on an account unchecked, even a pretend one. Between the decision and the card there is a small piece of ordinary code:

def guardrail(decision):
    action, risk = decision["action"], decision.get("risk_score", 50)
    if action == "ALLOW" and risk >= 70:
        return "STEP_UP_AUTH", f"policy: ALLOW with risk score {risk} is upgraded to step-up verification"
    if action in ("BLOCK_CARD", "FREEZE_ACCOUNT") and risk <= 30:
        return "STEP_UP_AUTH", f"policy: {action} with risk score {risk} is downgraded to step-up verification"
    return action, None

If the model contradicts itself (“allow”, but with a risk of 85), the service picks step-up instead, the one action a real customer can undo in seconds. The same goes for a model that never produces a valid decision in six rounds: the agent falls back to step-up. In the run below neither safety net was needed, but they are there.

What I saw in one run

These numbers are from the dashboard about ten minutes after a fresh start, the same session as the clip and the screenshots.

Purchases through Kafka and Flink	32,223
Alerts raised by Flink	57
Fraud cases caught	44 of 44
Genuine customers cleared	10 of 10, none blocked, none asked to verify
Time from the purchase to the enforced action	8.3 seconds (median)
The model’s share of that	6.2 seconds (median), from 3.6 to 15.6
Model rounds per case	3 in most cases, 2 in the rest
Tokens per case	about 10,800 read, about 980 written (medians)
Fraud declined after a decision	$38,239
Fraud approved before the decision landed	$38,313

The last two rows are the most honest ones on the dashboard. The loop takes about eight seconds, and a fraudster who retries every few seconds gets one or two purchases through in that time. Half the money was already gone before the decision landed. A faster model, or a cheap automatic hold while the model thinks, would move that line. That is the kind of trade-off I wanted the demo to make visible.

What is in the box

Everything except the model runs in Docker on one Linux machine: twelve containers from one Compose file.

Piece	Version	Role
Apache Kafka	3.9.1, one broker, no ZooKeeper	the event log
Apache Flink	1.20.2, two SQL jobs	enrich and detect
PostgreSQL	17	customers, merchants, cards, history, past decisions
Python	3.12	simulator, agent, action service, dashboard
Qwen 3.8	served by SGLang on a separate workstation	the reasoning

It is a demo, and it cuts the corners a demo can: one Kafka broker, no security anywhere, a dashboard that forgets everything when it restarts, and an agent that drops an alert if the model call fails.

The obvious next step is to sink the enriched stream into Apache Iceberg, so “what normally happens” can grow past what a PostgreSQL table should hold.