AI-Powered Checkout Personalization

Introduction

30 Second Summary

Every time you shop online, the offers you see at checkout feel eerily relevant. Behind the scenes, real-time personalization engines analyze your behavior and decide which offer to show you in milliseconds.

In this project, you will build a personalization engine with four microservices that process user behavior in real-time and use an AI agent to pick the best offer for each customer. You'll containerize everything with Docker and deploy the system to AWS with a public demo URL.

What You'll Build

An event-driven platform with four FastAPI microservices communicating through Apache Kafka, storing data in PostgreSQL, caching in Redis, and making intelligent decisions via a LangChain-powered AI agent running on Groq's free Llama 3.3 70B model.

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

• A 4-service microservices architecture with event-driven communication, real-time feature computation, and an ML pipeline that builds user profiles from behavior streams.
• An agentic AI system that reasons about offer selection using multiple tools, then explains its decisions in plain language.
• A containerized deployment on AWS EC2 free tier where anyone can interact with the checkout personalization system via a public URL.
• Secret Mission: an A/B testing framework that compares the AI agent's decisions against random selection with statistical significance.

Are there any prerequisites?

This project assumes familiarity with Python, REST APIs, and basic database concepts. Theoretical knowledge of Kafka, Redis, and Docker is helpful but not required.

Before We Start

You're about to build a production-grade personalization engine made up of multiple microservices, event streams, and an AI agent. Before any code gets written, you need a solid foundation of tools on your machine.

In this step, you'll confirm your Python installation, grab your Groq API key, install Docker Desktop and Git, and create the project directory. Once this is done, every tool you need for the rest of the project will be ready.

In this step, get ready to:

• Confirm Python 3.11+ is installed and get your Groq API key ready.
• Install Docker Desktop and verify it runs correctly.
• Install Git, configure it, and create your project repository.

Confirm your Python version and Groq API key

This project uses Python 3.11+ for all four microservices. You need to confirm your version before moving on.

• Open your terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
• Check your Python version by running this command:

macOS

python3 --version

Windows

python --version

You should see a version number like Python 3.11.x or higher.

✔️ I see version 3.11 or higher

Your Python installation is ready. Move on to getting your Groq API key below.

ⓧ I see an older version

Your system has an older Python version that won't work with this project's dependencies.

• Download Python 3.11 or newer from python.org/downloads.
• Run the installer and make sure to check the option to add Python to your PATH.
• Close and reopen your terminal, then run the version command again to confirm.

macOS default Python

macOS may ship with an older Python version. After installing a newer version, you might need to use python3.11 or python3.12 directly instead of python3. Alternatively, use a version manager like pyenv to set the default.

ⓧ Command not found

Python is not installed on your machine yet.

• Download Python 3.11 or newer from python.org/downloads.
• Run the installer. On Windows, check Add Python to PATH before clicking Install Now.
• Close and reopen your terminal, then run the version command again to confirm.

Next, you need your Groq API key. This key authenticates your microservices with Groq's LLM inference API so the AI agent can reason about personalization decisions.

• Open your browser and navigate to console.groq.com/keys.
• Log in with your existing Groq account.
• Click Create API Key.
• Give your key a name like smart-checkout and confirm.
• Copy the key and save it somewhere secure (a password manager or a local note). You will not be able to see the full key again after this page.

Why save the key now?

Groq only shows your full API key once at creation time. If you lose it, you'll need to create a new one. You'll use this key in a later step when configuring your AI agent service.

What is Groq?

Groq provides ultra-fast LLM inference on their custom hardware. The free tier gives you access to models like Llama 3.3 70B with no credit card required. If you exceed the rate limits, you'll get a 429 error (not a bill).

Install Docker Desktop

Your personalization engine runs 7+ containers locally (microservices, Kafka, PostgreSQL, Redis). Docker Desktop packages everything you need to build and run containers on your machine.

• Check if Docker is already installed by running this command in your terminal:

docker --version

✔️ I see a version number

Docker is installed. Make sure Docker Desktop is running (look for the whale icon in your system tray or menu bar).

• If Docker Desktop is not running, open it. Press Cmd+Space (macOS) or the Windows key (Windows), type Docker Desktop, and press Enter.

ⓧ Command not found

Docker Desktop is not installed. Follow the steps below to install it.

• Open docs.docker.com/get-started/get-docker in your browser.
• Select your operating system (macOS, Windows, or Linux) and download the installer.
• Run the installer and follow the on-screen prompts. Accept the default settings.
• Once installed, launch Docker Desktop. On first launch, accept the Docker Subscription Service Agreement to proceed.
• Wait for the Docker engine to start. You'll see a green indicator in the Docker Desktop window when it's ready.
• Close and reopen your terminal, then run docker --version again to confirm.

Docker Desktop won't start on Windows?

Docker Desktop on Windows requires WSL 2. If you see an error about WSL, open PowerShell as Administrator and run wsl --install. Restart your machine, then try launching Docker Desktop again.

Help me fix Docker Desktop startup issues on Windows.

What is Docker Desktop?

Docker Desktop is an application that lets you build and run containers on your local machine. It bundles the Docker engine, Docker Compose, and a GUI for managing containers. You'll use Docker Compose to start all 7+ services with a single command.

Install Git and create the project directory

Git tracks every change you make to your code. You'll use it throughout this project to commit progress and eventually push to a repository for deployment.

• Check if Git is already installed by running:

git --version

✔️ I see a version number

Git is ready. Skip ahead to configuring it below.

ⓧ Command not found

Git is not installed. Install it for your platform:

macOS

• Install Git using Homebrew by running:

brew install git

If you don't have Homebrew, you can also install Git via Xcode Command Line Tools:

xcode-select --install

Windows

• Install Git using winget by running this in PowerShell:

winget install --id Git.Git -e --source winget

Alternatively, download the installer from git-scm.com/download/win and run it with default settings.

• Close and reopen your terminal after installing.
• Run git --version again to confirm.

Now configure Git with your identity. Every commit you make will include this information.

• Set your Git username by running:

git config --global user.name [[YOURNAME="your full name"]]

• Set your Git email by running:

git config --global user.email [[YOUREMAIL="your email address"]]

Why configure Git identity?

Git attaches your name and email to every commit. This is required for proper attribution and matters when you push code to GitHub or other platforms later.

Finally, create your project directory and initialize it as a Git repository.

• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir smart-checkout
cd smart-checkout

• Initialize a Git repository by running:

git init

You should see a message like Initialized empty Git repository in .../smart-checkout/.git/.

• Confirm the folder exists by running:

ls

You won't see any files listed yet (the .git folder is hidden). That's expected for a fresh repository.

Don't see the initialized message?

Make sure you ran cd smart-checkout before running git init. The init command should run inside your project folder, not on the Desktop.

Help me troubleshoot my Git init.

Now open your project in VS Code.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type VS Code and press Enter to open it.
• In VS Code, click File in the top menu bar.
• Click Open Folder.
• Navigate to your Desktop and select the smart-checkout folder, then click Open.

VS Code now shows your empty smart-checkout project. You're ready to start building.

Your development environment is fully ready. Next up, you'll define the infrastructure layer with Docker Compose, spinning up PostgreSQL, Redis, and Kafka in containers.

Set Up Your Environment

Your project directory exists, but right now it's just an empty folder with a Git repository. A microservices project needs a clear organizational structure where each service lives in its own directory with its own dependencies.

In this step, you'll create that structure. You'll organize the project into service directories, configure your environment secrets, and scaffold the dependency files each service will need when you build them.

In this step, get ready to:

• Create the microservices directory structure with separate folders for each service.
• Set up environment configuration and Git ignore rules.
• Scaffold requirements files for all four services.

Create the project directory structure

Each microservice in this project gets its own folder with its own code and dependencies. This isolation is the core principle of microservices architecture. You'll also create folders for shared infrastructure configuration and the demo frontend.

• In VS Code, open the smart-checkout folder by clicking File in the top menu bar.
• Click Open Folder.
• Select the smart-checkout folder on your Desktop, then click Open.

• Open the integrated terminal by pressing Ctrl+` (macOS) or Ctrl+` (Windows).
• Create the service directories and shared folders by running these commands:

mkdir -p services/event-ingestion services/profile-builder services/offer-agent services/api-gateway infra frontend

What does this command do?

The -p flag creates parent directories as needed. This single command creates six folders:

• services/event-ingestion receives user behavior events via REST endpoints.
• services/profile-builder consumes events from Kafka and computes real-time user features.
• services/offer-agent uses LangChain to make intelligent offer decisions.
• services/api-gateway acts as the single entry point for all client requests.
• infra/ holds shared infrastructure config like Docker Compose and database schemas.
• frontend/ holds the interactive demo checkout page.

• Confirm the directories were created by running:

ls services

You should see api-gateway, event-ingestion, offer-agent, and profile-builder listed in the output.

Don't see all four directories?

Make sure you ran the mkdir command from the root of smart-checkout. Run pwd to check your current directory. If you're somewhere else, run cd ~/Desktop/smart-checkout and try again.

Help me troubleshoot my directory structure.

Configure your environment files

Your project needs two configuration files at the root level. The .env file stores your Groq API key so your services can authenticate with the LLM. The .gitignore file keeps secrets and generated files out of version control.

• Create the .env file by running this command:

echo "GROQ_API_KEY=your-api-key-here" > .env

• Open the newly created .env file in the VS Code Explorer sidebar (you may need to click the refresh icon to see it).
• Replace your-api-key-here with the actual Groq API key from your Groq account.
• Save the file.

Why not use a variable placeholder for the API key?

API keys are sensitive credentials. You should never store them in version control or share them. The .gitignore file you'll create next ensures .env never gets committed to your Git repository.

• Create the .gitignore file in the root of smart-checkout with the following content:

__pycache__/
*.pyc
.env
*.egg-info
.venv/
dist/
build/

What does this file do?

• __pycache__/ and *.pyc are Python bytecode files generated when you run scripts.
• .env keeps your API key out of Git history.
• .venv/, dist/, and build/ are generated directories from Python packaging tools.

• Save .gitignore.
• Verify Git respects your ignore rules by running:

git status

You should see .gitignore listed as an untracked file, but .env should NOT appear. That confirms your secrets are hidden from Git.

Seeing .env in git status?

Make sure your .gitignore file is saved and that the line .env appears exactly as written (no leading spaces or extra characters). If you already staged .env before creating the ignore file, run git rm --cached .env to unstage it.

Help me fix my .gitignore so .env is hidden.

Scaffold service dependencies

Each service has its own requirements.txt file listing the Python packages it needs. This keeps services decoupled. The API gateway doesn't need Kafka libraries, and the profile builder doesn't need web server packages.

• Create the Event Ingestion service's dependencies by running:

cat > services/event-ingestion/requirements.txt << EOF
fastapi
uvicorn
aiokafka
pydantic
python-dotenv
EOF

• Create the Profile Builder service's dependencies by running:

cat > services/profile-builder/requirements.txt << EOF
aiokafka
asyncpg
redis
pydantic
python-dotenv
EOF

• Create the Offer Agent service's dependencies by running:

cat > services/offer-agent/requirements.txt << EOF
fastapi
uvicorn
langchain
langchain-groq
asyncpg
redis
pydantic
python-dotenv
EOF

• Create the API Gateway service's dependencies by running:

cat > services/api-gateway/requirements.txt << EOF
fastapi
uvicorn
httpx
pydantic
python-dotenv
EOF

Why different dependencies per service?

This is a key microservices principle: each service installs only what it needs. The event-ingestion service uses aiokafka to publish events but never talks to PostgreSQL directly. The offer-agent needs langchain and langchain-groq for AI reasoning, but doesn't produce Kafka messages.

Shared across all services: pydantic for data validation and python-dotenv for loading environment variables from .env.

• Verify all four requirements files were created by running:

