Build a Local AI Product Review Stream

Enrich local product-review events with AI and display a live dashboard.

Introduction

30 Second Summary

Customer feedback rarely arrives in a neat queue. A sudden complaint can disappear inside a growing pile of reviews before a product team spots it.

In this project, you will build a local Product Review Pulse that turns incoming reviews into operational insight. Redpanda Community Edition retains each event before Ministral 3 enriches the batch through Ollama for a Streamlit dashboard.

What You'll Build

You will watch each review burst change the browser's sentiment totals before the latest customer story surfaces with a recommended action.

By the end of this project, you'll have:

  • A repeatable Kafka-compatible event stream that returns each review with broker metadata from the retained log.
  • Five-review micro-batches that emerge with sentiment labels, themes, a concise summary, and one recommended product action.
  • A live operational dashboard where sentiment metrics refresh every two seconds after each committed batch.
  • Secret Mission: Replay retained events with a fresh consumer group to rebuild the SQLite projection from scratch.

Are there any prerequisites?

This project assumes an experienced data-engineering background. You also need a Windows device that passes every hardware check in Step 1.

Before We Start

The pipeline needs a clear operational question before its implementation details matter. That question connects a current product-review signal to one concrete action for the product team.

Set Up the Local Toolchain

Your dashboard needs a broker, a local language model, Python workers, and a browser app to work together. A reproducible Windows foundation makes later failures easier to isolate.

This step prepares Python, Visual Studio Code, Docker Desktop, Ollama, and PowerShell for the local pipeline. You will finish with the project dependencies installed and the Ministral model ready.

In this step, get ready to:
  • Install the required Windows tools at their pinned releases.
  • Create the project folder and its isolated Python environment.
  • Download the local model and verify the complete toolchain.
Install and verify the Windows tools

Each tool supports one part of the local architecture. Checking them first means you only install what your workstation is missing.

Check the Windows baseline

  • Use Windows 10 22H2 or a newer supported Windows release.
  • Confirm that hardware virtualization is enabled on your workstation.
  • Confirm that your workstation has at least 8 GB of system memory.
  • Use WSL 2 version 2.1.5 or newer for the Docker backend.
  • Keep at least 4 GB of memory free for the later Redpanda broker.
  • Press the Windows key to open Windows search.
  • Type PowerShell into the search box.
  • Press Enter to open PowerShell.
  • Check the four system tools by running:
python --version
code --version
docker compose version
ollama version

What do these checks prove?

  • The first command confirms which Python interpreter PowerShell can find.
  • The second command confirms that the Visual Studio Code command is available.
  • The third command confirms that Docker Compose can communicate with Docker Desktop.
  • The fourth command reports the installed Ollama release.

Choose the tab that matches the Python result.

✔️ I see the required Python release

Python 3.14.8 is ready to create the project environment.

ⓧ I see an older Python release

The project pins Python 3.14.8 so every dependency uses the tested interpreter.

  • Download Python 3.14.8 from the official Python downloads page.
  • Run the downloaded Python installer.
  • Select the installer option that makes Python available from the terminal.
  • Complete the installation.

ⓧ Python is not found

PowerShell needs Python 3.14.8 on its executable path before it can create the project environment.

  • Download Python 3.14.8 from the official Python downloads page.
  • Run the downloaded Python installer.
  • Select the installer option that makes Python available from the terminal.
  • Complete the installation.

Choose the tab that matches the Visual Studio Code result.

✔️ I see Visual Studio Code details

Visual Studio Code is installed with its command available to PowerShell.

ⓧ The code command is not found

The recommended User setup installs Visual Studio Code without requiring administrator access. It also makes the editor command available from PowerShell.

Check Docker Desktop licensing

Docker Desktop is free for personal learning, education, non-commercial open source work, and qualifying small businesses. A paid subscription is required for larger enterprises and government entities.

Use an approved licensed installation if your workstation belongs to one of those organizations.

Choose the tab that matches the Docker Compose result.

✔️ I see the required Docker Compose release

Docker Compose v5.6.0 confirms that Docker Desktop 4.95.0 is ready.

ⓧ I see an older Docker Compose release

Upgrade to Docker Desktop 4.95.0 to use the tested Docker Compose release.

  • Download Docker Desktop from the official Windows installation guide.
  • Run the downloaded installer.
  • Keep the WSL 2 backend selected during setup.
  • Approve the Windows permission request if it appears.
  • Complete the installation.
  • Press the Windows key to open Windows search.
  • Type Docker Desktop into the search box.
  • Press Enter to start Docker Desktop.

ⓧ Docker Compose is not found

Docker Desktop provides the Docker Compose command used to run the local broker.

  • Download Docker Desktop 4.95.0 from the official Windows installation guide.
  • Run the downloaded installer.
  • Keep the WSL 2 backend selected during setup.
  • Approve the Windows permission request if it appears.
  • Complete the installation.
  • Press the Windows key to open Windows search.
  • Type Docker Desktop into the search box.
  • Press Enter to start Docker Desktop.

Choose the tab that matches the Ollama result.

✔️ I see the required Ollama release

Ollama 0.40.4 is ready to run local models.

ⓧ I see an older Ollama release

Upgrade to Ollama 0.40.4 before downloading the model.

ⓧ Ollama is not found

The Windows installer runs Ollama in the background. It also adds the Ollama command to PowerShell.

  • Close the current PowerShell window.
  • Press the Windows key to reopen Windows search.
  • Type PowerShell into the search box.
  • Press Enter to open a fresh PowerShell session.
  • Confirm that every system tool now responds by running:
python --version
code --version
docker compose version
ollama version

What should you see?

PowerShell should report Python 3.14.8, Docker Compose v5.6.0, and Ollama 0.40.4. Visual Studio Code prints its installed version, commit, and architecture.

Is a tool still unavailable?

  • Confirm that you opened a new PowerShell session after completing the installer.
  • Confirm that Docker Desktop has finished starting before checking Docker Compose.
  • Restart Windows if a completed installer has not updated your executable path.
  • Ask for targeted help by using help me diagnose which local tool PowerShell cannot find.
Create the pinned project environment

Every project file stays inside local-review-pulse on your Desktop. Keeping PowerShell in this folder ensures that later relative paths such as reviews.db always point to the same location.

  • Move PowerShell to your Desktop by running:
Set-Location -Path "$env:USERPROFILE\Desktop" -PassThru

What does this command do?

PowerShell changes its current location to your Windows Desktop. The output shows the full Desktop path so you can confirm where the project folder will be created.

  • Create the local-review-pulse folder by running:
New-Item -ItemType Directory -Path "local-review-pulse"

What does this command do?

PowerShell creates the project folder on your Desktop. The returned directory details confirm that the folder now exists.

