How to add ticket routing to a LangChain support bot

Building a ticket routing langchain support bot means moving beyond simple Q&A into structured triage: classify the issue, enrich it with context, and dispatch it to the right queue or human. This tutorial walks through a production-ready pattern using LangChain’s expression language (LCEL), a lightweight classifier chain, and a pluggable router that works with any ticketing backend — Jira, Linear, Zendesk, or a database table.

Step 1: Define the routing schema and categories

Before writing code, enumerate the categories your bot will route to. Keep this list small and mutually exclusive; each additional category increases classifier confusion. A typical support taxonomy looks like:

# routing/schema.py
from enum import Enum
from pydantic import BaseModel, Field
from typing import Literal

class TicketCategory(str, Enum):
    BILLING = "billing"
    TECHNICAL = "technical"
    ACCOUNT = "account"
    FEATURE_REQUEST = "feature_request"
    GENERAL = "general"

class RoutingDecision(BaseModel):
    category: TicketCategory
    confidence: float = Field(ge=0.0, le=1.0)
    reasoning: str
    suggested_priority: Literal["low", "medium", "high", "urgent"] = "medium"
    requires_human: bool = False
    metadata: dict = Field(default_factory=dict)

The confidence field lets you build fallback logic later. The metadata bag carries extracted entities (order ID, account ID, error codes) that the downstream ticket creator needs.

Step 2: Build the classifier chain

Use a structured-output LLM chain to classify incoming messages. LCEL makes this composable and testable.

# routing/classifier.py
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from langchain_openai import ChatOpenAI
from .schema import RoutingDecision, TicketCategory

CATEGORIES = ", ".join([c.value for c in TicketCategory])

CLASSIFIER_PROMPT = ChatPromptTemplate.from_messages([
    ("system", f"""You are a support ticket classifier. Categorize the user's message into exactly one of: {CATEGORIES}.

Return a JSON object matching the RoutingDecision schema. 
- confidence: your certainty (0.0-1.0)
- reasoning: one sentence explaining the choice
- suggested_priority: low/medium/high/urgent based on urgency signals
- requires_human: true if the issue needs human judgment (fraud, legal, data deletion, etc.)
- metadata: extracted entities like order_id, account_id, error_code, product_area"""),
    ("human", "{message}"),
    ("human", "Conversation history (most recent first):\n{history}"),
])

def build_classifier_chain(model: str = "gpt-4o-mini", temperature: float = 0.0):
    llm = ChatOpenAI(model=model, temperature=temperature)
    parser = PydanticOutputParser(pydantic_object=RoutingDecision)
    return CLASSIFIER_PROMPT | llm | parser

Why gpt-4o-mini? It’s fast, cheap, and handles structured output reliably. Swap the model parameter if your organization standardizes on a different provider — this chain works with any ChatOpenAI-compatible endpoint.

Step 3: Add conversation context extraction

Routing accuracy improves when the classifier sees recent history. Extract the last N turns and format them compactly.

# routing/context.py
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from typing import Sequence

def format_history(messages: Sequence[BaseMessage], max_turns: int = 4) -> str:
    """Format recent history for the classifier prompt."""
    relevant = messages[-max_turns * 2:]  # each turn = human + ai
    lines = []
    for msg in relevant:
        role = "User" if isinstance(msg, HumanMessage) else "Assistant"
        content = msg.content[:200]  # truncate long responses
        lines.append(f"{role}: {content}")
    return "\n".join(lines) if lines else "(no prior context)"

Pass the output of format_history into the classifier chain as the history variable.

Step 4: Implement the router with fallback logic

The router receives the RoutingDecision and dispatches to the appropriate handler. Design it as a protocol so you can swap ticketing backends without touching classification logic.

# routing/router.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from .schema import RoutingDecision, TicketCategory
from .context import format_history
from .classifier import build_classifier_chain
from langchain_core.messages import BaseMessage

@dataclass
class TicketPayload:
    category: TicketCategory
    title: str
    priority: str
    metadata: dict
    source_conversation_id: str

class TicketBackend(ABC):
    @abstractmethod
    def create_ticket(self, payload: TicketPayload) -> str:
        """Returns the created ticket ID."""
        pass

