从维修方案到自动执行——当 AI 开始动手

关键词：FED、自动执行、工单闭环、CMMS、数字孪生、人机协同、安全围栏、工业智能体、Qoder
适用人群：工业 AI 工程师、FED/FDE、维修数字化架构师、自动化工程师
⚠️ 免责声明：本文案例为典型场景综合改编，非特定企业真实事件；文中技术方案需结合实际工况评估；涉及工控操作须遵守 LOTO 规范；AI 不直接控制物理设备，仅生成工单和辅助决策，执行动作由人工确认后完成。

---

一、“方案写得好好的，然后呢？”

第四篇上线后，系统能生成一份像模像样的维修方案。但用了一周，小李提了个问题：

“方案是挺清楚的，但我还得自己手动建工单、自己填备件领料单、自己标记完成。这些能不能也自动弄了？”

老张补了一句：“还有，上次我按方案修了，结果备件库没货，白跑一趟。”

这两个问题指向同一件事：方案生成只是中间态，真正的价值在闭环。

FED 的交付物不是一份 PDF 方案，而是从故障发现到修复验证的完整闭环。

FED 经验 #5：AI 系统的价值不在于输出多漂亮的报告，在于它能不能嵌入现有工作流，让一线人员少做重复劳动。

---

二、什么是"自动执行"——先划清边界

在工业场景里，"自动执行"四个字必须慎之又慎。我给这次迭代划了三条红线：

红线	原则	原因
红线一	AI 不直接下发控制指令	不操作 PLC、不控制继电器、不远程启停设备
红线二	物理动作必须人工确认	拆装、更换、润滑等动作由人执行，AI 只生成指令和检查清单
红线三	安全相关操作强制 LOTO	涉及能源隔离的操作，系统只生成 LOTO 检查单，不自动执行

所以本篇的"自动执行"，实际是三个层面的自动化：

1. 流程自动化：自动创建工单、自动预留备件、自动通知相关人员
2. 信息自动化：自动回填维修记录、自动更新知识图谱、自动生成报告
3. 辅助执行：AR 步骤指引、扭矩/间隙参数实时校验、电子签名确认

---

三、FED 执行闭环架构

┌─────────────────────────────────────────────────────────────┐
│                    推理引擎（第四篇）                         │
│              输出：结构化维修方案                             │
└───────────────────────┬─────────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────────────────┐
│              执行编排器（Execution Orchestrator）             │
│  方案 → 工单拆解 → 步骤序列化 → 安全校验                     │
└───────────────────────┬─────────────────────────────────────┘
                        │
        ┌───────────────┼───────────────┬───────────────┐
        ▼               ▼               ▼               ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│  工单系统     │ │  备件系统     │ │  人员调度     │ │  知识闭环     │
│  (CMMS)      │ │  (ERP/MES)   │ │  (排班/通知)  │ │  (图谱更新)   │
│              │ │              │ │              │ │              │
│ 自动建工单   │ │ 自动预留备件 │ │ 推送任务到    │ │ 维修结果回填  │ │
│ 步骤拆解     │ │ 检查库存     │ │ 维修工终端    │ │ 因果链验证    │ │
│ 状态流转     │ │ 缺货预警     │ │ 超时提醒      │ │ 新知识沉淀    │ │
└──────┬───────┘ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘
       │                │                │                │
       ▼                ▼                ▼                ▼
┌─────────────────────────────────────────────────────────────┐
│                    维修执行（人机协同）                       │
│  1. 维修工接收任务（手机/平板）                               │
│  2. 逐步执行，每步电子签名确认                               │
│  3. 关键参数实时录入（扭矩/间隙/温度）                        │
│  4. LOTO 确认（拍照+电子签名）                               │
│  5. 完成验收 → 自动关闭工单                                  │
└─────────────────────────────────────────────────────────────┘

---

四、核心代码实现

4.1 执行编排器（Execution Orchestrator）

把第四篇输出的维修方案拆成可执行的工单步骤：

