简介数字化工厂建设正在从单点系统集成走向AI大模型驱动的全链路协同这份PPT方案面向制造业管理者、数字化转型规划人员及智能制造从业者系统展示了如何融合MES、WMS、SCADA与IoT技术构建人机一体化的智慧工厂。内容以工业4.0为背景从MESWMSSCADAIoT系统联动入手贯穿AI数据驱动闭环、动态排产与产能预测、AGV路径规划与库存优化、全流程异常预警、安全风险识别及实施保障等关键环节可帮助企业打通信息孤岛、优化资源配置并提升响应效率。资源为1个pptx文件压缩包仅614KB结构清晰、章节完整适合在项目汇报或内部培训中直接参考改编。已有37人浏览学习对于正在规划数字化工厂或撰写技术方案的人来说可快速获取一套涵盖技术架构、算法应用与实施路径的框架性思路。1. AI大模型赋能数字化工厂从PPT到产线先想清楚这三件事AI大模型赋能数字化工厂MESWMSSCADAIoT人机一体化智慧解决方案听起来是“给工厂装个脑子”实际做起来是“给老系统配一个会说话的分析师”。大模型不直接拧阀门、不下发工单它通过统一接口读MES的工单和返修记录、读WMS的库存库位、读SCADA的实时点位、读IoT的时序趋势然后把结果用对话方式给到操作员和工艺员真正执行的还是原来的系统。这套方案适合正在做数字化工厂规划的甲方、总包集成商和MES/WMS实施工程师。我给你的第一句实话是别急着选大模型和显卡先确认四套系统的数据能不能在一个地方汇齐这决定了项目是三个月交付还是三年烂尾。2. 先解决数据底座MES/WMS/SCADA/IoT 怎么变成大模型能读的统一接口2.1 四套系统的数据长什么样为什么不能直接喂给大模型做个最简单的盘点四套系统的数据特性和访问方式完全不同这也是方案里最容易被低估的部分。系统典型数据常见访问方式大模型直接用的痛点MES工单、工艺路线、返工返修记录、报工MySQL/Oracle 表或 WebService表结构复杂、字段缩写多不同厂商的MES字段命名差异非常大WMS物料库存、库位、出入库单、批次单体服务 API问一句“料在哪”要跨库存、库位、订单三张表才能回答SCADA设备点位、实时值、报警、事件OPC UA / Modbus TCP / 组态软件数据库点位编号七零八落报警文本机器味重模型直接读就乱说IoT设备时序、能耗、温湿度MQTT / Kafka数据量大数据是百万级的时间点塞不进模型上下文所以常见做法不是让大模型直连数据库而是建一层统一数据服务。我一般会把每个系统做成只读授权账号通过视图或 API 暴露“业务上能理解”的字段比如把 F01002 映射成 workorder_no把 QTY 映射成 qty。这层“翻译”工作才是整个方案里最花时间的部分。2.2 统一数据服务层用 REST API 把老系统“翻译”给大模型这层服务不需要多复杂的框架Flask 或 FastAPI 都行关键是收敛接口、翻译字段、控制权限。下面是 MES 返修查询和 SCADA 点位查询的最小实现。# unified_data_service.py from flask import Flask, jsonify, request from db import get_mes_conn, get_scada_cache app Flask(__name__) app.route(/api/mes/rework_orders) def mes_rework_orders(): MES返修记录查询只暴露业务字段不暴露库表结构 line request.args.get(line, ).strip() shift request.args.get(shift, ).strip() if not line: return jsonify({error: line参数必填}), 400 sql SELECT workorder_no, product_code, process_code, rework_reason, status, operator, created_at FROM mes_rework_order WHERE product_line %s # 只读账号连接MES库带参数化防注入 rows get_mes_conn().query(sql, (line,)) return jsonify({items: [dict(r) for r in rows]}) app.route(/api/scada/current) def scada_current(): SCADA实时点位走采集服务缓存不直连PLC point_id request.args.get(point_id, ).strip() if not point_id: return jsonify({error: point_id参数必填}), 400 # 从缓存里拿最近值避免大模型对话直接把PLC通讯打满 value get_scada_cache().get_latest(point_id) return jsonify({point_id: point_id, value: value[value], ts: value[ts]})这段代码的意图是三层字段翻译、只读安全控制、接口收敛。MES 库一律用只读账号所有 SQL 参数化返回的 JSON 字段名要和业务术语一致比如 workorder_no 而不是 WON。SCADA 点位必须走缓存或采集服务不要直连 PLC否则操作员问几句话控制器通讯就被占满了。为了让大模型知道调用这个 API还需要给一份符合 OpenAI 兼容格式的工具描述告诉模型“什么时候调、传什么参数”。tools [ { type: function, function: { name: query_mes_rework, description: 按产线和班次查询MES返修工单返回工单号、返修原因、状态, parameters: { type: object, properties: { line: {type: string, description: 产线名如 A线}, shift: {type: string, description: 班次如 早班/晚班} }, required: [line] } } } ]工具描述里的字段名必须和统一数据服务返回的字段名完全一致这是防幻觉的第一道闸门。模型拿到这个描述才会去调 API而不是自己“脑补”工单状态。2.3 时序数据进时序库工艺文档走向量库别全塞给大模型IoT 和 SCADA 的数据是时间序列不适合都堆进 MySQL否则表膨胀得厉害。常见做法是搭一个 TDengine 或 InfluxDB 存点位历史按设备 ID 加点位打标签。工艺文件、SOP、设备手册这类文档则进向量库供 RAG 检索。数据类型存储位置喂给大模型的方式结构化业务数据MES/WMSPostgreSQL/MySQL统一数据服务 REST API function calling实时点位与报警SCADARedis 缓存 TDengine 历史先查最近值和趋势再格式化成文本时序设备数据IoTTDengine / Kafka滑动窗口聚合后再送模型工艺文档/设备手册向量数据库RAG 切块后只取 Top K我一般会劝项目少做“把所有业务数据喂给大模型做训练”的尝试成本高、周期长除非你已经有几千条带标注的问答对。绝大多数工厂更适合“API 取数 RAG 查文档 大模型组织语言”的组合改动小、见效快。2.4 部署选型API 模型先跑通流程本地部署大模型再谈安全很多人第一句就问“用什么卡”我的建议正相反先用 API 把流程跑通确定场景有价值了再评估要不要本地部署大模型。方案数据边界延迟成本适合阶段云端大模型 API数据出内网信息安全评审难过快但受网络影响按 token 付费初期便宜原型验证、场景打磨本地部署大模型数据不出厂取决于显卡和模型大小一次性硬件投入生产试用、长期稳定本地小模型7B 量化数据不出厂快一张 24G 显存卡就能跑简单问答、告警摘要本地大模型13B/14B 以上量化数据不出厂较慢需要更大显存或多卡复杂业务决策辅助拿本地部署大模型的常见配置来说14B 量化后大概需要 14GB 显存单张 RTX 4090 能跑日常问答要是同时接几十个工位的并发请求单卡顶不住前面还得加网关做队列和限流。记住这个顺序先把业务逻辑调通再谈硬件投入。3. 用SSE流式输出和abort把大模型嵌进MES操作台交互层怎么封装3.1 为什么选 SSE 而不是 WebSocket 或轮询大模型回复是“一串文字按顺序生成”的最适合用 SSE。SSE 本质是 HTTP 长连接服务端持续输出data:行前端边收边渲染。原生 EventSource 只支持 GET生产环境通常用 fetch POST 加 ReadableStream 模拟 SSE这样可以把用户问题和请求参数放进 POST 体里也方便后续做 abort 取消。WebSocket 双向长连接在产线网络里要额外维护心跳和反向推送轮询则浪费资源都不如 SSE 直接。3.2 前端用 fetch 流式渲染大模型回答停止回答就 abortMES 操作台右侧的 AI 助手面板核心逻辑是“发问题、流式收、增量渲染、随时停止”。// mes_chat.js —— MES操作台AI助手面板 let currentController null; async function sendMessage() { const text document.getElementById(user-input).value.trim(); if (!text || currentController) return; const controller new AbortController(); currentController controller; const resp await fetch(/api/chat/stream, { method: POST, signal: controller.signal, headers: { Content-Type: application/json }, body: JSON.stringify({ message: text, sessionId: mes-op-001, userId: zhang }) }); const reader resp.body.getReader(); const decoder new TextDecoder(utf-8); let answer ; while (true) { const { done, value } await reader.read(); if (done) break; answer decoder.decode(value, { stream: true }); // 用innerText而不是innerHTML防止模型输出里带标签注入到页面 document.getElementById(chat-content).innerText answer; } currentController null; saveToHistory(answer); } function stopAnswer() { // 用户点了“停止回答”断开前端读取后端随后也会感知 currentController?.abort(); currentController null; }逻辑说明每次发送新问题时先 abort 上一次未结束的请求避免两个流同时在界面上抢着渲染。signal: controller.signal必须传给 fetch否则 abort 无效。解码时用decoder.decode(value, { stream: true })否则中文字符在分块时会被截断出现乱码。渲染用 innerText 而不是 innerHTML这一步能在前端兜底防住模型输出内容里的 HTML 注入。3.3 后端FastAPI 转发大模型流abort 要一路传到底后端只是一个转发层拿到的用户问题拼好系统提示词再转发给推理服务。这里的关键是客户端断开时转发请求也要一起取消否则模型还在后台继续生成白烧显存。# chat_stream.py —— 统一聊天入口鉴权、拼装prompt、转发SSE import httpx from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app FastAPI() async def forward_llm_stream(messages: list): 把请求转发给本地推理服务逐行透传SSE url http://127.0.0.1:8000/v1/chat/completions payload { model: factory-14b, messages: messages, stream: True, temperature: 0.2, max_tokens: 1024 } async with httpx.AsyncClient(timeoutNone) as client: try: async with client.stream(POST, url, jsonpayload) as resp: async for line in resp.aiter_lines(): if line.startswith(data:): yield line \n except asyncio.CancelledError: # 客户端断开连接停止转发避免下游推理服务空转 raise app.post(/api/chat/stream) async def chat_stream(request: Request): body await request.json() user_msg body.get(message, ).strip() if not user_msg: return {error: message不能为空} # 服务端拼装系统提示词避免模型被奇怪输入带偏 messages [ {role: system, content: 你是工厂运营助手回答必须基于工具返回的数据不编造工单号和点位值。}, {role: user, content: user_msg} ] return StreamingResponse( forward_llm_stream(messages), media_typetext/event-stream )逻辑说明FastAPI 的 StreamingResponse 在客户端断开时会向生成器抛 CancelledError这里补一个捕获并重新抛出保证取消信号能传播到下游。如果你用的推理服务不支持流式中断客户端关掉连接后模型可能还在继续生成血泪经验是一定要在推理网关前面加一层“会话取消”逻辑检测到连接断开就标记对应推理请求为取消。技术栈封装 AI 交互逻辑的时候这一层最容易被漏掉。3.4 必调参数温度、超时、上下文窗口和并发数参数推荐值说明temperature0.1~0.3工厂业务要确定性温度高了同一个问题答案忽东忽西车间主任会直接不认max_tokens512~1024MES 场景回答不需要长篇大论超过 1024 大多是在绕首字超时20 秒超过就给“系统繁忙请先查看 SCADA 组态图”的兜底文案上下文窗口最近 10 轮超过就把旧消息压缩成摘要再继续省 token 也省显存并发连接数单卡 2~4 路按 GPU 显存算其他请求排队不要无脑往上怼这里要特别强调温度。生产问答必须低温我习惯固定在 0.2 左右。温度设到 0.7同一条“A 线还有几个返修单”模型上午答 3 个下午答 5 个这种黑匣子行为在工厂完全不可接受。4. AI大模型在工厂里的四个落地方向返修查询、告警语义化、找料、异常预判4.1 MES 返工返修查询让模型只会“说话”不会“编数”操作员问“A 线早班还有几个返修工单没处理”这背后要查 MES 返修表还要按产线、班次、状态过滤。我的方案是先用 function calling 让模型把自然语言转成参数后端代码执行查询模型看到的只有查询结果不接触原始表。# mes_rework_agent.py 核心片段 from openai import OpenAI import requests client OpenAI(base_urlhttp://127.0.0.1:8000/v1, api_keyfactory-local) def query_rework_by_line(line: str, shift: str ) - str: 调用统一数据服务的MES返修接口返回JSON字符串 params {line: line} if shift: params[shift] shift resp requests.get( http://10.0.1.20:8801/api/mes/rework_orders, paramsparams, timeout3 ) return resp.text tools [{ type: function, function: { name: query_rework_by_line, description: 查询MES返修工单列表返回工单号、状态、返修原因, parameters: { type: object, properties: { line: {type: string, description: 产线编号如 A线}, shift: {type: string, description: 班次如 早班/晚班} }, required: [line] } } }] resp client.chat.completions.create( modelfactory-14b, messages[{role: user, content: A线早班还有几个返修工单没处理}], toolstools, tool_choiceauto, temperature0.2, max_tokens512 )逻辑说明这里的返回不是最终答案而是模型选中了query_rework_by_line这个函数。你要自己去调这个函数把真实结果再送回去让模型组织自然语言。这样即使模型后面说错工单号和数量也都来自 MES 返回不会凭空编造。关键参数是 temperature0.2如果模型在这步开始“发挥”参数就抽不对整个链路就废了。还有个细节工具描述里的“返修”要和 MES 业务术语完全一致否则模型会去查一个不存在的字段名。4.2 SCADA 告警语义化千条报警变成一句人话SCADA 一天产生上千条报警值班室不可能逐条看。AI 大模型接收原始报警记录输出类似“三号空压机排气压力偏高过去 1 小时从 0.72 升到 0.86建议检查冷却风扇”。这个场景的关键是点位映射表和防幻觉提示词。# scada_alert_prompt.py PROMPT 你是车间设备诊断助手。下面是一条SCADA报警记录请按以下规则输出 1. 只使用提供的字段和点位映射表不猜测设备或点位编号 2. 如果点位映射表中查不到该点位回答点位未知请查看SCADA组态图 3. 输出控制在50字以内给出建议检查项不给操作命令 报警记录 {alert_json} 点位映射表 {point_mapping} 点位映射表要建全包含点位编号、设备名、点位名称、单位、正常范围。模型一旦查到映射表之外必须让它停下来。生成之后还要加一道后置校验把答案里出现的设备名、点位编号和原始报警记录逐一比对对不上就打回重生成。这个场景我吃过亏模型把“气缸没到位”答成“传感器损坏”老师傅拆了半天最后发现只是机械卡滞。所以告警语义化只做“信息翻译”不做“终极诊断”。4.3 WMS 对话式找料查询只在库位列表里选不允许模型自由发挥仓库管理员问“这批工单要用的 Φ10 铜管在哪个库位、有多少”正确的链路是大模型解析出物料编码后端去 WMS 查库存结果直接映射为“库位Z3-02可用180 件”库存不够时再提示替代批次。这里有一个实际边界同一个物料编码可能有多个批次不同批次的库位和数量都不一样。模型必须在返回的批次列表里选不能在提示词里自作主张选一个。单位换算也是箱和件的换算由 WMS 接口返回模型不负责计算。凡是在对话里出现数量、库位、批次都必须问一句“数据来自哪次查询”拿不出来就打回。4.4 IoT 异常预判滑动窗口先筛模型只写摘要电机振动值持续走高把原始时序数据直接丢给大模型不现实token 不够模型也看不出规律。我一般做法是先用规则引擎做滑动窗口聚合比如连续 15 分钟内振动均值超过阈值 5%判定为“预警”。然后把聚合特征喂给模型让它结合历史维修记录给一句话建议。参数推荐值说明滑动窗口15 分钟太短误报多太长响应慢特征字段均值、最大值、斜率不传原始点传特征能省大量 token触发条件均值超阈值 5% 且持续 2 个窗口先保守再放宽稳定了再降阈值输出要求1 句结论 1 个建议动作防止生成没人看的长篇报告模型输出“振动趋势上升建议周末保养时检查轴承”这是决策辅助。真正触发保养工单还是 MES 的计划模块来执行。这个边界在方案里要写清楚大模型的判断只是参考不是自动执行指令。5. 数字化工厂接大模型必踩的五个坑现象、原因、后悔药5.1 模型把工单号编出来了现象让大模型查“MES-20250609-0013 这批铜管的返修记录”它回答得有板有眼工单号、操作员、时间全都有实际上这个工单根本不存在。原因生成式模型为了流畅会在没数据的地方把细节“补全”。只要没有强制约束“回答内容必须出现在工具返回结果里”它就觉得编一个更顺。解决所有涉及 MES/WMS/SCADA 数据的回答要求模型只能复述工具返回里的字段。后端再加一道过滤把回答中出现的工单号、物料编码、点位编号与真实返回值逐一比对不在列表里的直接打回重生成。这道后置校验是必须写的别让前端裸奔。5.2 SCADA 告警对话上线一周没人敢用现象接线师傅问“#3 空压机为什么报警”模型答复“传感器损坏建议更换”师傅拆开一看只是冷却风扇轮片被异物卡住。原因模型只拿到一条报警编号没有上下文。它不知道这台设备前 1 小时的压力趋势也不知道上一次故障是怎么处理的。单点信息喂给再好的模型也只能猜。解决把告警事件做成一包数据连同设备编号、点位、当前值、最近 1 小时趋势、最近一次维护记录一起丢进 prompt。同时限制输出为“建议检查项”而不是“诊断结论”诊断结论留给老师傅拍板。这一步做完可信度会明显不一样。5.3 SSE 在车间弱网下频繁中断操作台白屏现象车间角落 Wi-Fi 信号差AI 回答到一半连接断开界面上只留下半句话再点发送也没反应。原因abort 了上一次连接但没重置状态也没做断线重连。前端没捕获读取异常一个抛出就把整个交互线程打挂了。解决fetch 的 ReadableStream 读取要加异常捕获和重试断线时提示“网络不稳定已显示部分内容”重试最多 3 次。每次请求必须新建 AbortController请求结束时及时置空避免下一次发送被上一次的“取消状态”卡住。SSE 流收完第一帧后可以加一个 15 秒心跳检查没有新数据就主动重连。5.4 本地部署小模型答非所问架子搭了白搭现象为了省成本上了个 7B 量化模型结果接 SQL 工具时参数都抽不对返回结果也是错乱最后只能把整个方案否掉。原因工厂场景不是通用对话是“从自然语言到工具调用”的专项任务。7B 模型的指令遵循能力在这个难度的任务上不够用不是 prompt 写得不好是模型确实理解不了复杂工具链。解决如果坚持本地部署大模型至少用 13B/14B 量化后的版本。更稳的路径是先用 API 大模型做功能验证把工具链和产品流程调通再评估要不要本地化。数据敏感没法出内网的情况可以先用本地小模型做“告警摘要”这类简单的单点任务复杂推理再另寻路径。别指望一个模型包打天下这是做项目最大的误区。5.5 知识库刚更新模型却按旧工艺回答现象上午工艺员更新了返工 SOP下午问同样的问题模型给的还是旧温度参数差点导致操作工按错误工艺作业。原因RAG 的向量库更新要重新切块和构建索引。很多项目做知识库增量更新时旧向量没被覆盖或者检索时没有按版本过滤旧的片段照样被命中。解决文档入库时带上版本号和生效日期检索条件强制“版本号最新且生效日期今天”。对时效敏感的参数比如温度、压力范围不要走向量检索直接走结构化字段查询。最后每天凌晨做一次向量库全量重建最迟三天一次。老化的知识库比没有知识库更危险。6. 验证这套方案值不值三个验收动作定生死6.1 先跑一个最小评测脚本别用肉眼判断效果方案上线前先用脚本量化“回答准不准”。# eval_factory_agent.py queries [ {q: A线早班有几个返修工单, expect: 3}, {q: PT-101点位当前值是多少, expect: 0.86}, {q: Φ10铜管在哪个库位, expect: Z3-02}, ] def run(): hit 0 for item in queries: answer call_agent(item[q]) # 你的Agent入口 # 关键数字出现在回答里就算命中不能让模型光说“查到了” hit 1 if item[expect] in answer else 0 print(f命中 {hit}/{len(queries)})逻辑说明评测不能只看“有没有回答”要看回答里的关键数字是否出现。我建议每个系统至少准备 25 条带标准答案的回归集每次升级模型或改提示词都跑一遍。没跑过回归就直接上线的都是赌博。6.2 通过验收的三个硬指标验收指标目标值测量方式业务准确率关键字段命中率不低于 95%用上面的评测脚本对回归集评分首字响应延迟SSE 首个字符不超过 2 秒前端打点记录从发送到第一帧渲染的时间现场使用率连续 5 天每天 10 次以上单个场景使用率超过 70%操作台埋点统计说实话我见过太多方案卡在第一步项目启动会上大家讨论“要不要自己训练大模型”最后才发现 MES 的接口文档还没拿到。先把数据底座、工具链路、评测脚本这三样东西立起来模型换哪个都只是换脑子不换骨架。大模型是那个会说话的实习生老师傅是 MES/WMS/SCADA实习生的价值是让老师傅的经验被快速找到、说给人听。希望帮到你。本文还有配套的精品资源点击获取 AI大模型赋能数字化工厂：统一数据接口与SSE流式问答实战
简介数字化工厂建设正在从单点系统集成走向AI大模型驱动的全链路协同这份PPT方案面向制造业管理者、数字化转型规划人员及智能制造从业者系统展示了如何融合MES、WMS、SCADA与IoT技术构建人机一体化的智慧工厂。内容以工业4.0为背景从MESWMSSCADAIoT系统联动入手贯穿AI数据驱动闭环、动态排产与产能预测、AGV路径规划与库存优化、全流程异常预警、安全风险识别及实施保障等关键环节可帮助企业打通信息孤岛、优化资源配置并提升响应效率。资源为1个pptx文件压缩包仅614KB结构清晰、章节完整适合在项目汇报或内部培训中直接参考改编。已有37人浏览学习对于正在规划数字化工厂或撰写技术方案的人来说可快速获取一套涵盖技术架构、算法应用与实施路径的框架性思路。1. AI大模型赋能数字化工厂从PPT到产线先想清楚这三件事AI大模型赋能数字化工厂MESWMSSCADAIoT人机一体化智慧解决方案听起来是“给工厂装个脑子”实际做起来是“给老系统配一个会说话的分析师”。大模型不直接拧阀门、不下发工单它通过统一接口读MES的工单和返修记录、读WMS的库存库位、读SCADA的实时点位、读IoT的时序趋势然后把结果用对话方式给到操作员和工艺员真正执行的还是原来的系统。这套方案适合正在做数字化工厂规划的甲方、总包集成商和MES/WMS实施工程师。我给你的第一句实话是别急着选大模型和显卡先确认四套系统的数据能不能在一个地方汇齐这决定了项目是三个月交付还是三年烂尾。2. 先解决数据底座MES/WMS/SCADA/IoT 怎么变成大模型能读的统一接口2.1 四套系统的数据长什么样为什么不能直接喂给大模型做个最简单的盘点四套系统的数据特性和访问方式完全不同这也是方案里最容易被低估的部分。系统典型数据常见访问方式大模型直接用的痛点MES工单、工艺路线、返工返修记录、报工MySQL/Oracle 表或 WebService表结构复杂、字段缩写多不同厂商的MES字段命名差异非常大WMS物料库存、库位、出入库单、批次单体服务 API问一句“料在哪”要跨库存、库位、订单三张表才能回答SCADA设备点位、实时值、报警、事件OPC UA / Modbus TCP / 组态软件数据库点位编号七零八落报警文本机器味重模型直接读就乱说IoT设备时序、能耗、温湿度MQTT / Kafka数据量大数据是百万级的时间点塞不进模型上下文所以常见做法不是让大模型直连数据库而是建一层统一数据服务。我一般会把每个系统做成只读授权账号通过视图或 API 暴露“业务上能理解”的字段比如把 F01002 映射成 workorder_no把 QTY 映射成 qty。这层“翻译”工作才是整个方案里最花时间的部分。2.2 统一数据服务层用 REST API 把老系统“翻译”给大模型这层服务不需要多复杂的框架Flask 或 FastAPI 都行关键是收敛接口、翻译字段、控制权限。下面是 MES 返修查询和 SCADA 点位查询的最小实现。# unified_data_service.py from flask import Flask, jsonify, request from db import get_mes_conn, get_scada_cache app Flask(__name__) app.route(/api/mes/rework_orders) def mes_rework_orders(): MES返修记录查询只暴露业务字段不暴露库表结构 line request.args.get(line, ).strip() shift request.args.get(shift, ).strip() if not line: return jsonify({error: line参数必填}), 400 sql SELECT workorder_no, product_code, process_code, rework_reason, status, operator, created_at FROM mes_rework_order WHERE product_line %s # 只读账号连接MES库带参数化防注入 rows get_mes_conn().query(sql, (line,)) return jsonify({items: [dict(r) for r in rows]}) app.route(/api/scada/current) def scada_current(): SCADA实时点位走采集服务缓存不直连PLC point_id request.args.get(point_id, ).strip() if not point_id: return jsonify({error: point_id参数必填}), 400 # 从缓存里拿最近值避免大模型对话直接把PLC通讯打满 value get_scada_cache().get_latest(point_id) return jsonify({point_id: point_id, value: value[value], ts: value[ts]})这段代码的意图是三层字段翻译、只读安全控制、接口收敛。MES 库一律用只读账号所有 SQL 参数化返回的 JSON 字段名要和业务术语一致比如 workorder_no 而不是 WON。SCADA 点位必须走缓存或采集服务不要直连 PLC否则操作员问几句话控制器通讯就被占满了。为了让大模型知道调用这个 API还需要给一份符合 OpenAI 兼容格式的工具描述告诉模型“什么时候调、传什么参数”。tools [ { type: function, function: { name: query_mes_rework, description: 按产线和班次查询MES返修工单返回工单号、返修原因、状态, parameters: { type: object, properties: { line: {type: string, description: 产线名如 A线}, shift: {type: string, description: 班次如 早班/晚班} }, required: [line] } } } ]工具描述里的字段名必须和统一数据服务返回的字段名完全一致这是防幻觉的第一道闸门。模型拿到这个描述才会去调 API而不是自己“脑补”工单状态。2.3 时序数据进时序库工艺文档走向量库别全塞给大模型IoT 和 SCADA 的数据是时间序列不适合都堆进 MySQL否则表膨胀得厉害。常见做法是搭一个 TDengine 或 InfluxDB 存点位历史按设备 ID 加点位打标签。工艺文件、SOP、设备手册这类文档则进向量库供 RAG 检索。数据类型存储位置喂给大模型的方式结构化业务数据MES/WMSPostgreSQL/MySQL统一数据服务 REST API function calling实时点位与报警SCADARedis 缓存 TDengine 历史先查最近值和趋势再格式化成文本时序设备数据IoTTDengine / Kafka滑动窗口聚合后再送模型工艺文档/设备手册向量数据库RAG 切块后只取 Top K我一般会劝项目少做“把所有业务数据喂给大模型做训练”的尝试成本高、周期长除非你已经有几千条带标注的问答对。绝大多数工厂更适合“API 取数 RAG 查文档 大模型组织语言”的组合改动小、见效快。2.4 部署选型API 模型先跑通流程本地部署大模型再谈安全很多人第一句就问“用什么卡”我的建议正相反先用 API 把流程跑通确定场景有价值了再评估要不要本地部署大模型。方案数据边界延迟成本适合阶段云端大模型 API数据出内网信息安全评审难过快但受网络影响按 token 付费初期便宜原型验证、场景打磨本地部署大模型数据不出厂取决于显卡和模型大小一次性硬件投入生产试用、长期稳定本地小模型7B 量化数据不出厂快一张 24G 显存卡就能跑简单问答、告警摘要本地大模型13B/14B 以上量化数据不出厂较慢需要更大显存或多卡复杂业务决策辅助拿本地部署大模型的常见配置来说14B 量化后大概需要 14GB 显存单张 RTX 4090 能跑日常问答要是同时接几十个工位的并发请求单卡顶不住前面还得加网关做队列和限流。记住这个顺序先把业务逻辑调通再谈硬件投入。3. 用SSE流式输出和abort把大模型嵌进MES操作台交互层怎么封装3.1 为什么选 SSE 而不是 WebSocket 或轮询大模型回复是“一串文字按顺序生成”的最适合用 SSE。SSE 本质是 HTTP 长连接服务端持续输出data:行前端边收边渲染。原生 EventSource 只支持 GET生产环境通常用 fetch POST 加 ReadableStream 模拟 SSE这样可以把用户问题和请求参数放进 POST 体里也方便后续做 abort 取消。WebSocket 双向长连接在产线网络里要额外维护心跳和反向推送轮询则浪费资源都不如 SSE 直接。3.2 前端用 fetch 流式渲染大模型回答停止回答就 abortMES 操作台右侧的 AI 助手面板核心逻辑是“发问题、流式收、增量渲染、随时停止”。// mes_chat.js —— MES操作台AI助手面板 let currentController null; async function sendMessage() { const text document.getElementById(user-input).value.trim(); if (!text || currentController) return; const controller new AbortController(); currentController controller; const resp await fetch(/api/chat/stream, { method: POST, signal: controller.signal, headers: { Content-Type: application/json }, body: JSON.stringify({ message: text, sessionId: mes-op-001, userId: zhang }) }); const reader resp.body.getReader(); const decoder new TextDecoder(utf-8); let answer ; while (true) { const { done, value } await reader.read(); if (done) break; answer decoder.decode(value, { stream: true }); // 用innerText而不是innerHTML防止模型输出里带标签注入到页面 document.getElementById(chat-content).innerText answer; } currentController null; saveToHistory(answer); } function stopAnswer() { // 用户点了“停止回答”断开前端读取后端随后也会感知 currentController?.abort(); currentController null; }逻辑说明每次发送新问题时先 abort 上一次未结束的请求避免两个流同时在界面上抢着渲染。signal: controller.signal必须传给 fetch否则 abort 无效。解码时用decoder.decode(value, { stream: true })否则中文字符在分块时会被截断出现乱码。渲染用 innerText 而不是 innerHTML这一步能在前端兜底防住模型输出内容里的 HTML 注入。3.3 后端FastAPI 转发大模型流abort 要一路传到底后端只是一个转发层拿到的用户问题拼好系统提示词再转发给推理服务。这里的关键是客户端断开时转发请求也要一起取消否则模型还在后台继续生成白烧显存。# chat_stream.py —— 统一聊天入口鉴权、拼装prompt、转发SSE import httpx from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app FastAPI() async def forward_llm_stream(messages: list): 把请求转发给本地推理服务逐行透传SSE url http://127.0.0.1:8000/v1/chat/completions payload { model: factory-14b, messages: messages, stream: True, temperature: 0.2, max_tokens: 1024 } async with httpx.AsyncClient(timeoutNone) as client: try: async with client.stream(POST, url, jsonpayload) as resp: async for line in resp.aiter_lines(): if line.startswith(data:): yield line \n except asyncio.CancelledError: # 客户端断开连接停止转发避免下游推理服务空转 raise app.post(/api/chat/stream) async def chat_stream(request: Request): body await request.json() user_msg body.get(message, ).strip() if not user_msg: return {error: message不能为空} # 服务端拼装系统提示词避免模型被奇怪输入带偏 messages [ {role: system, content: 你是工厂运营助手回答必须基于工具返回的数据不编造工单号和点位值。}, {role: user, content: user_msg} ] return StreamingResponse( forward_llm_stream(messages), media_typetext/event-stream )逻辑说明FastAPI 的 StreamingResponse 在客户端断开时会向生成器抛 CancelledError这里补一个捕获并重新抛出保证取消信号能传播到下游。如果你用的推理服务不支持流式中断客户端关掉连接后模型可能还在继续生成血泪经验是一定要在推理网关前面加一层“会话取消”逻辑检测到连接断开就标记对应推理请求为取消。技术栈封装 AI 交互逻辑的时候这一层最容易被漏掉。3.4 必调参数温度、超时、上下文窗口和并发数参数推荐值说明temperature0.1~0.3工厂业务要确定性温度高了同一个问题答案忽东忽西车间主任会直接不认max_tokens512~1024MES 场景回答不需要长篇大论超过 1024 大多是在绕首字超时20 秒超过就给“系统繁忙请先查看 SCADA 组态图”的兜底文案上下文窗口最近 10 轮超过就把旧消息压缩成摘要再继续省 token 也省显存并发连接数单卡 2~4 路按 GPU 显存算其他请求排队不要无脑往上怼这里要特别强调温度。生产问答必须低温我习惯固定在 0.2 左右。温度设到 0.7同一条“A 线还有几个返修单”模型上午答 3 个下午答 5 个这种黑匣子行为在工厂完全不可接受。4. AI大模型在工厂里的四个落地方向返修查询、告警语义化、找料、异常预判4.1 MES 返工返修查询让模型只会“说话”不会“编数”操作员问“A 线早班还有几个返修工单没处理”这背后要查 MES 返修表还要按产线、班次、状态过滤。我的方案是先用 function calling 让模型把自然语言转成参数后端代码执行查询模型看到的只有查询结果不接触原始表。# mes_rework_agent.py 核心片段 from openai import OpenAI import requests client OpenAI(base_urlhttp://127.0.0.1:8000/v1, api_keyfactory-local) def query_rework_by_line(line: str, shift: str ) - str: 调用统一数据服务的MES返修接口返回JSON字符串 params {line: line} if shift: params[shift] shift resp requests.get( http://10.0.1.20:8801/api/mes/rework_orders, paramsparams, timeout3 ) return resp.text tools [{ type: function, function: { name: query_rework_by_line, description: 查询MES返修工单列表返回工单号、状态、返修原因, parameters: { type: object, properties: { line: {type: string, description: 产线编号如 A线}, shift: {type: string, description: 班次如 早班/晚班} }, required: [line] } } }] resp client.chat.completions.create( modelfactory-14b, messages[{role: user, content: A线早班还有几个返修工单没处理}], toolstools, tool_choiceauto, temperature0.2, max_tokens512 )逻辑说明这里的返回不是最终答案而是模型选中了query_rework_by_line这个函数。你要自己去调这个函数把真实结果再送回去让模型组织自然语言。这样即使模型后面说错工单号和数量也都来自 MES 返回不会凭空编造。关键参数是 temperature0.2如果模型在这步开始“发挥”参数就抽不对整个链路就废了。还有个细节工具描述里的“返修”要和 MES 业务术语完全一致否则模型会去查一个不存在的字段名。4.2 SCADA 告警语义化千条报警变成一句人话SCADA 一天产生上千条报警值班室不可能逐条看。AI 大模型接收原始报警记录输出类似“三号空压机排气压力偏高过去 1 小时从 0.72 升到 0.86建议检查冷却风扇”。这个场景的关键是点位映射表和防幻觉提示词。# scada_alert_prompt.py PROMPT 你是车间设备诊断助手。下面是一条SCADA报警记录请按以下规则输出 1. 只使用提供的字段和点位映射表不猜测设备或点位编号 2. 如果点位映射表中查不到该点位回答点位未知请查看SCADA组态图 3. 输出控制在50字以内给出建议检查项不给操作命令 报警记录 {alert_json} 点位映射表 {point_mapping} 点位映射表要建全包含点位编号、设备名、点位名称、单位、正常范围。模型一旦查到映射表之外必须让它停下来。生成之后还要加一道后置校验把答案里出现的设备名、点位编号和原始报警记录逐一比对对不上就打回重生成。这个场景我吃过亏模型把“气缸没到位”答成“传感器损坏”老师傅拆了半天最后发现只是机械卡滞。所以告警语义化只做“信息翻译”不做“终极诊断”。4.3 WMS 对话式找料查询只在库位列表里选不允许模型自由发挥仓库管理员问“这批工单要用的 Φ10 铜管在哪个库位、有多少”正确的链路是大模型解析出物料编码后端去 WMS 查库存结果直接映射为“库位Z3-02可用180 件”库存不够时再提示替代批次。这里有一个实际边界同一个物料编码可能有多个批次不同批次的库位和数量都不一样。模型必须在返回的批次列表里选不能在提示词里自作主张选一个。单位换算也是箱和件的换算由 WMS 接口返回模型不负责计算。凡是在对话里出现数量、库位、批次都必须问一句“数据来自哪次查询”拿不出来就打回。4.4 IoT 异常预判滑动窗口先筛模型只写摘要电机振动值持续走高把原始时序数据直接丢给大模型不现实token 不够模型也看不出规律。我一般做法是先用规则引擎做滑动窗口聚合比如连续 15 分钟内振动均值超过阈值 5%判定为“预警”。然后把聚合特征喂给模型让它结合历史维修记录给一句话建议。参数推荐值说明滑动窗口15 分钟太短误报多太长响应慢特征字段均值、最大值、斜率不传原始点传特征能省大量 token触发条件均值超阈值 5% 且持续 2 个窗口先保守再放宽稳定了再降阈值输出要求1 句结论 1 个建议动作防止生成没人看的长篇报告模型输出“振动趋势上升建议周末保养时检查轴承”这是决策辅助。真正触发保养工单还是 MES 的计划模块来执行。这个边界在方案里要写清楚大模型的判断只是参考不是自动执行指令。5. 数字化工厂接大模型必踩的五个坑现象、原因、后悔药5.1 模型把工单号编出来了现象让大模型查“MES-20250609-0013 这批铜管的返修记录”它回答得有板有眼工单号、操作员、时间全都有实际上这个工单根本不存在。原因生成式模型为了流畅会在没数据的地方把细节“补全”。只要没有强制约束“回答内容必须出现在工具返回结果里”它就觉得编一个更顺。解决所有涉及 MES/WMS/SCADA 数据的回答要求模型只能复述工具返回里的字段。后端再加一道过滤把回答中出现的工单号、物料编码、点位编号与真实返回值逐一比对不在列表里的直接打回重生成。这道后置校验是必须写的别让前端裸奔。5.2 SCADA 告警对话上线一周没人敢用现象接线师傅问“#3 空压机为什么报警”模型答复“传感器损坏建议更换”师傅拆开一看只是冷却风扇轮片被异物卡住。原因模型只拿到一条报警编号没有上下文。它不知道这台设备前 1 小时的压力趋势也不知道上一次故障是怎么处理的。单点信息喂给再好的模型也只能猜。解决把告警事件做成一包数据连同设备编号、点位、当前值、最近 1 小时趋势、最近一次维护记录一起丢进 prompt。同时限制输出为“建议检查项”而不是“诊断结论”诊断结论留给老师傅拍板。这一步做完可信度会明显不一样。5.3 SSE 在车间弱网下频繁中断操作台白屏现象车间角落 Wi-Fi 信号差AI 回答到一半连接断开界面上只留下半句话再点发送也没反应。原因abort 了上一次连接但没重置状态也没做断线重连。前端没捕获读取异常一个抛出就把整个交互线程打挂了。解决fetch 的 ReadableStream 读取要加异常捕获和重试断线时提示“网络不稳定已显示部分内容”重试最多 3 次。每次请求必须新建 AbortController请求结束时及时置空避免下一次发送被上一次的“取消状态”卡住。SSE 流收完第一帧后可以加一个 15 秒心跳检查没有新数据就主动重连。5.4 本地部署小模型答非所问架子搭了白搭现象为了省成本上了个 7B 量化模型结果接 SQL 工具时参数都抽不对返回结果也是错乱最后只能把整个方案否掉。原因工厂场景不是通用对话是“从自然语言到工具调用”的专项任务。7B 模型的指令遵循能力在这个难度的任务上不够用不是 prompt 写得不好是模型确实理解不了复杂工具链。解决如果坚持本地部署大模型至少用 13B/14B 量化后的版本。更稳的路径是先用 API 大模型做功能验证把工具链和产品流程调通再评估要不要本地化。数据敏感没法出内网的情况可以先用本地小模型做“告警摘要”这类简单的单点任务复杂推理再另寻路径。别指望一个模型包打天下这是做项目最大的误区。5.5 知识库刚更新模型却按旧工艺回答现象上午工艺员更新了返工 SOP下午问同样的问题模型给的还是旧温度参数差点导致操作工按错误工艺作业。原因RAG 的向量库更新要重新切块和构建索引。很多项目做知识库增量更新时旧向量没被覆盖或者检索时没有按版本过滤旧的片段照样被命中。解决文档入库时带上版本号和生效日期检索条件强制“版本号最新且生效日期今天”。对时效敏感的参数比如温度、压力范围不要走向量检索直接走结构化字段查询。最后每天凌晨做一次向量库全量重建最迟三天一次。老化的知识库比没有知识库更危险。6. 验证这套方案值不值三个验收动作定生死6.1 先跑一个最小评测脚本别用肉眼判断效果方案上线前先用脚本量化“回答准不准”。# eval_factory_agent.py queries [ {q: A线早班有几个返修工单, expect: 3}, {q: PT-101点位当前值是多少, expect: 0.86}, {q: Φ10铜管在哪个库位, expect: Z3-02}, ] def run(): hit 0 for item in queries: answer call_agent(item[q]) # 你的Agent入口 # 关键数字出现在回答里就算命中不能让模型光说“查到了” hit 1 if item[expect] in answer else 0 print(f命中 {hit}/{len(queries)})逻辑说明评测不能只看“有没有回答”要看回答里的关键数字是否出现。我建议每个系统至少准备 25 条带标准答案的回归集每次升级模型或改提示词都跑一遍。没跑过回归就直接上线的都是赌博。6.2 通过验收的三个硬指标验收指标目标值测量方式业务准确率关键字段命中率不低于 95%用上面的评测脚本对回归集评分首字响应延迟SSE 首个字符不超过 2 秒前端打点记录从发送到第一帧渲染的时间现场使用率连续 5 天每天 10 次以上单个场景使用率超过 70%操作台埋点统计说实话我见过太多方案卡在第一步项目启动会上大家讨论“要不要自己训练大模型”最后才发现 MES 的接口文档还没拿到。先把数据底座、工具链路、评测脚本这三样东西立起来模型换哪个都只是换脑子不换骨架。大模型是那个会说话的实习生老师傅是 MES/WMS/SCADA实习生的价值是让老师傅的经验被快速找到、说给人听。希望帮到你。本文还有配套的精品资源点击获取