class SupportRouter:
    def __init__(
        self,
        backend: TicketBackend,
        classifier_model: str = "gpt-4o-mini",
        low_confidence_threshold: float = 0.65,
    ):
        self.backend = backend
        self.classifier = build_classifier_chain(model=classifier_model)
        self.low_confidence_threshold = low_confidence_threshold

    def route(
        self,
        message: str,
        conversation_id: str,
        history: list[BaseMessage],
    ) -> tuple[str, RoutingDecision]:
        """Classify and create ticket. Returns (ticket_id, decision)."""
        decision = self.classifier.invoke({
            "message": message,
            "history": format_history(history),
        })

        # Low confidence -> route to general queue with human review flag
        if decision.confidence < self.low_confidence_threshold:
            decision.category = TicketCategory.GENERAL
            decision.requires_human = True
            decision.reasoning += f" [Low confidence ({decision.confidence:.2f}), escalated for review]"

        payload = TicketPayload(
            category=decision.category,
            title=self._generate_title(message, decision),
            description=message,
            priority=decision.suggested_priority,
            metadata=decision.metadata,
            source_conversation_id=conversation_id,
        )

        ticket_id = self.backend.create_ticket(payload)
        return ticket_id, decision

    def _generate_title(self, message: str, decision: RoutingDecision) -> str:
        # Truncate to first 80 chars, append category tag
        prefix = message[:80].strip()
        return f"[{decision.category.value.upper()}] {prefix}"

The low_confidence_threshold is your safety valve. Tune it based on evaluation data — start at 0.65 and adjust after reviewing misclassifications.

Step 5: Wire a concrete ticket backend

Here’s a minimal Jira backend using the jira Python package. Replace with your system’s API.

# routing/backends/jira_backend.py
from jira import JIRA
from jira.exceptions import JIRAError
from ..router import TicketBackend, TicketPayload
from ..schema import TicketCategory
import os

JIRA_PROJECT_KEY = os.getenv("JIRA_PROJECT_KEY", "SUP")
CATEGORY_TO_COMPONENT = {
    TicketCategory.BILLING: "Billing",
    TicketCategory.TECHNICAL: "Engineering",
    TicketCategory.ACCOUNT: "Account Management",
    TicketCategory.FEATURE_REQUEST: "Product",
    TicketCategory.GENERAL: "General Support",
}

class JiraBackend(TicketBackend):
    def __init__(self):
        self.client = JIRA(
            server=os.getenv("JIRA_SERVER"),
            basic_auth=(os.getenv("JIRA_EMAIL"), os.getenv("JIRA_API_TOKEN")),
        )

    def create_ticket(self, payload: TicketPayload) -> str:
        issue_dict = {
            "project": {"key": JIRA_PROJECT_KEY},
            "summary": payload.title,
            "description": self._format_description(payload),
            "issuetype": {"name": "Task"},
            "components": [{"name": CATEGORY_TO_COMPONENT[payload.category]}],
            "priority": {"name": payload.priority.capitalize()},
            "labels": ["auto-routed", f"source:{payload.source_conversation_id}"],
        }
        # Add custom fields from metadata
        for key, value in payload.metadata.items():
            if key.startswith("cf_"):  # convention: cf_ = custom field
                issue_dict[key] = value

        try:
            issue = self.client.create_issue(fields=issue_dict)
            return issue.key
        except JIRAError as e:
            raise RuntimeError(f"Jira ticket creation failed: {e.text}") from e

    def _format_description(self, payload: TicketPayload) -> str:
        lines = [
            f"*Category:* {payload.category.value}",
            f"*Priority:* {payload.priority}",
            f"*Source Conversation:* {payload.source_conversation_id}",
            "",
            "*User Message:*",
            payload.description,
        ]
        if payload.metadata:
            lines.extend(["", "*Extracted Metadata:*"])
            for k, v in payload.metadata.items():
                lines.append(f"- {k}: {v}")
        return "\n".join(lines)

For Zendesk, Linear, or a Postgres table, implement the same TicketBackend interface. The router doesn’t care.

Step 6: Integrate into your LangGraph or LCEL graph

If you’re using LangGraph for the support bot, add a routing node that triggers on specific intents or explicit user requests.

# bot/graph.py
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage
from typing import TypedDict, Annotated
from operator import add
from routing.router import SupportRouter
from routing.backends.jira_backend import JiraBackend

class SupportState(TypedDict):
    messages: Annotated[list[BaseMessage], add]
    conversation_id: str
    ticket_id: str | None

router = SupportRouter(backend=JiraBackend())

def should_route(state: SupportState) -> bool:
    """Route if user explicitly asks or if agent detects escalation need."""
    last_msg = state["messages"][-1]
    # Simple heuristic: route on "create ticket", "talk to human", "escalate"
    # In production, use an intent classifier node instead.
    trigger_phrases = ["create ticket", "talk to human", "escalate", "file a bug"]
    return any(phrase in last_msg.content.lower() for phrase in trigger_phrases)

def route_node(state: SupportState) -> SupportState:
    last_msg = state["messages"][-1]
    ticket_id, decision = router.route(
        message=last_msg.content,
        conversation_id=state["conversation_id"],
        history=state["messages"][:-1],
    )
    return {
        **state,
        "ticket_id": ticket_id,
        "messages": [
            *state["messages"],
            # Inform user ticket was created
            AIMessage(content=f"Created ticket {ticket_id} ({decision.category.value}, {decision.suggested_priority} priority). A team member will follow up.")
        ],
    }

