Триаж флота 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 — результат аудита с использованием ИИ в роли судьи

2. Станции выполнения

Каждый узел представляет собой функцию (FleetState) -> partial update:

• triage_node — сканирует на наличие критических признаков (смягчение торможения, утечка жидкости, отсутствие тормозов) и устанавливает значение is_emergency.
• emergency_node — применяется только в чрезвычайных ситуациях; пропускает поиск векторов и использование ИИ; возвращает команду о немедленной остановке и уведомляет службу управления флотом.
• 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 превращает процессы сортировки, поиска, генерации и аудита в видимую топологию. Состояние буфера обмена позволяет проверять каждое принятое решение, а механизм чрезвычайных ситуаций гарантирует, что безопасность никогда не будет зависеть от результатов обработки данных.

Документируйте предположения, касающиеся потока управления, рядом с кодом, чтобы будущие изменения не могли незаметно удалить какую-либо грань или фильтр. Предпочитайте утверждения, проверяемые машиной, вместо знаний, передаваемых только в чатах. Проводите тренировки на случай сбоев каждый раз, когда меняются правила топологии или идентичности. Сохраняйте версии наборов данных вместе с графом, чтобы проблемы проявлялись раньше, чем это сделают клиенты. Рассматривайте отказы и неполные запуски как важные результаты, заслуживающие отдельных панелей управления.