1. 为什么我要用 LangGraph 和 MCP 重做质量分析系统

在制造业做质量分析超过五年的朋友大概率都有同一个感受：数据不是没有，而是散落在太多地方。SPC 控制图跑在 Minitab 或者某个自研的看板里，工艺参数躺在 MES 的时序数据库，缺陷记录在 QMS，设备维保日志又在另一个系统。每次产线出现异常，质量工程师要做的第一件事不是分析，而是"找数据"——导出、对齐时间戳、手动拼表，一套流程下来半天就没了，等真正开始做根因分析，异常批次可能已经流到下游了。

我这次要聊的项目，就是拿LangGraph做流程编排、用MCP（Model Context Protocol）做工具和数据源的统一接入层，搭一套能自动串起"异常检测 → 多源数据取证 → 根因推理 → 报告生成"的 AI-SPC 智能分析系统。核心目标很明确：把质量工程师从"数据搬运工"的角色里解放出来，让系统自己去打通质量根因链条。

先说清楚这套东西适合谁看。如果你是完全不懂代码的质量工程师，这篇文章能帮你理解这类系统大概是怎么运转的，方便你跟 IT 提需求；如果你是做数据平台或 AI 应用的开发者，那这篇基本可以当成一份落地参考，我会把架构选型、节点设计、MCP 工具封装、状态管理这些关键环节都拆开讲，包括我踩过的坑。前置知识上，你最好对 Python 异步编程、SPC 的基本概念（Cpk、UCL/LCL、Western Electric 判异规则）有个大概印象，不熟也没关系，涉及的地方我会补一句。

为什么非得是 LangGraph 而不是普通的 LangChain Chain，或者干脆写个脚本？这是我在项目初期纠结最久的问题，后面会专门用一节讲清楚。简单说一句：质量根因分析天然是一个有状态、有分支、可能需要回退重查的流程，线性 Chain 撑不住这种复杂度，而 LangGraph 的图结构和检查点机制刚好对得上。至于 MCP，它解决的是另一个维度的痛点——工具和数据的接入标准化，让"加一个新数据源"从改代码变成加配置。

2. 系统整体架构与核心技术选型拆解

2.1 从业务视角看这套系统到底在干什么

先把业务链路讲清楚，不然后面聊技术容易飘。一条典型的产线质量异常处理流程是这样的：SPC 系统监控到某个关键特性（比如某零件的孔径尺寸）连续 7 点上升，触发判异规则告警。传统做法是质量工程师收到告警，去查这批产品对应的设备、班次、原材料批次、环境温湿度，然后凭经验判断可能是刀具磨损还是来料波动。

AI-SPC 系统要做的就是把这个过程自动化。它接收告警事件后，会依次完成几件事：确认异常是否真实（排除测量系统误报）、拉取关联的多源数据、基于历史相似案例和工艺知识做推理、给出根因假设并按可能性排序、最后生成一份带证据链的分析报告。整个过程里，LangGraph 负责"怎么走"——先查什么、什么条件下要回退、哪一步需要人工介入；MCP 负责"能查什么"——把 MES、QMS、SPC 数据库、知识库这些异构数据源统一成标准工具供图节点调用。

2.2 为什么是 LangGraph：状态、分支与可回溯

普通 Chain 的问题在于它是单向的。质量分析里有个很典型的场景：系统初步判断是"设备刀具磨损"，但拉取维保记录后发现刀具三天前刚换过，这时候必须回退到上一个推理节点，把"刀具磨损"这个假设排除掉，重新走另一条分支去查"来料批次差异"。这种带条件跳转和循环的流程，用 Chain 写会变成一堆 if-else 嵌套，维护起来是灾难。

LangGraph 的三个特性正好对症：

• 显式状态（State）：整个分析过程的中间结果都存在一个共享的 State 对象里，每个节点读取并更新它。这意味着"已排除的假设""已查询的数据源"这些信息不会丢，回退重查时不用从头再来。
• 条件边（Conditional Edge）：可以根据当前 State 动态决定下一步走哪个节点。比如"数据完整性检查"节点之后，如果关键数据缺失就走"人工补录"分支，否则走"自动推理"分支。
• 检查点（Checkpointer）：每个节点执行完都会持久化状态。分析跑到一半服务挂了，重启后能从断点继续，这对动辄要跑几分钟的多源查询流程太重要了。

