LangGraph fleet triage: emergency short-circuit before RAG diagnostics

Fleet repair chatbots need more than a linear RAG chain. A driver message arriving over WhatsApp may be a mild squeal—or a brake failure that demands an immediate stop order. LangGraph’s stateful nodes and conditional edges turn that triage into an explicit state machine instead of a hopeful script.

Use case. An inbound WhatsApp message hits an Infobip webhook, gets sanitized and validated, then enters the agentic workflow. The workflow must triage first: short-circuit true emergencies, otherwise run retrieval over Daihatsu Mira manuals, generate structured diagnostics, and audit faithfulness.

from pathlib import Path
from typing import TypedDict, List
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_chroma import Chroma
from langchain_ollama import OllamaLLM
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.documents import Document
from langgraph.graph import StateGraph, START, END

# ------------------------------------------------------------------------------
# 1. SETUP & INGESTION (Offline Preparation)
# ------------------------------------------------------------------------------
PDF_PATH = Path(__file__).with_name("brake_pad_maintenance_manual.pdf")

loader = PyPDFLoader(str(PDF_PATH))
docs = loader.load()
splits = RecursiveCharacterTextSplitter(chunk_size=1200, chunk_overlap=150).split_documents(docs)

embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
vectorstore = Chroma.from_documents(documents=splits, embedding=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

llm = OllamaLLM(model="llama3.2", temperature=0)


# ------------------------------------------------------------------------------
# 2. STATE & GRAPH DEFINITION
# ------------------------------------------------------------------------------
class FleetState(TypedDict):
    driver_message: str
    is_emergency: bool
    documents: List[Document]
    response: str
    eval_verdict: str


# Node 1: Triage/Safety Router
def triage_node(state: FleetState):
    msg = state["driver_message"].lower()
    emergency_terms = ["soft pedal", "fluid leaking", "no brakes", "spongy", "pedal to floor"]
    is_hazard = any(term in msg for term in emergency_terms)
    return {"is_emergency": is_hazard}


# Node 2A: Emergency Path (Bypasses RAG completely)
def emergency_node(state: FleetState):
    response = (
        "🚨 **CRITICAL SAFETY ALERT** 🚨\n\n"
        "Stop driving immediately! A soft or leaking brake system indicates hydraulic failure.\n"
        "Fleet operations (David) has been automatically notified. Pull over safely and await support."
    )
    return {"response": response}


# Node 2B: Standard RAG Retrieval Path
def retrieve_node(state: FleetState):
    docs = retriever.invoke(state["driver_message"])
    return {"documents": docs}


# Node 3: Generation Path (LangChain prompt execution)
def generate_node(state: FleetState):
    context = "\n\n".join(
        f"[{doc.metadata.get('source')}, page {doc.metadata.get('page') + 1}]\n{doc.page_content}"
        for doc in state["documents"]
    )
    prompt_template = """
    You are a cautious fleet mechanic assistant. Answer strictly from the manual passages.
    For a reported symptom, use four headings: Assessment, Possible causes, Diagnostic checks, and Safety.

    === CONTEXT ===
    {context}

    === QUERY ===
    {question}
    """
    prompt = ChatPromptTemplate.from_template(prompt_template)
    response = llm.invoke(prompt.format(context=context, question=state["driver_message"]))
    return {"response": response}


# Node 4: Evaluation Node (Audits Context & Response Quality)
def evaluate_node(state: FleetState):
    if state.get("is_emergency"):
        return {"eval_verdict": "SKIPPED (Emergency Routing Overrode RAG)"}

    eval_prompt_template = """
    Audit the mechanic assistant's answer against the retrieved context:
    1. RETRIEVAL CHECK: Are the retrieved documents relevant to the query? (PASS/FAIL)
    2. FAITHFULNESS CHECK: Does the answer introduce any outside facts or unstated specs? (PASS/FAIL)

    Context: {context}
    Answer: {response}

    Verdict:
    """
    eval_prompt = ChatPromptTemplate.from_template(eval_prompt_template)
    context = "\n\n".join([d.page_content for d in state["documents"]])
    verdict = llm.invoke(eval_prompt.format(context=context, response=state["response"]))
    return {"eval_verdict": verdict}


# ------------------------------------------------------------------------------
# 3. GRAPH ASSEMBLY
# ------------------------------------------------------------------------------
workflow = StateGraph(FleetState)

workflow.add_node("triage", triage_node)
workflow.add_node("emergency", emergency_node)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)
workflow.add_node("evaluate", evaluate_node)