Visual Studio Code may ask whether you trust the files in this folder. Confirm the prompt because you created the empty folder yourself.

  • Enter the project folder and open it in Visual Studio Code by running:
Set-Location -Path "local-review-pulse" -PassThru
code .

Why open the folder this way?

The location command makes local-review-pulse the current PowerShell folder. The editor command opens that exact folder so its integrated terminal starts in the project root.

A virtual environment keeps this project's Python packages separate from packages used by other projects. The .venv folder stores that isolated interpreter and its installed commands.

  • Click Terminal in the Visual Studio Code top menu.
  • Click New Terminal.
  • Create .venv with Python 3.14.8 by running:
python -m venv .venv

What does this command do?

Python creates an isolated environment named .venv inside local-review-pulse. You should see the new folder appear in the Visual Studio Code Explorer.

  • Activate the PowerShell environment by running:
.\.venv\Scripts\Activate.ps1

What does activation change?

Activation places the environment's Scripts directory at the front of the current terminal path. Python and package commands now resolve inside .venv.

Your terminal prompt should now begin with (.venv). That prefix confirms that subsequent packages stay inside the project environment.

Did PowerShell block the activation script?

The current-user execution policy can allow locally created scripts without changing the policy for every user. PowerShell may ask you to confirm this change.

  • Approve the policy change if PowerShell requests confirmation.
  • Apply the documented current-user policy and activate the environment by running:
Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser
.\.venv\Scripts\Activate.ps1

The prompt should begin with (.venv) after the second command completes.

Ask for help by using help me activate my PowerShell virtual environment safely.

  • Click the New File control in the Visual Studio Code Explorer.
  • Enter requirements.txt as the file name.

You should now see an empty requirements.txt file at the same level as .venv.

  • Paste the pinned project dependencies into requirements.txt using the following content:
confluent-kafka==2.16.0
streamlit==1.65.0

Why pin these dependencies?

  • The confluent-kafka package provides the producer and consumer APIs used later in the pipeline.
  • The streamlit package provides the browser dashboard and its timed fragment refresh.
  • The exact versions keep every learner on the tested package behavior.
  • Save requirements.txt by pressing Ctrl+S.
  • Install the pinned dependencies into the active environment by running:
pip install -r requirements.txt

What does this command do?

The package installer reads both version pins from requirements.txt. It installs confluent-kafka version 2.16.0 and streamlit version 1.65.0 inside .venv.

The first installation can take a little while as PowerShell downloads the packages. The final output should show that installation completed without a package failure.

Did dependency installation fail?

  • Confirm that your terminal prompt begins with (.venv).
  • Confirm that requirements.txt contains both lines exactly as shown.
  • Check your internet connection before retrying the installation.
  • Ask for help by using help me diagnose my pinned dependency installation.

✔️ Awesome, I've got everything!

  • Save requirements.txt before continuing.

ⓧ I'd like to double check the full code

Compare your complete requirements.txt file with this reference.

confluent-kafka==2.16.0
streamlit==1.65.0
Download the model and verify the toolchain

The enrichment worker needs a local model that can return structured review analysis. Ministral 3 provides that model through Ollama without a cloud account.

The model download is about 3.0 GB, so the first run can take several minutes. A quiet terminal during the download does not mean the process has stalled.

  • Download and start ministral-3:3b by running:
ollama run ministral-3:3b

What does this command do?

Ollama downloads ministral-3:3b when the model is absent. It then opens an interactive model session in the same terminal.

When the interactive prompt is ready, you can test whether the local model understands a simple product-review task.

  • Enter this short test prompt into the interactive model session:
Summarize why customer reviews matter in one sentence.

What does this test prove?

A coherent response confirms that the model is stored locally and can perform text generation. The consumer later sends structured batches through the same Ollama service.

You should receive a short response about learning from customer feedback. That response proves the local model can accept a prompt and generate text.

  • Press Ctrl+C to stop the interactive model session.

The final command starts the Streamlit Hello app and opens it in your browser. The integrated terminal remains occupied until you stop the app.

Before you run these checks, do you expect every tool to resolve from the same activated project terminal?

  • Verify the complete local toolchain by running:
python --version
docker compose version
code --version
ollama ls
streamlit hello

What should you see?

  • The Python line reports Python 3.14.8.
  • The Docker line reports Docker Compose v5.6.0.
  • The Visual Studio Code check reports its version, commit, and architecture.
  • The Ollama model list includes ministral-3:3b.
  • Your browser opens the Streamlit Hello app.

You have now proven that the editor, container runtime, local model, Python environment, and browser app all work from one project folder. The workstation is ready for the event stream.

  • Return to the Visual Studio Code terminal.
  • Press Ctrl+C to stop the Streamlit Hello app.

Your local toolchain is installed and verified. Next, you will start Redpanda and put the first retained product review onto a real event stream.

Stream a Smoke-Test Review

Your local data tools are ready. The next risk is whether review data survives outside the process that publishes it.

A Kafka-compatible event stream records each review in a broker log. This gives another process a durable event to read later.

In this step, get ready to:
  • Start a single Redpanda broker with Docker Compose.
  • Create a topic for retained product-review events.
  • Complete a retained round trip for one JSON review.
Start the Redpanda broker

The broker needs a repeatable definition for its image, listeners, ports, and storage. The docker-compose.yml file keeps that definition beside your project files.

  • Return to the Explorer sidebar in Visual Studio Code.
  • Select the new-file icon in the Explorer sidebar.
  • Enter docker-compose.yml as the file name.

You should see docker-compose.yml at the root of local-review-pulse.

  • Configure the broker by pasting this definition into docker-compose.yml:
name: local-review-pulse

services:
  redpanda:
    image: docker.redpanda.com/redpandadata/redpanda:v26.2.4
    container_name: review-pulse-redpanda
    command:
      - redpanda
      - start
      - --kafka-addr internal://0.0.0.0:9092,external://0.0.0.0:19092
      - --advertise-kafka-addr internal://redpanda:9092,external://localhost:19092
      - --rpc-addr redpanda:33145
      - --advertise-rpc-addr redpanda:33145
      - --mode dev-container
      - --smp 1
      - --default-log-level=info
    ports:
      - "19092:19092"
      - "19644:9644"
    volumes:
      - redpanda-data:/var/lib/redpanda/data

volumes:
  redpanda-data:

What does this configuration do?

  • The name field gives the Compose project a stable name.
  • The redpanda service runs the pinned development broker image.
  • The startup arguments configure internal and host-facing broker addresses.
  • The port mappings expose Kafka at localhost:19092.
  • The second port mapping exposes the admin port at localhost:19644.
  • The redpanda-data volume preserves the broker log across container restarts.
  • Save docker-compose.yml by pressing Ctrl+S.