ls services/*/requirements.txt

You should see all four paths listed: services/api-gateway/requirements.txt, services/event-ingestion/requirements.txt, services/offer-agent/requirements.txt, and services/profile-builder/requirements.txt.

Missing one or more requirements files?

If a file is missing, run the corresponding cat > command again. Make sure you type EOF on its own line at the end. If you see a > prompt waiting for more input, type EOF and press Enter to close the heredoc.

Help me create my requirements files correctly.

✔️ Awesome, I've got everything!

Great. Double check you've saved all your files. Your project structure is ready to go.

ⓧ I'd like to double check the full code

Here is what each file should contain:

__pycache__/
*.pyc
.env
*.egg-info
.venv/
dist/
build/

GROQ_API_KEY=your-api-key-here

fastapi
uvicorn
aiokafka
pydantic
python-dotenv

aiokafka
asyncpg
redis
pydantic
python-dotenv

fastapi
uvicorn
langchain
langchain-groq
asyncpg
redis
pydantic
python-dotenv

fastapi
uvicorn
httpx
pydantic
python-dotenv

Your project skeleton is in place. You have six directories, secure environment configuration, and dependency manifests ready for each service. Next up, you'll build the infrastructure layer with Docker Compose to run Kafka, PostgreSQL, and Redis locally.

Build the Infrastructure Layer

Your microservices need three pieces of infrastructure to function: a database for persistent storage, a cache for fast lookups, and a message broker for real-time event streaming.

In this step, you will define all three as containers in a single Docker Compose file. One command will spin up PostgreSQL, Redis, and Apache Kafka together on a shared network.

In this step, get ready to:

• Create a Docker Compose file that defines PostgreSQL, Redis, and Kafka.
• Write a SQL initialization script that sets up your database schema.
• Verify all infrastructure containers start and pass health checks.

Create the Docker Compose file

Docker Compose lets you define multiple containers, their configuration, and how they connect to each other in a single YAML file. Your personalization engine needs three infrastructure services running before any application code can work.

• In VS Code, create a new file called docker-compose.yml in the root of your smart-checkout folder.
• Paste the following configuration into docker-compose.yml:

services:
  # PostgreSQL: persistent storage for user profiles, offers, and events
  postgres:
    image: postgres:15
    container_name: checkout-postgres
    environment:
      POSTGRES_DB: checkout_db
      POSTGRES_USER: checkout_user
      POSTGRES_PASSWORD: checkout_pass
    ports:
      - "5432:5432"
    volumes:
      - ./infra/init.sql:/docker-entrypoint-initdb.d/init.sql
    networks:
      - checkout-net
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U checkout_user -d checkout_db"]
      interval: 10s
      timeout: 5s
      retries: 5

  # Redis: in-memory cache for fast profile lookups
  redis:
    image: redis:7-alpine
    container_name: checkout-redis
    ports:
      - "6379:6379"
    networks:
      - checkout-net
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5

  # Apache Kafka: event streaming in KRaft mode (no Zookeeper)
  kafka:
    image: apache/kafka:3.7.0
    container_name: checkout-kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_NUM_PARTITIONS: 3
    networks:
      - checkout-net
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092"]
      interval: 15s
      timeout: 10s
      retries: 5
      start_period: 30s

networks:
  checkout-net:
    driver: bridge

What does this configuration do?

This file defines three services that will run as Docker containers on a shared network called checkout-net.

• postgres runs PostgreSQL 15 with a pre-created database (checkout_db). The volumes line mounts your SQL init script so the database schema is created automatically on first startup.
• redis runs Redis 7 on Alpine Linux. The healthcheck uses redis-cli ping to confirm the cache is responsive.
• kafka runs in KRaft mode, which eliminates the need for a separate Zookeeper container. The KAFKA_PROCESS_ROLES setting makes this single node act as both broker and controller.

Why KRaft mode instead of Zookeeper?

Traditional Kafka deployments required a separate Zookeeper cluster for metadata management. KRaft (Kafka Raft) removes that dependency entirely, reducing your container count and simplifying local development.

KRaft is the recommended approach for all new Kafka deployments since version 3.3. Companies like Rokt and Netflix use Kafka for real-time event streams in their personalization systems.

• Save docker-compose.yml.

YAML syntax errors?

YAML is whitespace-sensitive. Make sure you use spaces (not tabs) for indentation, and that each nested level is indented by exactly 2 spaces.

Help me fix a YAML indentation error in my docker-compose.yml.

Write the database initialization script

PostgreSQL automatically runs any .sql files mounted into /docker-entrypoint-initdb.d/ when the container starts for the first time. This is where you define your tables and seed data.

• Create a new file called init.sql inside the infra/ directory.
• Paste the following SQL into infra/init.sql:

-- User profiles table: stores computed ML features per user
CREATE TABLE user_profiles (
    user_id VARCHAR PRIMARY KEY,
    purchase_frequency FLOAT,
    avg_cart_value FLOAT,
    category_affinity JSONB,
    recency_score FLOAT,
    session_depth INT,
    updated_at TIMESTAMP DEFAULT NOW()
);

-- Offers catalog: available personalization offers
CREATE TABLE offers_catalog (
    offer_id VARCHAR PRIMARY KEY,
    name VARCHAR,
    discount_pct FLOAT,
    category VARCHAR,
    min_cart_value FLOAT,
    active BOOLEAN DEFAULT TRUE
);

-- Events table: raw user behavior events
CREATE TABLE events (
    event_id SERIAL PRIMARY KEY,
    user_id VARCHAR,
    event_type VARCHAR,
    payload JSONB,
    created_at TIMESTAMP DEFAULT NOW()
);

-- Seed the offers catalog with sample data
INSERT INTO offers_catalog VALUES
    ('OFF001', '10% Off Electronics', 10, 'electronics', 50, true),
    ('OFF002', 'Free Shipping', 0, 'all', 0, true),
    ('OFF003', '15% Off Apparel', 15, 'apparel', 30, true),
    ('OFF004', '$5 Off Cart', 5, 'all', 40, true);

What do these tables store?

• user_profiles holds computed ML features like purchase frequency, average cart value, and category preferences. The Profile Builder service will write to this table.
• offers_catalog contains the available offers that the AI agent can choose from. The seed data gives you four realistic offers to work with immediately.
• events stores raw user behavior events (page views, add-to-cart actions, checkout starts). The JSONB payload column lets each event type carry different data without schema changes.

• Save infra/init.sql.

✔️ Awesome, I've got everything!

Great. Double check you have saved both docker-compose.yml and infra/init.sql.

ⓧ I'd like to double check the full code

Here is the complete docker-compose.yml:

services:
  # PostgreSQL: persistent storage for user profiles, offers, and events
  postgres:
    image: postgres:15
    container_name: checkout-postgres
    environment:
      POSTGRES_DB: checkout_db
      POSTGRES_USER: checkout_user
      POSTGRES_PASSWORD: checkout_pass
    ports:
      - "5432:5432"
    volumes:
      - ./infra/init.sql:/docker-entrypoint-initdb.d/init.sql
    networks:
      - checkout-net
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U checkout_user -d checkout_db"]
      interval: 10s
      timeout: 5s
      retries: 5

  # Redis: in-memory cache for fast profile lookups
  redis:
    image: redis:7-alpine
    container_name: checkout-redis
    ports:
      - "6379:6379"
    networks:
      - checkout-net
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5

  # Apache Kafka: event streaming in KRaft mode (no Zookeeper)
  kafka:
    image: apache/kafka:3.7.0
    container_name: checkout-kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_NUM_PARTITIONS: 3
    networks:
      - checkout-net
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092"]
      interval: 15s
      timeout: 10s
      retries: 5
      start_period: 30s

networks:
  checkout-net:
    driver: bridge

And here is the complete infra/init.sql:

-- User profiles table: stores computed ML features per user
CREATE TABLE user_profiles (
    user_id VARCHAR PRIMARY KEY,
    purchase_frequency FLOAT,
    avg_cart_value FLOAT,
    category_affinity JSONB,
    recency_score FLOAT,
    session_depth INT,
    updated_at TIMESTAMP DEFAULT NOW()
);

-- Offers catalog: available personalization offers
CREATE TABLE offers_catalog (
    offer_id VARCHAR PRIMARY KEY,
    name VARCHAR,
    discount_pct FLOAT,
    category VARCHAR,
    min_cart_value FLOAT,
    active BOOLEAN DEFAULT TRUE
);

-- Events table: raw user behavior events
CREATE TABLE events (
    event_id SERIAL PRIMARY KEY,
    user_id VARCHAR,
    event_type VARCHAR,
    payload JSONB,
    created_at TIMESTAMP DEFAULT NOW()
);

-- Seed the offers catalog with sample data
INSERT INTO offers_catalog VALUES
    ('OFF001', '10% Off Electronics', 10, 'electronics', 50, true),
    ('OFF002', 'Free Shipping', 0, 'all', 0, true),
    ('OFF003', '15% Off Apparel', 15, 'apparel', 30, true),
    ('OFF004', '$5 Off Cart', 5, 'all', 40, true);

Verify all infrastructure containers start correctly

Time to bring the entire infrastructure stack online. Docker Compose will pull the images (if not already cached), create the network, and start all three containers.

• In your VS Code terminal, make sure you are in the smart-checkout folder.
• Start all infrastructure containers by running:

docker compose up -d

What does this command do?

The -d flag runs containers in detached mode (in the background). Docker Compose reads your docker-compose.yml, pulls any images it does not have locally, creates the checkout-net network, and starts all three services.

The first run will take a minute or two while Docker downloads the PostgreSQL, Redis, and Kafka images. Subsequent runs will be nearly instant.

• Check the status of your containers by running:

docker compose ps

You should see three containers listed: checkout-postgres, checkout-redis, and checkout-kafka. All three should show a status of Up with (healthy) next to them.

Kafka takes longer to become healthy

The Kafka container has a start_period: 30s in its healthcheck. If you run docker compose ps immediately, Kafka may show as (health: starting). Wait 30-45 seconds and check again.

• Verify PostgreSQL accepted the schema by running:

docker exec checkout-postgres psql -U checkout_user -d checkout_db -c "\dt"

You should see a table listing user_profiles, offers_catalog, and events.

• Verify the seed data loaded correctly by running:

docker exec checkout-postgres psql -U checkout_user -d checkout_db -c "SELECT * FROM offers_catalog;"

You should see four rows of offer data including 10% Off Electronics and Free Shipping.

• Verify Redis is responsive by running:

docker exec checkout-redis redis-cli ping

You should see PONG in the output.

• Verify Kafka is running by listing its topics:

docker exec checkout-kafka /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092

You should see an empty list or internal Kafka topics. No errors means the broker is accepting connections.

Container exited or won't start?

Check the container logs to see what went wrong:

docker compose logs postgres
docker compose logs kafka

• If Postgres shows a permission error on init.sql, make sure the file exists at infra/init.sql relative to your docker-compose.yml.
• If Kafka keeps restarting, ensure Docker Desktop has at least 4 GB of memory allocated in its settings.
• If port 5432 or 6379 is already in use, stop any local PostgreSQL or Redis instances first.

Help me debug why my Docker Compose containers are not starting.

Your infrastructure layer is live. PostgreSQL, Redis, and Kafka are all running on a shared Docker network, ready to accept connections from your microservices. Next up, you will build the first microservice that ingests user behavior events into Kafka.

Build the Event Ingestion Service

Your infrastructure layer is up and running. Kafka, PostgreSQL, and Redis are all healthy and communicating inside Docker Compose.

Now you need a way to get data into the system. The Event Ingestion Service is the front door of the personalization engine. It receives user behavior events (page views, items added to cart, checkout starts) and publishes them to Kafka for downstream processing.

In this step, get ready to:

• Build a FastAPI microservice that accepts user behavior events via a REST endpoint.
• Implement a Kafka producer that publishes events to a user-events topic.
• Containerize the service and wire it into Docker Compose.

Define the event models

Every event entering the system needs a consistent shape. Pydantic models enforce this structure at the API boundary, rejecting malformed data before it ever touches Kafka.

• Create the service directory structure by running these commands:

mkdir -p services/event-ingestion
cd services/event-ingestion

• Confirm the directory exists by running:

ls services/event-ingestion

You should see an empty directory listed.

• Create a new file called models.py inside services/event-ingestion/.
• Add the Pydantic models by pasting this code into models.py:

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Literal

# Defines the shape of incoming event data
class EventPayload(BaseModel):
    user_id: str
    event_type: Literal['page_view', 'add_to_cart', 'checkout_start']
    payload: dict
    timestamp: datetime = Field(default_factory=datetime.now)

# Defines the shape of the API response
class EventResponse(BaseModel):
    event_id: str
    status: str

What does this code do?

EventPayload defines the exact shape of every event entering the system. The Literal type restricts event_type to only three valid values. Any request with a different event type gets rejected automatically by FastAPI.

EventResponse defines what the API sends back. The caller gets a unique event_id and a status confirming the event was accepted.

• Save models.py.

• Create a requirements.txt file in the same services/event-ingestion/ folder and paste these dependencies:

fastapi
uvicorn
aiokafka
pydantic

What are these packages?

• fastapi is the async web framework for your microservice endpoints.
• uvicorn is the ASGI server that runs the FastAPI app.
• aiokafka is the async Kafka client for producing messages.
• pydantic provides the data validation models.

• Save requirements.txt.

Don't see the files?

Make sure you created both files inside services/event-ingestion/, not in the project root. Check the Explorer sidebar in VS Code to confirm the folder structure.

Help me debug my project folder structure.

Build the FastAPI service with Kafka producer

The main application file connects two pieces together. FastAPI handles incoming HTTP requests, and an AIOKafkaProducer publishes each event to Kafka asynchronously.

• Create a file called main.py inside services/event-ingestion/.
• Paste the application setup and Kafka producer initialization:

import os
import json
import uuid
import logging
from datetime import datetime
from contextlib import asynccontextmanager

from fastapi import FastAPI
from aiokafka import AIOKafkaProducer

from models import EventPayload, EventResponse

# Configure structured logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("event-ingestion")

# Read Kafka broker address from environment variable
KAFKA_BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")

# Global producer instance managed by the lifespan
producer: AIOKafkaProducer = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Start the Kafka producer on app startup, stop it on shutdown."""
    global producer
    producer = AIOKafkaProducer(
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        value_serializer=lambda v: json.dumps(v).encode("utf-8")
    )
    await producer.start()
    logger.info("Kafka producer connected to %s", KAFKA_BOOTSTRAP_SERVERS)
    yield
    await producer.stop()
    logger.info("Kafka producer stopped")

app = FastAPI(title="Event Ingestion Service", lifespan=lifespan)

What does this code do?

• The lifespan context manager starts the Kafka producer before the app serves any requests. It stops the producer cleanly when the app shuts down.
• KAFKA_BOOTSTRAP_SERVERS reads the broker address from an environment variable. Inside Docker Compose, this resolves to the Kafka container's hostname.
• The value_serializer converts Python dicts to JSON bytes automatically before sending to Kafka.

Why use a lifespan context manager?

The Kafka producer needs a persistent connection to the broker. Creating it once at startup (before any requests arrive) and reusing it across all requests avoids the overhead of connecting on every single event. The lifespan pattern is FastAPI's recommended approach for managing shared resources.

• Now add the route handlers below the app = FastAPI(...) line in the same file:

@app.post("/events", response_model=EventResponse)
async def ingest_event(event: EventPayload):
    """Receive a user behavior event and publish it to Kafka."""
    event_id = str(uuid.uuid4())

    # Build the event message for Kafka
    event_data = {
        "event_id": event_id,
        "user_id": event.user_id,
        "event_type": event.event_type,
        "payload": event.payload,
        "timestamp": event.timestamp.isoformat()
    }

    # Publish to the user-events topic
    await producer.send_and_wait("user-events", value=event_data)

    logger.info("Event received: %s from user %s", event.event_type, event.user_id)

    return EventResponse(event_id=event_id, status="accepted")

@app.get("/health")
async def health_check():
    """Simple health check endpoint."""
    return {"status": "ok"}

What does this code do?

• The POST /events endpoint accepts an EventPayload, generates a unique ID, serializes the event, and publishes it to the user-events Kafka topic.
• send_and_wait ensures the message is acknowledged by Kafka before responding to the client. This guarantees no silent message loss.
• The GET /health endpoint lets Docker Compose verify the service is alive.

• Save main.py.

You won't be able to run this file directly yet. It needs Kafka running, which you will get via Docker Compose in the next substep.

✔️ Awesome, I've got everything!

Great. Double check you saved main.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete main.py file:

import os
import json
import uuid
import logging
from datetime import datetime
from contextlib import asynccontextmanager

from fastapi import FastAPI
from aiokafka import AIOKafkaProducer

from models import EventPayload, EventResponse

# Configure structured logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("event-ingestion")

# Read Kafka broker address from environment variable
KAFKA_BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")

# Global producer instance managed by the lifespan
producer: AIOKafkaProducer = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Start the Kafka producer on app startup, stop it on shutdown."""
    global producer
    producer = AIOKafkaProducer(
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        value_serializer=lambda v: json.dumps(v).encode("utf-8")
    )
    await producer.start()
    logger.info("Kafka producer connected to %s", KAFKA_BOOTSTRAP_SERVERS)
    yield
    await producer.stop()
    logger.info("Kafka producer stopped")

app = FastAPI(title="Event Ingestion Service", lifespan=lifespan)

@app.post("/events", response_model=EventResponse)
async def ingest_event(event: EventPayload):
    """Receive a user behavior event and publish it to Kafka."""
    event_id = str(uuid.uuid4())

    # Build the event message for Kafka
    event_data = {
        "event_id": event_id,
        "user_id": event.user_id,
        "event_type": event.event_type,
        "payload": event.payload,
        "timestamp": event.timestamp.isoformat()
    }

    # Publish to the user-events topic
    await producer.send_and_wait("user-events", value=event_data)

    logger.info("Event received: %s from user %s", event.event_type, event.user_id)

    return EventResponse(event_id=event_id, status="accepted")

@app.get("/health")
async def health_check():
    """Simple health check endpoint."""
    return {"status": "ok"}

Seeing import errors in VS Code?

VS Code may underline aiokafka or fastapi with red squiggles. This is expected because you have not installed these packages locally. They will be installed inside the Docker container when you build it in the next substep.

Help me understand why VS Code shows import errors for packages I haven't installed locally.

Containerize and wire into Docker Compose

The service needs its own Docker image so it can run alongside Kafka and the other infrastructure containers. You will create a Dockerfile and then update docker-compose.yml to include the new service.

• Create a file called Dockerfile inside services/event-ingestion/ and paste this:

FROM python:3.11-slim

WORKDIR /app

# Install Python dependencies
COPY requirements.txt .
RUN pip install -r requirements.txt

# Copy application code
COPY . .

# Run the FastAPI app on port 8001
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8001"]

What does this Dockerfile do?

• Starts from the official python:3.11-slim image (small footprint, production-friendly).
• Copies and installs dependencies first (Docker caches this layer so rebuilds are fast when only code changes).
• Runs uvicorn on port 8001 to serve the FastAPI app.

• Save Dockerfile.

• Switch back to docker-compose.yml in your project root.
• Add the event-ingestion service definition below your existing kafka service block:

  event-ingestion:
    build: ./services/event-ingestion
    ports:
      - "8001:8001"
    environment:
      - KAFKA_BOOTSTRAP_SERVERS=kafka:9092
    depends_on:
      kafka:
        condition: service_healthy
    networks:
      - checkout-net

What does this service definition do?

• build points Docker Compose to the Dockerfile in ./services/event-ingestion.
• KAFKA_BOOTSTRAP_SERVERS=kafka:9092 tells the service how to reach Kafka. Inside Docker Compose, containers use service names as hostnames.
• depends_on with service_healthy ensures the event-ingestion service only starts after Kafka passes its health check.

• Save docker-compose.yml.
• Rebuild and start the full stack by running:

docker compose up --build -d

• Wait about 30 seconds for Kafka to become healthy, then check that the event-ingestion service is running:

docker compose ps

You should see the event-ingestion service listed with a status of Up or Running.

• Test the service by sending a sample event:

curl -X POST http://localhost:8001/events \
  -H "Content-Type: application/json" \
  -d '{"user_id": "user_123", "event_type": "page_view", "payload": {"page": "/products/shoes"}}'

You should see a JSON response with an event_id and "status": "accepted".

• Verify the health endpoint works by running:

curl http://localhost:8001/health

You should see {"status":"ok"} in the response.

Service won't start or curl fails?

• Run docker compose logs event-ingestion to see the service logs. Look for connection errors to Kafka.
• If you see a KafkaConnectionError, Kafka may still be starting. Wait 30 more seconds and check docker compose ps to confirm the kafka container is healthy.
• If the build fails, check that requirements.txt is in the same folder as Dockerfile (both inside services/event-ingestion/).

Help me debug my event-ingestion service that won't connect to Kafka in Docker Compose.

Your first microservice is live and publishing events to Kafka. Next up, you will build the Profile Builder service that consumes these events and computes real-time user features.

Build the Profile Builder Service

Your Event Ingestion service is now publishing raw user behavior events to Apache Kafka. Raw events alone are not useful for personalization decisions.

In this step, you will build the Profile Builder service. This service consumes events from Kafka and transforms them into rich user profiles with computed features like purchase frequency, cart value, and category affinity. These profiles get stored in PostgreSQL and cached in Redis for low-latency retrieval.

In this step, get ready to:

• Build a Kafka consumer that reads from the user-events topic and computes real-time user features.
• Store computed profiles in PostgreSQL and cache them in Redis with a 300-second TTL.
• Add a batch recomputation job and containerize the service with Docker.

Create the feature computation module

The feature computation module is the brain of your Profile Builder. It takes raw events and distills them into numeric signals the AI agent can reason about.

• Create a new folder for the service by running these commands:

mkdir -p services/profile-builder
touch services/profile-builder/features.py

• Confirm the files exist by running:

ls services/profile-builder/

You should see features.py listed in the output.

• In VS Code, open services/profile-builder/features.py and add the following code:

import json
from datetime import datetime, timezone


async def compute_features(user_id: str, event: dict, db_conn) -> dict:
    """Compute real-time features for a user from their event history."""
    # Query all events for this user from the events table
    rows = await db_conn.fetch(
        "SELECT event_type, payload, timestamp FROM events WHERE user_id = $1 ORDER BY timestamp",
        user_id
    )

    if not rows:
        return {
            "user_id": user_id,
            "purchase_frequency": 0.0,
            "avg_cart_value": 0.0,
            "category_affinity": {},
            "recency_score": 0.0,
            "session_depth": 0
        }

    # Purchase frequency: checkout_start count / days active
    checkout_count = sum(1 for r in rows if r["event_type"] == "checkout_start")
    timestamps = [r["timestamp"] for r in rows]
    days_active = max((max(timestamps) - min(timestamps)).days, 1)
    purchase_frequency = checkout_count / days_active

    # Average cart value: mean of add_to_cart payload.cart_value
    cart_values = []
    for r in rows:
        if r["event_type"] == "add_to_cart":
            payload = json.loads(r["payload"]) if isinstance(r["payload"], str) else r["payload"]
            if "cart_value" in payload:
                cart_values.append(float(payload["cart_value"]))
    avg_cart_value = sum(cart_values) / len(cart_values) if cart_values else 0.0

    # Category affinity: normalized score from page_view events
    category_counts = {}
    for r in rows:
        if r["event_type"] == "page_view":
            payload = json.loads(r["payload"]) if isinstance(r["payload"], str) else r["payload"]
            category = payload.get("category", "unknown")
            category_counts[category] = category_counts.get(category, 0) + 1
    total_views = sum(category_counts.values()) if category_counts else 1
    category_affinity = {k: v / total_views for k, v in category_counts.items()}

    # Recency score: 1 / (1 + days since last event)
    last_event_time = max(timestamps)
    days_since_last = (datetime.now(timezone.utc) - last_event_time).days
    recency_score = 1.0 / (1.0 + days_since_last)

    # Session depth: unique session count
    session_ids = set()
    for r in rows:
        payload = json.loads(r["payload"]) if isinstance(r["payload"], str) else r["payload"]
        if "session_id" in payload:
            session_ids.add(payload["session_id"])
    session_depth = len(session_ids) if session_ids else 1

    return {
        "user_id": user_id,
        "purchase_frequency": round(purchase_frequency, 4),
        "avg_cart_value": round(avg_cart_value, 2),
        "category_affinity": category_affinity,
        "recency_score": round(recency_score, 4),
        "session_depth": session_depth
    }

What does this code do?

The compute_features function queries all of a user's historical events from PostgreSQL, then derives five real-time signals:

• purchase_frequency: how often the user starts a checkout relative to their active days.
• avg_cart_value: the average dollar value of items added to cart.
• category_affinity: a normalized dictionary showing which product categories the user browses most.
• recency_score: a decay signal that is higher when the user was active recently.
• session_depth: how many unique sessions the user has had, indicating engagement depth.

• Save features.py.

You will not see visible output yet. The next chunk connects this module to the Kafka consumer.

Seeing a syntax error?

Make sure the return statement at the end of the function is indented at the same level as the if not rows block above it. Python is sensitive to whitespace alignment.

help me fix a syntax error in my features.py compute_features function

Build the Kafka consumer with PostgreSQL and Redis storage

Now you need a main loop that subscribes to the Kafka topic, calls your feature computation on each event, and persists the result to both PostgreSQL and Redis.

• Open services/profile-builder/main.py and add the imports, configuration, and helper functions:

import asyncio
import json
import os
import logging

import asyncpg
import redis.asyncio as redis
from aiokafka import AIOKafkaConsumer

from features import compute_features

# Configure structured logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

# Environment configuration
KAFKA_BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://checkout_user:checkout_pass@localhost:5432/checkout_db")
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
PROFILE_CACHE_TTL = 300  # 300 seconds TTL for cached profiles


async def upsert_profile(db_conn, profile: dict):
    """Upsert a computed user profile into PostgreSQL."""
    await db_conn.execute(
        """
        INSERT INTO user_profiles (user_id, purchase_frequency, avg_cart_value,
            category_affinity, recency_score, session_depth, updated_at)
        VALUES ($1, $2, $3, $4, $5, $6, NOW())
        ON CONFLICT (user_id) DO UPDATE SET
            purchase_frequency = EXCLUDED.purchase_frequency,
            avg_cart_value = EXCLUDED.avg_cart_value,
            category_affinity = EXCLUDED.category_affinity,
            recency_score = EXCLUDED.recency_score,
            session_depth = EXCLUDED.session_depth,
            updated_at = NOW()
        """,
        profile["user_id"],
        profile["purchase_frequency"],
        profile["avg_cart_value"],
        json.dumps(profile["category_affinity"]),
        profile["recency_score"],
        profile["session_depth"]
    )


async def cache_profile(redis_client, profile: dict):
    """Cache a user profile in Redis with TTL."""
    key = f"profile:{profile['user_id']}"
    await redis_client.set(key, json.dumps(profile), ex=PROFILE_CACHE_TTL)

What does this code do?

The top section sets up environment-based configuration so the service adapts whether running locally or inside Docker. The PROFILE_CACHE_TTL constant sets Redis cache expiry to 300 seconds (5 minutes).

upsert_profile uses PostgreSQL's ON CONFLICT clause to insert a new profile or update an existing one. cache_profile stores the profile JSON in Redis with the key pattern profile:{user_id} and the 300-second TTL.

• Now add the batch recomputation job and main consumer loop below the helper functions in the same file:

async def batch_recompute(db_conn, redis_client):
    """Recompute all profiles from historical events."""
    logger.info("Starting batch recomputation of all user profiles")
    # Get all unique user IDs from the events table
    rows = await db_conn.fetch("SELECT DISTINCT user_id FROM events")
    for row in rows:
        user_id = row["user_id"]
        profile = await compute_features(user_id, {}, db_conn)
        await upsert_profile(db_conn, profile)
        await cache_profile(redis_client, profile)
        logger.info(f"Recomputed profile for user: {user_id}")
    logger.info(f"Batch recomputation complete. Processed {len(rows)} users.")


async def main():
    """Main entry point: consume events and build profiles."""
    logger.info("Profile Builder starting up...")

    # Connect to PostgreSQL
    db_conn = await asyncpg.connect(DATABASE_URL)
    logger.info("Connected to PostgreSQL")

    # Connect to Redis
    redis_client = redis.from_url(REDIS_URL)
    logger.info("Connected to Redis")

    # Create Kafka consumer
    consumer = AIOKafkaConsumer(
        "user-events",
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        group_id="profile-builder",
        value_deserializer=lambda m: json.loads(m.decode("utf-8"))
    )
    await consumer.start()
    logger.info("Kafka consumer started on topic 'user-events'")

    try:
        async for message in consumer:
            event = message.value
            user_id = event.get("user_id")
            logger.info(f"Processing event for user: {user_id}, type: {event.get('event_type')}")

            # Compute updated features for this user
            profile = await compute_features(user_id, event, db_conn)

            # Store in PostgreSQL
            await upsert_profile(db_conn, profile)

            # Cache in Redis with TTL
            await cache_profile(redis_client, profile)

            logger.info(f"Profile updated for user: {user_id}")
    finally:
        await consumer.stop()
        await db_conn.close()
        await redis_client.close()
        logger.info("Profile Builder shut down")


if __name__ == "__main__":
    asyncio.run(main())

What does this code do?

batch_recompute queries every unique user from the events table and recalculates their full profile from scratch. This is useful for backfilling profiles when the service first starts or after a schema change.

The main function connects to all three infrastructure services (PostgreSQL, Redis, Kafka), then enters an infinite loop consuming from the user-events topic. Each event triggers a full feature recomputation for that user. The group_id="profile-builder" ensures Kafka tracks this consumer's offset so it does not re-process events after a restart.

• Save main.py.

You will not see visible output yet. The service needs to run inside Docker to connect to Kafka and PostgreSQL.

Import error on features module?

If you see an import error for from features import compute_features, make sure features.py and main.py are both in the same services/profile-builder/ directory. The import uses a relative path within the container's working directory.

help me fix my import error in the profile-builder service

Containerize and integrate the service

With the application code ready, you need to package the service as a Docker container and wire it into your existing Compose stack.

• Create the requirements.txt for the Profile Builder by running:

cat > services/profile-builder/requirements.txt << 'EOF'
aiokafka
asyncpg
redis
EOF

• Create the Dockerfile at services/profile-builder/Dockerfile with the following content:

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["python", "main.py"]

Why python main.py instead of uvicorn?

Unlike the Event Ingestion service which is a web server (FastAPI + uvicorn), the Profile Builder is a long-running consumer process. It does not expose HTTP endpoints. It simply runs an asyncio event loop that continuously reads from Kafka, so python main.py is the correct entrypoint.

• Open docker-compose.yml and add the profile-builder service below the existing event-ingestion service definition:

  profile-builder:
    build: ./services/profile-builder
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy
      kafka:
        condition: service_healthy
    environment:
      - KAFKA_BOOTSTRAP_SERVERS=kafka:9092
      - DATABASE_URL=postgresql://checkout_user:checkout_pass@postgres:5432/checkout_db
      - REDIS_URL=redis://redis:6379
    networks:
      - checkout-net

What does this configuration do?

The depends_on with service_healthy conditions ensures the Profile Builder only starts after PostgreSQL, Redis, and Kafka have passed their health checks. The environment variables inject the connection strings that your Python code reads via os.getenv.

• Save docker-compose.yml.
• Rebuild and start the full stack by running:

docker compose up --build -d

• Check that the profile-builder container is running by running:

docker compose logs profile-builder --tail 20

You should see log lines showing the service connected to PostgreSQL and Redis, then started consuming from the user-events topic.

Profile Builder keeps restarting?

If the container exits immediately, check the logs for a connection error to PostgreSQL or Kafka. The most common cause is the depends_on conditions not matching the health check names in your docker-compose.yml. Verify the service names postgres, redis, and kafka match exactly.

help me debug why my profile-builder container keeps restarting

• Test the end-to-end flow by sending a test event through the Event Ingestion service:

curl -X POST http://localhost:8001/events \
  -H "Content-Type: application/json" \
  -d '{"user_id": "test-user-1", "event_type": "page_view", "payload": {"category": "electronics", "session_id": "sess-001"}}'

• Check the Profile Builder logs again to confirm it processed the event:

docker compose logs profile-builder --tail 5

You should see a log line containing Processing event for user: test-user-1 followed by Profile updated for user: test-user-1.

No log output after sending the event?

Make sure the Event Ingestion service is also running and healthy. Run docker compose ps to check all container statuses. The event must flow through Kafka from the ingestion service before the Profile Builder can consume it.

help me debug why my profile-builder is not receiving events from Kafka

✔️ Awesome, I've got everything!

Great. Double check you have saved both features.py and main.py in the services/profile-builder/ directory.

ⓧ I'd like to double check the full code

Here is the complete features.py:

import json
from datetime import datetime, timezone


async def compute_features(user_id: str, event: dict, db_conn) -> dict:
    """Compute real-time features for a user from their event history."""
    # Query all events for this user from the events table
    rows = await db_conn.fetch(
        "SELECT event_type, payload, timestamp FROM events WHERE user_id = $1 ORDER BY timestamp",
        user_id
    )

    if not rows:
        return {
            "user_id": user_id,
            "purchase_frequency": 0.0,
            "avg_cart_value": 0.0,
            "category_affinity": {},
            "recency_score": 0.0,
            "session_depth": 0
        }

    # Purchase frequency: checkout_start count / days active
    checkout_count = sum(1 for r in rows if r["event_type"] == "checkout_start")
    timestamps = [r["timestamp"] for r in rows]
    days_active = max((max(timestamps) - min(timestamps)).days, 1)
    purchase_frequency = checkout_count / days_active

    # Average cart value: mean of add_to_cart payload.cart_value
    cart_values = []
    for r in rows:
        if r["event_type"] == "add_to_cart":
            payload = json.loads(r["payload"]) if isinstance(r["payload"], str) else r["payload"]
            if "cart_value" in payload:
                cart_values.append(float(payload["cart_value"]))
    avg_cart_value = sum(cart_values) / len(cart_values) if cart_values else 0.0

    # Category affinity: normalized score from page_view events
    category_counts = {}
    for r in rows:
        if r["event_type"] == "page_view":
            payload = json.loads(r["payload"]) if isinstance(r["payload"], str) else r["payload"]
            category = payload.get("category", "unknown")
            category_counts[category] = category_counts.get(category, 0) + 1
    total_views = sum(category_counts.values()) if category_counts else 1
    category_affinity = {k: v / total_views for k, v in category_counts.items()}

    # Recency score: 1 / (1 + days since last event)
    last_event_time = max(timestamps)
    days_since_last = (datetime.now(timezone.utc) - last_event_time).days
    recency_score = 1.0 / (1.0 + days_since_last)

    # Session depth: unique session count
    session_ids = set()
    for r in rows:
        payload = json.loads(r["payload"]) if isinstance(r["payload"], str) else r["payload"]
        if "session_id" in payload:
            session_ids.add(payload["session_id"])
    session_depth = len(session_ids) if session_ids else 1

    return {
        "user_id": user_id,
        "purchase_frequency": round(purchase_frequency, 4),
        "avg_cart_value": round(avg_cart_value, 2),
        "category_affinity": category_affinity,
        "recency_score": round(recency_score, 4),
        "session_depth": session_depth
    }

Here is the complete main.py:

import asyncio
import json
import os
import logging

import asyncpg
import redis.asyncio as redis
from aiokafka import AIOKafkaConsumer

from features import compute_features

# Configure structured logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

# Environment configuration
KAFKA_BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://checkout_user:checkout_pass@localhost:5432/checkout_db")
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
PROFILE_CACHE_TTL = 300  # 300 seconds TTL for cached profiles


async def upsert_profile(db_conn, profile: dict):
    """Upsert a computed user profile into PostgreSQL."""
    await db_conn.execute(
        """
        INSERT INTO user_profiles (user_id, purchase_frequency, avg_cart_value,
            category_affinity, recency_score, session_depth, updated_at)
        VALUES ($1, $2, $3, $4, $5, $6, NOW())
        ON CONFLICT (user_id) DO UPDATE SET
            purchase_frequency = EXCLUDED.purchase_frequency,
            avg_cart_value = EXCLUDED.avg_cart_value,
            category_affinity = EXCLUDED.category_affinity,
            recency_score = EXCLUDED.recency_score,
            session_depth = EXCLUDED.session_depth,
            updated_at = NOW()
        """,
        profile["user_id"],
        profile["purchase_frequency"],
        profile["avg_cart_value"],
        json.dumps(profile["category_affinity"]),
        profile["recency_score"],
        profile["session_depth"]
    )


async def cache_profile(redis_client, profile: dict):
    """Cache a user profile in Redis with TTL."""
    key = f"profile:{profile['user_id']}"
    await redis_client.set(key, json.dumps(profile), ex=PROFILE_CACHE_TTL)


async def batch_recompute(db_conn, redis_client):
    """Recompute all profiles from historical events."""
    logger.info("Starting batch recomputation of all user profiles")
    # Get all unique user IDs from the events table
    rows = await db_conn.fetch("SELECT DISTINCT user_id FROM events")
    for row in rows:
        user_id = row["user_id"]
        profile = await compute_features(user_id, {}, db_conn)
        await upsert_profile(db_conn, profile)
        await cache_profile(redis_client, profile)
        logger.info(f"Recomputed profile for user: {user_id}")
    logger.info(f"Batch recomputation complete. Processed {len(rows)} users.")


async def main():
    """Main entry point: consume events and build profiles."""
    logger.info("Profile Builder starting up...")

    # Connect to PostgreSQL
    db_conn = await asyncpg.connect(DATABASE_URL)
    logger.info("Connected to PostgreSQL")

    # Connect to Redis
    redis_client = redis.from_url(REDIS_URL)
    logger.info("Connected to Redis")

    # Create Kafka consumer
    consumer = AIOKafkaConsumer(
        "user-events",
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        group_id="profile-builder",
        value_deserializer=lambda m: json.loads(m.decode("utf-8"))
    )
    await consumer.start()
    logger.info("Kafka consumer started on topic 'user-events'")

    try:
        async for message in consumer:
            event = message.value
            user_id = event.get("user_id")
            logger.info(f"Processing event for user: {user_id}, type: {event.get('event_type')}")

            # Compute updated features for this user
            profile = await compute_features(user_id, event, db_conn)

            # Store in PostgreSQL
            await upsert_profile(db_conn, profile)

            # Cache in Redis with TTL
            await cache_profile(redis_client, profile)

            logger.info(f"Profile updated for user: {user_id}")
    finally:
        await consumer.stop()
        await db_conn.close()
        await redis_client.close()
        logger.info("Profile Builder shut down")


if __name__ == "__main__":
    asyncio.run(main())

Here is the complete Dockerfile:

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["python", "main.py"]

Here is the complete requirements.txt:

aiokafka
asyncpg
redis

Your ML feature pipeline is now live. Every time a user event arrives, their profile gets recomputed and cached for instant access. Next up, you will build the AI agent that uses these profiles to make personalization decisions.

Build the Offer Decision Agent

Your Profile Builder is now computing real-time features from user behavior events. The next question is: what do you do with those features?

In this step, you will build the brain of the system. The Offer Decision Agent is a LangChain-powered AI agent that uses Groq's Llama 3.3 70B model to reason about which offer to show each customer at checkout. Instead of hard-coded rules, the agent calls tools to gather data, evaluates multiple candidates, and explains its decision.

In this step, get ready to:

• Implement four agent tools for data access and scoring.
• Build the LangChain agent with a system prompt and Groq LLM.
• Expose the agent via a FastAPI endpoint and containerize the service.

Implement the agent tools

The agent needs access to data and scoring logic to make informed decisions. In LangChain, you give an agent this access through tools. Each tool is a Python function decorated with @tool that the agent can call during its reasoning loop.

• Create the service directory by running these commands:

mkdir -p services/offer-agent
cd services/offer-agent

• Create a file called requirements.txt inside services/offer-agent/ with the following content:

fastapi
uvicorn
langchain
langchain-groq
psycopg2-binary
redis
pydantic

Why these packages?

The langchain package provides the agent framework (create_agent, @tool). The langchain-groq package provides ChatGroq, the LangChain integration for Groq's inference API.

The tools use psycopg2-binary for synchronous PostgreSQL access and redis for synchronous cache lookups.

• Create a file called tools.py inside services/offer-agent/ and add the following code:

import json
import psycopg2
import redis
from langchain.tools import tool

# Module-level connections initialized at startup
_db_conn = None
_redis_client = None


def init_tools(database_url: str, redis_url: str):
    """Initialize database and cache connections for tools."""
    global _db_conn, _redis_client
    _db_conn = psycopg2.connect(database_url)
    _db_conn.autocommit = True
    _redis_client = redis.from_url(redis_url)


@tool
def get_user_profile(user_id: str) -> str:
    """Retrieve a user's behavioral profile including purchase frequency,
    average cart value, category affinity scores, and recency signals.
    Checks Redis cache first, falls back to PostgreSQL."""
    # Check Redis cache first for low-latency access
    cached = _redis_client.get(f"profile:{user_id}")
    if cached:
        return cached.decode("utf-8")

    # Fall back to PostgreSQL
    cur = _db_conn.cursor()
    cur.execute(
        "SELECT features FROM user_profiles WHERE user_id = %s",
        (user_id,)
    )
    row = cur.fetchone()
    cur.close()

    if row:
        return row[0] if isinstance(row[0], str) else json.dumps(row[0])
    return json.dumps({"error": "Profile not found"})


@tool
def get_eligible_offers(cart_value: float) -> str:
    """Get all active offers eligible for the given cart value.
    Returns offers where the minimum cart threshold is met."""
    cur = _db_conn.cursor()
    cur.execute(
        """SELECT offer_id, name, discount_pct, category, min_cart_value
           FROM offers_catalog
           WHERE active = TRUE AND min_cart_value <= %s""",
        (cart_value,)
    )
    rows = cur.fetchall()
    cur.close()

    offers = [
        {
            "offer_id": r[0], "name": r[1], "discount_pct": float(r[2]),
            "category": r[3], "min_cart_value": float(r[4])
        }
        for r in rows
    ]
    return json.dumps(offers)


@tool
def check_business_constraints(offer_id: str, user_id: str) -> str:
    """Check if a user is eligible for a specific offer based on
    business rules like redemption history and frequency caps."""
    # Placeholder: in production, check redemption history and caps
    return json.dumps({
        "eligible": True,
        "reason": "No constraints violated"
    })


@tool
def score_relevance(user_profile_json: str, offer_json: str) -> str:
    """Compute a relevance score between a user profile and an offer
    using the dot product of category affinity with offer category."""
    profile = json.loads(user_profile_json)
    offer = json.loads(offer_json)

    # Extract category affinity from profile
    category_affinity = profile.get("category_affinity", {})
    offer_category = offer.get("category", "")

    # Dot product: score is how much the user likes this category
    score = category_affinity.get(offer_category, 0.0)
    return json.dumps({"score": round(float(score), 4)})

What does this code do?

Each function decorated with @tool becomes a callable tool the LangChain agent can invoke during its reasoning process. The agent reads each tool's docstring to decide when to use it.

• get_user_profile checks Redis first for cached data (sub-millisecond), then falls back to PostgreSQL.
• get_eligible_offers queries the offers_catalog table for active offers matching the cart value threshold.
• check_business_constraints is a placeholder that always returns eligible. In production, you would check redemption history and frequency caps.
• score_relevance computes a numerical affinity score by matching the offer's category against the user's category preferences.

• Save tools.py.

Seeing import errors in your editor?

You may see red underlines for psycopg2 or langchain in VS Code. This is expected because the packages are not installed in your local environment. They will be installed inside the Docker container when you build it later in this step.

Help me debug import errors in my offer-agent tools.py

Build the LangChain agent with the system prompt

Now you will wire the tools into a LangChain agent powered by Groq's llama-3.3-70b-versatile model. The agent uses a ReAct-style reasoning loop: it reads the user context, decides which tools to call, interprets the results, and produces a final answer.

• Create a file called agent.py inside services/offer-agent/ and add the following code:

from langchain.agents import create_agent
from langchain_groq import ChatGroq
from tools import (
    get_user_profile,
    get_eligible_offers,
    check_business_constraints,
    score_relevance,
)

# System prompt that guides the agent's reasoning
SYSTEM_PROMPT = (
    "You are a senior personalization strategist at a top e-commerce company. "
    "Your goal is to select the single best offer for a customer at checkout "
    "to maximize conversion without eroding margin. "
    "Use the available tools to gather data, reason step by step, "
    "and explain your decision."
)


def build_agent(db_pool, redis_client):
    """Build the LangChain agent with Groq LLM and personalization tools."""
    # Initialize the Groq LLM with Llama 3.3 70B
    llm = ChatGroq(
        model="llama-3.3-70b-versatile",
        temperature=0,
    )

    # Define the tools the agent can use
    tools = [
        get_user_profile,
        get_eligible_offers,
        check_business_constraints,
        score_relevance,
    ]

    # Create the agent with model, tools, and system prompt
    agent = create_agent(
        model=llm,
        tools=tools,
        system_prompt=SYSTEM_PROMPT,
    )

    return agent

What does this code do?

ChatGroq connects to Groq's free inference API using the llama-3.3-70b-versatile model. This is a 70B parameter model running at 280 tokens per second with a 131K context window. Setting temperature=0 makes the agent deterministic so it produces consistent offer decisions.

The create_agent function from LangChain assembles a ReAct agent that can call any of the four tools in a loop until it reaches a final answer. The system_prompt parameter guides the agent to think like a personalization strategist rather than a generic assistant.

Why an agentic approach instead of simple scoring?

A hard-coded scoring function picks the highest score every time without considering context. An agent can reason about trade-offs: is a 20% discount on electronics better than a 10% discount on the exact category the user loves?

The agent evaluates constraints, weighs alternatives, and produces a human-readable explanation. This mirrors how companies like Netflix and Rokt use multi-stage reasoning for real-time personalization.

• Save agent.py.

Create the FastAPI service and containerize

The agent needs a REST endpoint so other services can request personalization decisions. You will wrap it in a FastAPI app and create a Dockerfile to run it as part of the Docker Compose stack.

• Create a file called main.py inside services/offer-agent/ and add the following code:

import os
import json
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from tools import init_tools
from agent import build_agent


# Request and response models
class DecideRequest(BaseModel):
    user_id: str
    cart_value: float
    cart_categories: list[str]


class DecideResponse(BaseModel):
    offer_id: str
    offer_name: str
    discount_pct: float
    reasoning: str


# Module-level agent reference
_agent = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Initialize connections and agent on startup."""
    global _agent

    # Initialize tool dependencies
    database_url = os.environ.get(
        "DATABASE_URL",
        "postgresql://checkout_user:checkout_pass@postgres:5432/checkout_db"
    )
    redis_url = os.environ.get("REDIS_URL", "redis://redis:6379")
    init_tools(database_url, redis_url)

    # Build the LangChain agent
    _agent = build_agent(db_pool=None, redis_client=None)
    print("Offer Decision Agent ready")

    yield

    print("Shutting down Offer Decision Agent")


app = FastAPI(title="Offer Decision Agent", lifespan=lifespan)


@app.post("/decide", response_model=DecideResponse)
async def decide(request: DecideRequest):
    """Ask the AI agent to select the best offer for a user."""
    # Build the prompt with user context
    prompt = (
        f"Select the best offer for user '{request.user_id}' "
        f"who has a cart worth ${request.cart_value:.2f} "
        f"with items in categories: {request.cart_categories}. "
        f"Return your answer as JSON with keys: "
        f"offer_id, offer_name, discount_pct, reasoning."
    )

    try:
        # Invoke the agent
        result = _agent.invoke(
            {"messages": [{"role": "user", "content": prompt}]}
        )

        # Extract the final message content
        final_message = result["messages"][-1]
        content = final_message.content

        # Parse the JSON response from the agent
        if isinstance(content, str):
            try:
                data = json.loads(content)
            except json.JSONDecodeError:
                # If not pure JSON, return raw reasoning as fallback
                return DecideResponse(
                    offer_id="fallback",
                    offer_name="Default Offer",
                    discount_pct=5.0,
                    reasoning=content
                )
        else:
            data = {"reasoning": str(content)}

        return DecideResponse(
            offer_id=data.get("offer_id", "unknown"),
            offer_name=data.get("offer_name", "Unknown Offer"),
            discount_pct=float(data.get("discount_pct", 0)),
            reasoning=data.get("reasoning", "No reasoning provided")
        )

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/health")
async def health():
    """Health check endpoint."""
    return {"status": "ok"}

What does this code do?

The lifespan context manager runs once at startup. It initializes database and Redis connections for the tools, then builds the LangChain agent.

The POST /decide endpoint accepts a user ID, cart value, and cart categories. It constructs a prompt asking the agent to select the best offer, invokes the agent (which may call multiple tools in its reasoning loop), and parses the structured response.

The fallback logic handles cases where the LLM returns unstructured text instead of clean JSON. This makes the service resilient to occasional format variations from the model.

• Save main.py.

Now create the Dockerfile to containerize the service.

• Create a file called Dockerfile inside services/offer-agent/ and add the following:

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8002"]

• Save the Dockerfile.

Finally, add the offer-agent service to your Docker Compose configuration.

• Switch back to docker-compose.yml in the root of your smart-checkout folder and add the following service definition below the profile-builder service:

  offer-agent:
    build: ./services/offer-agent
    ports:
      - "8002:8002"
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy
    environment:
      - GROQ_API_KEY=${GROQ_API_KEY}
      - DATABASE_URL=postgresql://checkout_user:checkout_pass@postgres:5432/checkout_db
      - REDIS_URL=redis://redis:6379
    networks:
      - checkout-net

What does this configuration do?

The offer-agent service exposes port 8002 and depends on postgres and redis being healthy before starting.

The GROQ_API_KEY environment variable is pulled from your host machine's environment using ${GROQ_API_KEY} so you never hard-code secrets in the compose file.

• Save docker-compose.yml.
• Set your Groq API key as an environment variable before running the stack by running this command:

export GROQ_API_KEY=your-api-key-here

Get your Groq API key

If you do not have a Groq API key yet, visit console.groq.com and create a free account. No credit card is required. Generate an API key from the dashboard and paste it in place of your-api-key-here above.

• Rebuild and start the stack by running:

docker compose up --build -d

• Verify the offer-agent service is running by checking its health endpoint:

curl http://localhost:8002/health

You should see {"status":"ok"} in the response.

Service not starting or health check failing?

Check the container logs for error details by running docker compose logs offer-agent. Common issues:

• If you see a GROQ_API_KEY error, make sure you exported the variable in the same terminal session where you ran docker compose up.
• If the container exits immediately, check for Python syntax errors in your files by reviewing the log output.
• If PostgreSQL or Redis connections fail, ensure those containers are healthy with docker compose ps.

Help me debug my offer-agent container

✔️ Awesome, I've got everything!

Great. Double check you have saved all four files in services/offer-agent/: requirements.txt, tools.py, agent.py, main.py, and Dockerfile.

ⓧ I'd like to double check the full code

Here is the complete code for each file in services/offer-agent/:

import json
import psycopg2
import redis
from langchain.tools import tool

# Module-level connections initialized at startup
_db_conn = None
_redis_client = None


def init_tools(database_url: str, redis_url: str):
    """Initialize database and cache connections for tools."""
    global _db_conn, _redis_client
    _db_conn = psycopg2.connect(database_url)
    _db_conn.autocommit = True
    _redis_client = redis.from_url(redis_url)


@tool
def get_user_profile(user_id: str) -> str:
    """Retrieve a user's behavioral profile including purchase frequency,
    average cart value, category affinity scores, and recency signals.
    Checks Redis cache first, falls back to PostgreSQL."""
    # Check Redis cache first for low-latency access
    cached = _redis_client.get(f"profile:{user_id}")
    if cached:
        return cached.decode("utf-8")

    # Fall back to PostgreSQL
    cur = _db_conn.cursor()
    cur.execute(
        "SELECT features FROM user_profiles WHERE user_id = %s",
        (user_id,)
    )
    row = cur.fetchone()
    cur.close()

    if row:
        return row[0] if isinstance(row[0], str) else json.dumps(row[0])
    return json.dumps({"error": "Profile not found"})


@tool
def get_eligible_offers(cart_value: float) -> str:
    """Get all active offers eligible for the given cart value.
    Returns offers where the minimum cart threshold is met."""
    cur = _db_conn.cursor()
    cur.execute(
        """SELECT offer_id, name, discount_pct, category, min_cart_value
           FROM offers_catalog
           WHERE active = TRUE AND min_cart_value <= %s""",
        (cart_value,)
    )
    rows = cur.fetchall()
    cur.close()

    offers = [
        {
            "offer_id": r[0], "name": r[1], "discount_pct": float(r[2]),
            "category": r[3], "min_cart_value": float(r[4])
        }
        for r in rows
    ]
    return json.dumps(offers)


@tool
def check_business_constraints(offer_id: str, user_id: str) -> str:
    """Check if a user is eligible for a specific offer based on
    business rules like redemption history and frequency caps."""
    # Placeholder: in production, check redemption history and caps
    return json.dumps({
        "eligible": True,
        "reason": "No constraints violated"
    })


@tool
def score_relevance(user_profile_json: str, offer_json: str) -> str:
    """Compute a relevance score between a user profile and an offer
    using the dot product of category affinity with offer category."""
    profile = json.loads(user_profile_json)
    offer = json.loads(offer_json)

    # Extract category affinity from profile
    category_affinity = profile.get("category_affinity", {})
    offer_category = offer.get("category", "")

    # Dot product: score is how much the user likes this category
    score = category_affinity.get(offer_category, 0.0)
    return json.dumps({"score": round(float(score), 4)})

from langchain.agents import create_agent
from langchain_groq import ChatGroq
from tools import (
    get_user_profile,
    get_eligible_offers,
    check_business_constraints,
    score_relevance,
)

# System prompt that guides the agent's reasoning
SYSTEM_PROMPT = (
    "You are a senior personalization strategist at a top e-commerce company. "
    "Your goal is to select the single best offer for a customer at checkout "
    "to maximize conversion without eroding margin. "
    "Use the available tools to gather data, reason step by step, "
    "and explain your decision."
)


def build_agent(db_pool, redis_client):
    """Build the LangChain agent with Groq LLM and personalization tools."""
    # Initialize the Groq LLM with Llama 3.3 70B
    llm = ChatGroq(
        model="llama-3.3-70b-versatile",
        temperature=0,
    )

    # Define the tools the agent can use
    tools = [
        get_user_profile,
        get_eligible_offers,
        check_business_constraints,
        score_relevance,
    ]

    # Create the agent with model, tools, and system prompt
    agent = create_agent(
        model=llm,
        tools=tools,
        system_prompt=SYSTEM_PROMPT,
    )

    return agent

import os
import json
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from tools import init_tools
from agent import build_agent


# Request and response models
class DecideRequest(BaseModel):
    user_id: str
    cart_value: float
    cart_categories: list[str]


class DecideResponse(BaseModel):
    offer_id: str
    offer_name: str
    discount_pct: float
    reasoning: str


# Module-level agent reference
_agent = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Initialize connections and agent on startup."""
    global _agent

    # Initialize tool dependencies
    database_url = os.environ.get(
        "DATABASE_URL",
        "postgresql://checkout_user:checkout_pass@postgres:5432/checkout_db"
    )
    redis_url = os.environ.get("REDIS_URL", "redis://redis:6379")
    init_tools(database_url, redis_url)

    # Build the LangChain agent
    _agent = build_agent(db_pool=None, redis_client=None)
    print("Offer Decision Agent ready")

    yield

    print("Shutting down Offer Decision Agent")


app = FastAPI(title="Offer Decision Agent", lifespan=lifespan)


@app.post("/decide", response_model=DecideResponse)
async def decide(request: DecideRequest):
    """Ask the AI agent to select the best offer for a user."""
    # Build the prompt with user context
    prompt = (
        f"Select the best offer for user '{request.user_id}' "
        f"who has a cart worth ${request.cart_value:.2f} "
        f"with items in categories: {request.cart_categories}. "
        f"Return your answer as JSON with keys: "
        f"offer_id, offer_name, discount_pct, reasoning."
    )

    try:
        # Invoke the agent
        result = _agent.invoke(
            {"messages": [{"role": "user", "content": prompt}]}
        )

        # Extract the final message content
        final_message = result["messages"][-1]
        content = final_message.content

        # Parse the JSON response from the agent
        if isinstance(content, str):
            try:
                data = json.loads(content)
            except json.JSONDecodeError:
                # If not pure JSON, return raw reasoning as fallback
                return DecideResponse(
                    offer_id="fallback",
                    offer_name="Default Offer",
                    discount_pct=5.0,
                    reasoning=content
                )
        else:
            data = {"reasoning": str(content)}

        return DecideResponse(
            offer_id=data.get("offer_id", "unknown"),
            offer_name=data.get("offer_name", "Unknown Offer"),
            discount_pct=float(data.get("discount_pct", 0)),
            reasoning=data.get("reasoning", "No reasoning provided")
        )

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/health")
async def health():
    """Health check endpoint."""
    return {"status": "ok"}

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8002"]

Your AI-powered decision engine is now running. Next up, you will build the API Gateway that ties all the services together into a single orchestrated flow.

Build the API Gateway and Wire Everything Together

Your system now has three independent services: one that ingests events, one that builds user profiles, and one that makes AI-powered offer decisions. But there is no unified way for a client to trigger the full personalization flow.

The API Gateway pattern solves this by providing a single entry point that orchestrates all internal services behind one endpoint. In this step you will build the gateway, add production-grade features like rate limiting and request tracing, and wire the entire 7-container stack together.

In this step, get ready to:

• Build the API Gateway microservice with rate limiting and correlation ID tracing.
• Implement the /personalize endpoint that orchestrates the entire flow.
• Finalize docker-compose.yml and verify the full stack starts with a single command.

Create the API Gateway service files

The gateway needs its own service directory with a Dockerfile, dependencies list, and the application code.

• Create the services/api-gateway/ directory inside your smart-checkout project folder by running this command:

mkdir services/api-gateway

• Create the requirements.txt file inside services/api-gateway/ with the gateway's dependencies:

fastapi
uvicorn
httpx
slowapi
pydantic
python-dotenv

What are these dependencies?

• httpx is Python's modern async HTTP client. The gateway uses it to call the internal services without blocking the event loop.
• slowapi provides rate limiting for FastAPI. It prevents any single client from flooding the system with requests.
• The rest (fastapi, uvicorn, pydantic) are the same web framework stack the other services use.

• Create the Dockerfile inside services/api-gateway/ with the following content:

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

What does this Dockerfile do?

This is the same pattern used by the other services. It starts from a slim Python 3.11 image, installs dependencies, copies the app code, and runs uvicorn on port 8000.

• Save both files.
• Confirm the files exist by running:

ls services/api-gateway/

You should see Dockerfile and requirements.txt in the output.

Implement the gateway orchestration logic

The gateway's main.py is where all the production features live: rate limiting, correlation ID middleware, the /personalize orchestration endpoint, and health checks.

• Create services/api-gateway/main.py and add the imports, configuration, rate limiter, and correlation ID middleware:

import asyncio
import time
import uuid
import os

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from starlette.middleware.base import BaseHTTPMiddleware
import httpx

# Service URLs from environment variables
EVENT_INGESTION_URL = os.getenv("EVENT_INGESTION_URL", "http://event-ingestion:8001")
OFFER_AGENT_URL = os.getenv("OFFER_AGENT_URL", "http://offer-agent:8002")

# Rate limiter: 10 requests per minute per client IP
limiter = Limiter(key_func=get_remote_address)
app = FastAPI(title="Smart Checkout API Gateway")
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)


# Middleware that attaches a unique correlation ID to every request
class CorrelationIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        correlation_id = request.headers.get("X-Correlation-ID", str(uuid.uuid4()))
        request.state.correlation_id = correlation_id
        response = await call_next(request)
        response.headers["X-Correlation-ID"] = correlation_id
        return response


app.add_middleware(CorrelationIDMiddleware)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

What does this code do?

• The Limiter from slowapi tracks requests per client IP address. When a client exceeds 10 requests per minute, they receive a 429 Too Many Requests response.
• CorrelationIDMiddleware generates a unique UUID for each request and attaches it to both the request state and the response headers. This lets you trace a single request across all services.
• The EVENT_INGESTION_URL and OFFER_AGENT_URL environment variables point to the internal Docker service hostnames. Inside the Docker network, services resolve by container name.

• Below the middleware section, add the Pydantic request and response models, then the /personalize endpoint:

# Request/Response models
class PersonalizeRequest(BaseModel):
    user_id: str
    cart_value: float
    cart_categories: list[str]
    event_type: str = "checkout_start"


class PersonalizeResponse(BaseModel):
    user_id: str
    offer: dict
    reasoning: str
    correlation_id: str
    latency_ms: float


@app.post("/personalize")
@limiter.limit("10/minute")
async def personalize(request: Request, body: PersonalizeRequest):
    """Orchestrates the full personalization flow."""
    start = time.time()
    correlation_id = request.state.correlation_id

    async with httpx.AsyncClient() as client:
        # Step 1: Send the checkout event to the event-ingestion service
        await client.post(
            f"{EVENT_INGESTION_URL}/events",
            json={
                "user_id": body.user_id,
                "event_type": body.event_type,
                "properties": {
                    "cart_value": body.cart_value,
                    "cart_categories": body.cart_categories,
                },
            },
        )

        # Step 2: Brief pause to let the profile builder consume the event
        await asyncio.sleep(0.5)

        # Step 3: Ask the offer-agent for a personalization decision
        agent_response = await client.post(
            f"{OFFER_AGENT_URL}/decide",
            json={
                "user_id": body.user_id,
                "cart_value": body.cart_value,
                "cart_categories": body.cart_categories,
            },
        )
        decision = agent_response.json()

    latency_ms = (time.time() - start) * 1000

    return PersonalizeResponse(
        user_id=body.user_id,
        offer={
            "offer_id": decision.get("offer_id"),
            "offer_name": decision.get("offer_name"),
            "discount_pct": decision.get("discount_pct"),
        },
        reasoning=decision.get("reasoning", ""),
        correlation_id=correlation_id,
        latency_ms=round(latency_ms, 2),
    )

What does the /personalize endpoint do?

This single endpoint orchestrates the entire personalization pipeline in three steps:

• It publishes the checkout event to the event-ingestion service (which forwards it to Kafka).
• It waits 0.5 seconds for the profile-builder to process the event and update the user's features.
• It calls the offer-agent's /decide endpoint, which triggers the LangChain AI agent to reason about the best offer.

The response includes the chosen offer, the agent's reasoning, the correlation ID for tracing, and the total latency in milliseconds.

• Finally, add the two health check endpoints at the bottom of main.py:

@app.get("/health/liveness")
async def liveness():
    """Simple liveness probe - confirms the process is running."""
    return {"status": "alive"}


@app.get("/health/readiness")
async def readiness():
    """Readiness probe - checks that downstream services are reachable."""
    dependencies = {}

    async with httpx.AsyncClient() as client:
        # Check event-ingestion service
        try:
            r = await client.get(f"{EVENT_INGESTION_URL}/health", timeout=2.0)
            dependencies["event-ingestion"] = "healthy" if r.status_code == 200 else "unhealthy"
        except Exception:
            dependencies["event-ingestion"] = "unhealthy"

        # Check offer-agent service
        try:
            r = await client.get(f"{OFFER_AGENT_URL}/health", timeout=2.0)
            dependencies["offer-agent"] = "healthy" if r.status_code == 200 else "unhealthy"
        except Exception:
            dependencies["offer-agent"] = "unhealthy"

    all_healthy = all(v == "healthy" for v in dependencies.values())
    return {
        "status": "ready" if all_healthy else "degraded",
        "dependencies": dependencies,
    }

What is the difference between liveness and readiness?

Liveness tells the orchestrator "this process is alive." If it fails, the container should be restarted.

Readiness tells the orchestrator "this service can handle traffic." It actively checks that downstream dependencies (event-ingestion, offer-agent) are reachable before declaring itself ready. This prevents routing requests to a gateway that cannot reach its backends.

• Save services/api-gateway/main.py.

Seeing an import error for slowapi?

Make sure your requirements.txt includes slowapi on its own line. The Dockerfile installs dependencies from that file, so any missing entry causes an ImportError at runtime.

help me debug an ImportError in my API Gateway service

✔️ Awesome, I've got everything!

Great. Make sure you have saved main.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete services/api-gateway/main.py file:

import asyncio
import time
import uuid
import os

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from starlette.middleware.base import BaseHTTPMiddleware
import httpx

# Service URLs from environment variables
EVENT_INGESTION_URL = os.getenv("EVENT_INGESTION_URL", "http://event-ingestion:8001")
OFFER_AGENT_URL = os.getenv("OFFER_AGENT_URL", "http://offer-agent:8002")

# Rate limiter: 10 requests per minute per client IP
limiter = Limiter(key_func=get_remote_address)
app = FastAPI(title="Smart Checkout API Gateway")
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)


# Middleware that attaches a unique correlation ID to every request
class CorrelationIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        correlation_id = request.headers.get("X-Correlation-ID", str(uuid.uuid4()))
        request.state.correlation_id = correlation_id
        response = await call_next(request)
        response.headers["X-Correlation-ID"] = correlation_id
        return response


app.add_middleware(CorrelationIDMiddleware)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


# Request/Response models
class PersonalizeRequest(BaseModel):
    user_id: str
    cart_value: float
    cart_categories: list[str]
    event_type: str = "checkout_start"


class PersonalizeResponse(BaseModel):
    user_id: str
    offer: dict
    reasoning: str
    correlation_id: str
    latency_ms: float


@app.post("/personalize")
@limiter.limit("10/minute")
async def personalize(request: Request, body: PersonalizeRequest):
    """Orchestrates the full personalization flow."""
    start = time.time()
    correlation_id = request.state.correlation_id

    async with httpx.AsyncClient() as client:
        # Step 1: Send the checkout event to the event-ingestion service
        await client.post(
            f"{EVENT_INGESTION_URL}/events",
            json={
                "user_id": body.user_id,
                "event_type": body.event_type,
                "properties": {
                    "cart_value": body.cart_value,
                    "cart_categories": body.cart_categories,
                },
            },
        )

        # Step 2: Brief pause to let the profile builder consume the event
        await asyncio.sleep(0.5)

        # Step 3: Ask the offer-agent for a personalization decision
        agent_response = await client.post(
            f"{OFFER_AGENT_URL}/decide",
            json={
                "user_id": body.user_id,
                "cart_value": body.cart_value,
                "cart_categories": body.cart_categories,
            },
        )
        decision = agent_response.json()

    latency_ms = (time.time() - start) * 1000

    return PersonalizeResponse(
        user_id=body.user_id,
        offer={
            "offer_id": decision.get("offer_id"),
            "offer_name": decision.get("offer_name"),
            "discount_pct": decision.get("discount_pct"),
        },
        reasoning=decision.get("reasoning", ""),
        correlation_id=correlation_id,
        latency_ms=round(latency_ms, 2),
    )


@app.get("/health/liveness")
async def liveness():
    """Simple liveness probe - confirms the process is running."""
    return {"status": "alive"}


@app.get("/health/readiness")
async def readiness():
    """Readiness probe - checks that downstream services are reachable."""
    dependencies = {}

    async with httpx.AsyncClient() as client:
        # Check event-ingestion service
        try:
            r = await client.get(f"{EVENT_INGESTION_URL}/health", timeout=2.0)
            dependencies["event-ingestion"] = "healthy" if r.status_code == 200 else "unhealthy"
        except Exception:
            dependencies["event-ingestion"] = "unhealthy"

        # Check offer-agent service
        try:
            r = await client.get(f"{OFFER_AGENT_URL}/health", timeout=2.0)
            dependencies["offer-agent"] = "healthy" if r.status_code == 200 else "unhealthy"
        except Exception:
            dependencies["offer-agent"] = "unhealthy"

    all_healthy = all(v == "healthy" for v in dependencies.values())
    return {
        "status": "ready" if all_healthy else "degraded",
        "dependencies": dependencies,
    }

Wire all services in Docker Compose and verify the full stack

The last piece is adding the api-gateway service to your docker-compose.yml and verifying the entire 7-container system starts and communicates correctly.

• Open docker-compose.yml in VS Code.
• Add the api-gateway service after the offer-agent service block:

  api-gateway:
    build: ./services/api-gateway
    ports:
      - "8000:8000"
    depends_on:
      - event-ingestion
      - offer-agent
    environment:
      EVENT_INGESTION_URL: http://event-ingestion:8001
      OFFER_AGENT_URL: http://offer-agent:8002
    networks:
      - checkout-net

What does depends_on do here?

The depends_on field tells Docker Compose to start event-ingestion and offer-agent before the api-gateway. This ensures the gateway's downstream services are available when it boots.

The environment variables EVENT_INGESTION_URL and OFFER_AGENT_URL use the Docker Compose service names as hostnames. Inside the checkout-net network, event-ingestion resolves to that container's IP automatically.

• Save docker-compose.yml.
• Start the full stack by running:

docker compose up --build

You should see all 7 containers start: postgres, redis, kafka, event-ingestion, profile-builder, offer-agent, and api-gateway.

• Once the logs settle and you see Uvicorn running on http://0.0.0.0:8000 from the api-gateway, test the liveness endpoint by running this in a new terminal:

curl http://localhost:8000/health/liveness

You should see {"status":"alive"} in the response.

• Test the readiness probe to confirm inter-service communication:

curl http://localhost:8000/health/readiness

You should see both dependencies listed as healthy and the status as ready.

• Test the full end-to-end personalization flow by running:

curl -X POST http://localhost:8000/personalize \
  -H "Content-Type: application/json" \
  -d '{"user_id": "user_123", "cart_value": 85.50, "cart_categories": ["electronics", "accessories"]}'

You should receive a JSON response containing the AI agent's chosen offer, its reasoning explanation, a correlation ID, and the total latency in milliseconds.

Seeing connection refused or unhealthy dependencies?

• Check that all service containers are running with docker compose ps. If any exited, check its logs with docker compose logs <service-name>.
• Verify the environment variable URLs in docker-compose.yml match the exact service names (event-ingestion, offer-agent). A typo means Docker's DNS cannot resolve the hostname.
• Make sure you have a valid GROQ_API_KEY environment variable set, because the offer-agent needs it to start successfully.

help me debug why my API Gateway cannot reach the downstream services

Your entire personalization engine is now running behind a single unified API. Next up, you will build a frontend demo page so visitors can interact with the system visually.

Build the Demo Frontend

Your backend services are running and the personalization flow works end-to-end through the API Gateway. But right now, the only way to trigger it is by sending raw HTTP requests from the terminal.

In this step, you'll build an interactive checkout page that simulates a real e-commerce experience. Visitors will browse products, add items to their cart, and hit checkout to see the AI agent's personalized offer decision with its full reasoning explanation.

In this step, get ready to:

• Build a simulated checkout page with product browsing and cart functionality.
• Display the AI agent's personalized offer and reasoning in real-time.
• Add a system dashboard panel and serve everything through nginx.

Enable CORS on the API Gateway

Browsers block requests from one origin (like http://localhost) to a different port (like http://localhost:8000) unless the server explicitly allows it. This security feature is called CORS (Cross-Origin Resource Sharing).

• Open services/api-gateway/main.py in VS Code.
• Find the import line at the top of the file:

from fastapi import FastAPI

• Add this import directly below it:

from fastapi.middleware.cors import CORSMiddleware

• Find the line where your app is created (app = FastAPI()), then add this CORS configuration directly below it:

# Allow browser requests from the frontend
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

What does this code do?

CORSMiddleware tells the API Gateway to accept requests from your frontend at http://localhost.

Without this, the browser would block every fetch() call from the checkout page to port 8000, and your personalization requests would silently fail.

• Save services/api-gateway/main.py.

Create the interactive checkout page

The checkout page simulates a mini e-commerce store where visitors browse products, add items to a cart, and trigger the AI personalization flow on checkout.

• Create a frontend directory inside your smart-checkout project folder by running this command:

mkdir frontend

• Create a new file called index.html inside the frontend folder.
• Copy the following code into frontend/index.html:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Smart Checkout Demo</title>
    <link rel="stylesheet" href="style.css">
</head>
<body>
    <header>
        <h1>Smart Checkout Demo</h1>
    </header>

    <main>
        <!-- Product grid with 3 items -->
        <section class="product-grid" id="products">
            <div class="product-card" data-name="Laptop" data-price="899" data-category="electronics">
                <h3>Laptop</h3>
                <p class="price">$899</p>
                <button class="add-to-cart">Add to Cart</button>
            </div>
            <div class="product-card" data-name="T-Shirt" data-price="45" data-category="apparel">
                <h3>T-Shirt</h3>
                <p class="price">$45</p>
                <button class="add-to-cart">Add to Cart</button>
            </div>
            <div class="product-card" data-name="Headphones" data-price="199" data-category="electronics">
                <h3>Headphones</h3>
                <p class="price">$199</p>
                <button class="add-to-cart">Add to Cart</button>
            </div>
        </section>

        <!-- Cart summary -->
        <section id="cart">
            <h2>Cart Summary</h2>
            <ul id="cart-items"></ul>
            <p id="cart-total">Total: $0</p>
            <button id="checkout-btn" disabled>Proceed to Checkout</button>
        </section>

        <!-- AI offer display -->
        <section id="offer-panel" class="hidden">
            <h2>Personalized Offer</h2>
            <p id="offer-name"></p>
            <p id="offer-discount"></p>
            <p id="offer-reasoning"></p>
        </section>

        <!-- System dashboard -->
        <section id="metrics">
            <h2>System Dashboard</h2>
            <div id="dependency-status"></div>
        </section>
    </main>

    <script>
        // Generate or retrieve a persistent user ID
        let userId = localStorage.getItem('checkout_user_id');
        if (!userId) {
            userId = crypto.randomUUID();
            localStorage.setItem('checkout_user_id', userId);
        }

        // Cart state
        const cart = [];

        // Add to cart functionality
        document.querySelectorAll('.add-to-cart').forEach(button => {
            button.addEventListener('click', (e) => {
                const card = e.target.closest('.product-card');
                const name = card.dataset.name;
                const price = parseFloat(card.dataset.price);
                const category = card.dataset.category;
                cart.push({ name, price, category });
                updateCartDisplay();
            });
        });

        // Update the cart UI
        function updateCartDisplay() {
            const cartItems = document.getElementById('cart-items');
            const cartTotal = document.getElementById('cart-total');
            const checkoutBtn = document.getElementById('checkout-btn');

            cartItems.innerHTML = cart.map(item =>
                `<li>${item.name} - $${item.price}</li>`
            ).join('');

            const total = cart.reduce((sum, item) => sum + item.price, 0);
            cartTotal.textContent = `Total: $${total}`;
            checkoutBtn.disabled = cart.length === 0;
        }

        // Checkout flow - call the personalization API
        document.getElementById('checkout-btn').addEventListener('click', async () => {
            const cartValue = cart.reduce((sum, item) => sum + item.price, 0);
            const cartCategories = [...new Set(cart.map(item => item.category))];

            const offerPanel = document.getElementById('offer-panel');
            offerPanel.classList.remove('hidden');
            document.getElementById('offer-name').textContent = 'Loading...';
            document.getElementById('offer-discount').textContent = '';
            document.getElementById('offer-reasoning').textContent = '';

            try {
                const response = await fetch('http://localhost:8000/personalize', {
                    method: 'POST',
                    headers: { 'Content-Type': 'application/json' },
                    body: JSON.stringify({
                        user_id: userId,
                        cart_value: cartValue,
                        cart_categories: cartCategories,
                        event_type: 'checkout_start'
                    })
                });

                const data = await response.json();
                document.getElementById('offer-name').textContent = `Offer: ${data.offer.name}`;
                document.getElementById('offer-discount').textContent = `Discount: ${data.offer.discount}`;
                document.getElementById('offer-reasoning').textContent = data.reasoning;
            } catch (error) {
                document.getElementById('offer-name').textContent = 'Error fetching offer';
                document.getElementById('offer-reasoning').textContent = error.message;
            }
        });

        // Dashboard polling - check system health every 5 seconds
        async function pollMetrics() {
            try {
                const response = await fetch('http://localhost:8000/health/readiness');
                const data = await response.json();
                const statusDiv = document.getElementById('dependency-status');
                statusDiv.innerHTML = Object.entries(data.dependencies)
                    .map(([service, status]) =>
                        `<p class="status-item"><strong>${service}:</strong> ${status}</p>`
                    ).join('');
            } catch (error) {
                document.getElementById('dependency-status').innerHTML =
                    '<p class="status-item error">Unable to reach API Gateway</p>';
            }
        }

        // Start polling immediately and repeat every 5 seconds
        pollMetrics();
        setInterval(pollMetrics, 5000);
    </script>
</body>
</html>

What does this page do?

• The product grid displays three items (laptop, t-shirt, headphones) with "Add to Cart" buttons that populate the cart array.
• crypto.randomUUID() generates a unique user ID stored in localStorage so the same visitor keeps their identity across page reloads.
• The checkout button sends a POST to http://localhost:8000/personalize with the user ID, cart value, and categories. The response populates the green offer panel.
• pollMetrics() hits the readiness endpoint every 5 seconds and renders each dependency's status in the dashboard.

• Save frontend/index.html.

Now create the stylesheet that gives the page a clean, modern e-commerce look.

• Create a new file called style.css inside the frontend folder.
• Copy the following CSS into frontend/style.css:

/* Color palette and design tokens */
:root {
    --primary-color: #2c3e50;
    --accent-color: #27ae60;
    --background-color: #f5f6fa;
    --card-background: #ffffff;
    --text-color: #333333;
    --border-color: #e0e0e0;
}

