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:
- Maintain conversation state across multiple turns, including user identity, intent, and collected facts.
- Route dynamically based on intent classification, confidence scores, or business rules (e.g., billing vs. technical issues).
- Integrate tools such as knowledge bases, ticket systems, and CRMs without losing context.
- Pause for human escalation when the bot is uncertain or when a sensitive operation requires approval.
- Recover from failures using built-in checkpointing and replay capabilities.
Project Overview
In this tutorial, you will build a complete customer support bot that handles typical user inquiries. The bot will:
- Classify the user’s intent into categories such as faq, technical, billing, or urgent.
- Retrieve relevant information from a simulated knowledge base.
- Generate a helpful response using an LLM, incorporating the retrieved context and conversation history.
- Decide whether the issue needs human escalation, and if so, pause execution to request agent input before continuing.
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
- Use structured outputs for intent classification. Relying on raw text parsing is fragile. Use function calling or a Pydantic model to force a valid enum, then set the state accordingly.
- Keep state minimal but informative. Avoid storing redundant data; let messages carry the conversation flow, and use separate fields for flags and tool results.
- Implement proper error handling. Wrap LLM calls and tool executions in try/except blocks, and add a dedicated error node that logs the issue and routes to a safe fallback (e.g., human escalation).
- Leverage checkpointing for recovery. Always compile with a checkpointer (like
MemorySaverfor testing, or a persistent one likeSqliteSaverfor production) so conversations survive restarts and allow replay. - Set a maximum number of iterations. For complex graphs with cycles (e.g., retry loops), configure
recursion_limitin the graph invocation to prevent infinite loops. - Guard against prompt injection. Sanitize user inputs and never include raw user text in system prompts without escaping. Consider using a dedicated guardrails node before classification.
- Add observability. Integrate logging at each node to capture state transitions, timings, and LLM costs. LangGraph works with LangSmith for tracing.
- Plan for multi-turn clarification. If intent is ambiguous, design a loop that asks the user clarifying questions, updating state until confidence is high.
- Human-in-the-loop for sensitive actions. Always interrupt before executing refunds, account changes, or deleting data. The graph can resume with an approval flag.
- Deploy as an API. Wrap the graph in a FastAPI or Flask server, using the same thread ID to handle concurrent conversations. LangGraph’s
RunnableConfigmakes this straightforward.
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.