The first broker launch can take a while because Docker may need to download the image. The terminal continues reporting progress while that happens.

  • Start the broker in the background by running this command in the activated PowerShell terminal:
docker compose up -d

What does this command do?

Docker Compose reads docker-compose.yml and starts review-pulse-redpanda. Detached mode keeps the broker running after PowerShell returns to the prompt.

  • Wait until the terminal reports that the broker container has started.

Before you check the cluster, how many brokers do you expect this configuration to report?

  • Inspect the running cluster by running this command:
docker exec review-pulse-redpanda rpk cluster info

What does this command check?

The command runs rpk inside the broker container. The cluster information confirms that Redpanda is accepting administrative requests.

You should see one broker in the cluster information. That is the first working piece of your local streaming pipeline.

Broker not responding?

  • Confirm Docker Desktop is still running before retrying the broker command.
  • Check the indentation in docker-compose.yml against the reference below.
  • Confirm the container name in the file is review-pulse-redpanda.
  • Help me troubleshoot my local Redpanda broker startup.

✔️ Awesome, I've got everything!

Your saved broker configuration matches the project environment. Your existing Python dependencies remain pinned for the worker and dashboard steps.

ⓧ I'd like to double check the full code

Compare these files with the copies at the root of local-review-pulse.

name: local-review-pulse

services:
  redpanda:
    image: docker.redpanda.com/redpandadata/redpanda:v26.2.4
    container_name: review-pulse-redpanda
    command:
      - redpanda
      - start
      - --kafka-addr internal://0.0.0.0:9092,external://0.0.0.0:19092
      - --advertise-kafka-addr internal://redpanda:9092,external://localhost:19092
      - --rpc-addr redpanda:33145
      - --advertise-rpc-addr redpanda:33145
      - --mode dev-container
      - --smp 1
      - --default-log-level=info
    ports:
      - "19092:19092"
      - "19644:9644"
    volumes:
      - redpanda-data:/var/lib/redpanda/data

volumes:
  redpanda-data:

confluent-kafka==2.16.0
streamlit==1.65.0
Create the topic and publish the review

A topic is the named log that receives related events. Your smoke-test review needs a topic before the broker can retain it.

  • Create the product-review topic by running this command:
docker exec review-pulse-redpanda rpk topic create product-reviews

What does this command do?

The command creates product-reviews inside the running broker. Future review events can now be appended to this retained log.

You should see confirmation that the product-reviews topic was created.

Topic creation failed?

  • Confirm the earlier cluster check still reports one broker.
  • Check that the command targets review-pulse-redpanda.
  • Help me fix my Redpanda topic creation problem.

The next command opens an interactive producer. It stays attached to PowerShell while it waits for event values.

  • Start the interactive producer by running this command:
docker exec -it review-pulse-redpanda rpk topic produce product-reviews

What does this command do?

The command opens an interactive producer for product-reviews. Each submitted value becomes an event in the topic log.

  • Publish the smoke-test event by pasting this JSON into the waiting producer:
{"review_id":"smoke-1","product":"Trail Camera","rating":5,"review_text":"Setup was easy and the picture is sharp.","event_time":0}

What is in this event?

  • The review_id field identifies this smoke-test review.
  • The product field names the reviewed item.
  • The rating field carries the five-star score.
  • The review_text field contains the customer feedback.
  • The event_time field uses a fixed smoke-test value.
  • Press Enter to submit the event.
  • Press Ctrl+C to stop the interactive producer.

The review now sits in the broker log. You have published a real event outside the Python process that will eventually consume it.

Review not accepted?

  • Confirm the interactive producer is still waiting for input before pasting the JSON.
  • Check that every JSON field uses double quotation marks.
  • Confirm the topic name is product-reviews.
  • Help me troubleshoot my smoke-test event.
Read the retained event

The final check reads from the broker instead of reusing an in-memory object. Broker metadata also shows where Redpanda stored the event in the log.

Before you consume one event, do you expect the result to include only your JSON or broker-assigned metadata too?

  • Consume one retained event by running this command:
docker exec -it review-pulse-redpanda rpk topic consume product-reviews --num 1

What does this command prove?

The command reads one event from product-reviews. The returned metadata identifies its position in the retained broker log.

You should see {"review_id":"smoke-1","product":"Trail Camera","rating":5,"review_text":"Setup was easy and the picture is sharp.","event_time":0} together with its topic, partition, offset, and timestamp metadata.

  • Locate the JSON event value in the result.
  • Locate the broker-assigned offset in the result.
  • Leave the Redpanda broker running for the next step.

That round trip proves your review survives as a retained event. Next, you will send full review bursts through the broker and enrich them with local AI.

Enrich Review Batches

Your Redpanda broker can retain a review event. Raw reviews still leave the product team to interpret every comment by hand.

This step turns retained events into an operational signal. A producer sends realistic reviews while a consumer uses micro-batches, local AI enrichment, manual offset commits, and an idempotent SQLite projection.

In this step, get ready to:
  • Build a producer that publishes keyed JSON review events.
  • Build a consumer that enriches five-review batches with structured local AI output.
  • Run both workers to create the analytical projection in reviews.db.
Build the event producer

A producer converts review tuples into keyed JSON events. Each run uses a fresh timestamp in review_id, so repeated demonstrations create distinct events.

  • Return to Visual Studio Code from earlier.
  • Use the file-creation control in the Explorer sidebar to create producer.py inside local-review-pulse.
  • Add the producer settings plus the review catalog by pasting this code at the top of producer.py.
import json
import time

from confluent_kafka import Producer

BROKER = "localhost:19092"
TOPIC = "product-reviews"

REVIEWS = [
    ("Trail Camera", 5, "Setup was easy and the night picture is sharp."),
    ("Trail Camera", 2, "The battery door feels weak and opened in the rain."),
    ("Trail Camera", 4, "Motion detection is fast but the app pairing took two tries."),
    ("Trail Camera", 1, "It stopped recording after the first firmware update."),
    ("Trail Camera", 5, "Great image quality and three weeks of battery life."),
    ("Smart Kettle", 4, "Water heats quickly and the temperature presets are useful."),
    ("Smart Kettle", 2, "The mobile app disconnects every morning."),
    ("Smart Kettle", 3, "The kettle works well, but the lid is awkward to clean."),
    ("Smart Kettle", 1, "The handle became too hot after ten minutes."),
    ("Smart Kettle", 5, "Quiet, accurate, and much faster than my old kettle."),
    ("Desk Lamp", 5, "The light is even and the controls are intuitive."),
    ("Desk Lamp", 3, "Brightness is good, but the base takes too much desk space."),
    ("Desk Lamp", 2, "The USB port charges very slowly."),
    ("Desk Lamp", 1, "The hinge became loose during the first week."),
    ("Desk Lamp", 4, "Solid lamp with a useful warm-light setting."),
]