* {
    margin: 0;
    padding: 0;
    box-sizing: border-box;
}

body {
    font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
    background-color: var(--background-color);
    color: var(--text-color);
    line-height: 1.6;
}

/* Header bar */
header {
    background-color: var(--primary-color);
    color: white;
    padding: 1.5rem;
    text-align: center;
}

main {
    max-width: 1000px;
    margin: 2rem auto;
    padding: 0 1rem;
}

/* Flexbox product grid */
.product-grid {
    display: flex;
    gap: 1.5rem;
    flex-wrap: wrap;
    margin-bottom: 2rem;
}

/* Card components with hover shadows */
.product-card {
    background: var(--card-background);
    border: 1px solid var(--border-color);
    border-radius: 8px;
    padding: 1.5rem;
    flex: 1;
    min-width: 200px;
    text-align: center;
    transition: box-shadow 0.2s ease;
}

.product-card:hover {
    box-shadow: 0 4px 12px rgba(0, 0, 0, 0.1);
}

.product-card h3 {
    margin-bottom: 0.5rem;
}

.price {
    font-size: 1.25rem;
    font-weight: bold;
    color: var(--primary-color);
    margin-bottom: 1rem;
}

.add-to-cart {
    background-color: var(--primary-color);
    color: white;
    border: none;
    padding: 0.5rem 1rem;
    border-radius: 4px;
    cursor: pointer;
}