我实测下来，用 LangGraph 重写之后，整个分析流程的代码从原来近两千行的状态机缩到了六百多行，而且加新分支的成本极低。

2.3 MCP 的角色：把数据接入从"写代码"变成"配工具"

在没有 MCP 之前，每接一个新数据源，我都要在代码里写一个专门的 client 类，处理连接、鉴权、查询语句、结果解析。接了五六个源之后，代码里全是各种 client，改一个公共逻辑要动好几处。

MCP 的思路是把每个数据源或能力封装成一个标准化的工具（Tool），对外暴露统一的调用接口和参数 schema。LangGraph 的节点不需要知道底层是 SQL 还是 REST API，它只需要知道"有一个叫query_spc_history的工具，输入是特性名和时间范围，输出是时序数据"。这样一来：

• 新增数据源 = 写一个 MCP Server + 在配置里注册，主流程代码零改动。
• 工具的参数 schema 自带描述，LLM 在需要调用工具时能自动理解该传什么参数。
• 权限和审计可以统一在 MCP 层做，哪个分析任务查了哪些数据都有记录。

下面这张表是我当时做选型对比时整理的，供参考：

2.4 整体架构分层

系统从上到下分四层，我画不出图（这里也不放图），用文字描述清楚：

• 交互层：告警接收接口 + 分析报告输出 + 人工介入入口。
• 编排层：LangGraph 图定义，包含异常确认、数据取证、根因推理、报告生成等节点。
• 工具层：一组 MCP Server，分别对接 SPC 时序库、MES 工艺参数、QMS 缺陷库、设备维保库、工艺知识库。
• 数据层：底层各类数据库和文件存储。

分层的好处是每层可以独立演进。比如后来我们把知识库从本地向量库换成了另一个检索服务，只动了工具层的一个 MCP Server，编排层完全没感觉。

3. 核心节点设计与状态管理实操

3.1 State 结构设计：一开始就要想清楚

LangGraph 的 State 是整个系统的"记忆"，设计得好后面省心，设计得差后面到处打补丁。我第一版 State 设计得太随意，把所有东西塞进一个 dict，结果节点之间互相覆盖字段，调试到怀疑人生。第二版我改成了带类型定义的 TypedDict，并且明确区分了三类字段：

from typing import TypedDict, Annotated, List from operator import add class AnalysisState(TypedDict): # 输入类：分析任务的基本信息 alarm_id: str characteristic: str # 监控的特性名，如"孔径尺寸" alarm_time: str # 过程类：分析过程中累积的信息 confirmed_anomaly: bool evidence: Annotated[List[dict], add] # 证据链，用 add 累加而非覆盖 excluded_hypotheses: List[str] # 已排除的假设 candidate_root_causes: List[dict] # 候选根因及置信度 # 控制类：决定流程走向 data_completeness: float need_human_review: bool iteration_count: int

这里有个关键点：evidence字段用了Annotated[List[dict], add]，意思是每个节点往里追加证据而不是覆盖。质量分析最忌讳的就是"后面的结论把前面的证据冲掉了"，用累加语义能保证证据链完整。而excluded_hypotheses用普通 List，因为它是被显式更新的。

提示：State 字段能少则少，但"证据"和"已排除假设"这两类一定要单独留字段。我见过有人把排除逻辑藏在 prompt 里让 LLM 自己记，结果多轮之后 LLM 就开始"失忆"，重复排查已经排除的方向。

3.2 异常确认节点：先别急着分析，先确认异常是真的

很多质量分析系统一上来就冲去查根因，这是大忌。SPC 告警里有相当一部分是测量系统问题或数据录入错误导致的假异常。所以第一个节点必须是异常确认。

这个节点的逻辑是：拉取告警点前后各 30 个点的原始数据，检查数据质量（有没有缺失、有没有明显超出量程的野值），然后调用一个 MCP 工具去查该测量设备的近期校准记录和 GR&R 结果。如果校准过期或者 GR&R 不达标，直接把confirmed_anomaly置为 False，流程走"误报归档"分支结束。

