Order History Lookup Agent

"Where's my order?" is the most common customer service question. An Order History Agent can answer it instantly - and help with returns, reorders, and more.

Architecture: RAG + Database Queries

Unlike pure RAG, order lookup combines:

• Structured queries for order data (SQL/NoSQL)
• Vector search for product matching ("those headphones I bought")
• LLM generation for natural responses

from typing import Dict, List, Optional
from datetime import datetime, timedelta
from openai import OpenAI
import json

# Database models (example with SQLAlchemy)
class Order:
    id: str
    customer_id: str
    status: str  # pending, processing, shipped, delivered, cancelled
    total: float
    created_at: datetime
    shipping_address: Dict
    tracking_number: Optional[str]
    carrier: Optional[str]
    items: List[Dict]  # [{product_id, name, quantity, price}]


class OrderHistoryAgent:
    def __init__(self, db, openai_client, product_index):
        self.db = db
        self.client = openai_client
        self.product_index = product_index

    def handle_query(self, customer_id: str, query: str) -> Dict:
        """Handle an order-related query"""

        # Step 1: Understand what they're asking
        intent = self._classify_intent(query)

        # Step 2: Identify which order(s) they mean
        orders = self._identify_orders(customer_id, query, intent)

        # Step 3: Execute appropriate action
        if intent["type"] == "track_order":
            return self._handle_tracking(orders, query)
        elif intent["type"] == "order_history":
            return self._handle_history(orders, query)
        elif intent["type"] == "return_request":
            return self._handle_return(orders, query)
        elif intent["type"] == "reorder":
            return self._handle_reorder(orders, query)
        else:
            return self._handle_general(orders, query)

    def _classify_intent(self, query: str) -> Dict:
        """Classify the customer's intent"""

        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": """Classify the customer's order-related query.
Return JSON with:
- type: "track_order", "order_history", "return_request", "reorder", "general"
- product_hint: Any product mentioned (null if none)
- time_hint: Any time reference like "last week" (null if none)
- order_id_hint: Any order number mentioned (null if none)"""
                },
                {"role": "user", "content": query}
            ],
            response_format={"type": "json_object"}
        )

        return json.loads(response.choices[0].message.content)

    def _identify_orders(
        self,
        customer_id: str,
        query: str,
        intent: Dict
    ) -> List[Order]:
        """Identify which order(s) the customer is asking about"""

        # If they gave an order ID, use it directly
        if intent.get("order_id_hint"):
            order = self.db.get_order(intent["order_id_hint"])
            if order and order.customer_id == customer_id:
                return [order]

        # Otherwise, search their orders
        filters = {"customer_id": customer_id}

        # Apply time filter
        if intent.get("time_hint"):
            time_range = self._parse_time_hint(intent["time_hint"])
            filters["created_after"] = time_range["start"]
            filters["created_before"] = time_range["end"]

        # Get matching orders
        orders = self.db.get_orders(**filters)

        # If they mentioned a product, filter by that
        if intent.get("product_hint"):
            orders = self._filter_by_product(orders, intent["product_hint"])

        return orders[:10]  # Limit to 10 most recent

    def _filter_by_product(self, orders: List[Order], product_hint: str) -> List[Order]:
        """Filter orders by product description using semantic search"""

        matching_orders = []

        for order in orders:
            for item in order.items:
                # Use semantic similarity to match product
                similarity = self._check_product_match(item["name"], product_hint)
                if similarity > 0.7:
                    matching_orders.append(order)
                    break

        return matching_orders

    def _check_product_match(self, product_name: str, query: str) -> float:
        """Check semantic similarity between product and query"""
        # Simplified - in production use embeddings
        query_embedding = self.client.embeddings.create(
            model="text-embedding-3-small",
            input=query
        ).data[0].embedding

        product_embedding = self.client.embeddings.create(
            model="text-embedding-3-small",
            input=product_name
        ).data[0].embedding

        # Calculate cosine similarity
        import numpy as np
        return np.dot(query_embedding, product_embedding)

Handling Order Tracking

def _handle_tracking(self, orders: List[Order], query: str) -> Dict:
    """Handle 'where's my order' type queries"""

    if not orders:
        return {
            "response": "I couldn't find any orders matching your description. Could you provide your order number or more details about what you ordered?",
            "orders": [],
            "action": None
        }

    if len(orders) == 1:
        order = orders[0]
        return self._format_tracking_response(order)

    # Multiple possible orders - ask for clarification
    return self._ask_which_order(orders, "track")


def _format_tracking_response(self, order: Order) -> Dict:
    """Format a tracking response for a single order"""

    status_messages = {
        "pending": "Your order has been received and is being prepared.",
        "processing": "Your order is being packed and will ship soon.",
        "shipped": f"Great news! Your order is on its way.",
        "delivered": "Your order has been delivered!",
        "cancelled": "This order was cancelled."
    }

    response_parts = [
        f"**Order #{order.id}** - {order.status.upper()}",
        "",
        status_messages.get(order.status, "Status: " + order.status),
        ""
    ]

    # Add tracking info if shipped
    if order.tracking_number:
        response_parts.extend([
            f"📦 **Tracking:** {order.tracking_number}",
            f"🚚 **Carrier:** {order.carrier}",
            f"[Track Package]({self._get_tracking_url(order.carrier, order.tracking_number)})",
            ""
        ])

    # Add estimated delivery
    if order.status == "shipped" and order.estimated_delivery:
        response_parts.append(
            f"📅 **Estimated Delivery:** {order.estimated_delivery.strftime('%A, %B %d')}"
        )

    # List items
    response_parts.extend(["", "**Items:**"])
    for item in order.items:
        response_parts.append(f"- {item['name']} (x{item['quantity']})")

    return {
        "response": "\n".join(response_parts),
        "orders": [self._serialize_order(order)],
        "action": "tracking_shown"
    }


