Customer Support Agent

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langchain_core.tools import tool
 
class SupportState(TypedDict):
    messages: Annotated[list, add_messages]
    category: str  # "faq", "order", "escalate"
    order_info: str
 
@tool
def lookup_order(order_id: str) -> str:
    """Look up order details by order ID."""
    orders = {
        "12345": "Order #12345: Shipped Feb 20, arriving Feb 27. Item: Blue Widget.",
        "67890": "Order #67890: Processing. Item: Red Gadget. Est. ship: Mar 1.",
    }
    return orders.get(order_id, f"No order found with ID {order_id}.")
 
@tool
def search_faq(question: str) -> str:
    """Search the FAQ knowledge base for answers."""
    faqs = {
        "return": "Our return policy allows returns within 30 days with receipt.",
        "shipping": "Standard shipping: 5-7 business days. Express: 2-3 days.",
        "payment": "We accept Visa, Mastercard, AMEX, and PayPal.",
    }
    for key, answer in faqs.items():
        if key in question.lower():
            return answer
    return "No FAQ match found. Let me connect you with a team member."

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from langchain_anthropic import ChatAnthropic
 
llm = ChatAnthropic(model="claude-sonnet-4-5-20250929")
 
def classify(state: SupportState) -> dict:
    last_msg = state["messages"][-1].content
    response = llm.invoke(
        f"Classify this support message as 'faq', 'order', or 'escalate'. "
        f"Reply with just the category.\n\nMessage: {last_msg}"
    )
    return {"category": response.content.strip().lower()}
 
def handle_faq(state: SupportState) -> dict:
    answer = search_faq.invoke({"question": state["messages"][-1].content})
    return {"messages": [{"role": "assistant", "content": answer}]}
 
def handle_order(state: SupportState) -> dict:
    last_msg = state["messages"][-1].content
    extraction = llm.invoke(f"Extract the order ID from: {last_msg}. Reply with just the ID.")
    order_id = extraction.content.strip().replace("#", "")
    result = lookup_order.invoke({"order_id": order_id})
    return {"messages": [{"role": "assistant", "content": result}], "order_info": result}
 
def handle_escalation(state: SupportState) -> dict:
    human_response = interrupt({
        "reason": "Customer issue requires human attention",
        "conversation": [m.content for m in state["messages"][-3:]],
    })
    return {"messages": [{"role": "assistant", "content": human_response}]}
 
builder = StateGraph(SupportState)
builder.add_node("classify", classify)
builder.add_node("faq", handle_faq)
builder.add_node("order", handle_order)
builder.add_node("escalate", handle_escalation)
 
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", lambda s: s["category"], {
    "faq": "faq", "order": "order", "escalate": "escalate",
})
builder.add_edge("faq", END)
builder.add_edge("order", END)
builder.add_edge("escalate", END)
 
graph = builder.compile(checkpointer=MemorySaver())

# FAQ question -- routes to FAQ handler
c1 = {"configurable": {"thread_id": "session-1"}}
result = graph.invoke({"messages": [{"role": "user", "content": "What is your return policy?"}]}, c1)
print(result["messages"][-1].content)
 
# Order query -- routes to order handler
c2 = {"configurable": {"thread_id": "session-2"}}
result = graph.invoke({"messages": [{"role": "user", "content": "Where is my order #12345?"}]}, c2)
print(result["messages"][-1].content)
 
# Sensitive issue -- triggers human escalation, graph pauses with __interrupt__
c3 = {"configurable": {"thread_id": "session-3"}}
result = graph.invoke({"messages": [{"role": "user", "content": "Fraudulent charges on my account!"}]}, c3)
 
# Human agent responds via Command(resume=...)
final = graph.invoke(Command(resume="Account flagged. A specialist will call within 1 hour."), c3)
print(final["messages"][-1].content)