第 7 章：制造业大模型应用

凌晨三点，苏州某汽车零部件工厂的三号产线突然停机。值班工程师李伟打开MES系统，面对屏幕上密密麻麻的报警日志和传感器数据——温度异常、振动频率偏移、电流波动。往常他需要翻阅三本维修手册、打电话给退休的老师傅，再花两小时排查才能定位问题。但今晚，他在工厂的智能运维终端输入："三号线停机原因分析，报警日志如下……" 30秒后，系统返回了完整的诊断报告：主轴轴承内圈磨损导致振动异常，建议更换型号6210-2RS的轴承，附带过去6个月该部件的温度趋势图和采购库存信息。李伟确认后，系统自动生成工单并通知仓库备料。从停机到恢复生产，全程47分钟。这是2025年中国制造业智能化改造的一个缩影。

7.1 场景开篇：工厂里的"AI工程师"

上述场景并非科幻。2025年，中国制造业增加值占全球比重超过30%，连续15年位居世界第一，但制造业全员劳动生产率仅为美国的约三分之一。这个巨大差距的核心原因之一，是制造业中大量经验知识存在于老工程师的头脑中，而非可复用的数字化资产。

大模型正在改变这一局面。它像一个永远在线、拥有全厂知识的"AI工程师"，能够理解自然语言、解读传感器数据、分析图像、调用工业系统API，完成从故障诊断到质量分析再到排产优化的全链条任务。

以一个典型的智能工厂为例，传统工作模式与大模型赋能后的对比：

据麦肯锡2025年全球调研数据，制造业大模型的平均ROI已达到300-500%，其中预测性维护和视觉质检是ROI最高的两个场景。中国制造业已建成超过400家"灯塔工厂"和智能工厂试点，大模型项目招投标金额同比增长超过500%。

这些数字背后，是大模型与工业知识深度融合的技术架构在发挥作用。接下来，我们将从行业全景到技术细节，再到实战案例，逐步拆解制造业大模型应用的全貌。

7.2 行业全景：工业4.0与大模型

制造业数字化转型的阶段演进

中国制造业正处于从工业3.0向工业4.0跨越的关键窗口期。根据工信部2025年数据，全国已建成超过400家"灯塔工厂"和智能工厂试点，规模以上工业企业中数字化研发设计工具普及率约为85%，但关键工序数控化率仅为63%。这意味着：数据基础已经具备，但数据的智能化利用才刚刚开始。

工业1.0          工业2.0          工业3.0          工业4.0
机械化           电气化           信息化           智能化
(1784-)         (1870-)         (1969-)         (2015-)
  |               |               |               |
  v               v               v               v
蒸汽机          流水线           PLC/SCADA      大模型+IoT+数字孪生
手工-机器       机器-规模       自动化控制      自主决策与优化
                                    ^
                              大多数中国企业当前阶段

大模型的独特价值在于：它是工业3.0积累的海量数据与工业4.0智能决策能力之间的桥梁。SCADA系统采集的传感器数据、MES中的生产记录、ERP中的物料信息、EAM中的维修日志——这些分散在不同系统中的数据，通过大模型实现了语义层面的融合与理解。

政策环境："人工智能+"行动意见

2024年，国务院印发《新一代人工智能发展规划》的深化实施方案，明确提出"人工智能+"行动意见，将制造业列为重点赋能领域。核心政策要点包括：

• 智能制造示范工厂建设：到2026年，建设500个以上智能制造示范工厂，推广大模型在质量检测、设备运维、供应链优化等场景的应用
• 工业大模型专项：支持龙头企业与AI公司联合研发工业大模型，重点突破多模态融合、工业知识蒸馏、边缘部署等关键技术
• 数据要素市场化：推动工业数据分类分级、确权流通，为大模型训练提供高质量数据基础
• 中小企业普惠AI：建设区域性工业AI公共服务平台，降低中小企业使用大模型的门槛

全球制造业AI关键数据对比：

政策推动下，制造业已成为大模型落地的第二大行业（仅次于IT/互联网），2025年制造业大模型项目招投标金额同比增长超过500%。

工业大模型市场格局

制造业大模型应用成熟度矩阵：

7.3 核心应用场景

7.3.1 智能制造与自然语言数据查询

制造业最大的痛点之一是"数据孤岛"——生产数据在MES、质量数据在QMS、设备数据在SCADA、物料数据在ERP、维修数据在EAM，业务人员想回答一个跨系统的问题（如"上周A产品的良率和对应的主要设备参数异常有没有关联？"），需要向三个部门提需求、等一周才能拿到报告。

大模型通过**自然语言数据查询（NL2SQL + RAG）**技术，让业务人员用自然语言直接查询跨系统数据：

# 自然语言查询引擎 - 基于LangGraph的完整实现

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Optional
import json
from openai import OpenAI

client = OpenAI()