# execution_orchestrator.py
import json
import uuid
import re
from datetime import datetime
from enum import Enum

class StepStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"

class ExecutionOrchestrator:
    def __init__(self, cmms_client, erp_client, notification_client):
        self.cmms = cmms_client      # 工单系统接口
        self.erp = erp_client        # 备件系统接口
        self.notify = notification_client  # 通知系统接口
    
    def orchestrate(self, diagnosis_result: dict, solution_text: str) -> dict:
        """
        将维修方案编排为可执行工单
        """
        primary = diagnosis_result["diagnosis"]["primary_diagnosis"]
        device = diagnosis_result["parsed"].get("device", "未知设备")
        urgency = diagnosis_result["parsed"].get("urgency", "medium")
        
        # 1. 创建工单
        work_order = self._create_work_order(
            device=device,
            fault=primary["fault"] if primary else "未知",
            urgency=urgency,
            solution=solution_text
        )
        
        # 2. 拆解步骤
        steps = self._parse_steps_from_solution(solution_text)
        
        # 3. 备件预留
        parts_result = self._reserve_parts(steps, work_order["id"])
        
        # 4. 安全检查
        safety_check = self._generate_safety_checklist(steps)
        
        # 5. 分配人员
        assignee = self._assign_technician(device, urgency)
        
        # 6. 推送通知
        self._notify_assigned(assignee, work_order["id"], primary)
        
        return {
            "work_order": work_order,
            "steps": steps,
            "parts_reservation": parts_result,
            "safety_checklist": safety_check,
            "assignee": assignee,
            "status": "dispatched"
        }
    
    def _create_work_order(self, device, fault, urgency, solution):
        """创建 CMMS 工单"""
        wo_id = f"WO-{datetime.now().strftime('%Y%m%d')}-{str(uuid.uuid4())[:6].upper()}"
        work_order = {
            "id": wo_id,
            "device": device,
            "fault_description": fault,
            "urgency": urgency,
            "status": "open",
            "created_at": datetime.now().isoformat(),
            "solution_reference": solution[:200],  # 方案摘要
            "source": "FED_AI_System"
        }
        
        # 调用 CMMS API 创建工单
        result = self.cmms.create_work_order(work_order)
        return result if result else work_order
    
    def _parse_steps_from_solution(self, solution_text):
        """
        从方案文本中解析出结构化步骤
        修正：兼容 '### 步骤 N：' 和 '步骤 N：' 两种格式
        """
        steps = []
        
        # 按 '### 步骤' 或 '步骤 N：' 分割
        step_blocks = re.split(r'(?:###\s*)?步骤\s*\d+[：:]', solution_text)
        
        # 提取步骤编号
        step_numbers = re.findall(r'(?:###\s*)?步骤\s*(\d+)[：:]', solution_text)
        
        for i, (num, block) in enumerate(zip(step_numbers, step_blocks[1:]), 1):
            lines = [l.strip() for l in block.strip().split('\n') if l.strip()]
            if not lines:
                continue
            
            title = lines[0]
            operation = ""
            tools = []
            duration = None
            safety_notes = []
            
            for line in lines[1:]:
                if line.startswith('操作：') or line.startswith('- 操作：'):
                    operation = line.split('：', 1)[1] if '：' in line else line[4:]
                elif line.startswith('工具：') or line.startswith('- 工具：'):
                    tool_text = line.split('：', 1)[1] if '：' in line else line[4:]
                    tools = [t.strip() for t in re.split(r'[、,，]', tool_text)]
                elif line.startswith('预计时间：') or line.startswith('- 预计时间：'):
                    dur_text = line.split('：', 1)[1] if '：' in line else line[5:]
                    dur_match = re.search(r'(\d+)', dur_text)
                    duration = int(dur_match.group(1)) if dur_match else None
                elif line.startswith('⚠️ 安全：') or line.startswith('- ⚠️') or line.startswith('- 🔴') or line.startswith('- 🟡'):
                    safety_notes.append(line)
            
            steps.append({
                "step_number": int(num),
                "title": title,
                "operation": operation,
                "tools": tools,
                "estimated_duration_min": duration,
                "safety_notes": safety_notes,
                "status": StepStatus.PENDING.value,
                "requires_signature": True,
                "requires_loto": any("LOTO" in note for note in safety_notes)
            })
        
        return steps
    
    def _reserve_parts(self, steps, work_order_id):
        """从步骤中提取备件并预留"""
        parts_needed = []
        
        for step in steps:
            op_text = step.get("operation", "")
            # 提取备件型号（简单规则：大写字母+数字组合）
            part_codes = re.findall(r'[A-Z]+\s*\d+[-]?\d*', op_text)
            for code in part_codes:
                parts_needed.append({"code": code.strip(), "step": step["step_number"]})
        
        reservation_results = []
        for part in parts_needed:
            result = self.erp.reserve_part(
                part_code=part["code"],
                quantity=1,
                work_order_id=work_order_id
            )
            reservation_results.append({
                "part": part["code"],
                "step": part["step"],
                "reserved": result.get("success", False),
                "stock_available": result.get("stock_available", False),
                "message": result.get("message", "")
            })
        
        return reservation_results
    
    def _generate_safety_checklist(self, steps):
        """生成安全检查清单"""
        checklist = []
        
        checklist.append({
            "id": "SAF-001",
            "item": "执行 LOTO（锁定挂牌）",
            "mandatory": True,
            "verified": False,
            "verification_method": "拍照 + 电子签名"
        })
        
        for step in steps:
            if step.get("requires_loto"):
                checklist.append({
                    "id": f"SAF-STEP-{step['step_number']}",
                    "item": f"步骤 {step['step_number']} LOTO 确认",
                    "mandatory": True,
                    "verified": False,
                    "verification_method": "电子签名"
                })
        
        checklist.append({
            "id": "SAF-002",
            "item": "确认设备完全停止并冷却",
            "mandatory": True,
            "verified": False,
            "verification_method": "温度检测 + 目视确认"
        })
        
        return checklist
    
    def _assign_technician(self, device, urgency):
        """分配维修工"""
        priority_map = {"critical": 1, "high": 2, "medium": 3, "low": 4}
        priority = priority_map.get(urgency, 3)
        
        technicians = {
            "HMC-80": ["小李", "老王"],
            "default": ["小李"]
        }
        
        assigned = technicians.get(device, technicians["default"])[0]
        return {"name": assigned, "priority": priority, "notified": True}
    
    def _notify_assigned(self, assignee, wo_id, diagnosis):
        """推送通知"""
        message = f"新工单 {wo_id}\n设备：{diagnosis.get('fault', '未知')}\n优先级：{assignee['priority']}"
        self.notify.send(recipient=assignee["name"], title=f"新维修工单 {wo_id}", message=message)

