LangGraph-Flotten-Triage: Notfall-Short-Circuit vor RAG-Diagnose

Fleet-Reparatur-Chatsbots benötigen mehr als nur eine lineare RAG-Kette. Eine über WhatsApp eingehende Nachricht des Fahrers kann ein leises Quietschen sein – oder ein Bremsversagen, das einen sofortigen Anhaltbefehl erfordert. LangGraphs Zustandsknoten und bedingte Kanten verwandeln diese Priorisierung in eine explizite Zustandsmaschine statt in einen vagen Algorithmus.

Anwendungsfall: Eine eingehende WhatsApp-Nachricht erreicht einen Infobip-Webhook, wird gereinigt und überprüft, bevor sie in den agierenden Workflow gelangt. Zunächst muss dieser Workflow Prioritäten setzen: echte Notfälle sofort bearbeiten, ansonsten Informationen aus den Daihatsu Mira-Handbüchern abrufen, strukturierte Diagnosen erstellen und die Genauigkeit überprüfen.

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. Zentrale Speicherung: FleetState

Jeder Ablauf verwendet einen gemeinsamen Zustandsklipboard zwischen den Knoten:

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

• driver_message – Rohtext aus dem Webhook

is_emergency – Flagge zum Sicherheitsklassifikator

documents – abgerufte manuelle Textabschnitte

response – diagnostischer oder sicherheitstechnischer Text

eval_verdict – Prüfergebnis mit dem LLM als Richter

2. Ausführungsstationen

Jeder Knoten ist eine Funktion (FleetState) -> partial update:

• triage_node – sucht nach kritischen Anzeichen (weicher Pedal, Flüssigkeitsleck, keine Bremsen) und setzt is_emergency.
• emergency_node – nur im Notfall; überspringt die Vektorabfrage und das LLM; gibt einen sofortigen Stopps Befehl aus und informiert die Flottenverwaltung.
• retrieve_node – Nicht-Notfall-Pfad; Ähnlichkeitsabfrage in den ChromaDB-Manualen; fügt Abschnitte sowie Seitenmetadaten zu documents hinzu.
• generate_node – füllt ein Mechanik-Template (Bewertung, Mögliche Ursachen, Diagnoseprüfungen, Sicherheit) aus und ruft über Ollama den lokalen Llama 3.2 auf, um einen response zu erzeugen.
• evaluate_node – prüft die Relevanz und Genauigkeit des Kontexts (keine erfundenen Drehmomentwerte); speichert eval_verdict.

3. Verkabelung von StateGraph

Registrieren Sie die Knoten, legen Sie den Eingang bei der Triage fest und fügen Sie route_triage(state) hinzu:

• Notfall → emergency_node → ENDE
• Andernfalls → retrieve_node → generate_node → evaluate_node → ENDE

Kompilieren Sie mit workflow.compile() und rufen Sie es mit {"driver_message": query} auf. Ein Bericht über plötzliches Bremsen folgt dem RAG-Pfad; ein schwammiger Pedaldruck löst den Sicherheitskurzschluss aus. Dieser Unterschied ist der Produktionsunterschied zwischen einem Demo-Chattenbot und einem betriebsbereiten Triage-Graphen.

Warum Triage in den Graphen gehört

Eine reine LCEL-Kette holt stets Daten ab und erzeugt Ergebnisse. Das ist für die Sicherheit falsch: Jeder Millisekunde, die mit dem Einbetten einer „keine Bremsen“-Meldung verbracht wird, ist eine Verzögerung vor dem Stoppsignal. Die Einbeziehung der Klassifizierung in den ersten Knoten sowie die Gestaltung einer Notfallsparte als erstklassige Abzweigung kodieren die Richtlinien in der Topologie – nicht in der Hoffnung auf passende Eingaben.

Begründung für den Eval-Knoten

Auch auf dem RAG-Pfad können Handbücher abgerufen werden, während das Modell noch Zahlen erfindet. Der Judge-Node bewertet die Treue der abgerufenen Passagen vor der Übermittlung per WhatsApp. Misserfolgreiche Entscheidungen können zur Neugenerierung zurückgeführt werden oder an einen menschlichen Dispatcher eskalieren, anstatt eine fehlerhafte Halluzination zu senden.

Einsatzskizze

Webhook → Reinigung → app.invoke → Umwandlung der Response über Infobip zurück. Halten Sie Notfallvorlagen kurz und zweisprachig, falls Fahrer dies benötigen. Messen Sie die Triage-Rate, die Anzahl der Notfall-Abbrüche sowie die Fehlerquote getrennt von der allgemeinen Latenz – damit das Betriebspersonal den Sicherheitszustand erkennen kann, nicht nur die Anzahl der Tokens pro Sekunde.

Fehlermöglichkeiten zum Testen

• Falschpositive Notfall-Schlüsselwörter (Umgangssprache, die kein Fehler ist), sollten eine legitime RAG-Nutzung nicht blockieren.

Fehlende Handbücher zu einem Symptom sollten zu einer sicheren Reaktion „an die Werkstatt weiterleiten“ führen, nicht zu einem erfundenen Drehmomentwert.

Die Überprüfung der Ausfallzeit des Evaluationsknotens sollte fehlschlagen – es dürfen keine unüberprüften Diagnosen an Fahrer gesendet werden.

Webhook-Neversuche müssen idempotent sein, damit doppelte WhatsApp-Nachrichten die Flottenmanager nicht zweimal benachrichtigen.

Zusammenfassung

LangGraph verwandelt Triage, Abruf, Erstellung und Überprüfung in eine sichtbare Topologie. Der Zustand der Zwischenablage macht jede Entscheidung überprüfbar, und die Notfallfunktion sorgt dafür, dass die Sicherheit niemals auf Embeddings warten muss.

Dokumentieren Sie die Annahmen bezüglich des Kontrollflusses neben dem Code, damit zukünftige Änderungen nicht heimlich eine Kante oder einen Filter löschen können. Ziehen Sie maschinenprüfbare Aussagen vor, anstelle von nur in Chat-Threads geteiltem „Stammeswissen“. Üben Sie immer wieder Fehlerbehebungsübungen durch, wenn sich Topologie- oder Identitätsregeln ändern. Halten Sie die Bewertungssätze zusammen mit dem Graphen versioniert, damit Regressionen bereits vor den Kunden aufgedeckt werden. Behandeln Sie Ablehnungen und unvollständige Ausführungen als Ergebnisse erster Klasse, die eigene Dashboards verdienen.