Triaje de la flota LangGraph: cortocircuito de emergencia antes de los diagnósticos RAG

Los chatbots de reparación de flotas necesitan algo más que una cadena lineal RAG. Un mensaje del conductor enviado por WhatsApp puede ser un simple chirrido o una falla en los frenos que exige una orden de parada inmediata. Los nodos con estado y los bordes condicionales de LangGraph convierten ese proceso de triaje en una máquina de estados explícita en lugar de un script basado en expectativas.

Caso de uso. Un mensaje entrante por WhatsApp llega a un webhook de Infobip, es sanitizado y validado, y luego entra en el flujo de trabajo automatizado. Este flujo debe realizar primero un triaje: atender de inmediato las emergencias reales; de lo contrario, buscar información en los manuales del Daihatsu Mira, generar diagnósticos estructurados y verificar su precisión.

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. Memoria central: FleetState

Cada ejecución transfiere un portapapeles de estado compartido entre los nodos:

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

• driver_message — texto sin procesar del webhook

is_emergency — bandera del clasificador de seguridad

documents — pasajes manuales recuperados

response — texto de diagnóstico o de seguridad

eval_verdict — resultado de la auditoría con el LLM como juez

2. Estaciones de ejecución

Cada nodo es una función (FleetState) -> actualización parcial:

• triage_node — busca indicadores críticos (pedal suave, fuga de líquido, ausencia de frenos) y establece is_emergency.
• emergency_node — solo en caso de emergencia; omite la búsqueda vectorial y el LLM; devuelve una orden de parada inmediata e informa al equipo de operaciones de la flota.
• retrieve_node — ruta para situaciones no emergentes; búsqueda por similitud en los manuales de ChromaDB; agrega fragmentos y metadatos de página a documents.
• generate_node — rellena un modelo mecánico (Evaluación, Causas posibles, Verificaciones de diagnóstico, Seguridad) y llama al Llama 3.2 local a través de Ollama para obtener el response.
• evaluate_node — evalúa la relevancia y fidelidad del contexto (sin especificaciones de par inventadas); almacena eval_verdict.

3. Conexión de StateGraph

Registre los nodos, establezca el punto de entrada en la triaje y agregue route_triage(state):

• emergencia → emergency_node → FIN
• de lo contrario → retrieve_node → generate_node → evaluate_node → FIN

Compílelo con workflow.compile() e invoquelo con {"driver_message": query}. Un informe de frenos estridentes sigue el camino RAG; un pedal blando activa el cortocircuito de seguridad. Esa diferencia es la que marca la distinción entre un chatbot de demostración y un grafo de triaje operativo.

Por qué el triaje debe formar parte del grafo

Una cadena pura LCEL siempre recupera y genera información. Eso es incorrecto desde el punto de vista de la seguridad: cada milisegundo invertido en incrustar un mensaje de “sin frenos” representa una demora antes de la orden de detención. Al incluir la clasificación en el primer nodo y hacer que la emergencia sea un ramo de primera categoría, se codifica la política en la topología, no en esperanzas basadas en prompts.

Razones detrás del nodo de evaluación

Incluso en el camino RAG, se pueden recuperar manuales mientras el modelo sigue generando números. El nodo de evaluación mide la fidelidad de los pasajes recuperados antes de enviarlos por WhatsApp. Las decisiones fallidas pueden reiniciarse o escalarse a un operador humano en lugar de enviar una alucinación pulida.

Boceto de implementación

Webhook → sanitización → app.invoke → mapeo del response de vuelta a través de Infobip. Mantenga los modelos de emergencia breves y bilingües si los conductores lo necesitan. Mida por separado la tasa de triaje, el número de cortocircuitos de emergencia y la tasa de fallos, aparte de la latencia general; así, el equipo operativo podrá ver el estado de seguridad y no solo los tokens por segundo.

Modos de fallo a probar

• Los falsos positivos en palabras clave de emergencia (lenguaje coloquial que no constituye un fallo) no deben bloquear el funcionamiento normal de RAG.

La falta de manuales para un síntoma debe generar una respuesta segura de “escalificar al taller”, y no un valor de par inventado.

La evaluación del tiempo de inactividad del nodo debe fallar de forma definitiva: no se deben enviar diagnósticos sin verificar a los conductores.

Las reintentos de webhook deben ser idempotentes para que las entregas duplicadas por WhatsApp no notifiquen dos veces a los gerentes de flota.

Resumen

LangGraph convierte la triaje, recuperación, generación y auditoría en una topología visible. El estado del portapapeles permite inspeccionar cada decisión, y existe el mecanismo de emergencia para que la seguridad nunca dependa de los embeddings.

Documente las suposiciones sobre el flujo de control junto al código para que las ediciones futuras no puedan eliminar silenciosamente un enlace o un filtro. Prefiera afirmaciones verificables por máquinas sobre conocimientos compartidos únicamente en hilos de chat. Practique simulacros de fallos cada vez que cambien las reglas de topología o identidad. Mantenga los conjuntos de evaluación versionados junto con el grafo para que las regresiones se detecten antes de que lo hagan los clientes. Trate las rechazos y ejecuciones incompletas como resultados de primera categoría que merecen sus propios paneles de control.