4.2 CMMS 接口封装

# cmms_client.py
import requests

class CMMSClient:
    def __init__(self, base_url, api_key):
        self.base_url = base_url.rstrip('/')
        self.headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    
    def create_work_order(self, wo_data):
        try:
            response = requests.post(f"{self.base_url}/api/workorders", headers=self.headers, json=wo_data, timeout=10)
            if response.status_code in (200, 201):
                return response.json()
            return None
        except Exception as e:
            print(f"CMMS 创建工单失败: {e}")
            return None
    
    def close_work_order(self, wo_id, resolution, duration_min):
        try:
            response = requests.patch(
                f"{self.base_url}/api/workorders/{wo_id}",
                headers=self.headers,
                json={"status": "closed", "notes": f"解决方案: {resolution}\n实际耗时: {duration_min} 分钟"},
                timeout=10
            )
            return response.status_code in (200, 204)
        except Exception as e:
            print(f"CMMS 关闭工单失败: {e}")
            return False

4.3 备件系统接口封装

# erp_client.py
import requests

class ERPClient:
    def __init__(self, base_url, api_key):
        self.base_url = base_url.rstrip('/')
        self.headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    
    def check_stock(self, part_code):
        try:
            response = requests.get(f"{self.base_url}/api/parts/{part_code}/stock", headers=self.headers, timeout=10)
            if response.status_code == 200:
                data = response.json()
                return {"available": data.get("quantity", 0) > 0, "quantity": data.get("quantity", 0)}
            return {"available": False, "quantity": 0}
        except Exception as e:
            print(f"ERP 查询失败: {e}")
            return {"available": False, "quantity": 0}
    
    def reserve_part(self, part_code, quantity, work_order_id=None):
        try:
            payload = {"part_code": part_code, "quantity": quantity, "work_order_id": work_order_id}
            response = requests.post(f"{self.base_url}/api/reservations", headers=self.headers, json=payload, timeout=10)
            if response.status_code in (200, 201):
                return {"success": True, "stock_available": True, "message": "预留成功"}
            return {"success": False, "stock_available": False, "message": response.text}
        except Exception as e:
            return {"success": False, "stock_available": False, "message": str(e)}

