LangGraph集群分诊：在RAG诊断之前先处理紧急短路问题

车队维修聊天机器人需要的不仅仅是线性的RAG链。通过WhatsApp发送的驾驶员消息可能只是轻微的异响，也可能意味着刹车故障需要立即下达停车指令。LangGraph的状态节点和条件边将这种分类处理转化为明确的状态机，而非依赖运气运行的脚本。

应用场景。收到的WhatsApp消息会传送到Infobip的webhook，经过清洗和验证后进入自动化工作流。该工作流首先需进行紧急程度分类：立即处理真正的紧急情况，否则则检索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 — 填充机械故障处理模板（评估、可能原因、诊断检查、安全措施），并通过 Ollama 调用本地的 Llama 3.2 生成 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 → 通过Infobip将response传回。若司机需要，可将紧急模板设计得简短且包含双语内容。需将分诊率、紧急情况短路次数以及失败率与常规延迟指标分开统计，这样运营人员就能了解系统安全性，而不仅仅是每秒处理的任务数。

需测试的故障模式

• 紧急关键词的误报（非故障性的俚语）不应阻碍正常的RAG功能。

遇到缺少相关手册的症状时，应给出“转交维修站”这一安全处理方案，而非编造扭矩数值。

对节点停机时间的评估必须给出明确结论：不得向驾驶员发送未经验证的诊断信息。

Webhook重试操作必须具备幂等性，以避免重复的WhatsApp通知导致车队管理人员收到多次提醒。

总结

LangGraph将分诊、检索、生成和审计功能转化为可视化的拓扑结构。剪贴板状态使得每一项决策都可被审查，而紧急处理机制确保安全问题不会因嵌入模型处理延迟而被搁置。

在代码旁记录控制流假设，以免后续修改时悄悄删除某条路径或过滤器。优先使用可机器验证的断言，而非仅在聊天中交流的隐性知识。每当拓扑结构或身份规则发生变化时，都要进行故障演练。将评估集与图结构一起版本控制，以便在客户发现问题之前就发现回归问题。应将拒绝处理和运行不完整的情况视为重要结果，专门为其创建监控面板。