What does this code define?

  • BROKER points the producer at the broker endpoint exposed by the existing Redpanda container.
  • TOPIC targets the retained product-reviews stream from the previous step.
  • REVIEWS holds 15 realistic reviews across three products.
  • Each tuple keeps the product name, rating, and review text together.
  • Save producer.py.
  • Confirm that REVIEWS contains 15 tuples.

Missing part of the review catalog?

  • Check that the opening [ after REVIEWS = has a matching closing ].
  • Check that every review tuple ends with a comma.
  • help me compare my REVIEWS list with the expected 15 review tuples

Delivery callbacks expose whether the broker accepted each event. The callback keeps that feedback beside the producer logic.

  • Add the delivery callback below the closing bracket of REVIEWS.
def delivery_report(error, _message):
    if error is not None:
        print(f"Delivery failed: {error}")
    else:
        print("Delivered review event")

What does this callback do?

delivery_report() prints a failure when the broker rejects an event. A successful callback prints a delivery confirmation.

  • Save producer.py.
  • Confirm that delivery_report() contains a failure branch plus a success branch.

Is the callback indented incorrectly?

  • Align the if statement four spaces inside delivery_report().
  • Align the two print() calls inside their matching branches.
  • help me fix the indentation in delivery_report

The producer now needs to turn every tuple into the event contract. Its key matches review_id, which gives each review a stable identity throughout the pipeline.

  • Add the producer loop below delivery_report().
def main():
    producer = Producer({"bootstrap.servers": BROKER})
    run_id = int(time.time() * 1000)

    for index, (product, rating, review_text) in enumerate(REVIEWS, start=1):
        event = {
            "review_id": f"{run_id}-{index}",
            "product": product,
            "rating": rating,
            "review_text": review_text,
            "event_time": int(time.time() * 1000),
        }
        producer.produce(
            TOPIC,
            key=event["review_id"],
            value=json.dumps(event).encode("utf-8"),
            on_delivery=delivery_report,
        )
        producer.poll(0)
        time.sleep(0.25)

    remaining = producer.flush(10)
    if remaining:
        raise RuntimeError(f"{remaining} review events were not delivered")


if __name__ == "__main__":
    main()

How does the producer publish events?

  • run_id captures the current timestamp for this demonstration.
  • event follows the contract of review_id, product, rating, review_text, and event_time.
  • produce() serializes each event as UTF-8 JSON with the review ID as its key.
  • flush() waits for pending messages before the process exits.
  • Save producer.py.

Before you run the producer, how many delivery confirmations do you expect from the 15 review tuples?

  • Publish the first review burst by running this command in the activated PowerShell terminal:
python producer.py

What should you see?

You should see Delivered review event once for each accepted event. The topic retains the burst until a consumer reads it.

That first burst is through. Your broker now holds 15 keyed review events from producer.py.

No delivery confirmations?

  • Confirm that Docker Desktop from earlier is still running.
  • Confirm that the review-pulse-redpanda container is still available.
  • Check that BROKER remains localhost:19092.
  • help me troubleshoot missing producer delivery confirmations

✔️ Awesome, I've got everything!

Your producer.py file now publishes 15 keyed review events per run. Make sure the file is saved.

ⓧ I'd like to double check the full code

Compare your complete producer.py file with this reference.

import json
import time

from confluent_kafka import Producer

BROKER = "localhost:19092"
TOPIC = "product-reviews"

REVIEWS = [
    ("Trail Camera", 5, "Setup was easy and the night picture is sharp."),
    ("Trail Camera", 2, "The battery door feels weak and opened in the rain."),
    ("Trail Camera", 4, "Motion detection is fast but the app pairing took two tries."),
    ("Trail Camera", 1, "It stopped recording after the first firmware update."),
    ("Trail Camera", 5, "Great image quality and three weeks of battery life."),
    ("Smart Kettle", 4, "Water heats quickly and the temperature presets are useful."),
    ("Smart Kettle", 2, "The mobile app disconnects every morning."),
    ("Smart Kettle", 3, "The kettle works well, but the lid is awkward to clean."),
    ("Smart Kettle", 1, "The handle became too hot after ten minutes."),
    ("Smart Kettle", 5, "Quiet, accurate, and much faster than my old kettle."),
    ("Desk Lamp", 5, "The light is even and the controls are intuitive."),
    ("Desk Lamp", 3, "Brightness is good, but the base takes too much desk space."),
    ("Desk Lamp", 2, "The USB port charges very slowly."),
    ("Desk Lamp", 1, "The hinge became loose during the first week."),
    ("Desk Lamp", 4, "Solid lamp with a useful warm-light setting."),
]


def delivery_report(error, _message):
    if error is not None:
        print(f"Delivery failed: {error}")
    else:
        print("Delivered review event")


def main():
    producer = Producer({"bootstrap.servers": BROKER})
    run_id = int(time.time() * 1000)

    for index, (product, rating, review_text) in enumerate(REVIEWS, start=1):
        event = {
            "review_id": f"{run_id}-{index}",
            "product": product,
            "rating": rating,
            "review_text": review_text,
            "event_time": int(time.time() * 1000),
        }
        producer.produce(
            TOPIC,
            key=event["review_id"],
            value=json.dumps(event).encode("utf-8"),
            on_delivery=delivery_report,
        )
        producer.poll(0)
        time.sleep(0.25)

    remaining = producer.flush(10)
    if remaining:
        raise RuntimeError(f"{remaining} review events were not delivered")


if __name__ == "__main__":
    main()

What should match?

The file defines BROKER, TOPIC, REVIEWS, delivery_report(), and main().

Build the enrichment consumer

The consumer coordinates a consumer group with local model inference. It commits an offset only after the corresponding enrichment reaches SQLite.

Why commit after the database write?

An offset records how far a consumer group has progressed through the topic. Committing after the database write prevents a failed enrichment from being marked as complete.

INSERT OR REPLACE uses review_id as the primary key. Reprocessing the same event replaces its row instead of creating a duplicate.

  • Use the file-creation control in the Explorer sidebar to create consumer.py inside local-review-pulse.
  • Use the complete-file tab below as the source for consumer.py.
  • Save consumer.py after the complete file is in place.

✔️ Awesome, I've got everything!

Your consumer is ready when it contains the database setup, structured model request, idempotent write, batch processor, and polling loop.

ⓧ I'd like to double check the full code

Use this complete consumer.py file to build or compare your consumer.

import json
import sqlite3
import time
from urllib.request import Request, urlopen

from confluent_kafka import Consumer

