Case 1 — Build a Customer Support Agent with LangGraph

# Case 1 — Build a Customer Support Agent with LangGraph > *"Build a production-grade customer support agent. It should: look up orders, check refund eligibility, create tickets, escalate when stuck. Use LangGraph with Postgres checkpointer."* --- ## Step 1 — Define state ```python from typing import TypedDict, Annotated from langchain_core.messages import BaseMessage import operator class SupportState(TypedDict): messages: Annotated[list[BaseMessage], operator.add] customer_id: str iteration: int escalated: bool ``` `messages` accumulates conversation; `customer_id` identifies the user; `iteration` caps loops; `escalated` flags terminal failures. ## Step 2 — Define tools ```python from langchain_core.tools import tool @tool def get_order_status(order_id: str) -> dict: '''Look up the current status, tracking info, and ETA of a customer's order. Use when the customer asks about their order, shipping, delivery, or tracking. Args: order_id: Order ID, e.g., "12345". ''' order = db.query_orders(order_id=order_id) if not order: return {"error": f"Order {order_id} not found"} return { "status": order.status, "carrier": order.carrier, "tracking_number": order.tracking, "eta": order.estimated_delivery.isoformat(), } @tool def check_refund_eligibility(order_id: str) -> dict: '''Check if an order is eligible for refund. ALWAYS call before promising a refund. Args: order_id: Order ID. ''' order = db.query_orders(order_id=order_id) if not order: return {"eligible": False, "reason": "Order not found"} days_old = (datetime.utcnow() - order.created_at).days if days_old > 30: return {"eligible": False, "reason": f"Order is {days_old} days old; refund window is 30 days"} return {"eligible": True, "reason": "Within refund window"} @tool def initiate_refund(order_id: str, reason: str) -> dict: '''Initiate a refund. Only call after check_refund_eligibility confirms eligibility. Args: order_id: Order ID. reason: Customer's stated reason. ''' eligibility = check_refund_eligibility.invoke({"order_id": order_id}) if not eligibility["eligible"]: return {"success": False, "error": eligibility["reason"]} refund = refund_service.create(order_id=order_id, reason=reason) return {"success": True, "refund_id": refund.id, "eta_days": 3} @tool def create_support_ticket(customer_id: str, category: str, summary: str, priority: str) -> dict: '''Create a support ticket for issues that need human attention. Args: customer_id: Customer ID. category: One of: complaint, billing, technical, feature_request, other. summary: One-paragraph summary of the issue. priority: One of: low, medium, high, urgent. ''' ticket = support_service.create_ticket( customer_id=customer_id, category=category, summary=summary, priority=priority, ) return {"ticket_id": ticket.id, "created": True} TOOLS = [get_order_status, check_refund_eligibility, initiate_refund, create_support_ticket] TOOL_MAP = {t.name: t for t in TOOLS} ``` ## Step 3 — Define nodes ```python from langchain_anthropic import ChatAnthropic from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage llm = ChatAnthropic(model="claude-sonnet-4-6") llm_with_tools = llm.bind_tools(TOOLS) SYSTEM_PROMPT = '''You are a customer support agent for Acme. Be empathetic, concise, and helpful. Guidelines: - ALWAYS check refund eligibility before promising a refund. - For complex issues or frustrated customers, create a support ticket and explain a human will follow up. - Use tools to look up real data; never fabricate order numbers or tracking info. - If you're unsure how to proceed, escalate by creating a ticket. ''' def agent_node(state: SupportState) -> dict: messages = [SystemMessage(SYSTEM_PROMPT)] + state["messages"] response = llm_with_tools.invoke(messages) return {"messages": [response], "iteration": state["iteration"] + 1} def tool_node(state: SupportState) -> dict: last_msg = state["messages"][-1] tool_messages = [] for tc in last_msg.tool_calls: tool = TOOL_MAP.get(tc["name"]) try: result = tool.invoke(tc["args"]) except Exception as e: result = {"error": str(e)} tool_messages.append( ToolMessage(content=json.dumps(result), tool_call_id=tc["id"]) ) return {"messages": tool_messages} def escalate_node(state: SupportState) -> dict: # Hit iteration cap or other stuck state create_support_ticket.invoke({ "customer_id": state["customer_id"], "category": "other", "summary": f"Agent reached iteration cap. Last messages: {state['messages'][-3:]}", "priority": "high", }) return { "messages": [AIMessage("Let me create a