.add-to-cart:hover {
    opacity: 0.9;
}

/* Cart section */
#cart {
    background: var(--card-background);
    border: 1px solid var(--border-color);
    border-radius: 8px;
    padding: 1.5rem;
    margin-bottom: 2rem;
}

#cart h2 {
    margin-bottom: 1rem;
}

#cart-items {
    list-style: none;
    margin-bottom: 1rem;
}

#cart-items li {
    padding: 0.25rem 0;
}

#cart-total {
    font-weight: bold;
    font-size: 1.1rem;
    margin-bottom: 1rem;
}

/* Checkout button */
#checkout-btn {
    background-color: var(--accent-color);
    color: white;
    border: none;
    padding: 0.75rem 1.5rem;
    border-radius: 4px;
    font-size: 1rem;
    cursor: pointer;
}

#checkout-btn:disabled {
    background-color: #ccc;
    cursor: not-allowed;
}

/* Green offer highlight panel */
#offer-panel {
    background-color: #e8f8f0;
    border: 2px solid var(--accent-color);
    border-radius: 8px;
    padding: 1.5rem;
    margin-bottom: 2rem;
}

#offer-panel h2 {
    color: var(--accent-color);
    margin-bottom: 1rem;
}

#offer-panel.hidden {
    display: none;
}

