Building a Customer Service Agent

Customer service is where AI can deliver immediate ROI. Klarna reported their AI assistant handles 2/3 of all customer chats - equivalent to 700 full-time agents. Let's build a production-quality customer service system with LangGraph.

Architecture Overview

We'll build a router + specialist architecture where a router agent classifies incoming queries and routes them to specialized sub-agents:

                    Customer Message
                           │
                           ▼
                    ┌─────────────┐
                    │   Router    │
                    │   Agent     │
                    └──────┬──────┘
                           │
         ┌─────────┬───────┴───────┬─────────┐
         ▼         ▼               ▼         ▼
    ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
    │  Order  │ │ Returns │ │ Product │ │  Human  │
    │  Agent  │ │  Agent  │ │  Agent  │ │  Agent  │
    └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘
         │           │           │           │
         └───────────┴─────┬─────┴───────────┘
                           ▼
                    ┌─────────────┐
                    │  Response   │
                    │  Formatter  │
                    └─────────────┘

Step 1: Define the State

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

class CustomerServiceState(TypedDict):
    # Conversation
    messages: Annotated[list[BaseMessage], operator.add]

    # Customer context
    customer_id: str
    customer_name: str
    customer_email: str
    customer_tier: Literal["standard", "premium", "vip"]

    # Routing
    query_type: Optional[Literal["order", "return", "product", "general", "escalate"]]
    sentiment: Literal["positive", "neutral", "frustrated", "angry"]

    # Data fetched by agents
    order_data: Optional[dict]
    product_data: Optional[dict]
    return_data: Optional[dict]

    # Workflow control
    needs_human: bool
    response_ready: bool

Step 2: Build the Router Agent

The router classifies incoming messages and detects sentiment:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

router_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

ROUTER_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are a customer service router for an e-commerce store.

Analyze the customer message and determine:
1. query_type: order | return | product | general | escalate
2. sentiment: positive | neutral | frustrated | angry

Classification rules:
- "order": tracking, delivery, order status, shipping
- "return": returns, refunds, exchanges, damaged items
- "product": questions about products, availability, specs
- "general": greetings, store hours, policies
- "escalate": complaints, legal threats, requests for manager

ALWAYS escalate if sentiment is "angry" or message mentions "lawyer", "sue", "manager".

Customer tier: {customer_tier} (VIP customers get priority escalation)

Respond in JSON format:
{{"query_type": "...", "sentiment": "...", "reasoning": "..."}}"""),
    ("human", "{message}")
])

def route_query(state: CustomerServiceState) -> dict:
    """Classify the query and detect sentiment"""
    last_message = state["messages"][-1].content

    response = router_llm.invoke(
        ROUTER_PROMPT.format(
            customer_tier=state["customer_tier"],
            message=last_message
        )
    )

    # Parse JSON response
    import json
    result = json.loads(response.content)

    # VIP escalation for frustrated customers
    needs_human = (
        result["sentiment"] == "angry" or
        result["query_type"] == "escalate" or
        (state["customer_tier"] == "vip" and result["sentiment"] == "frustrated")
    )

    return {
        "query_type": result["query_type"],
        "sentiment": result["sentiment"],
        "needs_human": needs_human
    }

Step 3: Build Specialized Agents

Order Agent

from langchain_core.tools import tool

# Tools for the order agent
@tool
def lookup_order(order_id: str) -> dict:
    """Look up order details by order ID"""
    # In production: query your orders database
    orders_db = {
        "ORD-001": {
            "status": "shipped",
            "tracking": "1Z999AA10123456784",
            "carrier": "UPS",
            "estimated_delivery": "Dec 15, 2024",
            "items": [{"name": "Wireless Headphones", "price": 149.99}]
        },
        "ORD-002": {
            "status": "processing",
            "tracking": None,
            "carrier": None,
            "estimated_delivery": "Dec 18, 2024",
            "items": [{"name": "USB-C Hub", "price": 79.99}]
        }
    }
    return orders_db.get(order_id, {"error": "Order not found"})

@tool
def get_customer_orders(customer_id: str) -> list:
    """Get all orders for a customer"""
    # In production: query by customer_id
    return ["ORD-001", "ORD-002"]

ORDER_AGENT_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are an order specialist for ShopMax e-commerce.

Customer: {customer_name} ({customer_tier} tier)

You have access to:
- lookup_order: Get details for a specific order
- get_customer_orders: List all customer orders

Guidelines:
- Be empathetic and professional
- For VIP/Premium customers, offer expedited shipping if there are delays
- Always provide tracking links when available
- If order is delayed > 3 days, offer 10% discount on next order

Previous conversation:
{history}"""),
    ("human", "{message}")
])