BROKER = "localhost:19092"
TOPIC = "product-reviews"
GROUP_ID = "review-enricher-v1"
DB_PATH = "reviews.db"
OLLAMA_URL = "http://localhost:11434/api/chat"
MODEL = "ministral-3:3b"
BATCH_SIZE = 5
MAX_BATCH_WAIT_SECONDS = 2

OUTPUT_SCHEMA = {
    "type": "object",
    "properties": {
        "batch_summary": {"type": "string"},
        "recommended_action": {"type": "string"},
        "results": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "review_id": {"type": "string"},
                    "sentiment": {
                        "type": "string",
                        "enum": ["positive", "neutral", "negative"],
                    },
                    "theme": {"type": "string"},
                },
                "required": ["review_id", "sentiment", "theme"],
            },
        },
    },
    "required": ["batch_summary", "recommended_action", "results"],
}


def init_database():
    connection = sqlite3.connect(DB_PATH, timeout=5.0)
    try:
        connection.execute("PRAGMA journal_mode=WAL")
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS enriched_reviews (
                review_id TEXT PRIMARY KEY,
                product TEXT NOT NULL,
                rating INTEGER NOT NULL,
                review_text TEXT NOT NULL,
                event_time INTEGER NOT NULL,
                sentiment TEXT NOT NULL,
                theme TEXT NOT NULL,
                batch_summary TEXT NOT NULL,
                recommended_action TEXT NOT NULL,
                processed_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
            )
            """
        )
        connection.commit()
    finally:
        connection.close()


def analyze_batch(events):
    request_body = {
        "model": MODEL,
        "messages": [
            {
                "role": "system",
                "content": (
                    "Analyze product reviews. Return one result for every review_id. "
                    "Use concise themes, summarize the batch in two sentences or fewer, "
                    "and recommend one concrete product-team action."
                ),
            },
            {"role": "user", "content": json.dumps(events)},
        ],
        "stream": False,
        "format": OUTPUT_SCHEMA,
        "options": {"temperature": 0},
    }
    request = Request(
        OLLAMA_URL,
        data=json.dumps(request_body).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urlopen(request, timeout=180) as response:
        ollama_response = json.load(response)
    return json.loads(ollama_response["message"]["content"])


def save_batch(events, analysis):
    results_by_id = {
        result["review_id"]: result for result in analysis["results"]
    }
    rows = []
    for event in events:
        result = results_by_id[event["review_id"]]
        rows.append(
            (
                event["review_id"],
                event["product"],
                event["rating"],
                event["review_text"],
                event["event_time"],
                result["sentiment"],
                result["theme"],
                analysis["batch_summary"],
                analysis["recommended_action"],
            )
        )

    connection = sqlite3.connect(DB_PATH, timeout=5.0)
    try:
        connection.executemany(
            """
            INSERT OR REPLACE INTO enriched_reviews (
                review_id,
                product,
                rating,
                review_text,
                event_time,
                sentiment,
                theme,
                batch_summary,
                recommended_action
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            rows,
        )
        connection.commit()
    finally:
        connection.close()


def process_batch(batch, consumer):
    analysis = analyze_batch(batch)
    save_batch(batch, analysis)
    consumer.commit(asynchronous=False)
    print(f"Processed {len(batch)} review events")


def main():
    init_database()
    consumer = Consumer(
        {
            "bootstrap.servers": BROKER,
            "group.id": GROUP_ID,
            "auto.offset.reset": "earliest",
            "enable.auto.commit": False,
        }
    )
    consumer.subscribe([TOPIC])
    batch = []
    last_event_time = time.monotonic()
    print(f"Waiting for events on {TOPIC}")

    try:
        while True:
            message = consumer.poll(1.0)
            if message is None:
                if batch and time.monotonic() - last_event_time >= MAX_BATCH_WAIT_SECONDS:
                    try:
                        process_batch(batch, consumer)
                    except Exception as error:
                        print(f"Batch failed and offsets were not committed: {error}")
                        last_event_time = time.monotonic()
                    else:
                        batch.clear()
                continue

            if message.error():
                print(f"Consumer error: {message.error()}")
                continue

            batch.append(json.loads(message.value().decode("utf-8")))
            last_event_time = time.monotonic()

            if len(batch) >= BATCH_SIZE:
                try:
                    process_batch(batch, consumer)
                except Exception as error:
                    print(f"Batch failed and offsets were not committed: {error}")
                    last_event_time = time.monotonic()
                else:
                    batch.clear()
    except KeyboardInterrupt:
        print("Stopping consumer")
    finally:
        consumer.close()


if __name__ == "__main__":
    main()

How does the consumer protect progress?

  • init_database() enables WAL mode and creates the enriched_reviews table.
  • analyze_batch() sends each batch to Ollama with a structured output schema.
  • save_batch() persists the returned sentiment, theme, summary, and recommendation.
  • process_batch() commits the consumer offset only after the SQLite write succeeds.

The first local enrichment can take longer while Ministral 3 starts. A short pause here is normal.

Before you start the consumer, do you expect its new group to wait for future events or read the retained events from the earliest available offset?

  • Start the consumer in the activated PowerShell terminal by running:
python consumer.py

What should you see?

The consumer first prints Waiting for events on product-reviews. The new group then reads the retained smoke test plus the producer burst.

Complete groups of five produce Processed 5 review events. A remaining partial batch is processed after the two-second idle boundary.

  • Leave this consumer terminal running for the end-to-end check.

Consumer batch not completing?

  • Confirm that the Ollama service from earlier remains available.
  • Confirm that ministral-3:3b remains downloaded.
  • Check that OLLAMA_URL and MODEL match the complete file.
  • help me troubleshoot my local enrichment consumer
Run the end-to-end worker path

The running consumer is ready for another live burst. A second terminal lets you publish events while the first terminal keeps polling.

  • Keep the consumer running in the first PowerShell terminal.
  • Use the split-terminal control in the Visual Studio Code terminal panel to create a second PowerShell terminal.
  • Confirm that the second terminal prompt shows the activated .venv.
  • Activate .venv with the same PowerShell command from the setup step if the environment name is missing.

Before you publish this burst, how many complete five-review batches should 15 new events create?

  • Emit a fresh review burst from the second terminal by running:
python producer.py

What proves the pipeline works?

  • The producer prints a delivery confirmation for every review accepted by Redpanda.
  • The consumer prints Processed 5 review events at each complete batch boundary.
  • The reviews.db file appears in the local-review-pulse Explorer tree.
  • Each successful batch reaches SQLite before its offsets are committed.

That is the hard coordination working. Your retained review stream now becomes structured analytical state without cloud services.

Producer succeeds but consumer stays quiet?

  • Confirm that both terminals are running from the local-review-pulse folder.
  • Confirm that TOPIC is product-reviews in both files.
  • Check the consumer terminal for a batch failure mentioning the model response or database write.
  • help me trace events from producer.py to consumer.py