/* Dashboard metrics panel */
#metrics {
    background: var(--card-background);
    border: 1px solid var(--border-color);
    border-radius: 8px;
    padding: 1.5rem;
}

#metrics h2 {
    margin-bottom: 1rem;
}

.status-item {
    padding: 0.25rem 0;
}

.status-item.error {
    color: red;
}

What does this stylesheet do?

• CSS variables in :root define a consistent color palette across the page.
• The product grid uses display: flex with flex-wrap: wrap so cards reflow on smaller screens.
• Product cards have a hover shadow transition that gives the page a polished, interactive feel.
• The offer panel uses a green border and background to draw the visitor's attention to the personalized deal.

• Save frontend/style.css.
• Confirm both files exist by running:

ls frontend/

You should see index.html and style.css in the output.

Don't see both files?

Make sure you created the files inside the frontend/ directory, not the project root. Check your current directory with pwd to confirm you're inside smart-checkout.

Help me find where my frontend files ended up.

✔️ Awesome, I've got everything!

Great. Make sure both files are saved before moving on.

ⓧ I'd like to double check the full code

Here is the complete frontend/index.html:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Smart Checkout Demo</title>
    <link rel="stylesheet" href="style.css">
</head>
<body>
    <header>
        <h1>Smart Checkout Demo</h1>
    </header>

    <main>
        <!-- Product grid with 3 items -->
        <section class="product-grid" id="products">
            <div class="product-card" data-name="Laptop" data-price="899" data-category="electronics">
                <h3>Laptop</h3>
                <p class="price">$899</p>
                <button class="add-to-cart">Add to Cart</button>
            </div>
            <div class="product-card" data-name="T-Shirt" data-price="45" data-category="apparel">
                <h3>T-Shirt</h3>
                <p class="price">$45</p>
                <button class="add-to-cart">Add to Cart</button>
            </div>
            <div class="product-card" data-name="Headphones" data-price="199" data-category="electronics">
                <h3>Headphones</h3>
                <p class="price">$199</p>
                <button class="add-to-cart">Add to Cart</button>
            </div>
        </section>

        <!-- Cart summary -->
        <section id="cart">
            <h2>Cart Summary</h2>
            <ul id="cart-items"></ul>
            <p id="cart-total">Total: $0</p>
            <button id="checkout-btn" disabled>Proceed to Checkout</button>
        </section>

        <!-- AI offer display -->
        <section id="offer-panel" class="hidden">
            <h2>Personalized Offer</h2>
            <p id="offer-name"></p>
            <p id="offer-discount"></p>
            <p id="offer-reasoning"></p>
        </section>

        <!-- System dashboard -->
        <section id="metrics">
            <h2>System Dashboard</h2>
            <div id="dependency-status"></div>
        </section>
    </main>

    <script>
        // Generate or retrieve a persistent user ID
        let userId = localStorage.getItem('checkout_user_id');
        if (!userId) {
            userId = crypto.randomUUID();
            localStorage.setItem('checkout_user_id', userId);
        }

        // Cart state
        const cart = [];

        // Add to cart functionality
        document.querySelectorAll('.add-to-cart').forEach(button => {
            button.addEventListener('click', (e) => {
                const card = e.target.closest('.product-card');
                const name = card.dataset.name;
                const price = parseFloat(card.dataset.price);
                const category = card.dataset.category;
                cart.push({ name, price, category });
                updateCartDisplay();
            });
        });

        // Update the cart UI
        function updateCartDisplay() {
            const cartItems = document.getElementById('cart-items');
            const cartTotal = document.getElementById('cart-total');
            const checkoutBtn = document.getElementById('checkout-btn');

            cartItems.innerHTML = cart.map(item =>
                `<li>${item.name} - $${item.price}</li>`
            ).join('');

            const total = cart.reduce((sum, item) => sum + item.price, 0);
            cartTotal.textContent = `Total: $${total}`;
            checkoutBtn.disabled = cart.length === 0;
        }

        // Checkout flow - call the personalization API
        document.getElementById('checkout-btn').addEventListener('click', async () => {
            const cartValue = cart.reduce((sum, item) => sum + item.price, 0);
            const cartCategories = [...new Set(cart.map(item => item.category))];

            const offerPanel = document.getElementById('offer-panel');
            offerPanel.classList.remove('hidden');
            document.getElementById('offer-name').textContent = 'Loading...';
            document.getElementById('offer-discount').textContent = '';
            document.getElementById('offer-reasoning').textContent = '';

            try {
                const response = await fetch('http://localhost:8000/personalize', {
                    method: 'POST',
                    headers: { 'Content-Type': 'application/json' },
                    body: JSON.stringify({
                        user_id: userId,
                        cart_value: cartValue,
                        cart_categories: cartCategories,
                        event_type: 'checkout_start'
                    })
                });

                const data = await response.json();
                document.getElementById('offer-name').textContent = `Offer: ${data.offer.name}`;
                document.getElementById('offer-discount').textContent = `Discount: ${data.offer.discount}`;
                document.getElementById('offer-reasoning').textContent = data.reasoning;
            } catch (error) {
                document.getElementById('offer-name').textContent = 'Error fetching offer';
                document.getElementById('offer-reasoning').textContent = error.message;
            }
        });

        // Dashboard polling - check system health every 5 seconds
        async function pollMetrics() {
            try {
                const response = await fetch('http://localhost:8000/health/readiness');
                const data = await response.json();
                const statusDiv = document.getElementById('dependency-status');
                statusDiv.innerHTML = Object.entries(data.dependencies)
                    .map(([service, status]) =>
                        `<p class="status-item"><strong>${service}:</strong> ${status}</p>`
                    ).join('');
            } catch (error) {
                document.getElementById('dependency-status').innerHTML =
                    '<p class="status-item error">Unable to reach API Gateway</p>';
            }
        }

        // Start polling immediately and repeat every 5 seconds
        pollMetrics();
        setInterval(pollMetrics, 5000);
    </script>
