A langgraph customer support agent turns the messy logic of triaging tickets, fetching answers, and escalating edge cases into an explicit state machine. This guide builds one from scratch with LangGraph, using a single OpenAI-compatible endpoint that handles model routing and fallback. You’ll end up with a runnable graph that classifies intent, retrieves from a knowledge base, and hands off to a human when confidence drops.
Step 1: Install dependencies and configure credentials
Create a fresh virtual environment and install the packages you need. LangGraph ships as langgraph; we use langchain-openai to talk to the LLM because it implements the standard chat interface LangGraph expects.
python -m venv .venv
source .venv/bin/activate
pip install langgraph langchain-openai langchain-core python-dotenvPut your API key in a .env file. We point LangChain’s ChatOpenAI at the n4n.ai OpenAI-compatible endpoint so we get access to 240+ models and automatic provider failover without writing retry code.
from dotenv import load_dotenv
load_dotenv()
import os
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="openai/gpt-4o-mini",
base_url="https://api.n4n.ai/v1",
api_key=os.environ["N4N_API_KEY"],
temperature=0
)If you already have an OpenAI key, swap base_url and model to your own stack. The rest of the code is provider-agnostic.
Step 2: Define the graph state
LangGraph persists state between nodes in a typed object. For a langgraph customer support agent, you need the conversation, a parsed intent, any retrieved context, and an escalation flag. Use TypedDict with Annotated so the reducer behavior is clear.
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage
class SupportState(TypedDict):
messages: Annotated[List[BaseMessage], "conversation history"]
intent: str | None
retrieved_docs: List[str]
escalate: boolThe messages list accumulates the thread. Keeping intent and escalate as first-class fields lets later nodes branch without re-parsing the transcript.
Step 3: Build a retrieval node
Most support queries hit a fixed knowledge base. Mock it with a dict; swap in Pinecone or pgvector later. The node reads the last user message and returns matched docs.
KB = {
"refund": "Refunds are processed within 5 business days to original payment method.",
"cancel": "Subscriptions cancel immediately but stay active until period end.",
"unknown": "I couldn't find that in our docs."
}
def retrieve(state: SupportState):
query = state["messages"][-1].content.lower()
if "refund" in query:
docs = [KB["refund"]]
elif "cancel" in query:
docs = [KB["cancel"]]
else:
docs = [KB["unknown"]]
return {"retrieved_docs": docs}In production, replace the keyword scan with a vector search and return score alongside text. The graph contract stays identical.
Step 4: Classify intent and decide on escalation
Add a node that calls the LLM to label intent and set the escalate flag. Keep the prompt strict about JSON output. This is the brain of the langgraph customer support agent.
from langchain_core.messages import SystemMessage, HumanMessage
import json
def classify(state: SupportState):
sys = SystemMessage(content=(
"You are a routing classifier for a langgraph customer support agent. "
"Return JSON: {\"intent\": \"refund|cancel|other\", \"escalate\": bool}. "
"Escalate if the user is angry, threatens churn, or request is outside scope."
))
user = HumanMessage(content=state["messages"][-1].content)
resp = llm.invoke([sys, user])
try:
data = json.loads(resp.content)
except json.JSONDecodeError:
data = {"intent": "other", "escalate": True}
return {"intent": data.get("intent"), "escalate": data.get("escalate", False)}Because n4n.ai forwards provider cache-control hints and meters per token, you can attach caching to the system prompt if your underlying model supports it, and the gateway passes the hints through without extra code.
Step 5: Generate the reply
A final node synthesizes the answer from retrieved docs and conversation. If the classifier escalated, skip generation and return a handoff message.
def respond(state: SupportState):
if state["escalate"]:
return {"messages": [HumanMessage(content="Transferring you to a human agent now.")]}
ctx = "\n".join(state["retrieved_docs"])
sys = SystemMessage(content=f"Answer using only context:\n{ctx}")
out = llm.invoke([sys] + state["messages"])
return {"messages": [out]}Separating classify and respond keeps each node single-purpose and testable. You can unit-test respond with a stubbed LLM.
Step 6: Assemble the graph with conditional edges
Use StateGraph to link nodes. Route from classify to respond normally, but to a human-escalation terminal node if escalate is true.
from langgraph.graph import StateGraph, END
def route(state: SupportState):
return "escalate" if state["escalate"] else "respond"
g = StateGraph(SupportState)
g.add_node("retrieve", retrieve)
g.add_node("classify", classify)
g.add_node("respond", respond)
g.add_node("escalate_node", lambda s: {"messages": [HumanMessage(content="Queued for human.")]})
g.set_entry_point("retrieve")
g.add_edge("retrieve", "classify")
g.add_conditional_edges("classify", route, {"respond": "respond", "escalate": "escalate_node"})
g.add_edge("respond", END)
g.add_edge("escalate_node", END)
app = g.compile()The conditional edge is the key advantage over a linear chain: the branching is visible in the graph, not buried in a Python if inside a giant node.
Step 7: Run and verify the langgraph customer support agent
Invoke with a user question and inspect the state.
from langchain_core.messages import HumanMessage
result = app.invoke({"messages": [HumanMessage(content="I want a refund, this is ridiculous!")]})
print(result["intent"], result["escalate"], result["messages"][-1].content)Expected output for an angry query: other True Transferring you to a human agent now. (or refund if the classifier catches the keyword, but anger should still flip escalate). For a calm refund query "How do I get a refund?" you should see refund False and the KB answer.
Verify success by running both happy-path and escalation-path inputs. The graph should never raise on JSON parse errors because we default to escalate. Check token usage via the metering headers if you need cost visibility.
Step 8: Add persistence and streaming
For production, wrap the compiled graph with SqliteSaver so conversations survive restarts.
from langgraph.checkpoint.sqlite import SqliteSaver
with SqliteSaver.from_conn_string("support.db") as saver:
app = g.compile(checkpointer=saver)
app.invoke(
{"messages": [HumanMessage(content="Cancel my plan")]},
config={"configurable": {"thread_id": "u1"}}
)Streaming works by passing stream_mode="messages" to invoke or using app.stream. If you need to swap the LLM call to a cheaper model for classification, change the model string in Step 1. Your inference gateway resolves it and fails over automatically if that provider is degraded.
Why a graph beats a linear chain
A langgraph customer support agent makes the branching explicit. New intents become new nodes, not nested conditionals. Fallback across model providers is handled by the gateway, so the graph code stays focused on business logic.
When a new requirement appears—say, a billing dispute that needs a CRM lookup—you add a crm_lookup node and a conditional edge from classify. No existing node changes signature.
Testing checklist
- Unit test each node with a fake state dict and a stub LLM.
- Integration test the compiled graph with
pytestandapp.invoke. - Confirm escalation triggers on angry phrasing and unknown intent.
- Confirm retrieval returns KB text for known keywords.
- Monitor per-token metering to catch runaway loops or oversized prompts.
That is a complete, runnable foundation. Extend the nodes with real RAG, a ticketing API, and a feedback loop, and you have a production support workflow that is easy to reason about and cheap to operate.