class ManufacturingQueryState(TypedDict):
    user_question: str          # 用户自然语言问题
    intent: str                 # 解析后的意图
    target_tables: List[str]    # 需要查询的数据表
    sql_queries: List[str]      # 生成的SQL语句
    raw_results: List[dict]     # 查询结果
    analysis: str               # 分析结论
    visualization_spec: dict    # 可视化配置
    confidence: float           # 结果置信度

def intent_parser(state: ManufacturingQueryState) -> dict:
    """解析用户意图，识别需要查询的系统"""
    prompt = f"""你是一个制造业数据分析助手。分析以下问题，识别需要查询的数据源。

问题：{state['user_question']}

可用数据源：
- MES（制造执行系统）：产量、工单、工艺参数、设备状态
- QMS（质量管理系统）：检测结果、不良品记录、SPC数据
- SCADA（数据采集系统）：传感器实时数据、设备运行参数
- ERP（企业资源计划）：物料、采购、订单、库存、BOM
- EAM（设备管理系统）：维修记录、备件库存、保养计划

请以JSON格式输出：
{{
    "intent": "意图描述",
    "systems": ["需要查询的系统列表"],
    "time_range": "时间范围",
    "key_entities": ["关键实体"],
    "query_type": "trend|comparison|root_cause|status"
}}"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,
        response_format={"type": "json_object"}
    )
    result = json.loads(response.choices[0].message.content)
    return {
        "intent": result["intent"],
        "target_tables": result["systems"]
    }

def sql_generator(state: ManufacturingQueryState) -> dict:
    """根据意图生成跨系统SQL查询"""
    # 获取表结构元数据
    table_schemas = get_table_metadata(state["target_tables"])

    prompt = f"""基于以下表结构生成SQL查询：

表结构：{json.dumps(table_schemas, ensure_ascii=False, indent=2)}

用户问题：{state['user_question']}
意图：{state['intent']}

要求：
1. 输出标准SQL，只查询必要字段
2. 注意性能优化（合理使用索引、限制返回行数）
3. 如需跨表关联，明确指定JOIN条件
4. 日期范围默认为最近7天，除非用户指定"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0
    )
    return {"sql_queries": [response.choices[0].message.content]}

def execute_queries(state: ManufacturingQueryState) -> dict:
    """执行生成的SQL查询"""
    results = []
    for sql in state["sql_queries"]:
        # 安全检查：只允许SELECT语句
        if not sql.strip().upper().startswith("SELECT"):
            continue
        result = execute_sql_safely(sql)
        results.append(result)
    return {"raw_results": results}