</body>
</html>

And here is the complete frontend/style.css:

/* Color palette and design tokens */
:root {
    --primary-color: #2c3e50;
    --accent-color: #27ae60;
    --background-color: #f5f6fa;
    --card-background: #ffffff;
    --text-color: #333333;
    --border-color: #e0e0e0;
}

* {
    margin: 0;
    padding: 0;
    box-sizing: border-box;
}

body {
    font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
    background-color: var(--background-color);
    color: var(--text-color);
    line-height: 1.6;
}

/* Header bar */
header {
    background-color: var(--primary-color);
    color: white;
    padding: 1.5rem;
    text-align: center;
}

main {
    max-width: 1000px;
    margin: 2rem auto;
    padding: 0 1rem;
}

/* Flexbox product grid */
.product-grid {
    display: flex;
    gap: 1.5rem;
    flex-wrap: wrap;
    margin-bottom: 2rem;
}

/* Card components with hover shadows */
.product-card {
    background: var(--card-background);
    border: 1px solid var(--border-color);
    border-radius: 8px;
    padding: 1.5rem;
    flex: 1;
    min-width: 200px;
    text-align: center;
    transition: box-shadow 0.2s ease;
}

.product-card:hover {
    box-shadow: 0 4px 12px rgba(0, 0, 0, 0.1);
}

