LLM classifier with a keyword fast-path

The supervisor runs on every single user message. If it uses a big model, you pay for that call every turn. If it uses no model at all, you miss routing nuance. The fix is a two-stage design: a cheap keyword gate that catches obvious cases, then a small LLM classifier for everything else.

agent_graph.py
python
_VALID_ROUTES = {'policy', 'billing', 'claims', 'escalation'}

_KEYWORD_HINTS = {
    'policy': ['coverage', 'covered', 'deductible', 'policy', 'terms', 'exclusion', 'benefit'],
    'billing': ['bill', 'premium', 'invoice', 'payment', 'charge', 'balance', 'due', 'autopay'],
    'claims': ['claim', 'accident', 'damage', 'incident', 'file a', 'reimburse', 'payout'],
    'escalation': ['human', 'agent', 'manager', 'supervisor', 'complaint', 'angry', 'lawyer', 'lawsuit'],
}


def _keyword_route(text: str) -> str | None:
    lowered = text.lower()
    scores = {route: sum(1 for kw in kws if kw in lowered) for route, kws in _KEYWORD_HINTS.items()}
    best = max(scores, key=scores.get)
    return best if scores[best] > 0 else None

The keyword map is short and readable. Every word gets a point, and the route with the highest score wins. 'file a claim' hits two claims keywords, 'my lawyer says' hits escalation, and 'what is my balance' hits billing. No LLM call, zero latency beyond a string lookup.

agent_graph.py
python
SUPERVISOR_PROMPT = '''You are the supervisor agent for an insurance copilot.
Classify the user's latest message into exactly one of these routes:

- policy: questions about coverage, terms, deductibles, exclusions (use the RAG KB)
- billing: questions about premiums, invoices, balances, payments (use the SQL DB)
- claims: filing, checking, or disputing a claim (may use both KB and SQL)
- escalation: user is angry, asks for a human, legal threats, or anything unsafe

Respond with a single JSON object: {{"route": "policy|billing|claims|escalation"}}.

Conversation so far:
{history}

Latest user message:
{user_input}
'''

The supervisor prompt includes conversation history so it can disambiguate follow-ups like "what about mine?" The output is JSON, not prose, so parsing is a simple regex plus json.loads.

agent_graph.py
python
async def supervisor_node(state: ClaimState) -> Dict[str, Any]:
    user_input = state.get('user_input', '')
    history = _format_history(state.get('messages', []))

    # Cheap keyword gate catches obvious escalations before spending an LLM call.
    kw = _keyword_route(user_input)
    if kw == 'escalation':
        return {'route': 'escalation'}

    provider = get_llm_provider()
    prompt = SUPERVISOR_PROMPT.format(history=history, user_input=user_input)
    try:
        raw = await provider.generate_text(prompt, max_tokens=60, temperature=0)
        match = re.search(r'\{.*?\}', raw, re.DOTALL)
        if match:
            parsed = json.loads(match.group(0))
            route = parsed.get('route', '').strip().lower()
            if route in _VALID_ROUTES:
                return {'route': route}
    except Exception as exc:
        logger.warning('Supervisor LLM failed (%s); falling back to keyword routing.', exc)

    return {'route': kw or 'policy'}

The order matters. Escalation goes through the keyword gate FIRST, because angry users should never wait for an LLM call. Everything else goes through the classifier. If the classifier fails (LLM error, malformed JSON, invalid route), we fall back to the keyword route, and if that also fails, we default to policy because it is the safest RAG-grounded answer.

The supervisor returns a tiny JSON object with one field. Sixty tokens is more than enough. Temperature zero means the same message always routes the same way, which is critical for debugging and for consistent user experience. The goal is a deterministic classifier, not a creative one.

Quiz: Quiz

Loading practice…