← Back to DevBytes

Building a Customer Support Bot with LangGraph: Step-by-Step Tutorial

What Is LangGraph and Why Use It for Customer Support

LangGraph is a framework from the LangChain ecosystem for building stateful, multi-step agent applications. It lets you define workflows as graphs where each node represents a computation (LLM call, tool execution, human intervention) and edges represent the flow of data and control between them. Unlike simple LLM chains, LangGraph gives you explicit control over branching logic, persistent state, and the ability to pause execution for human input—all essential for customer support scenarios.

A customer support bot built with LangGraph can:

Project Overview

In this tutorial, you will build a complete customer support bot that handles typical user inquiries. The bot will:

By the end, you’ll have a modular, maintainable graph that you can extend with real APIs and deploy as a production service.

Step 1: Setting Up the Environment

Install the required packages. We'll use langgraph for graph orchestration, langchain-core for message types, and langchain-openai for LLM calls (you can swap this with any LangChain-compatible model). Run the following command:

pip install langgraph langchain-core langchain-openai

Now create a new Python file (e.g., support_bot.py) and add the necessary imports:

import os
from typing import TypedDict, Annotated, List, Union, Literal
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langgraph.errors import NodeInterrupt
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI

Set your OpenAI API key (or equivalent) as an environment variable or directly in the code for testing:

os.environ["OPENAI_API_KEY"] = "your-api-key-here"

Step 2: Defining the State Schema

LangGraph revolves around a shared state object that flows through all nodes. For a customer support bot, we need to track the conversation history, the classified intent, any retrieved knowledge, whether human escalation is required, and the human agent's response when they intervene.

Define the state using a Python TypedDict with annotations that tell LangGraph how to merge updates (e.g., appending messages instead of overwriting them):

class SupportState(TypedDict):
    messages: Annotated[List[Union[HumanMessage, AIMessage]], add_messages]
    intent: str
    knowledge: str
    needs_escalation: bool
    human_response: str

The add_messages annotation automatically appends new messages to the existing list, preserving full conversation history.

Step 3: Intent Classification Node

The first node reads the latest user message and uses an LLM to classify the intent. The result is stored in state["intent"] and used later for routing.

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

def classify_intent(state: SupportState) -> dict:
    """Classify user intent based on the last message."""
    last_message = state["messages"][-1]
    if not isinstance(last_message, HumanMessage):
        return {"intent": "unknown"}
    
    prompt = f"""
You are a customer support intent classifier.
Analyze the user message and respond with exactly one word from this list: faq, technical, billing, urgent, unknown.

User message: {last_message.content}
Intent:"""
    response = llm.invoke([SystemMessage(content="Classify intent."), HumanMessage(content=prompt)])
    intent = response.content.strip().lower()
    # Normalize to allowed values
    allowed = {"faq", "technical", "billing", "urgent", "unknown"}
    if intent not in allowed:
        intent = "unknown"
    return {"intent": intent}

In production, you would use a structured output parser or function calling for reliability. Here, we keep it simple.

Step 4: Knowledge Retrieval Node

This node simulates searching a knowledge base based on intent and user query. In a real application, you'd replace it with a vector store query or API call. We return a string with relevant articles or data.

def retrieve_knowledge(state: SupportState) -> dict:
    """Simulate knowledge base retrieval."""
    intent = state.get("intent", "unknown")
    last_message = state["messages"][-1].content if state["messages"] else ""
    
    # Mock knowledge base
    kb = {
        "faq": "Our return policy allows returns within 30 days with a receipt.",
        "technical": "Try restarting your device and router. If the issue persists, check our status page at status.example.com.",
        "billing": "You can view and pay invoices under 'Billing' in your account dashboard.",
        "urgent": "For immediate safety issues, please call our emergency line at 1-800-555-0199."
    }
    
    result = kb.get(intent, "No specific information found. I'll provide general assistance.")
    # Append user query context
    full_result = f"Knowledge base ({intent}): {result}\nUser query: {last_message}"
    return {"knowledge": full_result}

You can later extend this with multiple tool calls, fallback mechanisms, or semantic search.

Step 5: Response Generation Node

The main LLM node constructs a final answer using the conversation history, intent, and retrieved knowledge. It also decides whether the issue should be escalated to a human (e.g., for sensitive billing actions or unresolved technical problems).

def generate_response(state: SupportState) -> dict:
    """Generate a support response and assess escalation need."""
    messages = state["messages"] + [
        SystemMessage(content=f"""
You are a helpful customer support assistant.
Use the following knowledge to answer the user. If the issue is complex or sensitive, suggest escalation.

Knowledge: {state.get('knowledge', 'No specific knowledge available')}
Current intent: {state.get('intent', 'unknown')}

If you need human help, end your response with exactly 'ESCALATE'.
""")
    ]
    response = llm.invoke(messages)
    content = response.content
    
    needs_escalation = "ESCALATE" in content.upper()
    # Remove the marker from the user-facing message
    clean_content = content.replace("ESCALATE", "").strip()
    return {
        "messages": [AIMessage(content=clean_content)],
        "needs_escalation": needs_escalation
    }