async def confirm_anomaly_node(state: AnalysisState) -> dict: history = await mcp_call("query_spc_history", characteristic=state["characteristic"], window=60) quality = assess_data_quality(history) gauge_ok = await mcp_call("check_gauge_status", characteristic=state["characteristic"]) if quality["missing_rate"] > 0.1 or not gauge_ok["valid"]: return {"confirmed_anomaly": False, "evidence": [{"type": "data_quality", "detail": quality}]} return {"confirmed_anomaly": True, "evidence": [{"type": "data_quality", "detail": quality}]}

实测下来，这一步能过滤掉大约 15% 到 20% 的无效告警，省下的分析算力相当可观。

3.3 数据取证节点：并行拉取多源数据

确认异常为真之后，进入取证阶段。这一步要同时查 MES 的工艺参数、QMS 的缺陷记录、设备维保日志、来料批次信息。这些查询彼此独立，用asyncio.gather并行发起，能把总耗时从串行的十几秒压到三四秒。

async def gather_evidence_node(state: AnalysisState) -> dict: tasks = [ mcp_call("query_mes_params", time=state["alarm_time"]), mcp_call("query_qms_defects", characteristic=state["characteristic"]), mcp_call("query_maintenance_log", time=state["alarm_time"]), mcp_call("query_material_batch", time=state["alarm_time"]), ] results = await asyncio.gather(*tasks, return_exceptions=True) evidence = [] for r in results: if isinstance(r, Exception): continue # 单个源失败不阻断整体流程 evidence.append(r) completeness = len(evidence) / len(tasks) return {"evidence": evidence, "data_completeness": completeness}

注意这里对异常的处理：单个数据源查询失败不阻断整个流程，而是记录到完整性指标里。如果完整性低于阈值（我设的是 0.6），后续推理节点会降低结论置信度并标记需要人工复核。这个设计是从一次真实事故里学来的——当时 MES 接口临时抽风，整个分析直接崩了，其实其他三个源的数据已经够做初步判断了。

3.4 根因推理节点：LLM 加规则双通道

这是整个系统最核心也最难调的部分。我的做法是规则引擎和 LLM 并行出结果，再融合。纯靠 LLM 推理，容易一本正经地胡说；纯靠规则，覆盖不了复杂场景。

规则通道负责处理那些有明确判据的场景，比如"连续 7 点上升 + 刀具使用时长接近寿命阈值 → 刀具磨损"，这类用 if-else 就能判，快且准。LLM 通道则把证据链整理成结构化上下文，让模型给出候选根因和推理过程，处理那些规则覆盖不到的、需要跨域联想的场景。

async def infer_root_cause_node(state: AnalysisState) -> dict: rule_causes = rule_engine(state["evidence"], state["characteristic"]) llm_causes = await llm_infer(state["evidence"], excluded=state["excluded_hypotheses"]) merged = merge_causes(rule_causes, llm_causes) # 过滤掉已排除的假设 merged = [c for c in merged if c["name"] not in state["excluded_hypotheses"]] return {"candidate_root_causes": merged, "iteration_count": state["iteration_count"] + 1}

融合逻辑上，规则命中的根因给一个置信度加成，LLM 给出的根因按模型自评置信度排序，两者取并集去重。这里有个经验：让 LLM 输出结构化的 JSON，字段固定为name、confidence、reasoning、supporting_evidence，不要让它自由发挥写段落，否则后面解析会疯。

3.5 验证与回退：根因链条的闭环

推理出候选根因后，不能直接出报告，要有一个验证节点。它会针对排名第一的根因，去查有没有反证。比如根因是"刀具磨损"，验证节点会去查维保记录里这把刀最近有没有更换、当前加工时长是多少。如果发现刀具刚换过，就把这个根因加入excluded_hypotheses，然后通过条件边回退到推理节点重新推理。

def route_after_verify(state: AnalysisState) -> str: if state["candidate_root_causes"]: return "generate_report" if state["iteration_count"] >= 3: return "human_review" # 防止无限循环 return "infer_root_cause" # 回退重推