.product-card h3 {
    margin-bottom: 0.5rem;
}

.price {
    font-size: 1.25rem;
    font-weight: bold;
    color: var(--primary-color);
    margin-bottom: 1rem;
}

.add-to-cart {
    background-color: var(--primary-color);
    color: white;
    border: none;
    padding: 0.5rem 1rem;
    border-radius: 4px;
    cursor: pointer;
}

.add-to-cart:hover {
    opacity: 0.9;
}

/* Cart section */
#cart {
    background: var(--card-background);
    border: 1px solid var(--border-color);
    border-radius: 8px;
    padding: 1.5rem;
    margin-bottom: 2rem;
}

#cart h2 {
    margin-bottom: 1rem;
}

#cart-items {
    list-style: none;
    margin-bottom: 1rem;
}

#cart-items li {
    padding: 0.25rem 0;
}

#cart-total {
    font-weight: bold;
    font-size: 1.1rem;
    margin-bottom: 1rem;
}

/* Checkout button */
#checkout-btn {
    background-color: var(--accent-color);
    color: white;
    border: none;
    padding: 0.75rem 1.5rem;
    border-radius: 4px;
    font-size: 1rem;
    cursor: pointer;
}

#checkout-btn:disabled {
    background-color: #ccc;
    cursor: not-allowed;
}