order_agent = router_llm.bind_tools([lookup_order, get_customer_orders])

def handle_order_query(state: CustomerServiceState) -> dict:
    """Handle order-related queries"""
    messages = state["messages"]
    last_message = messages[-1].content

    # Format history
    history = "\n".join([
        f"{'Customer' if isinstance(m, HumanMessage) else 'Agent'}: {m.content}"
        for m in messages[:-1]
    ])

    # Invoke with tools
    response = order_agent.invoke(
        ORDER_AGENT_PROMPT.format(
            customer_name=state["customer_name"],
            customer_tier=state["customer_tier"],
            history=history,
            message=last_message
        )
    )

    # Process tool calls if any
    if response.tool_calls:
        tool_results = []
        for tool_call in response.tool_calls:
            if tool_call["name"] == "lookup_order":
                result = lookup_order.invoke(tool_call["args"])
                tool_results.append(result)
            elif tool_call["name"] == "get_customer_orders":
                result = get_customer_orders.invoke(tool_call["args"])
                tool_results.append(result)

        # Generate final response with tool results
        final_response = router_llm.invoke([
            {"role": "system", "content": f"Tool results: {tool_results}. Generate a helpful response."},
            {"role": "user", "content": last_message}
        ])
        return {
            "messages": [AIMessage(content=final_response.content)],
            "order_data": tool_results[0] if tool_results else None,
            "response_ready": True
        }

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

Returns Agent

@tool
def check_return_eligibility(order_id: str) -> dict:
    """Check if an order is eligible for return"""
    # In production: check order date, item condition policies
    return {
        "eligible": True,
        "reason": "Within 30-day return window",
        "refund_amount": 149.99,
        "return_label_url": "https://shop.com/returns/label/ORD-001"
    }

@tool
def initiate_return(order_id: str, reason: str) -> dict:
    """Start the return process"""
    return {
        "return_id": "RET-12345",
        "status": "initiated",
        "instructions": "Print label, pack item, drop at UPS"
    }

RETURNS_AGENT_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are a returns specialist for ShopMax.

Customer: {customer_name} ({customer_tier} tier)

Guidelines:
- Check eligibility before initiating return
- For VIP customers, offer instant refund (before item received)
- Be understanding about damaged/defective items
- Offer exchange as alternative to refund

Tools available:
- check_return_eligibility: Verify return is allowed
- initiate_return: Start the return process"""),
    ("human", "{message}")
])

returns_agent = router_llm.bind_tools([check_return_eligibility, initiate_return])

def handle_return_query(state: CustomerServiceState) -> dict:
    """Handle return and refund queries"""
    # Similar implementation to order agent
    response = returns_agent.invoke(
        RETURNS_AGENT_PROMPT.format(
            customer_name=state["customer_name"],
            customer_tier=state["customer_tier"],
            message=state["messages"][-1].content
        )
    )

    # Process tools and generate response...
    return {
        "messages": [AIMessage(content=response.content)],
        "response_ready": True
    }

Product Agent

@tool
def search_products(query: str) -> list:
    """Search product catalog"""
    products = [
        {"id": "PROD-001", "name": "Wireless Headphones", "price": 149.99, "in_stock": True},
        {"id": "PROD-002", "name": "Bluetooth Speaker", "price": 79.99, "in_stock": False},
        {"id": "PROD-003", "name": "USB-C Hub", "price": 49.99, "in_stock": True},
    ]
    return [p for p in products if query.lower() in p["name"].lower()]

@tool
def get_product_details(product_id: str) -> dict:
    """Get detailed product information"""
    return {
        "id": product_id,
        "name": "Wireless Headphones",
        "price": 149.99,
        "description": "Premium noise-canceling headphones with 30hr battery",
        "specs": {"battery": "30 hours", "driver": "40mm", "weight": "250g"},
        "reviews_avg": 4.7,
        "in_stock": True
    }

product_agent = router_llm.bind_tools([search_products, get_product_details])

def handle_product_query(state: CustomerServiceState) -> dict:
    """Handle product-related queries"""
    response = product_agent.invoke([
        {"role": "system", "content": "You are a product expert. Help customers find products."},
        {"role": "user", "content": state["messages"][-1].content}
    ])

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

Step 4: Build the Complete Graph

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

def route_to_specialist(state: CustomerServiceState) -> str:
    """Route to the appropriate specialist based on query type"""
    if state["needs_human"]:
        return "human_handoff"

    routing_map = {
        "order": "order_agent",
        "return": "returns_agent",
        "product": "product_agent",
        "general": "general_agent",
        "escalate": "human_handoff"
    }
    return routing_map.get(state["query_type"], "general_agent")

def handle_general_query(state: CustomerServiceState) -> dict:
    """Handle general queries (FAQs, store info)"""
    response = router_llm.invoke([
        {"role": "system", "content": """You are a helpful assistant for ShopMax.
        Store hours: 9am-9pm EST
        Free shipping on orders over $50
        30-day return policy"""},
        {"role": "user", "content": state["messages"][-1].content}
    ])
    return {
        "messages": [AIMessage(content=response.content)],
        "response_ready": True
    }

def human_handoff(state: CustomerServiceState) -> dict:
    """Prepare for human agent handoff"""
    summary = f"""