iteration_count这个字段就是防死循环的保险丝。我设的上限是 3 轮，超过就转人工。实测中绝大多数案例 1 到 2 轮就能收敛，3 轮还没结论的，基本也确实需要人来判断了。

4. MCP 工具封装与多源数据接入实战

4.1 MCP Server 的最小实现骨架

一个 MCP Server 本质上就是暴露若干工具函数的服务。以 SPC 时序查询为例，核心是定义好工具的输入 schema 和返回结构。下面是一个简化骨架：

from mcp.server import Server from mcp.types import Tool, TextContent import json app = Server("spc-server") @app.list_tools() async def list_tools(): return [Tool( name="query_spc_history", description="查询指定特性的SPC历史时序数据", inputSchema={ "type": "object", "properties": { "characteristic": {"type": "string"}, "window": {"type": "integer", "default": 60} }, "required": ["characteristic"] } )] @app.call_tool() async def call_tool(name: str, arguments: dict): if name == "query_spc_history": data = await fetch_spc(arguments["characteristic"], arguments.get("window", 60)) return [TextContent(type="text", text=json.dumps(data, ensure_ascii=False))]

关键在description和inputSchema要写清楚。LLM 决定调不调这个工具、传什么参数，全靠这两处。我踩过的坑是 description 写得太简略，比如只写"查数据"，结果 LLM 经常传错参数或者该调的时候不调。后来我把 description 改成"查询指定质量特性的SPC历史时序数据，返回时间戳和测量值列表，用于判异和趋势分析"，调用准确率明显上来了。

4.2 五个核心 MCP 工具的设计要点

系统里我封装了五个核心工具，每个都有设计上的小心思：

这里重点说query_mes_params。MES 的工艺参数是秒级采集的，一个时间窗口能返回几万条。如果直接丢给 LLM，token 直接爆炸。我的做法是在 MCP Server 内部就做聚合——按分钟取均值、最大值、标准差，只返回统计特征。这样既保留了波动信息，又把数据量压到了几十条。聚合逻辑放在工具层而不是推理层，是因为不同数据源的聚合方式不同，放在工具层各管各的最清晰。

4.3 工具调用的错误处理与降级

MCP 工具调用失败是常态，网络抖动、数据库连接池满、查询超时都会发生。我的处理原则是分级降级：

• 查询超时（超过 5 秒）：重试一次，仍失败则返回空结果并标记该源不可用。
• 数据源整体不可用：跳过该源，降低data_completeness。
• 参数错误：这类是 bug，直接抛异常让流程中断，因为继续跑下去结果也是错的。

async def mcp_call(tool_name: str, retries: int = 1, **kwargs): for attempt in range(retries + 1): try: return await client.call_tool(tool_name, kwargs, timeout=5) except TimeoutError: if attempt == retries: log_warning(f"{tool_name} 超时，降级处理") return None except ParamError: raise # 参数错误不重试

注意：降级返回None和返回空列表是两回事。None表示"这个源没查到"，空列表表示"查到了但没数据"。推理层对这两种情况的处理不同，前者降低置信度，后者可能反而是有效证据（比如"维保记录为空"说明近期没维护，可能支持设备老化假设）。

4.4 知识库检索的 MCP 封装

工艺知识库这块单独说一下，因为它跟结构化数据源不太一样。我用的是向量检索，把工艺规范、历史分析报告、专家经验文档切片后建索引。MCP 工具search_process_knowledge接收查询文本，返回 top-k 相关片段，每个片段带上出处（哪份文档、哪一节）。

这里的关键是检索 query 的构造。直接把告警信息丢进去检索，效果一般。我的做法是在推理节点里，先用候选根因拼一个检索 query，比如"孔径尺寸 连续上升 刀具磨损"，再去检索，命中率高很多。检索回来的知识片段会作为额外证据注入 LLM 上下文，让推理有据可依。

5. 常见问题排查与避坑经验实录

5.1 LLM 推理不稳定怎么办

这是被问得最多的问题。同样的证据链，跑两次结论不一样，质量工程师根本不敢信。我的应对有三招：

第一，降低温度参数。推理节点用的模型 temperature 设到 0.1 甚至 0，牺牲一点创造性换稳定性。质量分析本来就不需要创造性。

