1. 为什么我要用 LangGraph 和 MCP 重做质量分析系统在制造业做质量分析超过五年的朋友大概率都有同一个感受数据不是没有而是散落在太多地方。SPC 控制图跑在 Minitab 或者某个自研的看板里工艺参数躺在 MES 的时序数据库缺陷记录在 QMS设备维保日志又在另一个系统。每次产线出现异常质量工程师要做的第一件事不是分析而是找数据——导出、对齐时间戳、手动拼表一套流程下来半天就没了等真正开始做根因分析异常批次可能已经流到下游了。我这次要聊的项目就是拿LangGraph做流程编排、用MCPModel 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 层做哪个分析任务查了哪些数据都有记录。下面这张表是我当时做选型对比时整理的供参考维度传统自研 ClientLangChain ToolMCP 标准化工具新增数据源成本高需改主流程中需注册低独立 Server参数校验手写依赖 schema协议内置跨系统复用差一般好权限审计分散分散集中LLM 友好度低中高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 工具去查该测量设备的近期校准记录和 GRR 结果。如果校准过期或者 GRR 不达标直接把confirmed_anomaly置为 False流程走误报归档分支结束。async def confirm_anomaly_node(state: AnalysisState) - dict: history await mcp_call(query_spc_history, characteristicstate[characteristic], window60) quality assess_data_quality(history) gauge_ok await mcp_call(check_gauge_status, characteristicstate[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, timestate[alarm_time]), mcp_call(query_qms_defects, characteristicstate[characteristic]), mcp_call(query_maintenance_log, timestate[alarm_time]), mcp_call(query_material_batch, timestate[alarm_time]), ] results await asyncio.gather(*tasks, return_exceptionsTrue) 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], excludedstate[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( namequery_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(typetext, textjson.dumps(data, ensure_asciiFalse))]关键在description和inputSchema要写清楚。LLM 决定调不调这个工具、传什么参数全靠这两处。我踩过的坑是 description 写得太简略比如只写查数据结果 LLM 经常传错参数或者该调的时候不调。后来我把 description 改成查询指定质量特性的SPC历史时序数据返回时间戳和测量值列表用于判异和趋势分析调用准确率明显上来了。4.2 五个核心 MCP 工具的设计要点系统里我封装了五个核心工具每个都有设计上的小心思工具名数据源关键设计点query_spc_historySPC 时序库返回原始点控制限让推理层自己判异query_mes_paramsMES按时间窗口聚合避免返回海量原始点query_qms_defectsQMS支持按特性名模糊匹配兼容命名不一致query_maintenance_log维保系统返回结构化事件含更换件和操作人search_process_knowledge知识库向量检索返回带出处的知识片段这里重点说query_mes_params。MES 的工艺参数是秒级采集的一个时间窗口能返回几万条。如果直接丢给 LLMtoken 直接爆炸。我的做法是在 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, timeout5) 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_sec300): aligned [] for r in records: delta abs(r[ts] - base_time) if delta tolerance_sec: aligned.append(r) return aligned放宽窗口会引入一些无关数据但比起错位导致的错误结论这个代价可以接受。而且我在证据里保留了原始时间戳人工复核时能看出来。5.3 常见问题速查表现象可能原因排查方向解决分析结果每次不同LLM 温度过高检查推理节点参数温度降到 0.1根因明显不对时间未对齐对比各源时间戳加时间对齐预处理流程卡住不结束回退死循环看 iteration_count设上限转人工工具调用报参数错schema 描述不清看 MCP 工具定义完善 description报告缺证据某源查询失败看 data_completeness检查该源连接token 超限原始数据未聚合看工具返回量工具层做聚合5.4 几个不那么显然的经验经验一别让 LLM 直接看原始时序数据。我一开始把几百个数据点直接塞进 prompt结果模型要么数错点要么被噪声带偏。后来改成在工具层算好统计特征斜率、连续上升点数、是否超控制限把特征喂给模型准确率提升非常明显。LLM 擅长的是基于特征的推理不是数值计算。经验二证据链要可追溯。报告里每一条结论都要能点回到具体证据。我在 State 的 evidence 里给每条证据加了唯一 id报告生成时引用这些 id。这样工程师质疑某个结论时能立刻定位到是哪条数据支撑的。这个功能上线后工程师对系统的信任度明显提高。经验三人工介入不是失败。一开始我总想着让系统全自动后来发现有些复杂案例就是需要人判断。现在我把转人工设计成一个正常分支系统会把已收集的证据、已排除的假设、候选根因整理好交给工程师工程师只需要做最后的判断。这比让系统硬猜一个错误结论强得多。经验四MCP 工具的粒度要适中。太细比如每个字段一个工具会导致调用次数爆炸太粗一个大工具查所有又失去灵活性。我的经验是按数据域划分一个域一个工具比如 MES 一个、QMS 一个这样粒度刚好。6. 系统落地后的效果与后续扩展方向这套系统在我们内部试运行了几个月最直观的变化是异常响应时间。以前从告警到出初步分析报告平均要 4 到 6 小时现在系统自动跑完整个流程大概 2 到 3 分钟工程师拿到报告后做复核和确认整体压缩到半小时以内。更重要的是证据链是自动收集和留痕的不像以前靠人工记录经常漏项。从技术角度看LangGraph 加 MCP 这个组合最大的价值是可扩展性。后来我们要接入一个新的环境监控数据源只写了一个 MCP Server在配置里注册主流程一行没改。要加一个新的判异规则也只需要在规则引擎里加一条或者调整推理节点的 prompt。这种加东西不改核心的体验是之前用自研状态机时想都不敢想的。后续我打算往两个方向扩展。一是把分析结果反哺回知识库每次人工确认过的根因和证据自动沉淀成新的知识条目让系统越用越聪明。二是做跨产线的根因关联同一类异常在不同产线反复出现时系统能识别出这是共性问题而不是当成孤立事件处理。这两个方向都还在验证阶段等跑通了再单独写一篇。最后分享一个我在调试期用的小技巧LangGraph 的检查点机制可以拿来回放。把某次分析的状态快照导出来改改某个节点的逻辑从中间某个节点重新跑不用每次都从头来。这在调推理节点的时候特别省时间强烈建议你也在开发环境里用起来。
