1. 从一张失控的SPC控制图说起做过制造业质量分析的人大概都经历过这样的场景产线SPC控制图上某个关键尺寸的Cpk突然掉到0.8以下报警邮件叮叮当当响了一整天你打开Minitab或者自己写的Python脚本开始一层层往下钻——先看是哪条产线再看是哪台设备接着查是哪个班次最后定位到某个批次的原材料。整个过程快则两小时慢则一整天等根因报告出来不良品已经堆了半个仓库。这个链条之所以慢核心问题不在于数据不够而在于数据之间的语义断层。SPC系统知道X-bar图第37个点超出UCLMES系统知道这批活是3号机台B班做的ERP系统知道这批钢材来自供应商A的6月批次但没有任何一个系统能自动把这三条信息串成一句人话3号机台B班使用供应商A的6月批次钢材时刀具磨损速率异常导致尺寸漂移。我最近用LangGraph配合MCP协议搭了一套AI-SPC智能分析系统目标就是打通这条根因链条。它不是简单地把LLM接上数据库做问答而是用图状态机来编排整个分析推理流程用MCP工具协议来标准化每个数据源的接入方式让AI像一个有经验的质量工程师那样按步骤、有逻辑地完成根因定位。这套系统适合谁参考如果你正在做工业数据分析、质量管理系统、或者任何需要多源数据推理链条的AI应用这里面的架构思路和踩坑经验应该对你有用。如果你只是想了解LangGraph和MCP怎么配合我也会把关键代码和配置逻辑讲清楚。先说结论这套系统把根因分析的平均耗时从小时级压到了分钟级但过程中踩的坑比想象中多得多——尤其是状态管理和工具调用的边界问题后面会详细展开。2. 为什么是LangGraph加MCP而不是一条Chain走到底2.1 传统Chain式分析的死胡同最开始我试过用最朴素的方式一个LLM加几个工具函数让模型自己决定调什么。结果很快就崩了。质量根因分析不是查一个数那么简单它是一条有向推理链先确认异常点再判断异常模式是突变还是渐变然后根据模式选择不同的排查路径——突变往往指向设备故障或换料渐变往往指向刀具磨损或温漂。这两条路径要查的数据源、要做的统计检验完全不同。用LangChain的SequentialChain硬编码这条路径问题在于分支太多。光异常模式就有七八种每种模式下的排查路径又有三四个分叉写出来就是一棵巨大的决策树维护成本极高。而且实际分析中经常出现查到一半发现假设不成立需要回退的情况线性Chain根本处理不了。LangGraph的价值就在这里。它把分析流程建模成状态图每个节点是一个分析步骤边是条件跳转整个图共享一个状态对象。回退、循环、并行分支都是原生支持的。比如检查设备参数这个节点发现数据缺失可以直接跳回补充数据采集节点而不是抛异常终止。2.2 MCP解决的最后一公里问题有了LangGraph编排流程下一个问题是每个分析节点怎么拿到数据传统做法是给每个数据源写一个Python函数然后在节点里调用。这在小规模下没问题但SPC分析涉及的数据源太多了——实时数据库、历史库、MES、ERP、LIMS、甚至Excel台账。每接一个新数据源就要改代码、重新部署而且不同数据源的认证方式、查询语法、返回格式都不一样。MCPModel Context Protocol的思路是把每个数据源封装成一个标准化的工具服务对外暴露统一的接口描述。LangGraph的节点不需要知道底层是SQL还是REST API只需要知道有一个叫query_spc_data的工具接受产线ID和时间范围参数。这样新增数据源就是注册一个新的MCP Server主流程代码一行不用改。更重要的是MCP的工具描述是机器可读的。LLM可以根据工具的自然语言描述来判断该调哪个工具、传什么参数。这意味着我可以在LangGraph里加一个动态工具选择节点让模型根据当前分析状态自主决定下一步查什么而不是把所有分支都硬编码。2.3 两者配合的架构全景整个系统的分层是这样的层级组件职责编排层LangGraph StateGraph定义分析流程、状态流转、条件分支推理层LLM本地部署或API模式识别、假设生成、根因推理工具层MCP Client标准化工具调用、参数校验、结果解析服务层多个MCP Server封装SPC库、MES、ERP等数据源数据层各类数据库/API实际数据存储这个架构的关键设计决策是LLM只做推理不做数据搬运。所有数据查询通过MCP工具完成LLM看到的是结构化后的摘要信息而不是原始数据流。这既控制了token消耗也避免了模型被大量原始数据干扰判断。提示如果你的数据源少于3个其实不一定需要MCP直接写工具函数更简单。MCP的价值在数据源多、变化频繁的场景下才体现得明显。3. 状态图设计把质量工程师的思维过程画出来3.1 状态对象的字段设计LangGraph的核心是状态对象它贯穿整个图的执行过程。设计状态字段时我参考了一个资深质量工程师做根因分析时的工作记忆——他脑子里同时记着哪些信息from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, END import operator class SPCState(TypedDict): # 输入信息 alert_id: str # 报警ID product_line: str # 产线标识 characteristic: str # 监控特性如外径尺寸 alert_time: str # 报警时间 # 分析过程状态 anomaly_points: list[dict] # 异常点列表 anomaly_pattern: str # 异常模式sudden/drift/cycle/outlier current_hypothesis: str # 当前假设 hypothesis_stack: list[str] # 假设栈支持回退 evidence: Annotated[list[dict], operator.add] # 证据链累积 # 排查范围 suspect_equipment: list[str] # 可疑设备 suspect_material: list[str] # 可疑物料批次 suspect_shift: list[str] # 可疑班次 # 输出 root_cause: str # 最终根因 confidence: float # 置信度 report: str # 分析报告这里有几个设计细节值得展开。evidence字段用了operator.add作为reducer意味着每次节点往里面追加证据时是累加而不是覆盖这样整个分析过程的证据链是完整保留的。hypothesis_stack用栈结构是为了支持假设不成立时回退到上一个假设——这是实际分析中非常常见的操作线性流程做不到。3.2 节点划分与条件边整个图我划分了七个核心节点每个节点对应质量分析的一个标准步骤def build_analysis_graph(): graph StateGraph(SPCState) graph.add_node(fetch_anomaly, fetch_anomaly_data) graph.add_node(classify_pattern, classify_anomaly_pattern) graph.add_node(generate_hypothesis, generate_hypotheses) graph.add_node(investigate, investigate_hypothesis) graph.add_node(verify, verify_root_cause) graph.add_node(generate_report, generate_report) graph.add_node(request_more_data, request_additional_data) graph.set_entry_point(fetch_anomaly) graph.add_edge(fetch_anomaly, classify_pattern) graph.add_edge(classify_pattern, generate_hypothesis) graph.add_edge(generate_hypothesis, investigate) # 关键的条件边调查后决定是继续验证还是换假设 graph.add_conditional_edges( investigate, route_after_investigation, { verify: verify, next_hypothesis: generate_hypothesis, need_data: request_more_data, report: generate_report } ) graph.add_conditional_edges( verify, route_after_verification, { confirmed: generate_report, rejected: generate_hypothesis, inconclusive: request_more_data } ) graph.add_edge(request_more_data, investigate) graph.add_edge(generate_report, END) return graph.compile()route_after_investigation这个路由函数是整个系统的大脑它根据当前状态决定下一步走向。实现上它不调用LLM而是用规则判断——因为路由决策需要确定性不能让模型随机发挥。比如def route_after_investigation(state: SPCState) - str: # 证据充分且假设被验证 - 进入验证 if len(state[evidence]) 3 and state[current_hypothesis]: return verify # 还有未排查的假设 - 换下一个 if state[hypothesis_stack]: return next_hypothesis # 数据不足 - 请求补充 if not state[evidence]: return need_data # 兜底直接出报告 return report3.3 循环与回退的边界控制LangGraph支持循环但循环必须有终止条件否则会无限跑下去烧token。我设了三重保险第一最大迭代次数。在状态里加一个iteration_count字段每次进入investigate节点就加一超过10次强制跳转到报告生成。第二假设栈深度限制。假设栈最多存5个假设超过就丢弃最早的防止栈无限增长。第三证据去重。每次追加证据前检查是否已存在相同来源和内容的证据避免同一个查询反复执行。注意LangGraph的循环如果没有显式终止条件在LLM调用失败时会陷入死循环。我踩过这个坑跑了200多次调用才被手动停掉。一定要在路由函数里加硬性上限。4. MCP工具层的落地细节4.1 把SPC数据库封装成MCP ServerMCP Server的本质是一个遵循特定协议的服务进程它对外声明自己有哪些工具、每个工具接受什么参数、返回什么格式。我用Python的mcp库来实现核心是定义一个工具注册表from mcp.server import Server from mcp.types import Tool, TextContent import json app Server(spc-analysis-server) app.list_tools() async def list_tools() - list[Tool]: return [ Tool( namequery_spc_data, description查询指定产线和特性在时间范围内的SPC数据点返回时间序列和规格限, inputSchema{ type: object, properties: { product_line: {type: string, description: 产线标识}, characteristic: {type: string, description: 监控特性名称}, start_time: {type: string, description: 开始时间 ISO8601}, end_time: {type: string, description: 结束时间 ISO8601} }, required: [product_line, characteristic, start_time, end_time] } ), Tool( namequery_equipment_log, description查询设备运行日志包括参数设置、报警记录、维护记录, inputSchema{ type: object, properties: { equipment_id: {type: string}, start_time: {type: string}, end_time: {type: string} }, required: [equipment_id, start_time, end_time] } ), Tool( namequery_material_batch, description查询物料批次信息包括供应商、入库时间、检验结果, inputSchema{ type: object, properties: { batch_no: {type: string}, material_code: {type: string} }, required: [batch_no] } ) ]工具描述的文字很关键。LLM就是靠这段描述来判断该不该调这个工具的。我一开始写得太简略比如查询SPC数据结果模型经常在需要查设备日志的时候错误地调了SPC查询。后来把描述写详细明确说明返回时间序列和规格限模型的选择准确率明显提升。4.2 工具返回结果的压缩策略MCP工具返回的原始数据可能很大——一个SPC查询返回几百个数据点全塞给LLM既浪费token又干扰判断。我的做法是在MCP Server端做预聚合app.call_tool() async def call_tool(name: str, arguments: dict) - list[TextContent]: if name query_spc_data: raw_data await fetch_from_database(arguments) # 预聚合不返回原始点返回统计摘要异常点 summary { total_points: len(raw_data), mean: statistics.mean([d[value] for d in raw_data]), std: statistics.stdev([d[value] for d in raw_data]), ucl: raw_data[0][ucl], lcl: raw_data[0][lcl], violations: [ {index: i, value: d[value], rule: beyond_ucl} for i, d in enumerate(raw_data) if d[value] d[ucl] or d[value] d[lcl] ], trend: detect_trend(raw_data) # 返回上升/下降/平稳 } return [TextContent(typetext, textjson.dumps(summary, ensure_asciiFalse))]这样LLM看到的是200个数据点均值25.3标准差0.12UCL25.8LCL24.8有3个点超出UCL整体趋势上升——信息密度高模型容易推理。4.3 多Server的注册与路由实际系统里有四个MCP ServerSPC数据服务、设备日志服务、物料批次服务、工艺参数服务。LangGraph这边用一个MCP Client统一管理class MCPToolkit: def __init__(self): self.servers {} self.tool_index {} # tool_name - server_name async def register_server(self, name: str, command: list[str]): client MCPClient(command) await client.connect() self.servers[name] client tools await client.list_tools() for tool in tools: self.tool_index[tool.name] name async def call_tool(self, tool_name: str, arguments: dict): server_name self.tool_index.get(tool_name) if not server_name: raise ValueError(f未知工具: {tool_name}) return await self.servers[server_name].call_tool(tool_name, arguments)这个Toolkit在LangGraph的节点里被调用。节点函数拿到LLM决定的工具名和参数后通过Toolkit路由到对应的MCP Server执行。提示MCP Server的启动方式建议用子进程管理不要和主进程混在一起。我一开始图省事直接在同一个进程里跑结果某个Server的数据库连接池满了把主进程也拖死了。后来改成独立子进程通过stdio通信稳定性好很多。5. 推理节点的Prompt工程与输出约束5.1 模式识别节点的结构化输出classify_pattern节点负责判断异常属于哪种模式。这个节点的输出直接决定后续走哪条排查路径所以必须结构化、可校验。我用Pydantic定义输出schema然后让LLM按schema输出JSONfrom pydantic import BaseModel, Field from typing import Literal class PatternClassification(BaseModel): pattern: Literal[sudden_shift, gradual_drift, cyclic, single_outlier, increasing_variance] confidence: float Field(ge0, le1) reasoning: str Field(max_length200) key_evidence: list[str] Field(max_length5) PATTERN_PROMPT 你是一个SPC分析专家。根据以下异常数据摘要判断异常模式。 数据摘要 {data_summary} 异常模式定义 - sudden_shift: 均值突然跳变通常由换料、设备参数变更引起 - gradual_drift: 均值缓慢单向移动通常由刀具磨损、温漂引起 - cyclic: 呈现周期性波动通常由环境周期或操作周期引起 - single_outlier: 孤立异常点通常由测量误差或瞬时干扰引起 - increasing_variance: 方差增大但均值稳定通常由设备松动或来料不均引起 请严格按JSON格式输出不要添加任何其他文字。这里的关键是用定义约束模型。如果不给模式定义模型会自己发明分类标准不同批次的分析结果不可比。给了明确定义后分类一致性大幅提升。5.2 假设生成节点的发散与收敛generate_hypothesis节点需要根据异常模式生成排查假设。这个节点我用了两阶段策略先发散生成多个候选假设再收敛排序。HYPOTHESIS_PROMPT 基于以下信息生成3-5个可能的根因假设按可能性从高到低排序。 异常模式{pattern} 异常特征{characteristic} 产线{product_line} 已有证据{evidence} 每个假设需包含 1. 假设描述一句话 2. 需要验证的数据源 3. 预期验证结果 输出JSON数组格式。生成后路由函数会把假设压入hypothesis_stack然后investigate节点逐个验证。验证一个、排除一个、再验证下一个——这就是模拟质量工程师逐个排除的思维过程。5.3 证据链的累积与引用每次investigate节点执行后会把查询结果和推理结论追加到evidence列表。这个列表在生成报告时会被完整引用形成可追溯的证据链。证据的结构{ step: 3, hypothesis: 3号机台刀具磨损导致尺寸漂移, tool_called: query_equipment_log, tool_args: {equipment_id: EQ-003, start_time: ..., end_time: ...}, finding: 刀具更换记录显示上次更换在72小时前超过标准更换周期48小时, supports_hypothesis: True, confidence_delta: 0.3 }这个结构让最终报告可以精确地说第3步查询了设备日志发现刀具超期使用支持了假设置信度提升0.3——完全可审计。6. 实测中暴露的三个典型问题6.1 工具调用参数幻觉系统跑通后遇到的第一个问题是LLM经常编造不存在的设备ID或物料批次号。比如它会在参数里填EQ-999而实际设备只到EQ-010。MCP Server收到查询请求后返回空结果模型却把这个空结果当成设备无异常记录得出错误结论。修复方案是在MCP Server端加参数校验层查询前先验证设备ID是否存在不存在直接返回错误信息而不是空结果。错误信息里明确写设备EQ-999不存在可用设备列表EQ-001至EQ-010。模型收到这个错误后会重新选择正确的ID。async def query_equipment_log(equipment_id: str, start_time: str, end_time: str): valid_ids await get_valid_equipment_ids() if equipment_id not in valid_ids: return { error: f设备{equipment_id}不存在, available_ids: valid_ids[:20], # 只返回前20个避免过长 suggestion: 请从可用设备列表中选择 } # ... 正常查询逻辑6.2 状态字段的并发写入冲突LangGraph在并行执行分支时多个节点可能同时往evidence列表里写数据。虽然用了operator.add作为reducer但在高并发下仍然出现了证据顺序错乱的问题——第5步的证据排在了第3步前面。解决办法是给每条证据加时间戳和步骤号在生成报告前按步骤号排序。另外把并行分支改成串行执行——质量分析本来就是顺序推理的过程并行反而破坏了逻辑链条。6.3 LLM在长链条中的遗忘当分析链条超过6-7步后模型开始忘记早期的证据。比如第2步已经确认了异常模式是渐变漂移到第8步生成报告时却写成了突变。这是因为状态对象里的信息太多超出了模型的注意力窗口。我的应对策略是在每个节点的Prompt里显式注入关键状态摘要而不是让模型自己去状态里翻。具体做法是写一个summarize_state函数把当前最相关的5条信息压缩成一段话放在Prompt开头def summarize_state(state: SPCState) - str: parts [ f异常模式{state[anomaly_pattern]}, f当前假设{state[current_hypothesis]}, f已排除假设{len(state[hypothesis_stack])}个, f关键证据{state[evidence][-3:] if state[evidence] else 暂无}, f排查进度第{state.get(iteration_count, 0)}轮 ] return \n.join(parts)这个摘要放在每个Prompt的最前面确保模型始终记得核心上下文。7. 从报警到根因报告的完整链路演示7.1 触发与初始数据拉取假设产线PL-01的外径尺寸特性在14:30触发报警。系统入口接收报警IDfetch_anomaly节点调用query_spc_data工具拉取最近4小时的数据。MCP Server返回预聚合摘要200个点均值25.32标准差0.15UCL25.80LCL24.80有5个点超出UCL趋势为上升。7.2 模式判定与假设生成classify_pattern节点收到摘要后LLM判断为gradual_drift渐变漂移置信度0.85理由均值从25.1缓慢上升至25.6无突变点。generate_hypothesis节点基于渐变漂移模式生成三个假设刀具磨损可能性最高、温漂、来料批次差异。7.3 逐层排查与证据累积investigate节点先验证刀具磨损假设调用query_equipment_log查询3号机台日志。返回显示刀具已使用72小时超过标准更换周期48小时。证据追加置信度提升至0.7。接着验证温漂假设调用query_process_param查询温度记录显示温度稳定在22±0.5°C排除温漂。最后验证来料假设调用query_material_batch查询当前批次检验结果正常排除。7.4 根因确认与报告生成verify节点综合三条证据确认根因为3号机台刀具超期使用导致尺寸渐变漂移置信度0.92。generate_report节点生成结构化报告包含异常描述、分析过程、证据链、根因结论、建议措施立即更换刀具、调整更换周期至48小时、加强刀具寿命监控。整个流程从触发到报告生成实测平均耗时2分15秒其中LLM调用占约80秒MCP工具调用占约40秒其余为状态流转开销。8. 部署与扩展的几个实操建议8.1 本地模型与API的取舍如果数据敏感度高建议用本地部署的开源模型如Qwen系列。但要注意本地模型在结构化输出JSON格式上的稳定性不如大参数API模型。我的做法是在本地模型外面套一层输出解析器用正则提取JSON解析失败时自动重试并降低temperature。8.2 MCP Server的版本管理MCP工具的描述和参数schema会随着业务变化而调整。建议给每个Server加版本号LangGraph这边记录使用的版本。当Server升级时如果schema有breaking change主流程需要同步更新Prompt里的工具描述。我吃过这个亏——Server端加了一个必填参数但Prompt里没更新导致模型调用一直失败。8.3 分析结果的反馈闭环系统上线后我加了一个反馈机制质量工程师可以标记AI给出的根因是否正确。这些标记数据用来微调模式识别和假设生成的Prompt。跑了两个月后根因定位准确率从最初的68%提升到了87%。关键改进点是把工程师常用的排查经验比如先查换料记录再查设备日志写进了Prompt的few-shot示例里。这套系统目前还在持续迭代下一步计划是把SPC分析扩展到多特性联合分析——比如外径和圆度同时异常时根因可能指向夹具问题而不是刀具。那又是另一个故事了。
1. 从一张失控的SPC控制图说起做过制造业质量分析的人大概都经历过这样的场景产线SPC控制图上某个关键尺寸的Cpk突然掉到0.8以下报警邮件叮叮当当响了一整天你打开Minitab或者自己写的Python脚本开始一层层往下钻——先看是哪条产线再看是哪台设备接着查是哪个班次最后定位到某个批次的原材料。整个过程快则两小时慢则一整天等根因报告出来不良品已经堆了半个仓库。这个链条之所以慢核心问题不在于数据不够而在于数据之间的语义断层。SPC系统知道X-bar图第37个点超出UCLMES系统知道这批活是3号机台B班做的ERP系统知道这批钢材来自供应商A的6月批次但没有任何一个系统能自动把这三条信息串成一句人话3号机台B班使用供应商A的6月批次钢材时刀具磨损速率异常导致尺寸漂移。我最近用LangGraph配合MCP协议搭了一套AI-SPC智能分析系统目标就是打通这条根因链条。它不是简单地把LLM接上数据库做问答而是用图状态机来编排整个分析推理流程用MCP工具协议来标准化每个数据源的接入方式让AI像一个有经验的质量工程师那样按步骤、有逻辑地完成根因定位。这套系统适合谁参考如果你正在做工业数据分析、质量管理系统、或者任何需要多源数据推理链条的AI应用这里面的架构思路和踩坑经验应该对你有用。如果你只是想了解LangGraph和MCP怎么配合我也会把关键代码和配置逻辑讲清楚。先说结论这套系统把根因分析的平均耗时从小时级压到了分钟级但过程中踩的坑比想象中多得多——尤其是状态管理和工具调用的边界问题后面会详细展开。2. 为什么是LangGraph加MCP而不是一条Chain走到底2.1 传统Chain式分析的死胡同最开始我试过用最朴素的方式一个LLM加几个工具函数让模型自己决定调什么。结果很快就崩了。质量根因分析不是查一个数那么简单它是一条有向推理链先确认异常点再判断异常模式是突变还是渐变然后根据模式选择不同的排查路径——突变往往指向设备故障或换料渐变往往指向刀具磨损或温漂。这两条路径要查的数据源、要做的统计检验完全不同。用LangChain的SequentialChain硬编码这条路径问题在于分支太多。光异常模式就有七八种每种模式下的排查路径又有三四个分叉写出来就是一棵巨大的决策树维护成本极高。而且实际分析中经常出现查到一半发现假设不成立需要回退的情况线性Chain根本处理不了。LangGraph的价值就在这里。它把分析流程建模成状态图每个节点是一个分析步骤边是条件跳转整个图共享一个状态对象。回退、循环、并行分支都是原生支持的。比如检查设备参数这个节点发现数据缺失可以直接跳回补充数据采集节点而不是抛异常终止。2.2 MCP解决的最后一公里问题有了LangGraph编排流程下一个问题是每个分析节点怎么拿到数据传统做法是给每个数据源写一个Python函数然后在节点里调用。这在小规模下没问题但SPC分析涉及的数据源太多了——实时数据库、历史库、MES、ERP、LIMS、甚至Excel台账。每接一个新数据源就要改代码、重新部署而且不同数据源的认证方式、查询语法、返回格式都不一样。MCPModel Context Protocol的思路是把每个数据源封装成一个标准化的工具服务对外暴露统一的接口描述。LangGraph的节点不需要知道底层是SQL还是REST API只需要知道有一个叫query_spc_data的工具接受产线ID和时间范围参数。这样新增数据源就是注册一个新的MCP Server主流程代码一行不用改。更重要的是MCP的工具描述是机器可读的。LLM可以根据工具的自然语言描述来判断该调哪个工具、传什么参数。这意味着我可以在LangGraph里加一个动态工具选择节点让模型根据当前分析状态自主决定下一步查什么而不是把所有分支都硬编码。2.3 两者配合的架构全景整个系统的分层是这样的层级组件职责编排层LangGraph StateGraph定义分析流程、状态流转、条件分支推理层LLM本地部署或API模式识别、假设生成、根因推理工具层MCP Client标准化工具调用、参数校验、结果解析服务层多个MCP Server封装SPC库、MES、ERP等数据源数据层各类数据库/API实际数据存储这个架构的关键设计决策是LLM只做推理不做数据搬运。所有数据查询通过MCP工具完成LLM看到的是结构化后的摘要信息而不是原始数据流。这既控制了token消耗也避免了模型被大量原始数据干扰判断。提示如果你的数据源少于3个其实不一定需要MCP直接写工具函数更简单。MCP的价值在数据源多、变化频繁的场景下才体现得明显。3. 状态图设计把质量工程师的思维过程画出来3.1 状态对象的字段设计LangGraph的核心是状态对象它贯穿整个图的执行过程。设计状态字段时我参考了一个资深质量工程师做根因分析时的工作记忆——他脑子里同时记着哪些信息from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, END import operator class SPCState(TypedDict): # 输入信息 alert_id: str # 报警ID product_line: str # 产线标识 characteristic: str # 监控特性如外径尺寸 alert_time: str # 报警时间 # 分析过程状态 anomaly_points: list[dict] # 异常点列表 anomaly_pattern: str # 异常模式sudden/drift/cycle/outlier current_hypothesis: str # 当前假设 hypothesis_stack: list[str] # 假设栈支持回退 evidence: Annotated[list[dict], operator.add] # 证据链累积 # 排查范围 suspect_equipment: list[str] # 可疑设备 suspect_material: list[str] # 可疑物料批次 suspect_shift: list[str] # 可疑班次 # 输出 root_cause: str # 最终根因 confidence: float # 置信度 report: str # 分析报告这里有几个设计细节值得展开。evidence字段用了operator.add作为reducer意味着每次节点往里面追加证据时是累加而不是覆盖这样整个分析过程的证据链是完整保留的。hypothesis_stack用栈结构是为了支持假设不成立时回退到上一个假设——这是实际分析中非常常见的操作线性流程做不到。3.2 节点划分与条件边整个图我划分了七个核心节点每个节点对应质量分析的一个标准步骤def build_analysis_graph(): graph StateGraph(SPCState) graph.add_node(fetch_anomaly, fetch_anomaly_data) graph.add_node(classify_pattern, classify_anomaly_pattern) graph.add_node(generate_hypothesis, generate_hypotheses) graph.add_node(investigate, investigate_hypothesis) graph.add_node(verify, verify_root_cause) graph.add_node(generate_report, generate_report) graph.add_node(request_more_data, request_additional_data) graph.set_entry_point(fetch_anomaly) graph.add_edge(fetch_anomaly, classify_pattern) graph.add_edge(classify_pattern, generate_hypothesis) graph.add_edge(generate_hypothesis, investigate) # 关键的条件边调查后决定是继续验证还是换假设 graph.add_conditional_edges( investigate, route_after_investigation, { verify: verify, next_hypothesis: generate_hypothesis, need_data: request_more_data, report: generate_report } ) graph.add_conditional_edges( verify, route_after_verification, { confirmed: generate_report, rejected: generate_hypothesis, inconclusive: request_more_data } ) graph.add_edge(request_more_data, investigate) graph.add_edge(generate_report, END) return graph.compile()route_after_investigation这个路由函数是整个系统的大脑它根据当前状态决定下一步走向。实现上它不调用LLM而是用规则判断——因为路由决策需要确定性不能让模型随机发挥。比如def route_after_investigation(state: SPCState) - str: # 证据充分且假设被验证 - 进入验证 if len(state[evidence]) 3 and state[current_hypothesis]: return verify # 还有未排查的假设 - 换下一个 if state[hypothesis_stack]: return next_hypothesis # 数据不足 - 请求补充 if not state[evidence]: return need_data # 兜底直接出报告 return report3.3 循环与回退的边界控制LangGraph支持循环但循环必须有终止条件否则会无限跑下去烧token。我设了三重保险第一最大迭代次数。在状态里加一个iteration_count字段每次进入investigate节点就加一超过10次强制跳转到报告生成。第二假设栈深度限制。假设栈最多存5个假设超过就丢弃最早的防止栈无限增长。第三证据去重。每次追加证据前检查是否已存在相同来源和内容的证据避免同一个查询反复执行。注意LangGraph的循环如果没有显式终止条件在LLM调用失败时会陷入死循环。我踩过这个坑跑了200多次调用才被手动停掉。一定要在路由函数里加硬性上限。4. MCP工具层的落地细节4.1 把SPC数据库封装成MCP ServerMCP Server的本质是一个遵循特定协议的服务进程它对外声明自己有哪些工具、每个工具接受什么参数、返回什么格式。我用Python的mcp库来实现核心是定义一个工具注册表from mcp.server import Server from mcp.types import Tool, TextContent import json app Server(spc-analysis-server) app.list_tools() async def list_tools() - list[Tool]: return [ Tool( namequery_spc_data, description查询指定产线和特性在时间范围内的SPC数据点返回时间序列和规格限, inputSchema{ type: object, properties: { product_line: {type: string, description: 产线标识}, characteristic: {type: string, description: 监控特性名称}, start_time: {type: string, description: 开始时间 ISO8601}, end_time: {type: string, description: 结束时间 ISO8601} }, required: [product_line, characteristic, start_time, end_time] } ), Tool( namequery_equipment_log, description查询设备运行日志包括参数设置、报警记录、维护记录, inputSchema{ type: object, properties: { equipment_id: {type: string}, start_time: {type: string}, end_time: {type: string} }, required: [equipment_id, start_time, end_time] } ), Tool( namequery_material_batch, description查询物料批次信息包括供应商、入库时间、检验结果, inputSchema{ type: object, properties: { batch_no: {type: string}, material_code: {type: string} }, required: [batch_no] } ) ]工具描述的文字很关键。LLM就是靠这段描述来判断该不该调这个工具的。我一开始写得太简略比如查询SPC数据结果模型经常在需要查设备日志的时候错误地调了SPC查询。后来把描述写详细明确说明返回时间序列和规格限模型的选择准确率明显提升。4.2 工具返回结果的压缩策略MCP工具返回的原始数据可能很大——一个SPC查询返回几百个数据点全塞给LLM既浪费token又干扰判断。我的做法是在MCP Server端做预聚合app.call_tool() async def call_tool(name: str, arguments: dict) - list[TextContent]: if name query_spc_data: raw_data await fetch_from_database(arguments) # 预聚合不返回原始点返回统计摘要异常点 summary { total_points: len(raw_data), mean: statistics.mean([d[value] for d in raw_data]), std: statistics.stdev([d[value] for d in raw_data]), ucl: raw_data[0][ucl], lcl: raw_data[0][lcl], violations: [ {index: i, value: d[value], rule: beyond_ucl} for i, d in enumerate(raw_data) if d[value] d[ucl] or d[value] d[lcl] ], trend: detect_trend(raw_data) # 返回上升/下降/平稳 } return [TextContent(typetext, textjson.dumps(summary, ensure_asciiFalse))]这样LLM看到的是200个数据点均值25.3标准差0.12UCL25.8LCL24.8有3个点超出UCL整体趋势上升——信息密度高模型容易推理。4.3 多Server的注册与路由实际系统里有四个MCP ServerSPC数据服务、设备日志服务、物料批次服务、工艺参数服务。LangGraph这边用一个MCP Client统一管理class MCPToolkit: def __init__(self): self.servers {} self.tool_index {} # tool_name - server_name async def register_server(self, name: str, command: list[str]): client MCPClient(command) await client.connect() self.servers[name] client tools await client.list_tools() for tool in tools: self.tool_index[tool.name] name async def call_tool(self, tool_name: str, arguments: dict): server_name self.tool_index.get(tool_name) if not server_name: raise ValueError(f未知工具: {tool_name}) return await self.servers[server_name].call_tool(tool_name, arguments)这个Toolkit在LangGraph的节点里被调用。节点函数拿到LLM决定的工具名和参数后通过Toolkit路由到对应的MCP Server执行。提示MCP Server的启动方式建议用子进程管理不要和主进程混在一起。我一开始图省事直接在同一个进程里跑结果某个Server的数据库连接池满了把主进程也拖死了。后来改成独立子进程通过stdio通信稳定性好很多。5. 推理节点的Prompt工程与输出约束5.1 模式识别节点的结构化输出classify_pattern节点负责判断异常属于哪种模式。这个节点的输出直接决定后续走哪条排查路径所以必须结构化、可校验。我用Pydantic定义输出schema然后让LLM按schema输出JSONfrom pydantic import BaseModel, Field from typing import Literal class PatternClassification(BaseModel): pattern: Literal[sudden_shift, gradual_drift, cyclic, single_outlier, increasing_variance] confidence: float Field(ge0, le1) reasoning: str Field(max_length200) key_evidence: list[str] Field(max_length5) PATTERN_PROMPT 你是一个SPC分析专家。根据以下异常数据摘要判断异常模式。 数据摘要 {data_summary} 异常模式定义 - sudden_shift: 均值突然跳变通常由换料、设备参数变更引起 - gradual_drift: 均值缓慢单向移动通常由刀具磨损、温漂引起 - cyclic: 呈现周期性波动通常由环境周期或操作周期引起 - single_outlier: 孤立异常点通常由测量误差或瞬时干扰引起 - increasing_variance: 方差增大但均值稳定通常由设备松动或来料不均引起 请严格按JSON格式输出不要添加任何其他文字。这里的关键是用定义约束模型。如果不给模式定义模型会自己发明分类标准不同批次的分析结果不可比。给了明确定义后分类一致性大幅提升。5.2 假设生成节点的发散与收敛generate_hypothesis节点需要根据异常模式生成排查假设。这个节点我用了两阶段策略先发散生成多个候选假设再收敛排序。HYPOTHESIS_PROMPT 基于以下信息生成3-5个可能的根因假设按可能性从高到低排序。 异常模式{pattern} 异常特征{characteristic} 产线{product_line} 已有证据{evidence} 每个假设需包含 1. 假设描述一句话 2. 需要验证的数据源 3. 预期验证结果 输出JSON数组格式。生成后路由函数会把假设压入hypothesis_stack然后investigate节点逐个验证。验证一个、排除一个、再验证下一个——这就是模拟质量工程师逐个排除的思维过程。5.3 证据链的累积与引用每次investigate节点执行后会把查询结果和推理结论追加到evidence列表。这个列表在生成报告时会被完整引用形成可追溯的证据链。证据的结构{ step: 3, hypothesis: 3号机台刀具磨损导致尺寸漂移, tool_called: query_equipment_log, tool_args: {equipment_id: EQ-003, start_time: ..., end_time: ...}, finding: 刀具更换记录显示上次更换在72小时前超过标准更换周期48小时, supports_hypothesis: True, confidence_delta: 0.3 }这个结构让最终报告可以精确地说第3步查询了设备日志发现刀具超期使用支持了假设置信度提升0.3——完全可审计。6. 实测中暴露的三个典型问题6.1 工具调用参数幻觉系统跑通后遇到的第一个问题是LLM经常编造不存在的设备ID或物料批次号。比如它会在参数里填EQ-999而实际设备只到EQ-010。MCP Server收到查询请求后返回空结果模型却把这个空结果当成设备无异常记录得出错误结论。修复方案是在MCP Server端加参数校验层查询前先验证设备ID是否存在不存在直接返回错误信息而不是空结果。错误信息里明确写设备EQ-999不存在可用设备列表EQ-001至EQ-010。模型收到这个错误后会重新选择正确的ID。async def query_equipment_log(equipment_id: str, start_time: str, end_time: str): valid_ids await get_valid_equipment_ids() if equipment_id not in valid_ids: return { error: f设备{equipment_id}不存在, available_ids: valid_ids[:20], # 只返回前20个避免过长 suggestion: 请从可用设备列表中选择 } # ... 正常查询逻辑6.2 状态字段的并发写入冲突LangGraph在并行执行分支时多个节点可能同时往evidence列表里写数据。虽然用了operator.add作为reducer但在高并发下仍然出现了证据顺序错乱的问题——第5步的证据排在了第3步前面。解决办法是给每条证据加时间戳和步骤号在生成报告前按步骤号排序。另外把并行分支改成串行执行——质量分析本来就是顺序推理的过程并行反而破坏了逻辑链条。6.3 LLM在长链条中的遗忘当分析链条超过6-7步后模型开始忘记早期的证据。比如第2步已经确认了异常模式是渐变漂移到第8步生成报告时却写成了突变。这是因为状态对象里的信息太多超出了模型的注意力窗口。我的应对策略是在每个节点的Prompt里显式注入关键状态摘要而不是让模型自己去状态里翻。具体做法是写一个summarize_state函数把当前最相关的5条信息压缩成一段话放在Prompt开头def summarize_state(state: SPCState) - str: parts [ f异常模式{state[anomaly_pattern]}, f当前假设{state[current_hypothesis]}, f已排除假设{len(state[hypothesis_stack])}个, f关键证据{state[evidence][-3:] if state[evidence] else 暂无}, f排查进度第{state.get(iteration_count, 0)}轮 ] return \n.join(parts)这个摘要放在每个Prompt的最前面确保模型始终记得核心上下文。7. 从报警到根因报告的完整链路演示7.1 触发与初始数据拉取假设产线PL-01的外径尺寸特性在14:30触发报警。系统入口接收报警IDfetch_anomaly节点调用query_spc_data工具拉取最近4小时的数据。MCP Server返回预聚合摘要200个点均值25.32标准差0.15UCL25.80LCL24.80有5个点超出UCL趋势为上升。7.2 模式判定与假设生成classify_pattern节点收到摘要后LLM判断为gradual_drift渐变漂移置信度0.85理由均值从25.1缓慢上升至25.6无突变点。generate_hypothesis节点基于渐变漂移模式生成三个假设刀具磨损可能性最高、温漂、来料批次差异。7.3 逐层排查与证据累积investigate节点先验证刀具磨损假设调用query_equipment_log查询3号机台日志。返回显示刀具已使用72小时超过标准更换周期48小时。证据追加置信度提升至0.7。接着验证温漂假设调用query_process_param查询温度记录显示温度稳定在22±0.5°C排除温漂。最后验证来料假设调用query_material_batch查询当前批次检验结果正常排除。7.4 根因确认与报告生成verify节点综合三条证据确认根因为3号机台刀具超期使用导致尺寸渐变漂移置信度0.92。generate_report节点生成结构化报告包含异常描述、分析过程、证据链、根因结论、建议措施立即更换刀具、调整更换周期至48小时、加强刀具寿命监控。整个流程从触发到报告生成实测平均耗时2分15秒其中LLM调用占约80秒MCP工具调用占约40秒其余为状态流转开销。8. 部署与扩展的几个实操建议8.1 本地模型与API的取舍如果数据敏感度高建议用本地部署的开源模型如Qwen系列。但要注意本地模型在结构化输出JSON格式上的稳定性不如大参数API模型。我的做法是在本地模型外面套一层输出解析器用正则提取JSON解析失败时自动重试并降低temperature。8.2 MCP Server的版本管理MCP工具的描述和参数schema会随着业务变化而调整。建议给每个Server加版本号LangGraph这边记录使用的版本。当Server升级时如果schema有breaking change主流程需要同步更新Prompt里的工具描述。我吃过这个亏——Server端加了一个必填参数但Prompt里没更新导致模型调用一直失败。8.3 分析结果的反馈闭环系统上线后我加了一个反馈机制质量工程师可以标记AI给出的根因是否正确。这些标记数据用来微调模式识别和假设生成的Prompt。跑了两个月后根因定位准确率从最初的68%提升到了87%。关键改进点是把工程师常用的排查经验比如先查换料记录再查设备日志写进了Prompt的few-shot示例里。这套系统目前还在持续迭代下一步计划是把SPC分析扩展到多特性联合分析——比如外径和圆度同时异常时根因可能指向夹具问题而不是刀具。那又是另一个故事了。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