第二，结构化输出强约束。前面提过，强制 JSON 输出，并且用 Pydantic 做校验，格式不对就重试。重试两次还不行就退回规则通道的结果。

第三，关键结论加规则兜底。对于置信度最高的根因，如果规则引擎也命中了同类根因，就提升置信度；如果规则引擎完全没命中，就在报告里标注"此结论由模型推理得出，建议人工复核"。这个标注机制让工程师知道哪些结论更可信。

5.2 多源数据时间对齐的坑

不同系统的时钟不一定同步，这是血泪教训。有一次分析结果完全跑偏，查了半天发现 MES 服务器的时间比 SPC 系统快了将近两分钟，导致工艺参数和测量值错位了。后来我在取证节点加了一个时间对齐预处理：以 SPC 告警时间为基准，其他源查询时窗口前后各放宽 5 分钟，然后在结果里按时间戳就近匹配。

def align_time(records, base_time, tolerance_sec=300): aligned = [] for r in records: delta = abs(r["ts"] - base_time) if delta <= tolerance_sec: aligned.append(r) return aligned

放宽窗口会引入一些无关数据，但比起错位导致的错误结论，这个代价可以接受。而且我在证据里保留了原始时间戳，人工复核时能看出来。

5.3 常见问题速查表

5.4 几个不那么显然的经验

经验一：别让 LLM 直接看原始时序数据。我一开始把几百个数据点直接塞进 prompt，结果模型要么数错点，要么被噪声带偏。后来改成在工具层算好统计特征（斜率、连续上升点数、是否超控制限），把特征喂给模型，准确率提升非常明显。LLM 擅长的是基于特征的推理，不是数值计算。

经验二：证据链要可追溯。报告里每一条结论都要能点回到具体证据。我在 State 的 evidence 里给每条证据加了唯一 id，报告生成时引用这些 id。这样工程师质疑某个结论时，能立刻定位到是哪条数据支撑的。这个功能上线后，工程师对系统的信任度明显提高。

经验三：人工介入不是失败。一开始我总想着让系统全自动，后来发现有些复杂案例就是需要人判断。现在我把"转人工"设计成一个正常分支，系统会把已收集的证据、已排除的假设、候选根因整理好交给工程师，工程师只需要做最后的判断。这比让系统硬猜一个错误结论强得多。

经验四：MCP 工具的粒度要适中。太细（比如每个字段一个工具）会导致调用次数爆炸，太粗（一个大工具查所有）又失去灵活性。我的经验是按"数据域"划分，一个域一个工具，比如 MES 一个、QMS 一个，这样粒度刚好。

6. 系统落地后的效果与后续扩展方向

这套系统在我们内部试运行了几个月，最直观的变化是异常响应时间。以前从告警到出初步分析报告，平均要 4 到 6 小时，现在系统自动跑完整个流程大概 2 到 3 分钟，工程师拿到报告后做复核和确认，整体压缩到半小时以内。更重要的是，证据链是自动收集和留痕的，不像以前靠人工记录，经常漏项。

从技术角度看，LangGraph 加 MCP 这个组合最大的价值是可扩展性。后来我们要接入一个新的环境监控数据源，只写了一个 MCP Server，在配置里注册，主流程一行没改。要加一个新的判异规则，也只需要在规则引擎里加一条，或者调整推理节点的 prompt。这种"加东西不改核心"的体验，是之前用自研状态机时想都不敢想的。

后续我打算往两个方向扩展。一是把分析结果反哺回知识库，每次人工确认过的根因和证据，自动沉淀成新的知识条目，让系统越用越聪明。二是做跨产线的根因关联，同一类异常在不同产线反复出现时，系统能识别出这是共性问题，而不是当成孤立事件处理。这两个方向都还在验证阶段，等跑通了再单独写一篇。

最后分享一个我在调试期用的小技巧：LangGraph 的检查点机制可以拿来"回放"。把某次分析的状态快照导出来，改改某个节点的逻辑，从中间某个节点重新跑，不用每次都从头来。这在调推理节点的时候特别省时间，强烈建议你也在开发环境里用起来。