State Management for E-commerce

In e-commerce, context is everything. When a customer asks "What's the status of my order?" followed by "Can I change the shipping address?", your agent needs to remember which order they were discussing. This is where state management becomes critical.

State is Your Agent's Memory

Designing E-commerce State

A well-designed state schema captures everything your workflow needs:

from typing import TypedDict, Literal, Annotated, Optional
from langchain_core.messages import BaseMessage
import operator

class CustomerServiceState(TypedDict):
    # Conversation context
    messages: Annotated[list[BaseMessage], operator.add]  # Append reducer
    customer_id: str
    session_id: str

    # Current context
    current_order_id: Optional[str]
    current_topic: Literal["order", "return", "product", "general", None]

    # Customer data (fetched from DB)
    customer_name: str
    customer_tier: Literal["standard", "premium", "vip"]
    recent_orders: list[dict]

    # Workflow state
    sentiment: Literal["positive", "neutral", "frustrated", "angry"]
    needs_escalation: bool
    resolved: bool

State Reducers: Handling Updates

When multiple nodes update the same field, reducers define how to combine them.

The Problem: Overwriting

Without a reducer, returning {"messages": [new_msg]} would replace all messages:

# Node 1 returns
{"messages": [Message("Hello!")]}

# Node 2 returns
{"messages": [Message("How can I help?")]}

# Result WITHOUT reducer: only the second message exists
{"messages": [Message("How can I help?")]}  # "Hello!" is lost!

The Solution: Append Reducer

from typing import Annotated
import operator

class ChatState(TypedDict):
    # The operator.add reducer appends lists together
    messages: Annotated[list[BaseMessage], operator.add]

Now:

# Node 1 returns
{"messages": [Message("Hello!")]}

# Node 2 returns
{"messages": [Message("How can I help?")]}

# Result WITH reducer: both messages preserved
{"messages": [Message("Hello!"), Message("How can I help?")]}

Custom Reducers

For complex e-commerce scenarios, you can define custom reducers:

from typing import Annotated

def merge_order_updates(existing: dict, new: dict) -> dict:
    """
    Custom reducer for order data.
    New values override, but nested dicts are merged.
    """
    if existing is None:
        return new
    if new is None:
        return existing

    merged = existing.copy()
    for key, value in new.items():
        if isinstance(value, dict) and isinstance(merged.get(key), dict):
            merged[key] = {**merged[key], **value}
        else:
            merged[key] = value
    return merged

class OrderState(TypedDict):
    order_data: Annotated[dict, merge_order_updates]

Checkpointing: Persistent State

For real e-commerce, conversations span hours or days. Checkpointers save state to a database.

from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres import PostgresSaver

# For development: in-memory checkpointing
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# For production: PostgreSQL checkpointing
# DB_URI = "postgresql://user:pass@localhost/langgraph"
# with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
#     app = workflow.compile(checkpointer=checkpointer)

Thread IDs: Identifying Conversations

Each conversation gets a unique thread ID:

# First message from customer
config = {"configurable": {"thread_id": "customer_123_session_456"}}

result = app.invoke(
    {"messages": [HumanMessage("What's the status of order #789?")]},
    config=config
)

# ... hours later, same customer returns ...

# Continue the same conversation
result = app.invoke(
    {"messages": [HumanMessage("Can I return that item?")]},
    config=config  # Same thread_id - context preserved!
)

Practical Example: Order Inquiry Flow

Let's build a stateful order inquiry agent:

from langgraph.graph import StateGraph, START, END
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated, Optional, Literal
import operator

# State schema
class OrderInquiryState(TypedDict):
    messages: Annotated[list, operator.add]
    customer_id: str
    order_id: Optional[str]
    order_details: Optional[dict]
    inquiry_type: Optional[Literal["status", "modify", "cancel", "other"]]
    resolved: bool

