AI Chatbot for E-Commerce: Product Discovery, Cart Assistance, and Checkout Help

E-Commerce AI Beyond FAQ Bots

Most e-commerce chatbots are glorified FAQ pages that frustrate more than they help. A truly useful shopping assistant understands product catalogs, manages cart state, compares options side by side, and guides users through checkout — all through natural conversation. The difference is tool integration: the chatbot needs real-time access to your product database, cart API, and payment system.

Product Search and Discovery

The chatbot's most important capability is helping users find products. This means translating natural language queries ("I need a waterproof jacket for hiking under 200 dollars") into structured database queries.

from dataclasses import dataclass
from typing import Optional
import asyncpg


@dataclass
class ProductResult:
    id: str
    name: str
    price: float
    image_url: str
    rating: float
    review_count: int
    in_stock: bool
    short_description: str


async def search_products(
    pool: asyncpg.Pool,
    query: str,
    category: Optional[str] = None,
    min_price: Optional[float] = None,
    max_price: Optional[float] = None,
    sort_by: str = "relevance",
    limit: int = 5,
) -> list[ProductResult]:
    """Full-text search with filters."""
    conditions = ["p.active = true"]
    params = []
    idx = 1

    # Full-text search using PostgreSQL tsvector
    conditions.append(
        f"p.search_vector @@ plainto_tsquery('english', ${idx})"
    )
    params.append(query)
    idx += 1

    if category:
        conditions.append(f"p.category = ${idx}")
        params.append(category)
        idx += 1
    if min_price is not None:
        conditions.append(f"p.price >= ${idx}")
        params.append(min_price)
        idx += 1
    if max_price is not None:
        conditions.append(f"p.price <= ${idx}")
        params.append(max_price)
        idx += 1

    where = " AND ".join(conditions)

    order_map = {
        "relevance": f"ts_rank(p.search_vector, plainto_tsquery('english', $1)) DESC",
        "price_low": "p.price ASC",
        "price_high": "p.price DESC",
        "rating": "p.rating DESC",
    }
    order = order_map.get(sort_by, order_map["relevance"])

    sql = f"""
        SELECT p.id, p.name, p.price, p.image_url, p.rating,
               p.review_count, (p.stock_count > 0) as in_stock,
               p.short_description
        FROM products p
        WHERE {where}
        ORDER BY {order}
        LIMIT {limit}
    """
    rows = await pool.fetch(sql, *params)
    return [ProductResult(**dict(r)) for r in rows]

Product Comparison

When users are deciding between options, the chatbot should present a structured comparison rather than asking the user to remember details from previous messages.

async def compare_products(
    pool: asyncpg.Pool, product_ids: list[str]
) -> str:
    """Generate a comparison table for the given products."""
    rows = await pool.fetch(
        """SELECT id, name, price, rating, review_count,
                  brand, weight, dimensions, key_features
           FROM products WHERE id = ANY($1)""",
        product_ids,
    )

    if not rows:
        return "Could not find the requested products."

    # Build comparison text
    headers = ["Feature"] + [r["name"] for r in rows]
    comparison = {
        "Price": [f"${r['price']:.2f}" for r in rows],
        "Rating": [f"{r['rating']}/5 ({r['review_count']} reviews)" for r in rows],
        "Brand": [r["brand"] for r in rows],
        "Weight": [r["weight"] or "N/A" for r in rows],
    }

    lines = []
    for feature, values in comparison.items():
        lines.append(f"**{feature}**: " + " | ".join(values))
    return "\n".join(lines)

Cart Management

The chatbot maintains cart state through a session-based cart service. Users can add, remove, and modify items through natural conversation.

from dataclasses import dataclass, field


@dataclass
class CartItem:
    product_id: str
    product_name: str
    quantity: int
    unit_price: float

    @property
    def subtotal(self) -> float:
        return self.quantity * self.unit_price


@dataclass
class Cart:
    session_id: str
    items: list[CartItem] = field(default_factory=list)

    @property
    def total(self) -> float:
        return sum(item.subtotal for item in self.items)

    @property
    def item_count(self) -> int:
        return sum(item.quantity for item in self.items)


