Build a Customer Journey Agent for E-commerce in 45 Minutes

# journey_stages.py
from enum import Enum

class JourneyStage(str, Enum):
    DISCOVERY     = "discovery"      # First visit, no account
    CONSIDERATION = "consideration"  # Viewed 2+ products, no cart
    INTENT        = "intent"         # Added to cart, no checkout
    CHECKOUT      = "checkout"       # Started checkout
    RETENTION     = "retention"      # Past purchaser, returning

def infer_stage(events: list[dict]) -> JourneyStage:
    """
    Infer stage from raw event history.
    Events are ordered oldest-first.
    """
    event_types = {e["type"] for e in events}
    
    if "purchase_completed" in event_types:
        return JourneyStage.RETENTION
    if "checkout_started" in event_types:
        return JourneyStage.CHECKOUT
    if "cart_item_added" in event_types:
        return JourneyStage.INTENT
    
    product_views = [e for e in events if e["type"] == "product_viewed"]
    if len(product_views) >= 2:
        return JourneyStage.CONSIDERATION
    
    return JourneyStage.DISCOVERY

# tools.py
from langchain.tools import tool
from typing import Optional

@tool
def fetch_cart(customer_id: str) -> dict:
    """
    Returns the current cart contents for a customer.
    Use this before sending any cart-related message.
    """
    # Replace with your actual cart service call
    cart = db.query(
        "SELECT product_id, quantity, price FROM cart_items WHERE customer_id = %s",
        (customer_id,)
    )
    return {
        "items": cart,
        "total": sum(item["price"] * item["quantity"] for item in cart),
        "item_count": len(cart)
    }


@tool
def search_related_products(product_id: str, limit: int = 3) -> list[dict]:
    """
    Returns products frequently bought with the given product.
    Use when customer is in CONSIDERATION or INTENT stage.
    """
    return recommendation_engine.similar(product_id, k=limit)


@tool
def send_targeted_message(
    customer_id: str,
    channel: str,          # "email" | "sms" | "push"
    template_key: str,
    context: dict
) -> bool:
    """
    Sends a personalized message. Always call fetch_cart first
    if the template involves cart items.
    """
    template = load_template(template_key)
    rendered = template.render(**context)
    return messaging_service.send(customer_id, channel, rendered)

# agent.py
from langchain_anthropic import ChatAnthropic
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

from tools import fetch_cart, search_related_products, send_targeted_message
from journey_stages import JourneyStage

SYSTEM_PROMPT = """You are a customer journey agent for an e-commerce store.

Your job: given a customer's current journey stage and recent events, 
decide on the SINGLE best action to take right now.

Rules:
- DISCOVERY: Welcome them, don't push products yet
- CONSIDERATION: Surface related products based on what they viewed
- INTENT: Remind about cart if 30+ minutes since last activity
- CHECKOUT: Never interrupt — let them finish
- RETENTION: Reward loyalty, surface new arrivals in their preferred categories

Always call fetch_cart before referencing specific cart items.
If no action is needed, respond with "no_action" and explain why.
"""

def build_agent():
    llm = ChatAnthropic(model="claude-opus-4-6", temperature=0)
    tools = [fetch_cart, search_related_products, send_targeted_message]
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", SYSTEM_PROMPT),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ])
    
    agent = create_tool_calling_agent(llm, tools, prompt)
    return AgentExecutor(agent=agent, tools=tools, verbose=True)


def run_journey_agent(customer_id: str, stage: JourneyStage, events: list[dict]):
    agent = build_agent()
    
    input_text = f"""
    Customer ID: {customer_id}
    Current Stage: {stage.value}
    Recent Events (last 10): {events[-10:]}
    
    What action should we take right now?
    """
    
    return agent.invoke({"input": input_text})

# consumer.py
from kafka import KafkaConsumer
import json

from journey_stages import infer_stage
from agent import run_journey_agent

TRIGGER_EVENTS = {
    "cart_item_added",
    "checkout_abandoned",
    "product_viewed",
    "purchase_completed",
    "session_started",
}

consumer = KafkaConsumer(
    "storefront-events",
    bootstrap_servers="localhost:9092",
    value_deserializer=lambda m: json.loads(m.decode("utf-8"))
)

for message in consumer:
    event = message.value
    
    # Only run agent on meaningful trigger events
    if event["type"] not in TRIGGER_EVENTS:
        continue
    
    customer_id = event["customer_id"]
    history = fetch_event_history(customer_id, limit=50)
    stage = infer_stage(history)
    
    result = run_journey_agent(customer_id, stage, history)
    print(f"[{customer_id}] Stage: {stage} | Action: {result['output']}")

pip install pytest pytest-asyncio langchain-anthropic kafka-python
pytest tests/test_journey_agent.py -v