Your producer, broker, local model, and consumer now form a working enrichment pipeline. Next, you will turn reviews.db into a browser view with Streamlit.

Demonstrate the Stale Dashboard

Your Redpanda stream now carries review events into an idempotent SQLite projection. The consumer commits each offset only after the enriched rows are stored.

A Streamlit dashboard can turn those rows into an operational view. This step tests whether an already-open browser reflects database commits made after the page loads.

In this step, get ready to:
  • Create a dashboard that reads enriched reviews from reviews.db.
  • Display sentiment metrics with the latest batch insight.
  • Produce another review burst to test the open dashboard.
Create the static dashboard shell

The dashboard starts with its database path plus the page-level content. This gives you a runnable browser view before you add the analytical sections.

  • In the Visual Studio Code Explorer sidebar from earlier, create dashboard.py at the root of local-review-pulse.
  • Build the page shell in dashboard.py by pasting this code:
import sqlite3

import streamlit as st

DB_PATH = "reviews.db"

st.set_page_config(page_title="Product Review Pulse", layout="wide")
st.title("Product Review Pulse")
st.write("Local event streaming with Redpanda, Ministral 3, SQLite, and Streamlit")

What does this code do?

  • The sqlite3 import gives the dashboard access to the local analytical projection.
  • The DB_PATH value points every query at reviews.db.
  • The page configuration uses the available browser width for metrics plus the review table.
  • The title identifies the operational view when it opens in your browser.
  • Save dashboard.py.

The dashboard server occupies its PowerShell terminal while it runs. You can stop it later with Ctrl+C.

  • Create another PowerShell terminal in the Visual Studio Code terminal panel.
  • Confirm that the terminal starts in the local-review-pulse folder with .venv active.
  • Start the dashboard server by running this command:
streamlit run dashboard.py

What does this command do?

The command starts the local Streamlit server with dashboard.py as the app. It also connects your browser session to that running Python process.

You should see a browser page titled Product Review Pulse with the local streaming description underneath. That first browser view confirms the app can run from your project environment.

Dashboard page not opening?

  • Check that the PowerShell prompt showed .venv before you started the server.
  • Check that dashboard.py is saved at the root of local-review-pulse.
  • Use the local address printed in the terminal if the browser did not open automatically.
  • Help me troubleshoot my local Streamlit dashboard startup.
Render the enriched review state

The browser shell now needs a snapshot of the projection. The database query retrieves every field required by the metrics plus the latest insight and review table.

  • In dashboard.py, add this function below the page description:
def load_reviews():
    connection = sqlite3.connect(DB_PATH, timeout=5.0)
    connection.row_factory = sqlite3.Row
    try:
        rows = connection.execute(
            """
            SELECT
                review_id,
                product,
                rating,
                review_text,
                sentiment,
                theme,
                batch_summary,
                recommended_action,
                processed_at
            FROM enriched_reviews
            ORDER BY processed_at DESC, review_id DESC
            """
        ).fetchall()
    except sqlite3.OperationalError:
        rows = []
    finally:
        connection.close()
    return [dict(row) for row in rows]

How does the database read work?

  • The connection uses the shared DB_PATH value to read the consumer's projection.
  • The sqlite3.Row row factory keeps each selected field accessible by its column name.
  • The query sorts the newest processed rows first.
  • The exception path returns an empty list if the table is unavailable.
  • Save dashboard.py.
  • Refresh the browser page.

You should still see the Product Review Pulse heading with no Streamlit error panel. This confirms that the database function loads successfully.

Seeing a database error?

  • Check that reviews.db is at the root of local-review-pulse.
  • Check that DB_PATH remains set to reviews.db.
  • Help me diagnose the database read in my Streamlit dashboard.

The loaded rows become useful once the dashboard converts their sentiment values into counts. The empty-state branch also keeps the page informative when no enriched rows are available.

  • Add this function below load_reviews() in dashboard.py.
  • Place the final render_dashboard() call at the bottom of the file:
def render_dashboard():
    reviews = load_reviews()
    if not reviews:
        st.info("No enriched reviews yet. Start the consumer, then run the producer.")
        return

    total = len(reviews)
    positive = sum(review["sentiment"] == "positive" for review in reviews)
    neutral = sum(review["sentiment"] == "neutral" for review in reviews)
    negative = sum(review["sentiment"] == "negative" for review in reviews)

    total_column, positive_column, neutral_column, negative_column = st.columns(4)
    total_column.metric("Total reviews", total)
    positive_column.metric("Positive", positive)
    neutral_column.metric("Neutral", neutral)
    negative_column.metric("Negative", negative)


render_dashboard()

How are the metrics calculated?

  • The reviews list holds the current database snapshot.
  • Each sentiment total counts rows whose sentiment value matches one label.
  • The four columns keep the totals visible in one horizontal row.
  • The final function call renders the analytical section during the normal app run.
  • Save dashboard.py.
  • Refresh the browser page.

You should see four metrics for total, positive, neutral, and negative reviews. Those values come from the enriched rows already stored by your consumer.

Metrics missing from the page?

  • Check that render_dashboard() is aligned at the left edge of the final line.
  • Check that the sentiment labels remain lowercase inside the three count expressions.
  • Help me fix the missing sentiment metrics in my dashboard.

Counts show the shape of customer sentiment. The latest batch summary plus its recommended action turns that shape into a product decision.

  • In render_dashboard(), find the negative_column.metric("Negative", negative) line.
  • Insert this code immediately below that line while preserving its indentation:
    latest = reviews[0]
    st.subheader("Latest batch summary")
    st.write(latest["batch_summary"])
    st.info(latest["recommended_action"], title="Recommended action")

    st.subheader("Recent enriched reviews")
    st.dataframe(
        reviews,
        width="stretch",
        hide_index=True,
        column_order=[
            "processed_at",
            "product",
            "rating",
            "sentiment",
            "theme",
            "review_text",
        ],
    )

What completes the analytical view?

  • The first row is the latest processed review because load_reviews() sorts rows in descending order.
  • The latest row carries the AI-generated batch summary.
  • The recommended action gives the product team one concrete response.
  • The dataframe presents recent enriched reviews with the operational columns first.
  • Save dashboard.py.
  • Refresh the browser page.

You should now see the latest batch summary plus a recommended action. A recent enriched-review table should appear underneath.

Summary or table not appearing?

  • Check that the new lines remain indented inside render_dashboard().
  • Check that latest = reviews[0] appears after the four metric calls.
  • Help me troubleshoot the missing summary or dataframe in my dashboard.

✔️ Awesome, I've got everything!

  • Save dashboard.py.
  • Keep the Streamlit terminal running for the stale-dashboard test.

