A crewai support agent human escalation flow lets you automate tier-1 support while keeping a clean path to a person when the model hits its limits. This tutorial builds a minimal but production-shaped pipeline: a triage agent, a resolution agent, and an escalation tool that opens a human ticket. You will run it locally and see exactly where the human handoff triggers.
Prerequisites
- Python 3.10 or newer
pip install crewai crewai_tools openai- An API key for an OpenAI-compatible LLM. If you want automatic provider fallback when a model is rate-limited, point at the n4n.ai OpenAI-compatible endpoint instead of a single vendor.
- Familiarity with basic Python and environment variables
Set your key in the shell before running anything:
export OPENAI_API_KEY="sk-..."
# or for the gateway:
export N4N_API_KEY="sk-n4n-..."Step 1: Define the escalation tool
The handoff to a human is just a tool call. In a real system this would POST to Zendesk or an internal queue; here we return a fake ticket ID so the logic is observable.
from crewai_tools import tool
@tool("HumanEscalation")
def open_human_ticket(query: str, transcript: str) -> str:
"""Use this when the customer issue cannot be resolved automatically.
Pass the original query and the conversation transcript so far."""
ticket_id = "HUM-" + str(abs(hash(query)) % 100000)
# Imagine: requests.post("https://helpdesk.internal/tickets", json={...})
return f"Escalated to human. Ticket {ticket_id} created from transcript: {transcript[:60]}..."The decorator registers the function as a CrewAI tool with a name and description the LLM can reason about.
Step 2: Configure the LLM and agents
We instantiate one LLM and attach it to two agents. Using a gateway base URL means a degraded provider gets retried on a different one without code changes.
from crewai import LLM, Agent
llm = LLM(
model="openai/gpt-4o-mini",
base_url="https://api.n4n.ai/v1", # OpenAI-compatible, 240+ models, automatic fallback
api_key="sk-n4n-...",
temperature=0.2,
)
triage_agent = Agent(
role="Support Triage",
goal="Decide if the request is auto-resolvable or must reach a human.",
backstory="You enforce routing policy for a SaaS support desk.",
llm=llm,
tools=[open_human_ticket],
verbose=True,
)
resolution_agent = Agent(
role="Tier-1 Support",
goal="Resolve common billing and login issues using concise answers.",
backstory="You are a senior support rep who hates fluff.",
llm=llm,
verbose=True,
)The verbose=True flag prints agent thoughts so you can watch the escalation decision happen.
Step 3: Triage task and checkpoint output
We give the triage agent a strict policy: escalate refunds over $500 or any reported outage. Otherwise it returns RESOLVABLE.
from crewai import Task, Crew, Process
triage_task = Task(
description=(
"Review this support request: '{query}'. "
"If it mentions a refund over $500 or a service outage, call HumanEscalation. "
"Otherwise reply with exactly the word RESOLVABLE."
),
expected_output="A ticket confirmation string or the word RESOLVABLE.",
agent=triage_agent,
)
crew = Crew(
agents=[triage_agent],
tasks=[triage_task],
process=Process.sequential,
)
result = crew.kickoff(inputs={"query": "I was double charged $800 on my invoice"})
print("TRIAGE RESULT:", result)Expected output (abbreviated):
TRIAGE RESULT: Escalated to human. Ticket HUM-12345 created from transcript: I was double charged $800...If you change the input to "How do I reset my password?", the printed result is RESOLVABLE.
Step 4: Conditional resolution path
A real crewai support agent human escalation system should not run the resolution agent after escalation. Branch in code based on the triage string.
def run_support(query: str) -> str:
triage_out = crew.kickoff(inputs={"query": query})
if "RESOLVABLE" in str(triage_out):
resolve_task = Task(
description=f"Write a 2-sentence reply for: {query}",
expected_output="A customer-facing response.",
agent=resolution_agent,
)
resolve_crew = Crew(
agents=[resolution_agent],
tasks=[resolve_task],
process=Process.sequential,
)
return resolve_crew.kickoff()
return triage_out
print(run_support("How do I reset my password?"))Expected resolution output:
You can reset your password by clicking 'Forgot Password' on the login screen. A reset link will arrive in your email within a minute.Step 5: Add a human-in-the-loop confirmation
Before sending the auto-reply, a support lead should approve it. CrewAI supports this natively with human_input=True on a task.
confirm_task = Task(
description="Review the drafted reply for: {query}. Approve or rewrite.",
expected_output="Final reply text.",
agent=resolution_agent,
human_input=True,
)
confirm_crew = Crew(
agents=[resolution_agent],
tasks=[confirm_task],
process=Process.sequential,
)
final = confirm_crew.kickoff(inputs={"query": "How do I reset my password?"})
print("FINAL:", final)When run, the process pauses and prompts in the terminal:
[Human input requested] Review the drafted reply for: How do I reset my password?
> approve
FINAL: You can reset your password by clicking 'Forgot Password' on the login screen...Step 6: Full script
import os
from crewai import LLM, Agent, Task, Crew, Process
from crewai_tools import tool
@tool("HumanEscalation")
def open_human_ticket(query: str, transcript: str) -> str:
"""Use this when the customer issue cannot be resolved automatically."""
ticket_id = "HUM-" + str(abs(hash(query)) % 100000)
return f"Escalated to human. Ticket {ticket_id} created from transcript: {transcript[:60]}..."
llm = LLM(
model="openai/gpt-4o-mini",
base_url="https://api.n4n.ai/v1",
api_key=os.environ["N4N_API_KEY"],
temperature=0.2,
)
triage_agent = Agent(
role="Support Triage",
goal="Decide if the request is auto-resolvable or must reach a human.",
backstory="You enforce routing policy for a SaaS support desk.",
llm=llm,
tools=[open_human_ticket],
verbose=False,
)
resolution_agent = Agent(
role="Tier-1 Support",
goal="Resolve common billing and login issues using concise answers.",
backstory="You are a senior support rep who hates fluff.",
llm=llm,
verbose=False,
)
def run_support(query: str) -> str:
triage_task = Task(
description=(
"Review this support request: '{query}'. "
"If it mentions a refund over $500 or a service outage, call HumanEscalation. "
"Otherwise reply with exactly the word RESOLVABLE."
),
expected_output="A ticket confirmation or RESOLVABLE.",
agent=triage_agent,
)
triage_crew = Crew(agents=[triage_agent], tasks=[triage_task], process=Process.sequential)
out = triage_crew.kickoff(inputs={"query": query})
if "RESOLVABLE" not in str(out):
return str(out)
resolve_task = Task(
description=f"Write a 2-sentence reply for: {query}",
expected_output="A customer-facing response.",
agent=resolution_agent,
)
return str(Crew(agents=[resolution_agent], tasks=[resolve_task], process=Process.sequential).kickoff())
if __name__ == "__main__":
print(run_support("I was double charged $800 on my invoice"))
print(run_support("How do I reset my password?"))Production notes
The design above separates policy (triage) from execution (resolution) and makes the crewai support agent human escalation boundary explicit. Three things to harden before shipping:
- Tool side effects:
open_human_ticketmust be idempotent. Include a customer ID and dedupe on your side so a retried agent call does not spawn ten tickets. - LLM reliability: A single provider will rate-limit you. Routing through an OpenAI-compatible gateway that honors client routing directives and forwards cache-control hints keeps p95 latency flat when one backend degrades.
- Audit trail: Stream
verboselogs to your observability stack. The transcript string passed to the escalation tool is your handoff context; capture it.
If you need per-token metering across multiple models, a gateway that reports usage on each response saves you from writing billing glue code. That is the only part where the inference layer should leak into your agent code.
Build the triage prompt around concrete thresholds ($500, “outage”) rather than vague “complex” language; the agent will follow rules better than vibes.