def route_triage(state: FleetState):
    return "emergency" if state["is_emergency"] else "retrieve"


workflow.add_edge(START, "triage")
workflow.add_conditional_edges("triage", route_triage, {"emergency": "emergency", "retrieve": "retrieve"})
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "evaluate")
workflow.add_edge("evaluate", END)
workflow.add_edge("emergency", END)

app = workflow.compile()

# ------------------------------------------------------------------------------
# 4. RUNNING DEMO SCENARIOS
# ------------------------------------------------------------------------------
# Case A: Standard Maintenance Question
print("--- RUNNING CASE A: Standard Screeching Brakes ---")
res_a = app.invoke({"driver_message": "My brakes make a screeching noise when stopping."})
print(f"\n[RESPONSE]:\n{res_a['response']}")
print(f"\n[EVALUATION]:\n{res_a['eval_verdict']}")

# Case B: Emergency Incident
print("\n" + "=" * 60 + "\n")
print("--- RUNNING CASE B: Emergency Hydraulic Failure ---")
res_b = app.invoke({"driver_message": "My brake pedal feels soft and spongy and fluid is leaking!"})
print(f"\n[RESPONSE]:\n{res_b['response']}")
print(f"\n[EVALUATION]:\n{res_b['eval_verdict']}")

1. Central memory: FleetState

Every run carries a shared state clipboard between nodes:

class FleetState(TypedDict):
    driver_message: str
    is_emergency: bool
    documents: List[Document]
    response: str
    eval_verdict: str

• driver_message — raw text from the webhook
• is_emergency — safety classifier flag
• documents — retrieved manual passages
• response — diagnostic or safety text
• eval_verdict — LLM-as-judge audit result

2. Execution stations

Each node is a function (FleetState) -> partial update:

• triage_node — scans for critical cues (soft pedal, fluid leak, no brakes) and sets is_emergency.
• emergency_node — only if emergency; skips vector search and LLM; returns an immediate stop order and notifies fleet ops.
• retrieve_node — non-emergency path; similarity search over ChromaDB manuals; appends chunks and page metadata to documents.
• generate_node — fills a mechanic template (Assessment, Possible causes, Diagnostic checks, Safety) and calls local Llama 3.2 via Ollama into response.
• evaluate_node — judges context relevance and faithfulness (no invented torque specs); stores eval_verdict.

3. Wiring StateGraph

Register nodes, set entry at triage, and add route_triage(state):

• emergency → emergency_node → END
• otherwise → retrieve_node → generate_node → evaluate_node → END

Compile with workflow.compile() and invoke with {"driver_message": query}. A screeching-brake report follows the RAG path; a spongy pedal triggers the safety short circuit. That split is the production difference between a demo chatbot and an operational triage graph.

Why triage belongs in the graph

A pure LCEL chain always retrieves and generates. That is wrong for safety: every millisecond spent embedding a “no brakes” message is delay before the stop order. Putting classification in the first node and making emergency a first-class branch encodes policy in topology, not in prompt hope.

Eval node rationale

Even on the RAG path, manuals can be retrieved while the model still invents numbers. The judge node scores faithfulness against retrieved passages before WhatsApp delivery. Failed verdicts can loop to regenerate or escalate to a human dispatcher instead of sending a polished hallucination.

Deployment sketch

Webhook → sanitize → app.invoke → map response back through Infobip. Keep emergency templates short and bilingual if drivers need it. Metricize triage rate, emergency short-circuit count, and eval failure rate separately from generic latency—so ops sees safety health, not just tokens per second.

Failure modes to test

• Emergency keyword false positives (slang that is not a failure) should not block legitimate RAG.
• Missing manuals for a symptom should yield a safe “escalate to workshop” response, not an invented torque value.
• Eval node downtime should fail closed: do not send unaudited diagnostics to drivers.
• Webhook retries must be idempotent so duplicate WhatsApp deliveries do not double-notify fleet managers.

Summary

LangGraph turns triage, retrieval, generation, and audit into visible topology. The clipboard state makes each decision inspectable, and the emergency edge exists so safety never waits on embeddings.

Document the control-flow assumptions beside the code so future edits cannot silently delete an edge or a filter. Prefer machine-checkable assertions over tribal knowledge shared only in chat threads. Rehearse failure drills whenever topology or identity rules change. Keep evaluation sets versioned with the graph so regressions surface before customers do. Treat refusals and incomplete runs as first-class outcomes worth dashboards of their own.