4.4 维修执行终端（Streamlit 前端，修正 session_state key）

# maintenance_terminal.py
import streamlit as st
from datetime import datetime

st.set_page_config(page_title="维修执行终端", layout="wide")
st.title("🔧 维修执行终端")

if "current_wo" not in st.session_state:
    st.session_state.current_wo = None
if "current_step" not in st.session_state:
    st.session_state.current_step = 0
if "step_signatures" not in st.session_state:
    st.session_state.step_signatures = {}

def load_work_order(wo_id):
    return {
        "id": wo_id,
        "device": "HMC-80",
        "fault": "主轴轴承润滑不良（油雾喷嘴堵塞）",
        "steps": [
            {
                "step_number": 1,
                "title": "检查油雾喷嘴",
                "operation": "关闭油雾系统 → 拆下 3# 喷嘴 → 用酒精清洗 → 压缩空气吹通",
                "tools": ["内六角扳手（3mm）", "超声波清洗机", "压缩空气枪"],
                "estimated_duration_min": 20,
                "safety_notes": ["确认油雾系统已泄压"],
                "requires_loto": False,
                "status": "pending"
            },
            {
                "step_number": 2,
                "title": "检查轴承润滑状态",
                "operation": "手动转动主轴感受阻力 → 检查润滑油流量和颜色",
                "tools": ["听诊器", "流量计"],
                "estimated_duration_min": 15,
                "safety_notes": ["LOTO 锁定挂牌"],
                "requires_loto": True,
                "status": "pending"
            }
        ],
        "safety_checklist": [
            {"id": "SAF-001", "item": "执行 LOTO", "mandatory": True, "verified": False},
            {"id": "SAF-002", "item": "确认设备冷却", "mandatory": True, "verified": False}
        ]
    }

wo_id = st.text_input("工单号", placeholder="WO-20241115-ABC123")
if st.button("加载工单") and wo_id:
    st.session_state.current_wo = load_work_order(wo_id)
    st.session_state.current_step = 0
    st.session_state.step_signatures = {}