def result_analyzer(state: ManufacturingQueryState) -> dict:
    """分析查询结果，生成自然语言报告"""
    prompt = f"""分析以下数据查询结果，用制造业语言给出结论：

问题：{state['user_question']}
数据：{json.dumps(state['raw_results'], ensure_ascii=False, indent=2)}

要求：
1. 直接回答问题
2. 标注关键发现（用[关键]标记）
3. 如有异常，指出可能原因（用[异常]标记）
4. 给出改进建议（用[建议]标记）
5. 推荐可视化图表类型"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2
    )
    return {
        "analysis": response.choices[0].message.content,
        "confidence": 0.85  # 基于数据完整性的置信度评估
    }

# 构建查询工作流
workflow = StateGraph(ManufacturingQueryState)
workflow.add_node("parse_intent", intent_parser)
workflow.add_node("generate_sql", sql_generator)
workflow.add_node("execute", execute_queries)
workflow.add_node("analyze", result_analyzer)

workflow.set_entry_point("parse_intent")
workflow.add_edge("parse_intent", "generate_sql")
workflow.add_edge("generate_sql", "execute")
workflow.add_edge("execute", "analyze")
workflow.add_edge("analyze", END)

query_app = workflow.compile()

实际效果：某汽车零部件企业部署自然语言查询系统后，原本需要一周的数据分析需求缩短至分钟级。车间主任可以用手机直接问："今天CNC3的OEE为什么只有68%？"系统会自动关联设备停机记录、换模时间、故障日志，给出完整分析。系统上线3个月后，日均查询量从50次增长到800次，覆盖了90%的常见数据分析需求。

7.3.2 质量控制与缺陷检测（多模态）

质量控制是制造业大模型最具突破性的应用场景之一。传统机器视觉检测依赖"规则+特征工程"，每增加一种缺陷类型就需要重新标注数据、训练模型，部署周期长达数周。多模态大模型彻底改变了这一范式：

传统方法 vs 大模型方法对比：

技术实现架构：

产线相机 --→ 图像采集 --→ 边缘预处理 --→ 多模态大模型推理
                                                    |
                 +----------------------------------+
                 |                                  |
                 v                                  v
          缺陷检测与分类                      缺陷描述生成
          (小模型快速筛选)                    (大模型精细分析)
                 |                                  |
                 +-----------+----------------------+
                             |
                     +---------------+
                     |  根因分析Agent  |
                     |  关联工艺参数   |
                     |  追溯生产批次   |
                     |  生成改进建议   |
                     +---------------+

核心创新在于大模型+小模型协同：小模型（如YOLOv8等轻量级检测模型）在边缘端做高速初筛，只将疑似缺陷的图像发送给大模型进行精细分析和描述。这样既保证了产线上毫秒级的响应速度，又获得了大模型的深度分析能力。

7.3.3 设备预测性维护

设备非计划停机是制造业最大的成本黑洞之一。据西门子数据，大型制造企业因设备故障导致的年损失可达产值的5-10%。传统的维护策略要么是"坏了再修"（事后维修，损失最大），要么是"到点就修"（定期保养，浪费资源）。预测性维护通过分析设备运行数据，在故障发生前预警并指导维护。

大模型在预测性维护中的三层价值：

1. 数据理解层：理解传感器时序数据的语义——"主轴温度从65 C上升到82 C"不只是一个数值变化，大模型能关联到"轴承磨损导致摩擦增大，温度升高，可能在未来72小时内失效"
2. 知识融合层：融合设备手册、维修历史、工程师经验、同类设备故障案例等多源知识
3. 决策支持层：生成可执行的维护建议——不只是"设备异常"，而是"建议在48小时内更换主轴轴承（型号6210-2RS），当前库存3个，位于B区货架，预计维护时间2小时"

维护策略演进与成本对比：

# 预测性维护智能诊断Agent - LangGraph实现

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Dict, Optional
import json
from openai import OpenAI

client = OpenAI()

class MaintenanceState(TypedDict):
    device_id: str
    device_type: str
    sensor_data: Dict           # 传感器实时数据
    alarms: List[Dict]          # 当前报警列表
    anomaly_score: float        # 异常评分
    anomaly_details: Optional[Dict]  # 异常详情
    diagnosis: Optional[str]    # 诊断结论
    recommended_action: Optional[Dict]  # 建议措施
    confidence: float           # 置信度

def anomaly_detector(state: MaintenanceState) -> dict:
    """时序异常检测小模型：快速识别异常模式"""
    sensor_data = state["sensor_data"]

    # 模拟时序异常检测（实际部署中使用训练好的时序模型）
    # 检测温度、振动、电流等关键参数的异常
    anomaly_score = 0.0
    details = {}

    # 温度异常检测
    temp = sensor_data.get("temperature", 0)
    if temp > 80:
        anomaly_score += 0.4
        details["temperature"] = f"温度异常：{temp}°C（阈值80°C）"

    # 振动异常检测
    vibration = sensor_data.get("vibration", 0)
    if vibration > 5.0:
        anomaly_score += 0.3
        details["vibration"] = f"振动异常：{vibration}mm/s（阈值5.0mm/s）"

    # 电流异常检测
    current = sensor_data.get("current", 0)
    if current > 15:
        anomaly_score += 0.3
        details["current"] = f"电流异常：{current}A（阈值15A）"

    return {
        "anomaly_score": anomaly_score,
        "anomaly_details": details if anomaly_score > 0.3 else None
    }

def fault_diagnoser(state: MaintenanceState) -> dict:
    """大模型故障诊断：综合多源知识进行根因分析"""

    if state["anomaly_score"] < 0.3:
        return {"diagnosis": "设备运行正常，无需处理", "confidence": 0.95}

    # 检索设备手册和历史维修记录
    device_docs = search_knowledge_base(
        f"{state['device_id']} {state['device_type']} 维修手册 故障处理"
    )
    history_docs = search_knowledge_base(
        f"{state['device_id']} 历史维修记录"
    )

    prompt = f"""你是一个设备诊断专家。基于以下信息进行故障诊断：

【设备信息】{state['device_id']}，类型：{state['device_type']}
【异常检测结果】异常评分={state['anomaly_score']}，详情={state['anomaly_details']}
【当前报警】{json.dumps(state['alarms'], ensure_ascii=False)}
【传感器数据】温度={state['sensor_data'].get('temperature')}°C，
    振动={state['sensor_data'].get('vibration')}mm/s，
    电流={state['sensor_data'].get('current')}A
【设备手册相关章节】{device_docs}
【历史维修记录】{history_docs}

请输出：
1. 故障根因分析（可能的原因及概率，按概率降序排列）
2. 建议措施（紧急程度、所需备件、预计工时）
3. 风险评估（如不及时处理的后果）
4. 类似历史案例（如有）

以JSON格式输出。"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,
        response_format={"type": "json_object"}
    )

    diagnosis = json.loads(response.choices[0].message.content)

    return {
        "diagnosis": json.dumps(diagnosis, ensure_ascii=False),
        "recommended_action": diagnosis.get("recommended_action"),
        "confidence": min(0.95, state["anomaly_score"] + 0.5)
    }