/* Green offer highlight panel */
#offer-panel {
    background-color: #e8f8f0;
    border: 2px solid var(--accent-color);
    border-radius: 8px;
    padding: 1.5rem;
    margin-bottom: 2rem;
}

#offer-panel h2 {
    color: var(--accent-color);
    margin-bottom: 1rem;
}

#offer-panel.hidden {
    display: none;
}

/* Dashboard metrics panel */
#metrics {
    background: var(--card-background);
    border: 1px solid var(--border-color);
    border-radius: 8px;
    padding: 1.5rem;
}

#metrics h2 {
    margin-bottom: 1rem;
}

.status-item {
    padding: 0.25rem 0;
}

.status-item.error {
    color: red;
}

Serve the frontend with nginx

The frontend files need a web server to serve them to the browser. You'll add an nginx container to your Docker Compose stack that serves the frontend/ directory as static files on port 80.

• Open docker-compose.yml in VS Code.
• Add this nginx service at the bottom of the services: section, below the api-gateway service:

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./frontend:/usr/share/nginx/html:ro
    depends_on:
      - api-gateway
    networks:
      - checkout-net

What does this service do?

• nginx:alpine is a lightweight web server image perfect for serving static files.
• The volume mounts your local frontend/ folder into nginx's default document root (/usr/share/nginx/html) as read-only (:ro).
• Port 80 means the demo is accessible at plain http://localhost with no port number needed.

• Save docker-compose.yml.
• Rebuild and restart the entire stack by running this command:

docker compose up --build -d

• Wait for all 8 containers to start, then open http://localhost in your browser.

You should see the "Smart Checkout Demo" page with three product cards (Laptop $899, T-Shirt $45, Headphones $199), a Cart Summary section, and a System Dashboard at the bottom showing dependency statuses.

• Click Add to Cart on one or more products.
• Click Proceed to Checkout.

A green panel appears showing the AI agent's personalized offer name, discount, and its full reasoning explanation.

Page not loading or offer panel shows an error?

• Run docker compose ps to confirm all 8 containers are in the "Up" state. If nginx is not running, check for YAML indentation errors in your compose file.
• If the offer panel says "Error fetching offer", open DevTools (F12) and check the Console tab. A CORS error means the CORSMiddleware was not added correctly to services/api-gateway/main.py.
• If port 80 is already in use, stop any other web server on your machine or change the nginx port mapping to "8080:80" and visit http://localhost:8080 instead.

Help me debug why my checkout demo isn't working.

Your demo is live and fully interactive at http://localhost. Next up, you'll deploy this entire stack to AWS so anyone on the internet can see your personalization engine in action.

Deploy to AWS

Your personalization engine runs perfectly on your local machine. The next step is putting it somewhere the world can access it.

In this step, you will launch a cloud server on AWS, install Docker on it, and deploy your entire 8-container stack so anyone can visit your demo from a public URL.

In this step, get ready to:

• Launch an EC2 instance and configure its security group.
• Install Docker and Docker Compose on the instance.
• Deploy the full stack and verify end-to-end from a public URL.

Create an AWS account and launch an EC2 instance

EC2 (Elastic Compute Cloud) is a virtual server in the AWS cloud. You will launch a single t3.micro instance running Amazon Linux 2023, which is eligible for the AWS free tier.

• Open the AWS sign-up page at https://aws.amazon.com and create a free-tier account if you do not already have one.
• Sign in to the AWS Management Console.
• In the navigation bar at the top, select the region US East (N. Virginia) which corresponds to us-east-1.
• From the EC2 dashboard, click Launch instance.
• Under Name and tags, enter smart-checkout-demo as the instance name.
• Under Application and OS Images, choose Quick Start, then select Amazon Linux. Confirm the AMI is marked Free tier eligible and shows Amazon Linux 2023.
• Under Instance type, select t3.micro.
• Under Key pair (login), click Create new key pair.
• Name it smart-checkout-key, keep the type as RSA and format as .pem, then click Create key pair.
• Your browser downloads the .pem file. Save it somewhere safe. You will need it to SSH into your instance.

What is a key pair?

A key pair lets you securely connect to your EC2 instance via SSH. The private key (.pem file) stays on your machine. The public key lives on the instance. Without the private key, no one can log in.

Now configure the security group to allow the traffic your demo needs.

• Under Network settings, click Edit.
• Set Security group name to smart-checkout-sg.
• The first rule (SSH, port 22) is already listed. Keep it with source 0.0.0.0/0.
• Click Add security group rule. Set Type to HTTP (port 80). Leave source as 0.0.0.0/0.
• Click Add security group rule again. Set Type to Custom TCP, Port range to 8000, source 0.0.0.0/0.
• Check the Auto-assign public IP dropdown is set to Enable.
• Leave all other settings at their defaults and click Launch instance.

Why these three ports?

Port 22 allows SSH access so you can manage the server. Port 80 serves your demo website through nginx. Port 8000 exposes the API gateway so the frontend JavaScript can call /personalize directly.

• Wait for the instance state to show Running and the status checks to pass (1-2 minutes).
• Click on your instance name to open its details page.
• Copy the Public IPv4 address shown in the details panel. You will need this throughout the rest of the step.

Keep the IP handy

Paste the public IP into a text file or sticky note. You will use it for SSH, the frontend config, and final verification.

Instance not launching?

If you see a capacity error, try a different availability zone by editing the subnet setting. If you see a vCPU limit error, your account may need a service quota increase for t3 instances in us-east-1.

Help me troubleshoot my EC2 launch failure

Install Docker and Docker Compose on the EC2 instance

Your instance is running Amazon Linux 2023. Docker is available from the built-in package repository, but the Compose plugin requires a manual install step.

• Open a terminal on your local machine.
• Set the correct permissions on your key file by running this command:

chmod 400 [[PEMPATH="path/to/smart-checkout-key.pem"]]

• SSH into your EC2 instance by running this command:

ssh -i [[PEMPATH="path/to/smart-checkout-key.pem"]] ec2-user@[[EC2_PUBLIC_IP="your EC2 public IPv4 address"]]

What does this command do?

The ssh command opens a secure shell connection to your remote server. The -i flag points to your private key file. ec2-user is the default username on Amazon Linux instances.

• You will see a prompt asking if you want to continue connecting. Type yes and press Enter.
• Once connected, update system packages and install Docker by running these commands:

sudo yum update -y
sudo yum install -y docker git

• Start the Docker service and add your user to the Docker group by running:

sudo service docker start
sudo usermod -a -G docker ec2-user

• Install the Docker Compose plugin by running these commands:

sudo curl -SL https://github.com/docker/compose/releases/latest/download/docker-compose-linux-$(uname -m) -o /usr/libexec/docker/cli-plugins/docker-compose
sudo chmod +x /usr/libexec/docker/cli-plugins/docker-compose

Why install Compose separately?

Amazon Linux 2023's Docker package does not include the Compose plugin by default. The commands above download the latest Compose v2 binary and place it in Docker's CLI plugins directory so you can use docker compose (with a space).

• Apply the group permissions by logging out and back in. Run:

exit

• Reconnect via SSH using the same command from earlier:

ssh -i [[PEMPATH="path/to/smart-checkout-key.pem"]] ec2-user@[[EC2_PUBLIC_IP="your EC2 public IPv4 address"]]

• Verify Docker and Compose are working by running:

docker info
docker compose version

You should see Docker engine details and a Compose version number (v2.x.x).

Cannot connect to the Docker daemon?

If you see this error, the Docker service may not have started. Run sudo service docker start again. If the issue persists after reconnecting, reboot the instance with sudo reboot and SSH back in after a minute.

Help me fix Docker daemon connection issues on EC2

Deploy the stack and verify from a public URL

With Docker ready, you will clone your project, configure the environment, and bring up all 8 containers.

• Clone your project repository onto the EC2 instance by running:

git clone [[YOUR_REPO_URL="https://github.com/your-username/smart-checkout.git"]]
cd smart-checkout

• Create the .env file with your Groq API key by running:

echo "GROQ_API_KEY=your-api-key-here" > .env

Where do I find my Groq API key?

You created this key earlier in the project. Open https://console.groq.com/keys to view your existing keys. Paste the key value in place of your-api-key-here in the command above.

Before starting the stack, update the frontend so API calls use your EC2's public IP instead of localhost.

• Open frontend/index.html in a text editor on the instance:

nano frontend/index.html

• Find the line that sets the API base URL. It currently points to http://localhost:8000.
• Replace localhost with your EC2 public IP address. The line should now read something like http://YOUR_IP:8000 where YOUR_IP is the actual IP you copied earlier.
• Save the file with Ctrl+O, press Enter, then exit with Ctrl+X.

You also need to update the CORS middleware to accept requests from your public IP.

• Open the API gateway CORS configuration:

nano services/api-gateway/main.py

• Find the CORS allow_origins list. Add your EC2 public IP origin alongside the existing http://localhost entry. For example: ["http://localhost", "http://YOUR_EC2_IP"].
• Save and exit with Ctrl+O, Enter, Ctrl+X.

Now bring up the full stack in detached mode.

• Start all containers by running:

docker compose up -d

What does the -d flag do?

The -d flag runs containers in detached mode, meaning they continue running in the background after your terminal session ends. This is essential for a deployed service.

The first run pulls all Docker images, which takes 3-5 minutes depending on network speed.

• Monitor progress by running:

docker compose ps

You should see all 8 containers (postgres, redis, kafka, event-ingestion, profile-builder, offer-agent, api-gateway, nginx) with status Up.

Some containers keep restarting?

Check the logs for the failing container with docker compose logs <service-name>. Common issues include missing environment variables (check your .env file) and port conflicts. Make sure no other process is using port 80 or 8000 on the instance.

Help me debug containers that keep restarting on EC2

Time to verify the deployment from your browser.

• Open a new browser tab on your local computer.
• Navigate to http:// followed by your EC2 public IP address. You should see the Smart Checkout Demo page.
• Click on a product to add it to your cart.
• Click Proceed to Checkout.
• The offer panel should display a personalized offer with a discount and the AI agent's reasoning explanation.

What just happened?

Your browser sent a request to the nginx container on port 80, which served the frontend. The frontend JavaScript called port 8000 on your EC2 instance, hitting the API gateway. The gateway orchestrated the event ingestion, profile lookup, and agent decision, then returned a personalized offer with reasoning.

The full loop ran across 8 containers communicating over Kafka, Redis, and PostgreSQL on a real cloud server.

Your personalization engine is live on the internet. Anyone with the URL can interact with your AI-powered checkout demo. Next up, the Secret Mission will challenge you to prove the AI agent outperforms random selection.

Secret mission

Add an A/B Testing Framework

Your personalization engine is live. But how do you know the AI agent actually performs better than showing random offers? In the real world, every personalization system needs rigorous A/B testing to prove its value. Build a real-time experimentation framework that splits users into control and treatment groups, tracks conversion rates, and computes statistical significance using a chi-squared test.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Your AWS EC2 instance is covered by free tier credits, but charges begin after your free tier period expires (12 months for accounts created before July 15, 2025, or 6 months/$200 credits for newer accounts).

Cost warning

If your free tier credits expire or run out, the EC2 t3.micro instance will start incurring charges.

If you are not actively using the project, stop or terminate the instance to avoid unexpected bills. EBS storage charges still apply to stopped instances.

Resources you used:

• AWS EC2 t3.micro instance (cloud compute with attached EBS volume).
• Local project directory containing docker-compose.yml, services, frontend, and scripts.
• Docker Desktop (installed locally for running containers).
• Groq API key (free tier, no charges).

Keep everything running

No action needed. Choose this if you are still actively building or want to keep the demo live.

• Your EC2 instance continues running within free tier limits. Monitor usage in the AWS Billing Console to stay informed.
• Your local Docker containers keep running until you stop them or restart your machine.
• Docker Desktop remains installed and available for other projects.
• Your Groq API key stays active with no charges on the Free plan.

Pause - I'll come back to this later

Shut down running processes to free up resources and reduce costs while keeping everything restorable.

• Stop your EC2 instance: open the EC2 Console, select Instances, choose your instance, then click Instance state and select Stop instance. Compute charges stop, but a small EBS storage charge continues.
• Stop your local containers by opening a terminal in your project folder and running:

docker compose down

• Your project files, database data, and Docker images remain on disk for when you return.
• To resume later, start your EC2 instance from the console and run docker compose up -d on the instance.

Delete - I don't want to use this again

Remove all project resources and start fresh. This ensures zero ongoing charges.

Terminate the EC2 instance

• Open the EC2 Console and select Instances from the left sidebar.
• Select your instance, click Instance state, then select Terminate (delete) instance.
• Confirm the termination when prompted.
• After termination, navigate to Volumes in the left sidebar under Elastic Block Store. If any volumes remain in an available state, select them, click Actions, and choose Delete volume.

Delete local project files and Docker resources

• Stop and remove all project containers, networks, and volumes by running:

docker compose down -v

• Remove the Docker images built for this project by running:

docker compose down --rmi all

• Delete your project folder from the Desktop. On macOS or Linux:

cd ~/Desktop
rm -rf [[PROJECT_DIR="your project folder name"]]

On Windows (PowerShell):

cd ~\Desktop
Remove-Item -Recurse -Force [[PROJECT_DIR="your project folder name"]]

Uninstall Docker Desktop (optional)

• Open Settings, then select Apps, then Apps & features.
• Find Docker Desktop in the list and click Uninstall.
• Restart your computer to complete the removal.

Revoke your Groq API key (optional)

• Visit console.groq.com and navigate to your API Keys page.
• Delete the key you created for this project. This is optional since the Groq Free plan incurs no charges.

That's a Wrap!

That's a Wrap!

Nice work! You've just built a production-grade AI-powered personalization engine from scratch.

This isn't a tutorial app or a toy demo. It's a real distributed system with event streaming, real-time feature computation, and an AI agent that reasons about decisions.

You've learned how to:

• Design and implement a 4-service microservices architecture with event-driven communication using Apache Kafka, PostgreSQL, and Redis for real-time feature computation and caching.
• Build an agentic AI system with LangChain and Groq that uses multiple tools to reason about personalization decisions and explain its choices.
• Containerize and deploy a 7+ container system with Docker Compose to AWS EC2, giving anyone a public URL to interact with your personalization engine live.
• Built an A/B testing framework that proves the AI agent outperforms random offer selection with statistical significance.

Ready to quiz yourself?