**Escalation Summary**
Customer: {state["customer_name"]} ({state["customer_tier"]})
Sentiment: {state["sentiment"]}
Query Type: {state["query_type"]}

Conversation:
{chr(10).join(m.content for m in state["messages"])}
"""
    return {
        "messages": [AIMessage(content=f"I'm connecting you with a specialist who can better assist you. Please hold for a moment. Your reference number is ESC-{state['customer_id'][:8]}")],
        "response_ready": True
    }

# Build the graph
workflow = StateGraph(CustomerServiceState)

# Add nodes
workflow.add_node("router", route_query)
workflow.add_node("order_agent", handle_order_query)
workflow.add_node("returns_agent", handle_return_query)
workflow.add_node("product_agent", handle_product_query)
workflow.add_node("general_agent", handle_general_query)
workflow.add_node("human_handoff", human_handoff)

# Add edges
workflow.add_edge(START, "router")
workflow.add_conditional_edges(
    "router",
    route_to_specialist,
    {
        "order_agent": "order_agent",
        "returns_agent": "returns_agent",
        "product_agent": "product_agent",
        "general_agent": "general_agent",
        "human_handoff": "human_handoff"
    }
)

# All specialists lead to END
for agent in ["order_agent", "returns_agent", "product_agent", "general_agent", "human_handoff"]:
    workflow.add_edge(agent, END)

# Compile with memory
memory = MemorySaver()
customer_service_app = workflow.compile(checkpointer=memory)

Step 5: Run the Agent

# Initialize conversation
config = {"configurable": {"thread_id": "session_12345"}}

initial_state = {
    "messages": [],
    "customer_id": "CUST-001",
    "customer_name": "Sarah Johnson",
    "customer_email": "sarah@email.com",
    "customer_tier": "premium",
    "needs_human": False,
    "response_ready": False
}

# Customer asks about order
result = customer_service_app.invoke({
    **initial_state,
    "messages": [HumanMessage("Hi, where is my order ORD-001?")]
}, config)

print(result["messages"][-1].content)
# Output: "Hi Sarah! I found your order ORD-001. Great news - it's shipped!
#          Tracking: 1Z999AA10123456784 (UPS)
#          Estimated delivery: Dec 15, 2024..."

# Follow-up about return (context preserved!)
result = customer_service_app.invoke({
    "messages": [HumanMessage("Actually, I want to return it. The color is wrong.")]
}, config)

print(result["messages"][-1].content)
# Routes to returns_agent with full context

Conversation Flow Diagram

Customer: "Where's my order?"
    │
    ▼
Router: query_type="order", sentiment="neutral"
    │
    ▼
Order Agent: [lookup_order] → "Shipped! Tracking: 1Z999..."
    │
    ▼
Customer: "I want to return it"
    │
    ▼
Router: query_type="return", sentiment="neutral"
    │
    ▼
Returns Agent: [check_eligibility] → "You're eligible! Here's the label..."
  

Key Takeaways

Next up: Order Processing - Building multi-step order workflows with validation, payment, and fulfillment.