智联工坊实战：工业数据质量自动检测方案 3σ 原则 + Agent 编排 + 分层容错完整实践

博文导航-四阶段学习路径版持续更新 关于【智联工坊】那些事【文章摘要】针对制造业设备数据质量差导致 OEE 失真、人工排查效率低、不可复现的痛点本文基于智联工坊 50 万行真实压测数据演示从人工 2 天排查升级为 AI 自动巡检的完整落地方案。采用接入层 检测层 编排层 兜底层的四层架构覆盖空值、重复、异常、延迟、结构五类检测配套 Agent 工具调用双兜底防线附 9 类脏数据场景压测结果、选型权衡与边界局限所有结论均有注入基准可验证可直接复用到生产场景。目录开篇一个让 OEE 失真的真问题一、数据质量的代价业务问题二、人工排查的真实成本三、方案架构五项检测 Agent 编排 分层容错四、三步走实施一天落地五、真实数据验证从 1010 行到 50 万行5.1 基线验证1010 行注入基准 1:1 命中5.2 规模验证 ×503σ 误报的定量实锤5.3 压力验证50 万行「脏上加脏」9 类场景逐项过六、实践踩坑与生产级加固6.1 langchain 1.x 大版本漂移依赖只写下界pip 一升全红6.2 小模型漏调工具Agent「交卷」前没做完题6.3 压测造数的假覆盖测试全绿其实什么都没测到七、工具选型 trade-off7.1 大模型本地 Ollama Qwen2.5 vs 云 API7.2 调度APScheduler vs DolphinScheduler八、两个必须说清的局限九、实践成果与标准对标总结可复用的方法论 系列导航开篇一个让 OEE 失真的真问题本系列所有案例来自通用离散制造虚拟工厂——智联工坊。所有数据已脱敏不映射任何真实企业。上个月智联工坊的班组长小刘拿着一份 OEE 报表问我「老蒋设备上月停了两次、每次两小时OEE 怎么还能有 86%」我去翻底层数据发现传感器温度里混着 200℃ 和 -50℃ 的异常值振动数据有大段空值还有记录重复上报。用这样的数据算 OEE谁敢信发现这些问题靠数据工程师小张手动拉数据、Excel 筛空值、查重复、算标准差一轮下来快两天。本文要解决的是一个工程问题把数据质量巡检从「人工逐项排查」变成「Agent 标准化自动巡检」可复现、可定时、可审计。V1 版当时只有 1010 行的链路验证。这一版是完整实战收官工程从 1010 行跑到了50 万行主动注入了编码乱码、时间戳混乱、字段错位等 9 类「脏上加脏」场景做压力测试还揪出并修掉了小模型多工具编排的真实缺陷。全文所有数字都有注入基准可核对。【核心成果速览】・效率提升50 万行数据完整巡检从人工 2 天缩短至自动 26 秒・覆盖范围5 大类检测 9 类工业脏数据场景结构与内容分层处理・准确率验证注入基准与实跑检出 1:1 匹配结果可复现、可核对・生产级加固提示词硬约束 代码级兜底双防线解决 Agent 漏调工具问题一、数据质量的代价业务问题数据质量问题业务后果温度异常200℃/-50℃OEE 失真设备健康度误判振动整段空值预测性维护模型失效故障漏报工单重复上报产量统计虚高排产决策被误导上报延迟超阈值实时监控失效异常响应滞后编码乱码 / 字段错位数据接入即失真下游全错数据仓库和 AI 模型建得再好地基不牢都是空中楼阁。制造企业尤其要命这些数据直接进经营决策。二、人工排查的真实成本小张的手动流程单轮排查大致是这样分配的步骤操作大致耗时1SCADA 导出三天传感器数据1h2Excel 分列逐列查空值3h3条件格式标重复行2h4手算均值标准差标异常4h5查时间戳间隔找延迟2h6汇总问题写报告2h说明这是场景化估算用来说明「人力瓶颈」在哪。真正的问题不在 14 小时这个数字而在于不可复现、不持续、依赖个人经验——换个工程师换个手法结论可能不一样。三、方案架构五项检测 Agent 编排 分层容错架构在一句话里接入层管结构检测层管内容Agent 管编排兜底层管完整性。五项检测的分工检测项检测逻辑工具层级结构体检编码/BOM/列数/时间戳格式structure_checker接入层空值检测各列空值率超阈值报警null_checker检测层重复检测完全重复行占比超阈值报警duplicate_checker检测层异常值检测数值列 3σ 越界标记outlier_checker检测层延迟检测相邻时间戳间隔超 60s 报警delay_checker检测层为什么要分层压测时验证过的关键设计结构脏行编码错乱、缺列、错位发生在「读取」层空值、异常发生在「内容」层混在一起处理会同时误报与漏报比如float()会把空值行误判成错位行。分层之后可修复的修复留痕不可修复的隔离留样内容检测只吃同构的合法行。异常值检测是「模型侧判定」的代表把数值列按 3σ 规则标记越界任何一列命中即报警。下面是detectors/outlier_checker.py的核心实现标准库实现零第三方依赖列集合与 σ 值都从config注入换成 IQR/MAD 只需改判定段# detectors/outlier_checker.py节选 def check_outlier(csv_path: str, sigma: float OUTLIER_SIGMA) - Dict[str, Any]: 基于 3σ 原则检测数值列异常值。 功能描述: 对数值列按 3σ 规则标记越界值汇总异常总数。 设计思路: 先用 float 安全解析收集有效样本再按样本标准差判定越界标准差为 0 跳过。 Returns: 含 status / total_outliers / sigma / columns / message 的结构化字典 文件缺失时返回 {error, suggestion}V2.0 §二。 try: rows load_rows(csv_path) except FileNotFoundError: logger.error(异常值检测失败文件不存在 %s, csv_path) return file_error(csv_path) total_outliers 0 columns: List[Dict[str, Any]] [] for col in NUMERIC_COLUMNS: # 列集合来自 config values: List[float] [] for r in rows: raw str(r.get(col, )).strip() if raw : continue try: values.append(float(raw)) except ValueError: continue if len(values) 2: continue mean statistics.mean(values) std statistics.stdev(values) if std 0: continue lower mean - sigma * std upper mean sigma * std outliers [v for v in values if v lower or v upper] total_outliers len(outliers) columns.append( { column: col, count: len(outliers), mean: round(mean, 4), std: round(std, 4), lower: round(lower, 4), # 上下界一并返回便于回溯判定依据 upper: round(upper, 4), } ) status warning if total_outliers 0 else ok message f发现 {total_outliers} 个异常值 logger.info(异常值检测完成total_outliers%d, total_outliers) return { status: status, total_outliers: total_outliers, sigma: sigma, columns: columns, message: message, }注意两个细节检测列只取NUMERIC_COLUMNS不做全表扫描返回里带mean/std/lower/upper运维看到「温度 10096 个」时能直接判断是判定口径问题还是真故障。这也是 5.2 节误报分析能定量展开的前提。检测层之外模型调用与提示词也做了单点收口——工程里只有core/llm.py知道怎么连 Ollama只有core/prompt.py知道当前用哪版提示词业务代码不直接碰 SDK 与提示词文本# core/llm.py节选模型调用单点 try: # langchain-community 0.4ChatOllama 拆到独立包 from langchain_ollama import ChatOllama except ImportError: # pragma: no cover - 旧版本回退路径 from langchain_community.chat_models import ChatOllama # type: ignore[no-redef] def get_llm() - Any: 返回统一配置的 ChatOllama 实例模型调用单点。 功能描述: 收口大模型构建所有 Agent 经此获取 LLM。 设计思路: 参数全部来自 config便于在 .env 切换模型与温度而不改代码 ChatOllama 导入做新旧路径兼容防止 langchain 大版本升级直接 ImportError。 Returns: 已配置的 ChatOllama 实例。 logger.debug(构建 ChatOllama: model%s, base_url%s, LLM_MODEL_NAME, OLLAMA_BASE_URL) return ChatOllama( base_urlOLLAMA_BASE_URL, modelLLM_MODEL_NAME, temperatureLLM_TEMPERATURE, request_timeoutLLM_REQUEST_TIMEOUT, ) # core/prompt.py节选提示词单一来源按版本文件装配 SYSTEM_PROMPT_FILE: Path PROMPT_DIR / system_prompt_v1.0.1.md def build_react_prompt() - PromptTemplate: 装配 ReAct 系统提示词提示词单一来源 版本管理。 if not SYSTEM_PROMPT_FILE.exists(): raise FileNotFoundError( f系统提示词缺失: {SYSTEM_PROMPT_FILE}请确认 prompts/case05_data_quality/ 目录完整 ) template SYSTEM_PROMPT_FILE.read_text(encodingutf-8) return PromptTemplate.from_template(template)temperature固定为 0 是刻意的巡检是「按数据说话」的任务报告需要可复现不能让模型每次措辞都飘。提示词则放在prompts/case05_data_quality/下按版本号管理变更走 CHANGELOG调用方不动代码。【 适用边界说明】・适配场景离散制造业设备时序数据、工单数据的质量巡检可直接用于数仓 ODS 层前置校验・算法说明演示采用 3σ 原则生产环境建议替换为 IQR/MAD 稳健统计算法降低误报率・扩展说明本期为文件级巡检第二期可对接 Doris 数仓实现 SQL 级批量巡检。四、三步走实施一天落地没搞八周大排期三步走一天完成。环境完全复用已有资产Ollama Qwen2.5 零改动配置集中在config.py敏感参数走.env注入。# config.py节选所有配置一处收口支持环境变量覆盖 OLLAMA_BASE_URL: str os.getenv(OLLAMA_BASE_URL, http://localhost:11434) LLM_MODEL_NAME: str os.getenv(LLM_MODEL_NAME, qwen2.5:7b) NULL_THRESHOLD: float float(os.getenv(NULL_THRESHOLD, 0.01)) # 演示口径生产 5% MOCK_N_ROWS: int int(os.getenv(MOCK_N_ROWS, 1000)) # 数据量一行配置切换步骤产出耗时验证方式1. 数据就绪 检测工具01_generate_mock_data.py 五个检测工具2~3h手动调用与注入基准核对2. Agent 组装 CLI 测试agent/builder.py02_test_cli.py2h一句话触发全量巡检3. 定时巡检 报告落盘03_scheduled_check.pyAPScheduler1~2h--once单次验证 定时触发一键复现命令python 01_generate_mock_data.py # 生成含脏数据的传感器 CSV python tests/eval_script.py # 检测基准回归注入基准 vs 实跑检出 python 02_test_cli.py # 对话式巡检Agent 编排五工具 python 03_scheduled_check.py --once # 定时巡检单次验证报告落盘 reports/这套「配置收口 三步走」的回报马上就能看到后面从 1000 行扩到 50 万行零代码改动环境变量一盖就跑。五、真实数据验证从 1010 行到 50 万行5.1 基线验证1010 行注入基准 1:1 命中固定种子42生成 1010 行脱敏传感器数据刻意注入四类脏数据作为基准ground truth检测项注入基准实跑检出结论空值temperature 20 行20 行 / 1.98%命中重复10 行完全重复10 行 / 0.99%命中异常值20 个极端温度23 个20 注入 3 个高斯尾部命中含 3 个统计尾部延迟3 处时间 gap3 处 / 最大 120s命中关键交叉验证Agent 对话式巡检输出的四个数字与四工具独立检测结果完全一致证明 Agent 只做编排没有引入数值偏差。此处是基线阶段的四个内容检测工具结构体检是 50 万行阶段才加入的第五项见 5.3。5.2 规模验证 ×503σ 误报的定量实锤数据量从 1000 行扩到 50500 行出现了一个比「跑通」值钱得多的发现异常值误报从 3 个暴涨到256 个。这不是 bug是统计规律高斯分布 |z|3 理论占比 0.27%50000×0.27%≈135/列与实测振动 120 电流 136吻合。5.3 压力验证50 万行「脏上加脏」9 类场景逐项过50 万行规模下主动注入 9 类真实世界的脏数据场景场景注入量管线处理实测结果UTF-8 BOM 头全文件BOM 识别剥离✅ 507,500 行全读出GBK 编码分片5,000 行UTF-8 失败→回退 GBK✅ 回退路径真实触发时间戳格式变体10,000 行归一化时间值不变✅ 全部修复非法时间戳13月45日等1,000 行解析失败→隔离留样✅ 隔离 1,000物理缺列 / 多列 / 字段错位各 500 行列数/数值校验→隔离✅ 各隔离 500乱码产线名Latin-1 误码1,000 行Latin-1 往返还原✅ 全部还原四项内容检测隔离修复后的干净输入上见 5.1 口径标准检测✅ 合法行 505,000 精确、重复 5,000 精确、延迟 3 精确同构性验证结构脏行全部追加注入、接入层隔离后内容检测的输入与纯净场景完全一致基准回归一行测试没改、全绿通过。隔离层若多删一行回归立刻报错。性能数据环节耗时生成 50 万行 GBK 分片~5s单次读取 结构校验~4.5s一次完整巡检5 项检测 报告落盘~26sAgent 对话式巡检2m48sLLM 推理占 90% 以上结论检测不是瓶颈LLM 才是。定时巡检链路不调 LLM26 秒跑完对话式巡检才走 Agent。更大规模时应把检测层下沉到 Doris SQL第二期方案。六、实践踩坑与生产级加固这部分是整个实践里最有含金量的三个坑都有完整排坑笔记文末链接。坑点核心后果一句话解法LangChain 版本漂移依赖只写下界升级后全量 ImportError版本锁上下界 代码侧兼容导入双保险小模型漏调工具结构检测被跳过输入数据不可信提示词硬约束 编排层代码兜底双防线压测假覆盖测试全绿但实际未检测关闭注入逻辑对应断言必须变红验证6.1 langchain 1.x 大版本漂移依赖只写下界pip 一升全红requirements.txt只写langchain0.2pip 直接拉到 1.4.2两处 ImportErrorcreate_react_agent迁到langchain_classic.agentsChatOllama移出 community 包。修法双保险版本锁上下界 代码侧兼容导入。详见排坑笔记 1待发布。6.2 小模型漏调工具Agent「交卷」前没做完题50 万行实测qwen2.5:7b 拿到 4 项内容检测结果就直接输出 Final Answer跳过了structure_check。而结构问题恰恰发生在内容检测之前漏掉它意味着整条管线的输入都不可信。生产级修法分两层。第一道防线写在系统提示词 v1.0.1 里把「必须调齐五个工具」变成硬约束【硬性约束】必须依次调用全部工具各一次 null_check、 duplicate_check、 outlier_check、 delay_check、 structure_check。 五个工具全部返回结果之前禁止输出 Final Answer。 structure_check 负责编码/BOM/列结构/时间戳格式的结构体检 与内容检测同等重要绝不可跳过。第二道防线在编排层代码保证、不靠模型自觉。三个设计点值得留意补执行失败不中断主流程而是留{error, suggestion}占位保证报告维度齐全可审计合并走 LLM 但失败自动降级为原文追加无论是否触发都写fallback_executed留痕便于事后统计漏调率。加固后 50 万行实跑 5 工具全调齐报告含结构体检且数字与独立检测一致。详见排坑笔记 2待发布。6.3 压测造数的假覆盖测试全绿其实什么都没测到第一版压测造数有三个盲区缺列行被dict.get(col,)补成空串、ASCII 产线名让乱码注入恒等、纯 ASCII 的「GBK 文件」与 UTF-8 字节完全相同。四个坑的共同本质测试看起来在测实际什么都没测到。判定标准就一条把注入逻辑关掉对应断言必须变红。详见排坑笔记 3待发布。七、工具选型 trade-off7.1 大模型本地 Ollama Qwen2.5 vs 云 API维度本地 Ollama Qwen2.5:7b云 API成本一次性拉取本地推理零边际成本按 token 持续计费数据不出域符合制造数据安全要求出域敏感数据需脱敏能力7b 对复杂推理有限强适用工具编排 / 结构化输出足够复杂生成、长上下文取舍结论数据敏感 成本敏感 场景是「工具编排结构化汇总」本地模型够用数据不出域是制造业硬约束。实测也暴露了 7b 的边界漏调工具所以编排层兜底是必选项不是可选项。7.2 调度APScheduler vs DolphinScheduler维度APScheduler本期DolphinScheduler第二期部署进程内零独立部署独立集群持久化SQLAlchemyJobStore原生分布式调度对接成本直接 import 即用需开发 HTTP 调用接口适用单任务定时巡检多任务企业级编排取舍结论本期是「每小时巡检一次」的单点需求APScheduler 足够。第二期把巡检注册为 DS 工作流节点即可。八、两个必须说清的局限3σ 对高斯尾部会误报且随数据量线性放大。实测三级对照1000 行误报 3 个 → 50500 行 256 个 → 50 万行 2,734 个振动 1,339 电流 1,395占比从 13% 升到 21%。生产口径应换 IQR/MAD 稳健统计量 滑动窗口基线 按列聚合报警。日增百万行的产线不处理运维会被假警报淹没。演示阈值刻意低于生产阈值。本文空值 1%、重复 0.5% 是为直观看到报警生产口径是空值 5%、重复 2%见《制造数据治理巡检-设计方案》切换只改config.py一处。本地 demo 仅验证链路生产数据量级对接 Doris 取数是第二期。九、实践成果与标准对标维度人工方式Agent 方式V2.0 实测巡检机制出问题才查定时常态化26s/次50 万行可复现性依赖个人经验固定种子 注入基准 1:1 核对脏数据覆盖四类内容问题四类 9 类结构边缘场景编排可靠性无提示词约束 编排层兜底双防线报告手动整理自动结构化报告JSON 中文摘要最核心的价值从被动排查变成主动巡检且每一条检出都可溯源到注入基准。本方案对应 GB/T 39116-2020「数据资源」能力域辅域实现数据质量自动化巡检与问题预警为智能制造的数据驱动决策打地基。总结可复用的方法论五个检测项分两层结构体检管接入四项内容检测管数据本体互不重叠。结构化返回所有工具返回统一字典Agent 才能稳定识别与汇总。阈值与参数集中配置.env覆盖 config.py收口从 1000 行到 50 万行零代码改动。流程完整性由代码保证凡是靠模型自觉的环节都要有兜底Agent 漏调工具就是实测教训。选型讲 trade-off数据不出域优先于能力上限单点需求不上重型调度。案例数字必须可溯源注入基准 固定种子局限主动说清不假装完美。【有感而发】做了二十多年制造业数据最深的感受是数据质量从来都不是一劳永逸的事。把检测分层、把阈值收口、把兜底做足工具才能真正跑起来而不是摆样子给领导看。 系列导航专 栏制造业数据与AI落地实战 AI赋能数据开发工程手册系列文章#05 制造业MES工时异常检测AI帮我2小时干完Excel 2天的活#04 我用 WorkBuddy 分析了 30 篇 CSDN 博客发现 3 个反直觉的流量真相#03 MES 工时异常检测-升级版看板 脚本 提示词 排坑2 小时干完 2 天的活#02 还在翻 git log 写周报WorkBuddy 一键生成结构化周报附可复用 Prompt#01 源码首发-WorkBuddy实战CSDN后台数据分析系统完整代码排坑系列排坑笔记 1LangChain 1.x API 迁移create_react_agent 与 ChatOllama 导入错误完整解决方案排坑笔记 2数据健壮性测试假通过排查-注入有效性自检与变异式测试方法排坑笔记 3LangChain 多工具 Agent 完整性校验 return_intermediate_steps 事后核对方案关于作者制造业数据与AI践行者老蒋23 年 IT 老兵。聚焦离散制造业数据治理、工业数仓、AI Agent 工程化落地。只写自己跑通过的东西。全系列导航台账交流互动你在制造企业数据质量管理中遇到过哪些脏数据空值、重复、异常、延迟之外编码乱码和字段错位是不是也踩过欢迎评论区交流我会逐一回复。标签#制造业数据#数据质量巡检#AI Agent#数据治理#工业大数据#Python实战#智能工厂#脏数据检测