def action_planner(state: MaintenanceState) -> dict:
    """生成可执行的维护工单"""
    if not state["recommended_action"]:
        return {"confidence": state.get("confidence", 0.5)}

    action = state["recommended_action"]

    # 查询备件库存
    spare_parts_info = query_inventory(action.get("spare_parts", []))

    # 生成工单
    work_order = {
        "device_id": state["device_id"],
        "priority": action.get("urgency", "中"),
        "estimated_duration": action.get("estimated_hours", 2),
        "required_parts": spare_parts_info,
        "procedure": action.get("procedure", "参见设备手册"),
        "safety_notes": action.get("safety_notes", "断电操作，佩戴防护装备")
    }

    return {
        "recommended_action": work_order,
        "confidence": state.get("confidence", 0.5)
    }

# 构建预测性维护Agent工作流
workflow = StateGraph(MaintenanceState)
workflow.add_node("detect", anomaly_detector)
workflow.add_node("diagnose", fault_diagnoser)
workflow.add_node("plan", action_planner)

workflow.set_entry_point("detect")
workflow.add_conditional_edges(
    "detect",
    lambda state: END if state["anomaly_score"] < 0.3 else "diagnose"
)
workflow.add_edge("diagnose", "plan")
workflow.add_edge("plan", END)

maintenance_app = workflow.compile()

7.3.4 供应链与排产优化

排产调度是制造业最复杂的优化问题之一。一个中型工厂的日排产涉及数百个工单、数十台设备、数千种物料，约束条件包括设备能力、人员排班、物料齐套、交期优先级等。传统方法依赖运筹学求解器，但建模和调参需要高度专业的能力。

大模型的介入方式并非替代求解器，而是作为自然语言接口和智能建模助手：

用户需求（自然语言）
"下周有三个紧急订单，优先级高于常规订单，
 尽量减少换模次数，注意周二设备保养"
        |
        v
+------------------------+
|   大模型理解与建模层    |
|  - 提取约束条件         |
|  - 识别优化目标         |
|  - 生成求解器输入参数   |
+-----------+------------+
            |
            v
+------------------------+
|   OR求解器（CPLEX/     |
|   Gurobi/OR-Tools）     |
|  - 数学规划求解         |
|  - 约束满足与优化       |
+-----------+------------+
            |
            v
+------------------------+
|   大模型解释与可视化层  |
|  - 自然语言解释排产结果 |
|  - 识别瓶颈与风险       |
|  - 甘特图自动生成       |
+------------------------+

这种"大模型负责理解和解释、求解器负责计算"的协同模式，是目前制造业AI应用的最佳实践之一。它让排产从"OR专家的专利"变成了"车间主任的日常工具"。

实际效果：某大型离散制造企业引入排产优化助手后，排产周期从3天缩短至4小时，产能利用率从78%提升至89%，交期达成率从85%提升至96%。

7.4 技术架构深度解析

7.4.1 大模型+小模型协同架构

制造业对实时性和可靠性的要求，决定了不能完全依赖云端大模型。实际落地中，最有效的架构是大模型+小模型协同——云端大模型提供深度理解和知识推理能力，边缘端小模型提供高速推理和实时响应。

+---------------------------------------------------------+
|                     云端（数据中心）                       |
|                                                          |
|  +--------------+  +--------------+  +--------------+   |
|  |  工业大模型    |  |  知识库(RAG)  |  |  训练平台    |   |
|  |  72B参数      |  |  设备手册     |  |  小模型蒸馏  |   |
|  |  自然语言理解  |  |  工艺文档     |  |  持续学习    |   |
|  |  多模态分析    |  |  维修记录     |  |  A/B测试     |   |
|  +------+-------+  +------+-------+  +------+-------+   |
|         +----------------+------------------+           |
|                          |                              |
+--------------------------+------------------------------+
                           | 工业网络（TSN/5G专网）
                           |
+--------------------------+------------------------------+
|                     边缘层（车间网关/边缘服务器）          |
|                          |                              |
|  +--------------+  +-----+--------+  +--------------+   |
|  |  小模型集群    |  |  数据预处理   |  |  规则引擎    |   |
|  |  YOLOv8(质检) |  |  去噪+特征   |  |  安全兜底    |   |
|  |  时序异常检测  |  |  提取        |  |  阈值告警    |   |
|  |  NL2SQL轻量   |  |              |  |              |   |
|  +------+-------+  +--------------+  +--------------+   |
|         |                                              |
+---------+----------------------------------------------+
          |
+---------+----------------------------------------------+
|     设备层（PLC / 传感器 / 相机 / AGV）                  |
|                                                          |
|  +------+  +------+  +------+  +------+  +------+      |
|  | CNC  |  |机器人|  |相机  |  |AGV   |  |传感器|      |
|  +------+  +------+  +------+  +------+  +------+      |
+---------------------------------------------------------+

协同工作流程：

7.4.2 边缘-云协同部署

制造业场景对延迟、带宽和可靠性有严格要求。以汽车冲压线为例，每分钟生产15个零件，质检必须在500ms内完成，否则整条产线必须降速或停机等待。

边缘-云协同的关键设计：

                    云端                    边缘端
                    ----                    ----