# Mock database
ORDERS_DB = {
    "ORD-123": {
        "status": "shipped",
        "items": [{"name": "Wireless Headphones", "qty": 1}],
        "tracking": "1Z999AA10123456784",
        "estimated_delivery": "Dec 15, 2024"
    },
    "ORD-456": {
        "status": "processing",
        "items": [{"name": "USB-C Cable", "qty": 3}],
        "tracking": None,
        "estimated_delivery": "Dec 18, 2024"
    }
}

llm = ChatOpenAI(model="gpt-4o-mini")

# Node: Extract order ID from conversation
def extract_order_info(state: OrderInquiryState) -> dict:
    """Extract order ID and inquiry type from the conversation"""
    last_message = state["messages"][-1].content

    # Simple extraction (in production, use an LLM or NER)
    order_id = None
    for word in last_message.split():
        if word.upper().startswith("ORD-"):
            order_id = word.upper()
            break

    # Determine inquiry type
    inquiry_type = "status"  # default
    if "cancel" in last_message.lower():
        inquiry_type = "cancel"
    elif "change" in last_message.lower() or "modify" in last_message.lower():
        inquiry_type = "modify"

    return {
        "order_id": order_id,
        "inquiry_type": inquiry_type
    }

# Node: Fetch order from database
def fetch_order(state: OrderInquiryState) -> dict:
    """Look up order details from database"""
    order_id = state.get("order_id")
    if order_id and order_id in ORDERS_DB:
        return {"order_details": ORDERS_DB[order_id]}
    return {"order_details": None}

# Node: Generate response
def generate_response(state: OrderInquiryState) -> dict:
    """Generate contextual response based on order data"""
    order = state.get("order_details")
    inquiry = state.get("inquiry_type")

    if not state.get("order_id"):
        response = "I'd be happy to help! Could you please provide your order number? It starts with ORD-"
    elif not order:
        response = f"I couldn't find order {state['order_id']}. Please double-check the order number."
    elif inquiry == "status":
        if order["status"] == "shipped":
            response = f"""Great news! Your order {state['order_id']} has shipped!

**Tracking Number:** {order['tracking']}
**Estimated Delivery:** {order['estimated_delivery']}
**Items:** {', '.join(item['name'] for item in order['items'])}

Would you like me to help with anything else?"""
        else:
            response = f"""Your order {state['order_id']} is currently being processed.

**Status:** {order['status'].title()}
**Estimated Delivery:** {order['estimated_delivery']}

You'll receive a tracking number once it ships!"""
    elif inquiry == "cancel":
        if order["status"] == "shipped":
            response = f"Order {state['order_id']} has already shipped. Would you like to initiate a return instead?"
        else:
            response = f"I can help cancel order {state['order_id']}. This will refund ${sum(49.99 for _ in order['items']):.2f}. Should I proceed?"
    else:
        response = "I understand you want to modify your order. Let me connect you with our order specialist."

    return {
        "messages": [AIMessage(content=response)],
        "resolved": True
    }

# Build graph
workflow = StateGraph(OrderInquiryState)

workflow.add_node("extract", extract_order_info)
workflow.add_node("fetch", fetch_order)
workflow.add_node("respond", generate_response)

workflow.add_edge(START, "extract")
workflow.add_edge("extract", "fetch")
workflow.add_edge("fetch", "respond")
workflow.add_edge("respond", END)

# Compile with checkpointing
from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

Running the Conversation

# Session config
config = {"configurable": {"thread_id": "user_abc_session_1"}}

# First message
result = app.invoke({
    "messages": [HumanMessage("Hi, where's my order ORD-123?")],
    "customer_id": "user_abc",
    "resolved": False
}, config)

print(result["messages"][-1].content)
# Output: Great news! Your order ORD-123 has shipped!
#         Tracking Number: 1Z999AA10123456784
#         ...

# Follow-up (context preserved!)
result = app.invoke({
    "messages": [HumanMessage("Actually, can I cancel it?")]
}, config)

print(result["messages"][-1].content)
# Output: Order ORD-123 has already shipped. Would you like to initiate a return instead?

State Design Best Practices

Key Takeaways

Next up: Customer Service Agent - Building a production-ready support chatbot with routing, tools, and memory.