if st.session_state.current_wo:
    wo = st.session_state.current_wo
    
    col1, col2, col3 = st.columns(3)
    with col1:
        st.metric("设备", wo["device"])
    with col2:
        st.metric("故障", wo["fault"])
    with col3:
        st.metric("步骤数", len(wo["steps"]))
    
    st.divider()
    
    st.subheader("🛡️ 安全检查清单")
    for item in wo["safety_checklist"]:
        checked = st.checkbox(f"{item['item']} {'🔴 强制' if item['mandatory'] else ''}", key=f"safety_{item['id']}")
        if checked and not item.get("verified"):
            item["verified"] = True
            st.success(f"✅ {item['item']} 已确认")
    
    st.divider()
    
    st.subheader("📋 维修步骤")
    for step in wo["steps"]:
        with st.expander(f"步骤 {step['step_number']}: {step['title']}", expanded=(step['step_number'] - 1 == st.session_state.current_step)):
            col_a, col_b = st.columns([2, 1])
            with col_a:
                st.write(f"**操作**: {step['operation']}")
                st.write(f"**工具**: {', '.join(step['tools'])}")
                st.write(f"**预计时间**: {step['estimated_duration_min']} 分钟")
                for note in step['safety_notes']:
                    st.warning(note)
            with col_b:
                if step['status'] == 'pending':
                    if st.button(f"▶️ 开始步骤 {step['step_number']}", key=f"start_{step['step_number']}"):
                        step['status'] = 'in_progress'
                        st.session_state.current_step = step['step_number'] - 1
                        st.rerun()
                elif step['status'] == 'in_progress':
                    torque = st.number_input("扭矩值 (N·m)", key=f"torque_{step['step_number']}")
                    gap = st.number_input("间隙值 (mm)", key=f"gap_{step['step_number']}")
                    temp = st.number_input("温度 (°C)", key=f"temp_{step['step_number']}")
                    signature = st.text_input("电子签名（输入姓名确认）", key=f"sig_{step['step_number']}")
                    if st.button(f"✅ 完成步骤 {step['step_number']}", key=f"complete_{step['step_number']}"):
                        if signature:
                            step['status'] = 'completed'
                            st.session_state.step_signatures[step['step_number']] = {
                                "signature": signature,
                                "torque": torque,
                                "gap": gap,
                                "temp": temp,
                                "completed_at": datetime.now().isoformat()
                            }
                            st.success(f"步骤 {step['step_number']} 已完成")
                            st.session_state.current_step = min(step['step_number'], len(wo["steps"]) - 1)
                            st.rerun()
                        else:
                            st.error("需要电子签名确认")
                elif step['status'] == 'completed':
                    sig_info = st.session_state.step_signatures.get(step['step_number'], {})
                    st.success(f"✅ 已完成 (签名: {sig_info.get('signature', 'N/A')})")
    
    st.divider()
    
    all_completed = all(s['status'] == 'completed' for s in wo['steps'])
    safety_verified = all(s.get('verified', False) for s in wo['safety_checklist'])
    
    if all_completed and safety_verified:
        st.success("🎉 所有步骤已完成，安全检查已通过")
        if st.button("🔒 关闭工单", type="primary"):
            st.balloons()
            st.success("工单已关闭，维修记录已归档")
            st.info("📚 正在更新知识图谱...")
    elif not safety_verified:
        st.error("⚠️ 安全检查未完成，不能关闭工单")

4.5 知识闭环更新（修正 props 中排除 id）

# knowledge_feedback.py
from neo4j import GraphDatabase

class KnowledgeFeedback:
    def __init__(self, uri, user, password):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))
    
    def update_after_repair(self, work_order, execution_result):
        with self.driver.session() as session:
            # 排除 id 避免重复设置
            props = {k: v for k, v in {
                "device": work_order.get("device"),
                "fault": work_order.get("fault"),
                "completed_at": execution_result.get("completed_at"),
                "duration_min": execution_result.get("actual_duration", 0),
                "technician": execution_result.get("technician"),
                "source": "FED_System"
            }.items() if k != "id"}
            
            session.run("""
                MERGE (r:REPAIR_RECORD {id: $wo_id})
                SET r += $props
            """, wo_id=work_order["id"], props=props)
            
            session.run("""
                MATCH (r:REPAIR_RECORD {id: $wo_id})
                MATCH (f:FAULT {name: $fault})
                MERGE (r)-[:RESOLVED]->(f)
            """, wo_id=work_order["id"], fault=work_order["fault"])
            
            if execution_result.get("success", False):
                session.run("""
                    MATCH (f:FAULT {name: $fault})
                    SET f.confidence = CASE WHEN f.confidence IS NULL THEN 0.7 ELSE f.confidence + 0.05 END
                """, fault=work_order["fault"])
            
            print(f"✅ 知识图谱已更新，工单 {work_order['id']} 已归档")
    
    def close(self):
        self.driver.close()

4.6 端到端流程演示