模型             全量大模型(72B)          蒸馏小模型(1-3B)
响应时间          1-10秒                  <100毫秒
功能             深度分析、知识推理        实时检测、快速筛选
更新频率          按需更新                  定期从云端同步
离线能力          无要求                    必须支持离线运行
网络要求          高带宽                    低带宽可运行
硬件             GPU集群                  边缘GPU/NPU

模型蒸馏流水线：云端大模型定期向边缘端小模型"传授知识"：

# 工业知识蒸馏流程示意

def distill_for_edge(teacher_model, student_model, industrial_data):
    """
    将云端大模型的知识蒸馏到边缘端小模型

    Args:
        teacher_model: 云端工业大模型 (72B)
        student_model: 边缘端小模型 (1-3B)
        industrial_data: 工业场景标注数据
    """
    # Step 1: 大模型生成教师标签
    teacher_labels = []
    for batch in industrial_data:
        # 大模型对每个样本生成详细分析
        label = teacher_model.generate(
            prompt=f"分析以下工业数据：{batch}",
            max_tokens=512
        )
        teacher_labels.append(label)

    # Step 2: 小模型学习大模型的输出模式
    for epoch in range(num_epochs):
        for batch, teacher_label in zip(industrial_data, teacher_labels):
            student_output = student_model(batch)
            loss = distillation_loss(
                student_output,
                teacher_label,
                temperature=4.0  # 蒸馏温度
            )
            loss.backward()
            optimizer.step()

    # Step 3: 验证蒸馏效果
    metrics = evaluate_on_industrial_benchmarks(student_model)
    assert metrics["accuracy"] > 0.90, "蒸馏效果不达标，需要更多数据"

    # Step 4: 量化压缩，适配边缘硬件
    quantized_model = quantize(student_model, bits=4)
    return quantized_model

7.4.3 OT/IT数据融合

制造业数据的根本挑战在于OT（运营技术）数据与IT（信息技术）数据的割裂。OT数据来自PLC、传感器、DCS等控制系统，具有高频率、实时性、时序性特征；IT数据来自ERP、MES、CRM等管理系统，具有结构化、业务语义丰富的特征。

OT数据世界                    IT数据世界
---------                    ---------
传感器数据(毫秒级)            ERP订单数据
PLC控制信号                   MES生产工单
设备状态码                    QMS质量记录
振动/温度/电流                CRM客户需求
工艺参数(实时)                WMS库存信息

      |                           |
      v                           v
+-----------------------------------------+
|           OT/IT数据融合层                |
|                                         |
|  1. 时序数据标准化（统一时间戳、采样率） |
|  2. 语义映射（传感器ID - 设备 - 工位）   |
|  3. 上下文关联（OT异常 <-> IT事件）      |
|  4. 统一知识图谱（设备-工艺-质量-订单）  |
|                                         |
|  核心技术：                              |
|  - 时序大模型（理解传感器数据语义）       |
|  - 知识图谱（关联OT实体与IT实体）        |
|  - RAG索引（跨系统知识检索）             |
+------------------+----------------------+
                  |
                  v
          统一的数据语义层
      （大模型可直接理解和查询）

融合的核心难点与解法：

7.4.4 工业知识蒸馏

制造业最宝贵的资产不是设备，而是工程师的经验。一个工作30年的铸造工程师能听声音判断铸件质量，看火花判断切削参数——这些隐性知识如何转化为大模型的能力？

工业知识蒸馏的核心思路是将人类专家知识结构化，再注入大模型：

1. 经验文档化：通过大模型辅助的结构化访谈，将工程师经验转化为标准化文档
2. 案例库构建：收集历史故障案例及处理过程，构建"问题-分析-措施"三元组
3. 知识注入：通过RAG和微调，将结构化知识注入大模型
4. 持续学习：每次新的故障处理结果自动回流到知识库，模型持续进化

工程师经验 --+  结构化访谈 --+  知识三元组 --+  RAG知识库
                                                   |
设备手册 --+  文档解析 --+  条款级索引 --+---------+
                                                   |
维修记录 --+  案例提取 --+  故障-原因-措施库 --+---+
                                                   |
工艺规程 --+  参数提取 --+  工艺知识库 --+---------+
                                                   |
                                 大模型（RAG + 微调）
                                        |
                                        v
                                 工业智能助手
                               （问答/诊断/优化）

知识蒸馏的效果对比：

7.5 实战案例：工业视觉质检与智能运维系统

项目背景

某头部汽车零部件制造企业，拥有12条产线，年产3000万件产品。核心痛点：

• 漏检率高达3%：传统AOI（自动光学检测）系统对新型缺陷识别率低
• 非计划停机频繁：年均非计划停机1200小时，损失超过5000万元
• 专家经验断层：30%的高级工程师将在5年内退休，经验传承困难
• 数据利用率低：每月产生2TB传感器数据，但利用率不到5%

系统架构设计