graph = StateGraph(SupportState)
graph.add_node("route", route_node)
graph.add_conditional_edges("route", should_route, {True: "route", False: END})
graph.set_entry_point("route")
app = graph.compile()

If you’re not using LangGraph, the same router.route() call fits into any LCEL chain or FastAPI endpoint.

Step 7: Add observability and evaluation hooks

You cannot improve routing without measuring it. Log every decision with enough context to audit later.

# routing/observability.py
import json
import logging
from datetime import datetime
from .schema import RoutingDecision
from langchain_core.messages import BaseMessage

logger = logging.getLogger("ticket_routing")
logger.setLevel(logging.INFO)

def log_routing_decision(
    conversation_id: str,
    user_message: str,
    decision: RoutingDecision,
    ticket_id: str | None,
    history: list[BaseMessage],
) -> None:
    record = {
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "conversation_id": conversation_id,
        "user_message": user_message[:500],
        "category": decision.category.value,
        "confidence": decision.confidence,
        "reasoning": decision.reasoning,
        "priority": decision.suggested_priority,
        "requires_human": decision.requires_human,
        "metadata": decision.metadata,
        "ticket_id": ticket_id,
        "history_length": len(history),
    }
    logger.info(json.dumps(record))

Call log_routing_decision inside SupportRouter.route after ticket creation. Ship these logs to your observability stack (Datadog, Splunk, OpenTelemetry) and build a dashboard showing:

• Category distribution
• Confidence histogram
• Human escalation rate
• Ticket creation latency

Step 8: Build an evaluation harness

Create a labeled dataset of 100-200 real (anonymized) support messages with ground-truth categories. Run the classifier against it weekly.

# evals/routing_eval.py
import csv
from pathlib import Path
from routing.classifier import build_classifier_chain
from routing.context import format_history
from routing.schema import TicketCategory
from langchain_core.messages import HumanMessage

DATASET_PATH = Path(__file__).parent / "data" / "routing_labels.csv"

def load_dataset():
    with open(DATASET_PATH) as f:
        reader = csv.DictReader(f)
        return list(reader)

def evaluate():
    chain = build_classifier_chain()
    data = load_dataset()
    correct = 0
    total = len(data)
    confusion = {}

    for row in data:
        message = row["message"]
        expected = TicketCategory(row["category"])
        history = [HumanMessage(content=row["history"])] if row["history"] else []
        
        decision = chain.invoke({
            "message": message,
            "history": format_history(history),
        })
        
        predicted = decision.category
        if predicted == expected:
            correct += 1
        else:
            confusion.setdefault(expected.value, {}).setdefault(predicted.value, 0)
            confusion[expected.value][predicted.value] += 1

    accuracy = correct / total
    print(f"Accuracy: {accuracy:.2%} ({correct}/{total})")
    print("Confusion matrix:")
    for actual, preds in confusion.items():
        print(f"  {actual}: {preds}")

if __name__ == "__main__":
    evaluate()

Store the CSV in version control (with PII stripped). Add this to CI so regressions block merges.

Step 9: Verify end-to-end in staging

Deploy to a staging environment and run these manual checks:

1. Happy path: Send “I was charged twice for my September invoice” → verify a BILLING ticket appears in Jira with correct priority and metadata (amount, date if extracted).
2. Low confidence: Send “Thing is broken” → verify ticket lands in GENERAL queue with requires_human=true and the confidence note in reasoning.
3. Human escalation trigger: Send “I need to delete all my data per GDPR” → verify requires_human=true, category ACCOUNT, priority urgent.
4. Metadata extraction: Send “Order #ORD-12345 shows wrong items” → verify metadata.order_id == "ORD-12345" in the ticket description.
5. Idempotency: Submit the same message twice → verify only one ticket is created (implement deduplication in your backend if needed).

Check logs for the structured JSON output from Step 7. Confirm ticket IDs are returned to the user in the chat response.

Step 10: Harden for production

Before cutting over, address these operational concerns:

• Rate limiting: Wrap the classifier chain with a semaphore or token bucket. The LLM call is your bottleneck.
• Provider fallback: If you route classification through a gateway like n4n.ai, configure automatic fallback to a secondary model when the primary is degraded — this prevents routing outages during provider incidents.
• PII scrubbing: Run user messages through a PII detector (Presidio, AWS Comprehend) before they hit the classifier. Never log raw messages in production.
• Caching: For repeated identical queries (common in bot loops), cache the RoutingDecision keyed by message hash + recent history hash. TTL of 1 hour is safe.
• Canary rollout: Route 5% of traffic to the new classifier, compare category distribution and human escalation rate against the old rule-based router. Ramp over 48 hours.

You now have a ticket routing langchain support bot that classifies, enriches, and dispatches tickets with observable, testable, and swappable components. The classifier chain, router protocol, and evaluation harness form a foundation you can extend with intent detection, auto-reply suggestions, or SLA-based priority escalation without rewriting the core flow.