class CartService:
    def __init__(self, redis_client):
        self.redis = redis_client

    async def get_cart(self, session_id: str) -> Cart:
        import json
        data = await self.redis.get(f"cart:{session_id}")
        if not data:
            return Cart(session_id=session_id)
        cart_data = json.loads(data)
        items = [CartItem(**item) for item in cart_data.get("items", [])]
        return Cart(session_id=session_id, items=items)

    async def add_item(
        self, session_id: str, product_id: str,
        product_name: str, price: float, quantity: int = 1,
    ) -> Cart:
        import json
        cart = await self.get_cart(session_id)
        # Check if item already in cart
        for item in cart.items:
            if item.product_id == product_id:
                item.quantity += quantity
                break
        else:
            cart.items.append(CartItem(
                product_id=product_id,
                product_name=product_name,
                quantity=quantity,
                unit_price=price,
            ))
        await self.redis.set(
            f"cart:{session_id}",
            json.dumps({"items": [vars(i) for i in cart.items]}),
            ex=3600 * 24,  # 24-hour expiry
        )
        return cart

    async def remove_item(self, session_id: str, product_id: str) -> Cart:
        import json
        cart = await self.get_cart(session_id)
        cart.items = [i for i in cart.items if i.product_id != product_id]
        await self.redis.set(
            f"cart:{session_id}",
            json.dumps({"items": [vars(i) for i in cart.items]}),
            ex=3600 * 24,
        )
        return cart

Wiring the Chatbot Agent

The agent uses tools for each capability. The LLM decides which tool to call based on the user's message, maintaining a natural conversational flow.

from agents import Agent, function_tool


@function_tool
async def search(
    query: str, category: str = "", max_price: float = 0
) -> str:
    """Search for products by description, category, and price."""
    pool = await get_db_pool()
    results = await search_products(
        pool, query,
        category=category or None,
        max_price=max_price if max_price > 0 else None,
    )
    if not results:
        return "No products found matching your search."
    lines = []
    for p in results:
        stock = "In stock" if p.in_stock else "Out of stock"
        lines.append(
            f"- **{p.name}** (${p.price:.2f}) - {p.rating}/5 "
            f"({p.review_count} reviews) - {stock}\n"
            f"  {p.short_description}"
        )
    return "\n".join(lines)


@function_tool
async def add_to_cart(product_id: str, quantity: int = 1) -> str:
    """Add a product to the shopping cart."""
    pool = await get_db_pool()
    product = await pool.fetchrow(
        "SELECT name, price, stock_count FROM products WHERE id = $1",
        product_id,
    )
    if not product:
        return "Product not found."
    if product["stock_count"] < quantity:
        return f"Only {product['stock_count']} units available."
    cart_svc = CartService(await get_redis())
    cart = await cart_svc.add_item(
        get_session_id(), product_id,
        product["name"], float(product["price"]), quantity,
    )
    return (
        f"Added {quantity}x {product['name']} to cart. "
        f"Cart total: ${cart.total:.2f} ({cart.item_count} items)"
    )


@function_tool
async def view_cart() -> str:
    """Show current cart contents."""
    cart_svc = CartService(await get_redis())
    cart = await cart_svc.get_cart(get_session_id())
    if not cart.items:
        return "Your cart is empty."
    lines = [f"- {i.product_name} x{i.quantity} = ${i.subtotal:.2f}"
             for i in cart.items]
    lines.append(f"\n**Total: ${cart.total:.2f}**")
    return "\n".join(lines)


shopping_agent = Agent(
    name="ShoppingAssistant",
    instructions="""You are a helpful shopping assistant. Help customers
    find products, compare options, and manage their cart. Always confirm
    before adding items. Suggest related products when appropriate.""",
    tools=[search, add_to_cart, view_cart],
)

FAQ

How do I handle product IDs in conversation without exposing internal IDs to users?

Map products to short display references during the conversation (e.g., "Option A", "Option B"). Maintain an internal mapping from display references to product IDs within the session. When the user says "add Option A to my cart," resolve it to the actual product ID before calling the cart service.

How do I prevent the chatbot from recommending out-of-stock products?

Filter by stock availability at the query level (the WHERE stock_count > 0 clause) so out-of-stock products never appear in results. If a user specifically asks about an unavailable product, inform them it is out of stock and suggest similar in-stock alternatives using a follow-up search filtered to the same category.

How do I handle concurrent cart modifications from the same user across tabs?

Use Redis atomic operations like WATCH/MULTI/EXEC or Lua scripts to handle race conditions. Each cart operation should read-modify-write atomically. If a conflict is detected, re-read the cart state and retry the operation. For most e-commerce traffic, simple Redis serialization is sufficient without distributed locks.

---

#ECommerce #Chatbot #ProductSearch #CartManagement #Python #AgenticAI #LearnAI #AIEngineering