+--------------------------------------------------------------+
|                       用户交互层                               |
|  +----------+  +----------+  +----------+  +----------+      |
|  | 车间终端  |  | 管理看板  |  | 移动APP  |  | 大屏监控  |      |
|  +----------+  +----------+  +----------+  +----------+      |
+-----------------------------+--------------------------------+
                              |
+-----------------------------v--------------------------------+
|                       服务层                                   |
|  +-----------+  +-----------+  +-----------+                 |
|  | 质检服务   |  | Agent编排  |  | 运维服务   |                |
|  | (多模态)   |  | (LangGraph)|  | (预测性)   |                |
|  +-----+-----+  +-----+-----+  +-----+-----+                |
|        +---------------+---------------+                      |
|                  +-----v-----+                                |
|                  | 统一推理网关 |                                |
|                  +-----+-----+                                |
+-----------------------------+--------------------------------+
                              |
+-----------------------------v--------------------------------+
|                       模型层                                   |
|     +-----------+  +-----------+  +-----------+              |
|     | 云端大模型  |  | 知识库(RAG)|  | 边缘小模型  |              |
|     | 视觉+语言  |  | Milvus+   |  | YOLO+时序  |              |
|     | 72B       |  | Neo4j     |  | 实时推理    |              |
|     +-----------+  +-----------+  +-----------+              |
+--------------------------------------------------------------+
                              |
+-----------------------------v--------------------------------+
|                       数据层                                   |
|  +------+  +------+  +------+  +------+  +------+           |
|  |SCADA |  | MES  |  | ERP  |  | QMS  |  | EAM  |           |
|  +------+  +------+  +------+  +------+  +------+           |
+--------------------------------------------------------------+

关键实现：视觉质检与智能运维LangGraph工作流

"""
工业视觉质检与智能运维系统 - 完整LangGraph实现
三级质检架构 + 预测性维护联动
"""
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Dict, Optional
import json
from openai import OpenAI

client = OpenAI()


class InspectionMaintenanceState(TypedDict):
    """质检-运维联合状态"""
    # 质检相关
    image_data: Optional[str]           # 产品图像（base64）
    product_info: Dict                  # 产品规格信息
    l1_result: Optional[Dict]           # L1检测结果
    l2_result: Optional[Dict]           # L2检测结果
    l3_result: Optional[Dict]           # L3检测结果
    inspection_verdict: str             # 质检结论：PASS/NG/REVIEW
    defect_info: Optional[Dict]         # 缺陷详情

    # 运维相关
    device_id: str                      # 关联设备ID
    device_sensor_data: Dict            # 设备传感器数据
    maintenance_needed: bool            # 是否需要维护
    maintenance_action: Optional[Dict]  # 维护建议

    # 通用
    confidence: float
    escalation: bool                    # 是否需要人工介入


def l1_edge_detect(state: InspectionMaintenanceState) -> dict:
    """L1: 边缘端小模型快速筛选 (<50ms)"""
    # 模拟YOLOv8边缘端检测
    result = edge_model_detect(state["image_data"])
    return {
        "l1_result": result,
        "confidence": result.get("confidence", 0.0)
    }


def l2_multimodal_analyze(state: InspectionMaintenanceState) -> dict:
    """L2: 轻量级多模态分析 (<500ms)"""
    prompt = f"""分析以下产品图像，判断是否存在缺陷。

产品规格：{json.dumps(state['product_info'], ensure_ascii=False)}
L1检测结果：{json.dumps(state['l1_result'], ensure_ascii=False)}

已知缺陷类型：{state['product_info'].get('known_defects', [])}

请以JSON格式输出：
{{
    "defect_found": true/false,
    "defect_type": "缺陷类型",
    "severity": "轻微/中等/严重",
    "description": "缺陷描述",
    "confidence": 0.0-1.0
}}"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "user", "content": [
                {"type": "text", "text": prompt},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{state['image_data']}"}}
            ]}
        ],
        temperature=0.1,
        response_format={"type": "json_object"}
    )
    result = json.loads(response.choices[0].message.content)
    return {"l2_result": result}


def l3_deep_diagnose(state: InspectionMaintenanceState) -> dict:
    """L3: 云端大模型深度诊断 (<5s)"""
    # 检索历史缺陷数据
    history = search_defect_history(state["product_info"]["line_id"])

    prompt = f"""你是一个资深的工业质量工程师。请对以下缺陷进行深度分析。

L1检测结果：{json.dumps(state['l1_result'], ensure_ascii=False)}
L2分析结果：{json.dumps(state['l2_result'], ensure_ascii=False)}
产品信息：{json.dumps(state['product_info'], ensure_ascii=False)}
近期同类缺陷：{json.dumps(history[:5], ensure_ascii=False)}

请输出：
1. 缺陷根因分析（关联设备状态和工艺参数）
2. 缺陷等级评定（是否影响功能安全）
3. 改进建议（工艺参数调整/设备维护建议）
4. 是否需要关联设备维护