This node appends the assistant's reply to the message history. The needs_escalation flag will be used for conditional routing.

Step 6: Human Escalation Node with Interrupt

When escalation is needed, we want the graph to pause and wait for a human agent to provide input. LangGraph supports this through the interrupt mechanism. We define a node that reads the human response from state and adds it to the conversation.

def human_escalation_node(state: SupportState) -> dict:
    """Insert the human agent's response into the conversation."""
    human_reply = state.get("human_response", "No human input provided.")
    return {
        "messages": [AIMessage(content=f"[Human Agent]: {human_reply}")],
        "needs_escalation": False  # Reset after human handles
    }

To trigger the pause, we'll use interrupt before this node when compiling the graph. The actual human input is set externally via a subsequent invocation with the updated state.

Step 7: Routing Logic and Graph Assembly

Now we create the routing functions and the graph itself. The router after intent classification decides whether to go to retrieval, directly to human escalation (for urgent issues), or end if intent is unknown but trivial. After response generation, we route to human escalation if the flag is set, otherwise end.

def route_intent(state: SupportState) -> Literal["retrieve", "human_escalation", "end"]:
    intent = state.get("intent", "unknown")
    if intent == "urgent":
        return "human_escalation"
    elif intent == "unknown":
        # If truly unknown, we could end with a fallback; here we still try retrieval
        return "retrieve"
    else:
        return "retrieve"

def route_after_response(state: SupportState) -> Literal["end", "human_escalation"]:
    if state.get("needs_escalation", False):
        return "human_escalation"
    return "end"

Now build the graph with StateGraph, add nodes, and wire them with edges:

builder = StateGraph(SupportState)

# Add nodes
builder.add_node("classify_intent", classify_intent)
builder.add_node("retrieve_knowledge", retrieve_knowledge)
builder.add_node("generate_response", generate_response)
builder.add_node("human_escalation", human_escalation_node)

# Set entry point
builder.add_edge(START, "classify_intent")

# After intent classification
builder.add_conditional_edges(
    "classify_intent",
    route_intent,
    {
        "retrieve": "retrieve_knowledge",
        "human_escalation": "human_escalation",
        "end": END
    }
)

# From retrieval to response generation
builder.add_edge("retrieve_knowledge", "generate_response")

# After response generation
builder.add_conditional_edges(
    "generate_response",
    route_after_response,
    {
        "end": END,
        "human_escalation": "human_escalation"
    }
)

# Human escalation always goes to end (graph will pause there via interrupt)
builder.add_edge("human_escalation", END)

To enable human-in-the-loop, we compile with a checkpointer (to save state across pauses) and specify the node where execution should interrupt:

memory = MemorySaver()
app = builder.compile(
    checkpointer=memory,
    interrupt_before=["human_escalation"]  # Pause before entering this node
)

The interrupt_before argument tells LangGraph to stop execution just before the listed nodes. The graph will return the current state and allow us to update human_response before resuming.

Step 8: Running and Testing the Bot

We'll test with a typical technical support query. The graph is invoked with an initial state and a thread ID for conversation persistence.

# Start a new conversation thread
thread_id = "user-123-session-001"
config = {"configurable": {"thread_id": thread_id}}

initial_state = {
    "messages": [HumanMessage(content="My internet keeps dropping every 10 minutes.")],
    "intent": "",
    "knowledge": "",
    "needs_escalation": False,
    "human_response": ""
}

# First invocation – will run until before human_escalation (if needed)
result = app.invoke(initial_state, config)

Check the result. If needs_escalation became True, the graph stopped and returned the state. You can inspect the last message:

last_msg = result["messages"][-1]
print("Assistant:", last_msg.content)
print("Needs escalation?", result.get("needs_escalation"))

To provide human input, update the state with the human_response field and invoke again using the same thread ID (this resumes from the paused point):

if result.get("needs_escalation"):
    human_input = "Tell the user we've identified a regional outage and engineers are working on it."
    updated_state = {
        "messages": result["messages"],  # carry over
        "intent": result["intent"],
        "knowledge": result["knowledge"],
        "needs_escalation": result["needs_escalation"],
        "human_response": human_input
    }
    final_result = app.invoke(updated_state, config)
    print("Final assistant:", final_result["messages"][-1].content)

If no escalation was needed, the graph ended after response generation and you'll see the assistant's answer directly.

Best Practices for Customer Support Bots with LangGraph

Conclusion

You've built a complete customer support bot using LangGraph that classifies intent, retrieves knowledge, generates responses, and escalates to human agents when necessary. The graph-based architecture gives you fine-grained control over the conversation flow while keeping the code modular and testable. By following the patterns shown—state schema, conditional routing, and interrupt-based human-in-the-loop—you can adapt this foundation to real-world support systems, integrate with existing tools, and deploy reliably at scale. LangGraph’s explicit state management and checkpointing make it an ideal choice for any multi-step LLM application where reliability and traceability are paramount.

— Ad —

Google AdSense will appear here after approval

← Back to all articles