ⓧ I'd like to double check the full code

Compare your dashboard.py file with this complete static dashboard. The file ends with a normal render_dashboard() call.

import sqlite3

import streamlit as st

DB_PATH = "reviews.db"

st.set_page_config(page_title="Product Review Pulse", layout="wide")
st.title("Product Review Pulse")
st.write("Local event streaming with Redpanda, Ministral 3, SQLite, and Streamlit")


def load_reviews():
    connection = sqlite3.connect(DB_PATH, timeout=5.0)
    connection.row_factory = sqlite3.Row
    try:
        rows = connection.execute(
            """
            SELECT
                review_id,
                product,
                rating,
                review_text,
                sentiment,
                theme,
                batch_summary,
                recommended_action,
                processed_at
            FROM enriched_reviews
            ORDER BY processed_at DESC, review_id DESC
            """
        ).fetchall()
    except sqlite3.OperationalError:
        rows = []
    finally:
        connection.close()
    return [dict(row) for row in rows]


def render_dashboard():
    reviews = load_reviews()
    if not reviews:
        st.info("No enriched reviews yet. Start the consumer, then run the producer.")
        return

    total = len(reviews)
    positive = sum(review["sentiment"] == "positive" for review in reviews)
    neutral = sum(review["sentiment"] == "neutral" for review in reviews)
    negative = sum(review["sentiment"] == "negative" for review in reviews)

    total_column, positive_column, neutral_column, negative_column = st.columns(4)
    total_column.metric("Total reviews", total)
    positive_column.metric("Positive", positive)
    neutral_column.metric("Neutral", neutral)
    negative_column.metric("Negative", negative)

    latest = reviews[0]
    st.subheader("Latest batch summary")
    st.write(latest["batch_summary"])
    st.info(latest["recommended_action"], title="Recommended action")

    st.subheader("Recent enriched reviews")
    st.dataframe(
        reviews,
        width="stretch",
        hide_index=True,
        column_order=[
            "processed_at",
            "product",
            "rating",
            "sentiment",
            "theme",
            "review_text",
        ],
    )


render_dashboard()
Prove the browser becomes stale

The database and browser now show the same analytical snapshot. The useful test is what happens when the consumer commits another burst while that browser page remains open.

  • Return to the browser tab showing Product Review Pulse.
  • Check the current total plus the first row in the recent-review table.
  • Switch back to the PowerShell terminal running consumer.py from earlier.
  • Confirm that the consumer is still waiting for events on product-reviews.
  • Switch to another activated PowerShell terminal in local-review-pulse.

Before you run the producer, do you think the open browser metrics will change when new rows reach SQLite?

  • Produce another review burst by running this command:
python producer.py

What does this test change?

The producer assigns a fresh timestamp-based review_id to each review. The burst therefore creates new stream events plus new projection rows.

  • Watch the producer terminal until the delivery confirmations finish.
  • Return to the consumer terminal.
  • Wait for the local model to finish each batch.

You should see Processed 5 review events as each complete batch reaches SQLite. The new rows are now committed to the analytical projection.

  • Return to the browser without refreshing the page.
  • Compare the displayed metrics with the view you checked before the producer run.

The metrics plus recent-review table remain unchanged even though the consumer processed the new events. You have reproduced the intended shortfall: the open browser has become stale.

  • Refresh the browser page manually.

The total review count should increase after the refresh. You should also see newer rows plus the latest completed batch insight.

Dashboard changed before the test?

  • Check that dashboard.py does not contain the fragment decorator from the finished project code.
  • Repeat the test without editing or saving dashboard.py while the producer runs.
  • Help me verify the stale-dashboard behavior in my local pipeline.

You now have an analytical dashboard plus direct evidence of its refresh gap. Next up, you will make the live section reread committed insight without manually refreshing the full page.

Enable Live Dashboard Updates

You proved that the open dashboard stays stale after the SQLite projection changes. The Redpanda event pipeline is already processing each review burst.

A Streamlit fragment reruns one section of the page independently. Its two-second interval brings each committed batch into the browser without a manual refresh.

In this step, get ready to:
  • Add a two-second Streamlit fragment to render_dashboard().
  • Prove that every dashboard view updates after a new review burst.
Add the live fragment

The database read already happens inside render_dashboard(). Decorating that function makes every fragment rerun query reviews.db again.

  • In dashboard.py, scroll to the render_dashboard() definition near the bottom of the file.
  • Locate this current function header:
def render_dashboard():

What Is the Current State?

This header defines render_dashboard() as a normal function. Streamlit runs it during a full script run.

  • Add the decorator immediately above render_dashboard() by matching this result:
@st.fragment(run_every=2)
def render_dashboard():

What Does This Change Do?

  • The @st.fragment(run_every=2) decorator schedules an independent fragment rerun every two seconds.
  • Each rerun calls load_reviews() from inside render_dashboard().
  • The database query returns the latest committed rows for the fragment to render.
  • Save dashboard.py.
  • Return to the Streamlit browser from earlier.
  • Confirm that the four metrics remain visible.

Did the Dashboard Stop Rendering?

Check that @st.fragment(run_every=2) sits directly above def render_dashboard(): with no indentation.

Confirm that the final render_dashboard() call remains at the bottom of dashboard.py.

Help me debug my Streamlit fragment decorator placement.

✔️ Awesome, I've got everything!

The fragment decorator is saved. Your existing dashboard still renders its current analytical state.

ⓧ I'd like to double check the full code

  • Compare dashboard.py with the completed file below:
import sqlite3

import streamlit as st

DB_PATH = "reviews.db"

st.set_page_config(page_title="Product Review Pulse", layout="wide")
st.title("Product Review Pulse")
st.write("Local event streaming with Redpanda, Ministral 3, SQLite, and Streamlit")


def load_reviews():
    connection = sqlite3.connect(DB_PATH, timeout=5.0)
    connection.row_factory = sqlite3.Row
    try:
        rows = connection.execute(
            """
            SELECT
                review_id,
                product,
                rating,
                review_text,
                sentiment,
                theme,
                batch_summary,
                recommended_action,
                processed_at
            FROM enriched_reviews
            ORDER BY processed_at DESC, review_id DESC
            """
        ).fetchall()
    except sqlite3.OperationalError:
        rows = []
    finally:
        connection.close()
    return [dict(row) for row in rows]