以JSON格式输出。"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,
        response_format={"type": "json_object"}
    )
    result = json.loads(response.choices[0].message.content)

    return {
        "l3_result": result,
        "defect_info": result,
        "maintenance_needed": result.get("requires_maintenance", False)
    }


def maintenance_evaluator(state: InspectionMaintenanceState) -> dict:
    """评估是否需要联动设备维护"""
    if not state["maintenance_needed"]:
        return {"maintenance_needed": False}

    # 获取设备传感器数据
    sensor_data = get_device_sensors(state["device_id"])

    prompt = f"""基于质检发现的缺陷和设备传感器数据，评估设备是否需要维护。

缺陷信息：{json.dumps(state['defect_info'], ensure_ascii=False)}
设备传感器数据：{json.dumps(sensor_data, ensure_ascii=False)}

请输出维护建议，包含：
1. 维护紧急程度（立即/48小时内/下次保养时）
2. 建议措施
3. 所需备件
4. 预计工时

以JSON格式输出。"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,
        response_format={"type": "json_object"}
    )

    return {
        "maintenance_needed": True,
        "maintenance_action": json.loads(response.choices[0].message.content)
    }


def final_decision(state: InspectionMaintenanceState) -> dict:
    """最终决策：汇总质检和运维结果"""
    l2 = state.get("l2_result", {})
    l3 = state.get("l3_result", {})

    # 确定质检结论
    if l3:
        verdict = l3.get("verdict", "REVIEW")
    elif l2 and l2.get("confidence", 0) > 0.90:
        verdict = "NG" if l2.get("defect_found") else "PASS"
    else:
        verdict = "REVIEW"

    # 判断是否需要人工介入
    escalation = verdict == "REVIEW"
    if state.get("defect_info", {}).get("severity") == "严重":
        escalation = True

    return {
        "inspection_verdict": verdict,
        "escalation": escalation,
        "confidence": max(
            state.get("l1_result", {}).get("confidence", 0),
            l2.get("confidence", 0),
            l3.get("confidence", 0)
        )
    }


# 构建质检-运维联合工作流
workflow = StateGraph(InspectionMaintenanceState)

# 添加节点
workflow.add_node("l1_detect", l1_edge_detect)
workflow.add_node("l2_analyze", l2_multimodal_analyze)
workflow.add_node("l3_diagnose", l3_deep_diagnose)
workflow.add_node("eval_maintenance", maintenance_evaluator)
workflow.add_node("decide", final_decision)

# 定义边
workflow.set_entry_point("l1_detect")

# L1 -> L2（可疑时）或直接PASS
workflow.add_conditional_edges(
    "l1_detect",
    lambda state: END if state["l1_result"].get("status") == "OK" else "l2_analyze"
)

# L2 -> L3（不确定时）或直接判定
workflow.add_conditional_edges(
    "l2_analyze",
    lambda state: "decide" if state["l2_result"].get("confidence", 0) > 0.90 else "l3_diagnose"
)

# L3 -> 维护评估
workflow.add_edge("l3_diagnose", "eval_maintenance")

# 维护评估 -> 最终决策
workflow.add_edge("eval_maintenance", "decide")

# 最终决策 -> 结束
workflow.add_edge("decide", END)

inspection_app = workflow.compile()

效果数据

系统上线12个月后的效果：

经验教训

项目实施过程中的关键经验：

1. 数据先行：花了3个月做数据治理——统一设备命名、校准时钟、建立质量主数据，这比模型调优重要10倍
2. 渐进式部署：先在一条产线试点，跑通全流程后再推广到12条产线，避免"全面铺开、全面失败"
3. 人机协同：系统初期自动判定+人工复核并行，用人工复核数据持续优化模型，3个月后人工复核率从100%降到10%
4. 边缘优先：保证产线不停机的关键是边缘端必须能独立运行，云端断连不影响基本质检功能
5. 知识回流：每次人工修正的检测结果都回流到训练集，模型准确率持续提升

7.6 陷阱与最佳实践

陷阱一：忽视数据质量，迷信模型能力

典型症状：跳过数据治理，直接采购大模型，期望"AI能搞定一切"。

现实：一个传感器标签写错（如将"温度"标为"湿度"），就会导致整个预测性维护模型失效。制造业数据质量问题远比互联网行业严重——传感器漂移、采样率不统一、设备更换后参数重置、人工记录遗漏——这些问题不解决，再好的模型也是"垃圾进、垃圾出"。

最佳实践：

数据治理检查清单：
  [ ] 设备主数据字典建立（统一ID、命名规范）
  [ ] 传感器校准记录完整
  [ ] 各系统时钟同步（误差 <100ms）
  [ ] 历史数据缺失率 <5%
  [ ] 异常值处理策略明确
  [ ] 数据权限分级与脱敏策略
  [ ] 数据更新频率与保鲜机制

陷阱二：盲目追求全自动化，忽视人机协同

典型症状：试图让大模型直接控制产线设备，跳过人工确认环节。

现实：制造业的安全红线不容挑战。一个错误的指令可能导致设备损坏、产品报废甚至人员伤亡。大模型在制造业中的定位必须是"决策支持"而非"自主控制"。

最佳实践：建立三道防线：

陷阱三：忽视边缘离线能力

典型症状：所有推理都依赖云端，网络一断，系统瘫痪。

现实：工厂网络环境复杂——电磁干扰、网络切换、带宽争用——云端连接不稳定是常态。如果质检系统因为网络抖动停线，损失远超AI系统的全部价值。

最佳实践：

• 边缘端部署轻量级小模型，保证基本功能离线可用
• 云端断连时自动降级：大模型分析功能暂停，小模型检测功能持续运行
• 本地缓存关键知识库，支持离线查询
• 网络恢复后自动同步增量数据
• 边缘端必须支持至少24小时的独立运行

陷阱四：低估OT系统的复杂性和安全性要求

典型症状：用互联网思维做工业项目，忽视工控安全、实时性要求和行业认证。

现实：工业控制系统（PLC、DCS、SCADA）有严格的安全标准和实时性要求。大模型系统接入OT网络需要通过安全评估，不能直接"碰"控制层。

最佳实践：

OT安全接入架构：

大模型系统 -> IT网络 -> DMZ隔离区 -> OT网络 -> 控制层
                      |                    |
                      +-- 单向数据网关 ----+
                          （只读，不写控制指令）

关键原则：
1. 永远不要让AI直接写入PLC/DCS控制指令
2. 通过中间件层隔离IT和OT网络
3. 所有数据访问必须经过安全网关审计
4. 满足IEC 62443工控安全标准要求

陷阱五：项目贪大求全

典型症状：一上来就要做"全厂智能大脑"，覆盖所有场景，结果什么都做不好。

最佳实践：遵循"场景驱动、小步快跑"原则：

1. 选择一个高价值场景：优先选择痛点明确、数据基础好、ROI可量化的场景
2. 3个月MVP验证：快速搭建最小可行系统，验证技术可行性和业务价值
3. 6个月优化固化：基于MVP反馈优化模型和流程，固化最佳实践
4. 12个月规模化推广：将验证过的方案复制到更多产线和场景
5. 持续迭代：建立数据反馈闭环，模型持续进化

推荐优先级排序：

优先级1（低垂果实，快速见效）：
  - 设备知识问答（RAG）
  - 自然语言数据查询（NL2SQL）
  - 质检报告自动生成

优先级2（中等难度，高价值）：
  - 预测性维护
  - 多模态视觉质检
  - 排产优化助手

优先级3（高难度，长期投入）：
  - 工艺参数自优化
  - 全流程数字孪生
  - 自主决策控制

7.7 本章小结与延伸阅读

本章小结

制造业大模型应用正处于从"试点验证"到"规模落地"的关键拐点：

1. 场景价值明确：从预测性维护到多模态质检，从自然语言数据查询到排产优化，大模型在制造业的核心场景已经证明了显著的ROI——非计划停机减少65%、漏检率降低87%、故障定位效率提升93%。

2. 技术架构成熟：大模型+小模型协同、边缘-云协同部署、OT/IT数据融合、工业知识蒸馏——这四大技术支柱构成了制造业大模型落地的可靠架构。关键是不要追求"全用大模型"，而是让大模型做它最擅长的事（理解、推理、生成），小模型做它最擅长的事（快速、实时、轻量）。

3. 落地方法论清晰：数据治理先行、人机协同底线、边缘离线保障、OT安全隔离、小步快跑迭代——这五条最佳实践是无数项目踩坑后的经验总结。

4. 未来方向清晰：从单点应用到全流程协同，从辅助决策到自主优化，从数据驱动到知识驱动——制造业大模型的演进路径与工业4.0的终极目标高度一致。

制造业是中国经济的压舱石，也是大模型应用最大的蓝海。当我们能把30年经验的老工程师的知识注入大模型，让每一个车间工人都拥有一个全天候的AI专家顾问，制造业的效率革命才真正开始。

下一章我们将走进零售与电商行业——一个数据丰富、迭代快速、用户触达最直接的行业，看看大模型如何从"千人千面"进化到"千人千话"。

延伸阅读

• 工业互联网创新发展行动计划（2024-2026） -- 工信部，制造业数字化转型的核心政策文件。
• IDC：中国制造业AI应用市场分析2025 -- IDC研究报告，含制造业AI市场规模和增长率数据。
• Siemens Industrial Copilot技术白皮书 -- 西门子工业AI助手的技术架构与实施经验。
• McKinsey：AI in Manufacturing 2025 -- 麦肯锡制造业AI应用全球调研，含ROI分析和最佳实践。
• 中国信息通信研究院：工业大模型发展报告2025 -- 信通院工业大模型技术白皮书，含技术架构和评估体系。
• 亿欧智库：2026中国工业大模型发展洞察报告 -- 工业/制造业大模型深度分析。