# main_execution.py
import json
from symptom_parser import SymptomParser
from reasoner import MaintenanceReasoner
from solution_generator import SolutionGenerator
from execution_orchestrator import ExecutionOrchestrator
from knowledge_feedback import KnowledgeFeedback

class MockCMMS:
    def create_work_order(self, wo):
        print(f"  📋 CMMS: 创建工单 {wo['id']}")
        return wo

class MockERP:
    def reserve_part(self, part_code, quantity, work_order_id=None):
        print(f"  📦 ERP: 预留备件 {part_code} x{quantity} (工单: {work_order_id})")
        return {"success": True, "stock_available": True, "message": "预留成功"}

class MockNotify:
    def send(self, recipient, title, message):
        print(f"  📢 通知: 发送给 {recipient} - {title}")

def main():
    print("=" * 60)
    print("  FED 维修执行闭环系统 - 端到端演示")
    print("=" * 60)
    
    user_input = "HMC-80 主轴加速到 8000rpm 有嘶嘶声，AL-72 报警"
    print(f"\n👤 输入: {user_input}\n")
    
    print("─" * 40)
    print("  Step 1: 症状解析")
    print("─" * 40)
    parser = SymptomParser()
    parsed = parser.parse(user_input)
    print(f"  ✅ 解析完成\n")
    
    print("─" * 40)
    print("  Step 2: 故障推理")
    print("─" * 40)
    reasoner = MaintenanceReasoner("bolt://localhost:7687", "neo4j", "password")
    diagnosis = reasoner.diagnose(user_input, parsed)
    primary = diagnosis["diagnosis"]["primary_diagnosis"]
    print(f"  ✅ 诊断: {primary['fault'] if primary else 'N/A'}\n")
    
    print("─" * 40)
    print("  Step 3: 方案生成")
    print("─" * 40)
    generator = SolutionGenerator()
    solution = generator.generate(diagnosis)
    print(f"  ✅ 方案生成完成 ({len(solution)} 字符)\n")
    
    print("─" * 40)
    print("  Step 4: 执行编排")
    print("─" * 40)
    orchestrator = ExecutionOrchestrator(MockCMMS(), MockERP(), MockNotify())
    execution = orchestrator.orchestrate(diagnosis, solution)
    print(f"  ✅ 工单: {execution['work_order']['id']}")
    print(f"  ✅ 步骤: {len(execution['steps'])} 步")
    print(f"  ✅ 备件预留: {len(execution['parts_reservation'])} 项")
    print(f"  ✅ 分配: {execution['assignee']['name']}\n")
    
    print("─" * 40)
    print("  Step 5: 维修执行（模拟）")
    print("─" * 40)
    for step in execution['steps']:
        print(f"  ▶ 步骤 {step['step_number']}: {step['title']} ✅")
    print(f"\n  🔒 工单关闭\n")
    
    print("─" * 40)
    print("  Step 6: 知识闭环更新")
    print("─" * 40)
    feedback = KnowledgeFeedback("bolt://localhost:7687", "neo4j", "password")
    execution_result = {
        "completed_at": "2024-11-15T14:30:00",
        "actual_duration": 25,
        "technician": "小李",
        "success": True,
        "feedback": "方案准确，3# 喷嘴确实堵了",
        "rating": 5
    }
    feedback.update_after_repair(execution["work_order"], execution_result)
    feedback.close()
    
    print("\n" + "=" * 60)
    print("  ✅ 全流程完成：从故障描述到知识沉淀，闭环已通")
    print("=" * 60)
    
    reasoner.close()

if __name__ == "__main__":
    main()

---

五、实际运行效果

系统上线后，一个完整的维修流程时间线：

14:00  主轴异响，小李输入症状
14:01  系统解析 + 推理 + 生成方案（30秒）
14:01  自动创建工单 WO-20241115-001
14:01  自动预留备件（FAG 7014 ×2，喷嘴 ×1）
14:01  推送到小李手机："新工单已分配"
14:05  小李到达现场，打开平板确认工单
14:05  执行安全检查（LOTO 拍照确认）
14:06  步骤 1：拆检 3# 喷嘴（15分钟）
14:21  步骤 2：检查轴承润滑（5分钟，正常）
14:25  完成，电子签名，工单关闭
14:25  系统自动更新知识图谱
14:26  老张收到通知："HMC-80 维修完成，耗时 25 分钟"