@st.fragment(run_every=2)
def render_dashboard():
    reviews = load_reviews()
    if not reviews:
        st.info("No enriched reviews yet. Start the consumer, then run the producer.")
        return

    total = len(reviews)
    positive = sum(review["sentiment"] == "positive" for review in reviews)
    neutral = sum(review["sentiment"] == "neutral" for review in reviews)
    negative = sum(review["sentiment"] == "negative" for review in reviews)

    total_column, positive_column, neutral_column, negative_column = st.columns(4)
    total_column.metric("Total reviews", total)
    positive_column.metric("Positive", positive)
    neutral_column.metric("Neutral", neutral)
    negative_column.metric("Negative", negative)

    latest = reviews[0]
    st.subheader("Latest batch summary")
    st.write(latest["batch_summary"])
    st.info(latest["recommended_action"], title="Recommended action")

    st.subheader("Recent enriched reviews")
    st.dataframe(
        reviews,
        width="stretch",
        hide_index=True,
        column_order=[
            "processed_at",
            "product",
            "rating",
            "sentiment",
            "theme",
            "review_text",
        ],
    )


render_dashboard()

What Should You Compare?

The decorator sits immediately above render_dashboard(). The function still reads the database before calculating every dashboard value.

The final render_dashboard() call remains at the bottom of the file. That call places the fragment on the page.

Demonstrate the finished pipeline

The fragment proves its value when the analytical state changes. A fresh producer burst gives you a clear end-to-end test of the two-second refresh cycle.

  • Return to the PowerShell terminal running consumer.py from earlier.
  • Confirm that the consumer is still waiting for events on product-reviews.
  • Switch to the second PowerShell terminal from earlier.

Before you run the producer, which dashboard elements do you expect to change first after each consumer commit?

  • Emit another 15-review burst by running:
python producer.py

What Happens Across the Pipeline?

The producer sends keyed review events to product-reviews. The consumer handles the events in five-review batches.

Each successful batch updates reviews.db before the consumer commits its offsets. The fragment reads that committed analytical state during its next rerun.

  • Watch the producer terminal until it finishes printing Delivered review event confirmations.
  • Return to the consumer terminal after each batch completes.
  • Confirm that the consumer prints Processed 5 review events.
  • Switch back to the open dashboard in your browser.

Within two seconds of each consumer commit, you'll see the total and sentiment metrics change without a manual refresh. The latest summary also advances.

You'll see a new recommended action beneath the summary. The recent-review table fills with the newest enriched rows.

That closes the stale-dashboard gap. Your browser now tracks the latest committed analytical state on its own.

Are the Metrics Still Stale?

Confirm that the consumer printed Processed 5 review events. The dashboard only changes after enriched rows reach reviews.db.

Check that @st.fragment(run_every=2) remains directly above render_dashboard() in the file used by the running Streamlit process.

Help me trace why my live dashboard is not updating.

Secret mission

Replay the Stream to Rebuild State

Use a fresh consumer group to replay every retained product review. Reset the SQLite projection and watch the live dashboard rebuild from Redpanda without duplicate review IDs.

Clean Up Your Resources

Clean Up Your Resources

All project resources stay on your Windows workstation. Decide whether to keep them running, pause them to free up memory, or delete them entirely.

There are no usage-based cloud charges. Docker Desktop is free for personal use, education, non-commercial open source work, and small businesses with fewer than 250 employees plus less than $10 million in annual revenue. Larger enterprises and government entities require a paid subscription.

Resources you used:

  • The Redpanda Community Edition container review-pulse-redpanda plus the redpanda-data volume containing the retained product-reviews topic.
  • The 3.0 GB ministral-3:3b model stored by Ollama.
  • The local-review-pulse folder containing all local project files.

Keep everything running

No action is needed. Choose this option if you want to demonstrate the replay workflow again or continue experimenting with incoming reviews.

  • Leave Docker Desktop running with the review-pulse-redpanda container.
  • Leave the replay consumer running in its PowerShell window.
  • Leave the Streamlit dashboard running in its PowerShell window.
  • Keep the local-review-pulse folder plus the ministral-3:3b model for future demonstrations.

Pause - I'll come back to this later

Shut down the active processes to free up memory. Your retained events plus the rebuilt analytical state stay on disk for later use.

  • Stop the replay consumer by pressing Ctrl+C in its PowerShell window.
  • Stop the Streamlit app by pressing Ctrl+C in its PowerShell window.
  • Stop Redpanda while preserving its retained event data by running this command in the local-review-pulse PowerShell terminal:
docker compose down

What Stays on Disk?

This stops the broker container. The redpanda-data volume remains available with the retained topic.

Your reviews.db projection plus the ministral-3:3b model also remain available.

Your workstation now has the broker plus the Python processes stopped. The pipeline state is ready when you want to return.

  • Restart the broker when you return by running this command from the local-review-pulse PowerShell terminal:
docker compose up -d

What Starts Again?

This recreates the review-pulse-redpanda container from docker-compose.yml.

The broker reconnects to redpanda-data. Your retained product-reviews events become available again.

Delete - I don't want to use this again

Deleting these resources permanently removes the retained events, downloaded model, and local project files. Docker Desktop plus the Ollama application remain installed for other projects.

  • Stop any replay consumer that is still running by pressing Ctrl+C in its PowerShell window.
  • Stop any Streamlit app that is still running by pressing Ctrl+C in its PowerShell window.
  • Delete the Redpanda container plus its retained volume by running this command in the local-review-pulse PowerShell terminal:
docker compose down -v

What Does This Remove?

This removes the review-pulse-redpanda container plus the redpanda-data volume.

The retained product-reviews topic is deleted with the broker storage.

  • Delete the downloaded local model by running this command:
ollama rm ministral-3:3b

What Leaves Ollama?

This removes the 3.0 GB ministral-3:3b model from local storage. The Ollama application stays installed.

  • Move out of the project folder plus delete local-review-pulse by running these PowerShell commands:
cd ~/Desktop
Remove-Item -Recurse -Force .\local-review-pulse

What Leaves Your Desktop?

PowerShell first leaves the project folder. It then removes the source files, .venv environment, and reviews.db projection.

That is a clean reset. Your broker data, local model, virtual environment, and analytical projection are removed from the workstation.

Nice Work!

Nice Work!

You did it! Your Windows workstation now runs a complete product-review pipeline that turns retained review events into operational insight.

You've learned how to:

  • Built a local Kafka-compatible event stream with Redpanda. The producer sends keyed JSON review events into the retained product-reviews topic.
  • Created five-event micro-batches for local enrichment through Ollama running Ministral 3. Persisted structured sentiment results in an idempotent SQLite projection. Committed consumer offsets only after each successful database write.
  • Built a Streamlit operational dashboard that refreshes every two seconds. The browser shows live sentiment counts after each committed batch. The latest summary adds context. The recommended action points to the next product decision.
  • Secret Mission: Added operator-controlled replay through a fresh consumer group. The replay clears only the SQLite projection before rebuilding it from retained Redpanda events. Every review_id stays unique during the rebuild.

Ready to quiz yourself?