def _ask_which_order(self, orders: List[Order], action: str) -> Dict:
    """Ask customer to clarify which order"""

    response = f"I found {len(orders)} orders that might match. Which one are you asking about?\n\n"

    for i, order in enumerate(orders[:5], 1):
        items_preview = ", ".join([item["name"][:30] for item in order.items[:2]])
        if len(order.items) > 2:
            items_preview += f" (+{len(order.items) - 2} more)"

        response += f"{i}. **Order #{order.id}** ({order.created_at.strftime('%b %d')})\n"
        response += f"   {items_preview}\n"
        response += f"   Status: {order.status}\n\n"

    response += "Just reply with the order number or say '1', '2', etc."

    return {
        "response": response,
        "orders": [self._serialize_order(o) for o in orders[:5]],
        "action": "clarification_needed"
    }

Handling Order History

def _handle_history(self, orders: List[Order], query: str) -> Dict:
    """Handle order history queries"""

    if not orders:
        return {
            "response": "You don't have any orders yet. Ready to start shopping?",
            "orders": [],
            "action": None
        }

    # Generate a summary
    response = self.client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": """Summarize the customer's order history based on their question.
Be conversational and highlight relevant patterns.
If they asked about specific products, focus on those.
Format with clear sections and bullet points."""
            },
            {
                "role": "user",
                "content": f"""Question: {query}

Order history (most recent first):
{self._format_orders_for_llm(orders)}"""
            }
        ]
    )

    return {
        "response": response.choices[0].message.content,
        "orders": [self._serialize_order(o) for o in orders],
        "action": "history_shown"
    }


def _format_orders_for_llm(self, orders: List[Order]) -> str:
    """Format orders for LLM context"""
    formatted = []
    for order in orders:
        items = ", ".join([f"{item['name']} (x{item['quantity']})" for item in order.items])
        formatted.append(
            f"Order #{order.id} | {order.created_at.strftime('%Y-%m-%d')} | "
            f"{order.status} | ${order.total:.2f} | Items: {items}"
        )
    return "\n".join(formatted)

Handling Reorders

def _handle_reorder(self, orders: List[Order], query: str) -> Dict:
    """Handle reorder requests"""

    if not orders:
        return {
            "response": "I couldn't find the order you want to reorder. Could you describe what you'd like to buy again?",
            "orders": [],
            "action": None
        }

    order = orders[0]  # Most relevant order

    # Check item availability
    reorderable_items = []
    unavailable_items = []

    for item in order.items:
        product = self.db.get_product(item["product_id"])
        if product and product.in_stock:
            reorderable_items.append({
                **item,
                "current_price": product.price,
                "price_changed": product.price != item["price"]
            })
        else:
            unavailable_items.append(item)

    response_parts = ["I found that order! Here's what I can reorder for you:\n"]

    if reorderable_items:
        response_parts.append("**Available to reorder:**")
        for item in reorderable_items:
            line = f"- {item['name']} (x{item['quantity']}) - ${item['current_price']:.2f}"
            if item["price_changed"]:
                line += f" *(was ${item['price']:.2f})*"
            response_parts.append(line)

    if unavailable_items:
        response_parts.append("\n**Currently unavailable:**")
        for item in unavailable_items:
            response_parts.append(f"- {item['name']}")

    if reorderable_items:
        total = sum(i["current_price"] * i["quantity"] for i in reorderable_items)
        response_parts.extend([
            f"\n**Estimated total:** ${total:.2f}",
            "\nWould you like me to add these to your cart?"
        ])

    return {
        "response": "\n".join(response_parts),
        "orders": [self._serialize_order(order)],
        "reorderable_items": reorderable_items,
        "action": "reorder_ready"
    }

Example Conversations

Security Considerations

def verify_order_access(self, customer_id: str, order_id: str) -> bool:
    """Verify customer can access this order"""
    order = self.db.get_order(order_id)
    if not order:
        return False
    return order.customer_id == customer_id


def mask_sensitive_data(self, order: Order) -> Order:
    """Mask sensitive information for display"""
    masked = order.copy()

    # Mask address
    if masked.shipping_address:
        addr = masked.shipping_address
        masked.shipping_address = {
            "name": addr.get("name"),
            "city": addr.get("city"),
            "state": addr.get("state"),
            "zip": addr.get("zip", "")[:3] + "**"
        }

    # Never expose payment details
    masked.payment_method = "****" + str(order.payment_last_four)

    return masked

Key Takeaways

Next up: Product Comparison Agent - Build an agent that helps customers compare products side-by-side.