Building a langgraph support agent human handoff flow forces you to confront the messy reality of partial automation: the bot handles triage, but a human must take the wheel on edge cases. This tutorial walks through a minimal but production-shaped implementation using LangGraph’s interrupt primitive and a typed state graph.
Prerequisites
- Python 3.10 or newer.
langgraphandlangchain-openaiinstalled (pip install langgraph langchain-openai).- An OpenAI API key, or any OpenAI-compatible endpoint. If you do not want to juggle provider keys, point the client at n4n.ai’s OpenAI-compatible endpoint; it fronts 240+ models and automatically fails over when a provider is rate-limited or degraded.
- Basic comfort with LangChain message objects (
HumanMessage,AIMessage).
State design
A support conversation is just a message list plus routing metadata. We extend the state with a needs_human flag and a handoff_reason so the human node knows why it was invoked.
from typing import TypedDict, List
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
class SupportState(TypedDict):
messages: List[HumanMessage | AIMessage | SystemMessage]
needs_human: bool
handoff_reason: strKeep the state flat. Nesting invites serialization bugs when you persist to a checkpointer.
Triage node: the langgraph support agent human handoff decision point
The triage node is where the model decides whether to answer directly or escalate. We force JSON output and parse it. In production, use .with_structured_output(); here we parse manually to show the contract.
import json
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
TRIAGE_SYSTEM = SystemMessage(content=(
"You are a support triage agent. Classify the user request. "
"If it is a simple how-to, answer it. If it involves billing disputes, "
"legal, or angry customers, escalate. Respond ONLY with JSON: "
'{"action": "respond", "reply": "..."} or {"action": "handoff", "reason": "..."}'
))
def triage_node(state: SupportState):
user_msg = state["messages"][-1]
resp = llm.invoke([TRIAGE_SYSTEM, user_msg])
data = json.loads(resp.content)
if data["action"] == "respond":
return {
"messages": [AIMessage(content=data["reply"])],
"needs_human": False,
"handoff_reason": ""
}
return {
"needs_human": True,
"handoff_reason": data["reason"]
}The langgraph support agent human handoff logic lives entirely in that if branch. No separate classifier service required.
Human handoff via interrupt
LangGraph’s interrupt pauses the graph and returns control to the caller. Unlike a dummy “wait for API” node, it serializes cleanly and resumes without re-running upstream nodes.
from langgraph.types import interrupt
def human_handoff_node(state: SupportState):
payload = interrupt({
"reason": state["handoff_reason"],
"conversation": [m.content for m in state["messages"]]
})
# `payload` is whatever the human (or external system) resumes with.
return {"messages": [AIMessage(content=payload)]}Graph assembly and routing
We connect START to triage, then conditionally route to human_handoff or END.
from langgraph.graph import StateGraph, START, END
def route(state: SupportState):
if state["needs_human"]:
return "human_handoff"
return END
graph = StateGraph(SupportState)
graph.add_node("triage", triage_node)
graph.add_node("human_handoff", human_handoff_node)
graph.add_edge(START, "triage")
graph.add_conditional_edges(
"triage",
route,
{"human_handoff": "human_handoff", END: END}
)
graph.add_edge("human_handoff", END)
app = graph.compile()Checkpointing is not optional
Without a checkpointer, interrupt raises and you lose state. Use MemorySaver for local dev, Postgres for production.
from langgraph.checkpoint.memory import MemorySaver
app = graph.compile(checkpointer=MemorySaver())Run: automated resolution
Invoke with a simple question. The triage node answers, needs_human stays false, and the graph ends.
config = {"configurable": {"thread_id": "t1"}}
result = app.invoke({
"messages": [HumanMessage(content="How do I enable 2FA?")],
"needs_human": False,
"handoff_reason": ""
}, config)
print(result["messages"][-1].content)Expected output (model-generated, phrasing may vary):
You can enable 2FA under Settings > Security > Two-Factor Authentication.Run: handoff scenario
Now a billing complaint. The graph hits interrupt and pauses.
config = {"configurable": {"thread_id": "t2"}}
try:
app.invoke({
"messages": [HumanMessage(content="I was charged twice and demand a refund!")],
"needs_human": False,
"handoff_reason": ""
}, config)
except Exception as e:
print("PAUSED:", e.args[0])Expected output:
PAUSED: {'reason': 'billing dispute', 'conversation': ['I was charged twice and demand a refund!']}The process exits, but the checkpointer retains the state for thread t2.
Resume after human action
A human agent reviews the payload, then resumes the thread with a reply.
from langgraph.types import Command
resumed = app.invoke(
Command(resume="Hi, this is Sue from billing. I've refunded the duplicate charge."),
{"configurable": {"thread_id": "t2"}}
)
print(resumed["messages"][-1].content)Expected output:
Hi, this is Sue from billing. I've refunded the duplicate charge.That final AIMessage is the human’s text, injected by the handoff node. Your frontend can render it as an agent message.
Why this shape works in production
The langgraph support agent human handoff pattern decouples model inference from human latency. The LLM triages in milliseconds; the human step can take hours. Because the pause is explicit, you can store the thread in a queue, notify a Slack channel, or route to a CRM.
A few hard-won notes:
- Validate the triage JSON. A malformed response should default to
needs_human=True. Never let a parser error silently auto-respond. - Auth the resume call. Anyone who can post a
Commandto your thread id can impersonate an agent. - Set a TTL on interrupted threads. Stale handoffs waste memory and confuse agents.
Full code listing
import json
from typing import TypedDict, List
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
class SupportState(TypedDict):
messages: List[HumanMessage | AIMessage | SystemMessage]
needs_human: bool
handoff_reason: str
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
TRIAGE_SYSTEM = SystemMessage(content="You are a support triage agent. Respond ONLY with JSON: {\"action\": \"respond\", \"reply\": \"...\"} or {\"action\": \"handoff\", \"reason\": \"...\"}")
def triage_node(state: SupportState):
resp = llm.invoke([TRIAGE_SYSTEM, state["messages"][-1]])
data = json.loads(resp.content)
if data["action"] == "respond":
return {"messages": [AIMessage(content=data["reply"])], "needs_human": False, "handoff_reason": ""}
return {"needs_human": True, "handoff_reason": data["reason"]}
def human_handoff_node(state: SupportState):
payload = interrupt({"reason": state["handoff_reason"], "conversation": [m.content for m in state["messages"]]})
return {"messages": [AIMessage(content=payload)]}
def route(state: SupportState):
return "human_handoff" if state["needs_human"] else END
g = StateGraph(SupportState)
g.add_node("triage", triage_node)
g.add_node("human_handoff", human_handoff_node)
g.add_edge(START, "triage")
g.add_conditional_edges("triage", route, {"human_handoff": "human_handoff", END: END})
g.add_edge("human_handoff", END)
app = g.compile(checkpointer=MemorySaver())Swap the ChatOpenAI base URL to any gateway you trust. The rest of the graph does not care which model answered the triage call.
That’s a complete, runnable langgraph support agent human handoff loop. Build on it: add a feedback node, log token usage, or escalate to a different human queue based on handoff_reason.