对比之前：

环节	之前	现在	节省
故障判断	30-60 分钟	30 秒	~50 分钟
建工单	10 分钟	自动	10 分钟
备件领料	15 分钟	自动预留	15 分钟
方案查阅	翻笔记/问人	平板指引	~20 分钟
记录归档	事后补写	自动回填	10 分钟
总计	约 2 小时	约 30 分钟	约 1.5 小时

---

六、安全围栏设计

FED 经验 #6：工业 AI 的安全设计不是"加个确认按钮"就完了。需要分层防御。

层级	机制	实现
L1 物理层	LOTO 锁定挂牌	系统生成检查单，人工拍照确认
L2 系统层	操作权限控制	维修工只能看到分配给自己的工单
L3 逻辑层	步骤依赖	前一步未完成，后一步不可开始
L4 数据层	参数校验	扭矩/间隙超出范围时告警
L5 审计层	全链路日志	每一步操作记录时间、人员、参数

---

七、技术栈全景

环节	技术	说明
推理引擎	第四篇全部	症状解析 → 推理 → 方案生成
工单系统	CMMS API / 自研	工单创建、流转、关闭
备件系统	ERP API	库存查询、预留、扣减
通知系统	企业微信/钉钉 Webhook	任务推送、超时提醒
执行终端	Streamlit + ECharts	平板/手机端步骤指引
知识闭环	Neo4j	维修记录归档、置信度更新
安全审计	日志 + 电子签名	全链路可追溯

requirements.txt（完整版）：

# 前四篇依赖
paddleocr>=2.7.0
pdfplumber>=0.10.0
neo4j>=5.0.0
chromadb>=0.4.0
sentence-transformers>=2.2.0
streamlit>=1.30.0
openai>=1.0.0
python-dotenv>=1.0.0
pydantic>=2.0.0

# 本篇新增
requests>=2.31.0

---

八、FED 经验总结

8.1 自动执行的"三不原则"

1. 不替代人的判断：AI 生成方案，人做最终决策
2. 不跳过安全步骤：LOTO、双人确认等安全流程不可自动化
3. 不脱离现有系统：对接 CMMS/ERP，不另起炉灶

8.2 闭环设计的"三要原则"

1. 要反馈：每个方案执行后收集结果，用于知识更新
2. 要度量：跟踪 MTTR（平均修复时间）变化，证明价值
3. 要迭代：系统准确率从 60% 到 82% 不是终点，持续优化

8.3 落地策略的"三步走"

第一步：工单自动创建（最小闭环，1 周）
  → 证明价值，让车间看到效率提升

第二步：步骤指引 + 参数校验（2-3 周）
  → 提升执行质量，减少人为失误

第三步：知识闭环 + 置信度自更新（持续）
  → 系统越用越准，形成飞轮

---

九、结语

系统上线一个月后，老张找我：“你知道最让我意外的是什么吗？”

“什么？”

“小李现在处理故障，比老王退休前还快。”

我愣了一下。

“不是因为他比老王厉害。是因为他站在老王的肩膀上——加上你的系统。”

FED 交付的不是自动化，是放大。 把一个人的经验放大成整个团队的能力，把一次维修放大成可复用的知识。

王师傅退休那天说的话，现在每天在车间回响：“以后有问题问电脑。”

电脑里，是他的脑子。电脑外，是一群年轻人正在用它解决他没见过的故障。

⚠️ 本文案例为典型工业场景综合改编，技术方案需结合实际工况评估；涉及工控操作须遵守 LOTO 规范；AI 不直接控制物理设备，执行动作由人工确认后完成。

---

下一篇预告：《FED 实战：五篇回顾与工业 AI 落地方法论——从单点突破到体系化复制》