Тріаж флоту LangGraph: аварійний коротке замикання перед діагностикою RAG

Чат-боти для ремонту флоту потребують більшого, ніж просто лінійна послідовність RAG. Повідомлення водія, надіслане через WhatsApp, може бути лише легким звуком — або ж проблемою з гальмами, що вимагає негайного наказу про зупинку. Станові вузли та умовні краї LangGraph перетворюють цей процес сортування на чітку машину станів замість простої схеми дій.

Сценарій використання. Повідомлення з WhatsApp потрапляє на webhook Infobip, проходить очищення та перевірку, після чого потрапляє у робочий процес агента. Спочатку необхідно визначити ступінь терміновості: у разі справжньої надзвичайної ситуації діяти негайно, інакше провести пошук у посібниках Daihatsu Mira, створити структуровану діагностику та перевірити її точність.

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. Центральна пам’ять: FleetState

Кожен запуск передає спільний „блокнот стану“ між вузлами:

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

• driver_message — необроблений текст з webhook

is_emergency — прапорець класифікатора безпеки

documents — отримані ручні уривки

response — діагностичний або інформаційний текст про безпеку

eval_verdict — результат аудиту з використанням LLM як судді

2. Станції виконання

Кожен вузол — це функція (FleetState) -> partial update:

• triage_node — шукає критичні ознаки (відпуск педалі, витік рідини, відсутність гальм) та встановлює значення is_emergency.
• emergency_node — лише у разі надзвичайної ситуації; ігнорує пошук за векторами та використання LLM; надсилає негайний наказ про зупинку та повідомляє відділ керування флотом.
• retrieve_node — сценарій без надзвичайної ситуації; пошук за схожістю у посібниках ChromaDB; додає фрагменти та метадані сторінок до documents.
• generate_node — заповнює шаблон механіка (Оцінка, Можливі причини, Діагностичні перевірки, Безпека) та викликає локальний Llama 3.2 через Ollama, щоб отримати результат у response.
• evaluate_node — оцінює релевантність та точність контексту (без вигаданих характеристик крутного моменту); зберігає eval_verdict.

3. Підключення StateGraph

Реєструйте вузли, встановіть початок у розділі триажу та додайте route_triage(state):

• надзвичайна ситуація → emergency_node → КІНЕЦЬ
• інакше → retrieve_node → generate_node → evaluate_node → КІНЕЦЬ

Компілюйте за допомогою workflow.compile(), а викликайте за допомогою {"driver_message": query}. Звіт про різкий гальмування формується за маршрутом RAG; „губчаста“ педаль запускає захисний коротке замикання. Саме ця різниця визначає різницю між демонстраційним чат-ботом та функціональною схемою триажу.

Чому триаж має бути в схемі

Чистий ланцюг LCEL завжди отримує дані та генерує результати. Це неправильно з точки зору безпеки: кожен мілісекунду, витрачений на вбудовування повідомлення „без гальм“, — це затримка перед командою зупинки. Розміщення класифікації у першому вузлі та робота з надзвичайними ситуаціями як першокласної гілки дозволяють закодувати правила в топології, а не лише сподіватися на правильні запити.

Обґрунтування вузла оцінки

Навіть у режимі RAG можна отримувати посібники, поки модель все ще генерує числа. Вузол судді оцінює точність отриманих уривків перед надсиланням через WhatsApp. Невдалі рішення можуть бути перегенеровані або передані людині-оператору замість того, щоб надсилати спотворену інформацію.

Схема реалізації

Webhook → очищення → app.invoke → передача response назад через Infobip. Зберігайте шаблони для надзвичайних ситуацій короткими та багатомовними, якщо це потрібно водіям. Вимірюйте кількість триажу, кількість випадків швидкого реагування у надзвичайних ситуаціях та кількість невдалих спроб окремо від загальної затримки — щоб команда керування бачила стан безпеки, а не лише кількість токенів за секунду.

Способи збоїв для тестування

• Хибні позитивні результати щодо ключових слів у надзвичайних ситуаціях (сленг, який не є збоєм) не повинні блокувати нормальну роботу RAG.

Відсутність інструкцій до певного симптому має призводити до безпечної відповіді „передати на ремонтну майстерню“, а не до вигаданого значення крутного моменту.

Оцінка часу простою вузла має завершуватися успішно: не слід надсилати драйверам неперевірену діагностику.

Повторні спроби через Webhook мають бути ідемпотентними, щоб подвійна доставка повідомлень у WhatsApp не спричиняла подвійного оповіщення менеджерів флоту.

Підсумок

LangGraph перетворює процеси сортування, пошуку, генерації та аудиту на видиму топологію. Стан буфера копіювання дозволяє перевіряти кожне рішення, а механізм надзвичайних ситуацій гарантує, що безпека ніколи не залежатиме від ембеддингів.

Документуйте припущення щодо потоку керування поруч із кодом, щоб майбутні зміни не могли таємно видалити певну грань чи фільтр. Віддавайте перевагу твердженням, які можна перевірити за допомогою машин, замість інформації, яка передається лише в чатах. Проводьте тренування на випадок збоїв щоразу, коли змінюються правила топології чи ідентичності. Зберігайте набори даних для оцінки у версіях разом із графом, щоб проблеми проявлялися раніше, ніж це зроблять клієнти. Розглядайте відмови та неповні запуски як результати першого класу, які заслуговують на окремі панелі керування.