【导航台账】制造业数据与AI践行者老蒋的技术博客全系列文章汇总持续更新 文章摘要针对制造业传感器数据脏、质量难管控、导致 OEE 计算失真的痛点本文从零搭建数据质量智能巡检 Agent覆盖空值、重复值、异常值3σ 原则、时间延迟四大检测能力支持人工触发与定时巡检双模式配套 Web 可视化看板。方案对标 GB/T 39116-2020 智能制造标准可直接作为数据质量自动化的落地参考文章附完整工程源码。文章目录开篇痛点与场景正文方案设计核心实现Step 1生成模拟数据含脏数据注入Step 2核心检测工具Step 3Agent组装运行验证Step 4定时巡检Step 5Web看板总结价值提炼智能工厂国标落标系列导航互动与交流关于作者开篇痛点与场景兄弟们Case01到Case03我们搞定了制造Agent的“手脚”、“记忆”和“协同能力”。但班组长小刘在使用过程中提出了一个更深层次的问题。他说“老蒋你们搞数据的人天天喊‘数据驱动决策’。但我这边传感器数据经常出问题——温度突然飙到200度、振动值变成负数、有时候整行数据都是空的。这种数据拿去算OEE算出来的结果你敢信吗”我当时的第一反应是这不科学啊……但小刘说得对。在智联工坊的实际运营中数据质量问题无处不在空值传感器偶尔“罢工”上报空数据重复网络抖动导致同一条数据被重复上报异常值设备故障导致数值突变如温度瞬间飙到999℃数据延迟本应每5分钟上报一次的数据半小时没到我接触过的十几家制造工厂80% 都有这个问题 —— 花几百万上了传感器、MES 系统最后因为没人做数据质量管控算出来的报表没人信数据驱动成了一句空话。这些“脏数据”如果没被发现会直接影响OEE计算、质量分析、预测性维护等上层应用。数据质量是数据驱动决策的“地基”地基不牢上面盖什么都白搭。本文要解决的痛点构建一个数据质量智能巡检Agent能自动检测数据质量问题生成质量报告并在发现问题时主动通知相关人员。适合谁读制造业数据工程师、IT运维人员以及关心数据质量管理的开发者。脱敏声明本文基于虚拟工厂“智联工坊”场景所有数据均已脱敏处理。正文方案设计模块功能技术实现数据加载加载模拟传感器数据CSVPandas空值检测计算各列空值率超过阈值报警Pandas 阈值配置重复检测检测完全重复的行计算重复率Pandas duplicated异常值检测基于3σ原则检测数值列异常Numpy 标准差计算延迟检测检测时间戳间隔是否超过阈值Pandas diff 阈值配置定时巡检每分钟自动执行检测APScheduler核心实现工程目录结构cases/case04_data_quality_agent/ ├── config.py # 配置文件阈值、路径、端口 ├── tools/ # 检测工具集 │ ├── data_loader.py # 数据加载 │ ├── null_checker.py # 空值检测 │ ├── duplicate_checker.py # 重复检测 │ ├── outlier_checker.py # 异常值检测3σ原则 │ └── delay_checker.py # 延迟检测 ├── agent/ │ └── builder.py # 注册5个工具 ├── scripts/ │ ├── 01_generate_sensor_data.py # 生成模拟数据含脏数据 │ ├── 02_test_cli.py # CLI交互测试 │ └── 03_scheduled_check.py # 定时巡检演示 └── web/ # Web看板 ├── app.py └── templates/index.htmlStep 1生成模拟数据含脏数据注入运行scripts/01_generate_sensor_data.pypython 01_generate_sensor_data.py执行效果 开始生成模拟传感器数据... 注入脏数据: 共 500 行 (5%) - 空值: 199 行 - 重复: 99 行 - 异常值: 199 行 ✅ 数据已保存到: .../data/sensor_data.csv 总行数: 10000 列: [timestamp, line, temperature, vibration, current]核心代码异常值注入# 异常值注入逻辑 outlier_indices random.sample(range(num_rows), outlier_count) for idx in outlier_indices: col random.choice([temperature, vibration, current]) if col temperature: df.loc[idx, col] random.choice([200, -50]) # 温度瞬间飙到200或降到-50 elif col vibration: df.loc[idx, col] random.choice([10, -5]) else: # current df.loc[idx, col] random.choice([500, -100])Step 2核心检测工具空值检测def check_null(df: pd.DataFrame) - str: null_counts df.isnull().sum() total_rows len(df) for col in df.columns: null_rate null_counts[col] / total_rows if null_rate NULL_THRESHOLD: alert_columns.append(col) return json.dumps({ status: warning if alert_columns else ok, details: result, alert_columns: alert_columns, threshold: NULL_THRESHOLD, message: f发现 {len(alert_columns)} 列空值率超标 if alert_columns else 空值率正常 }, ensure_asciiFalse)异常值检测3σ原则def check_outlier(df: pd.DataFrame) - str: numeric_cols df.select_dtypes(include[np.number]).columns for col in numeric_cols: mean series.mean() std series.std() if std 0: continue lower mean - OUTLIER_SIGMA * std upper mean OUTLIER_SIGMA * std outliers series[(series lower) | (series upper)] total_outliers len(outliers) return json.dumps({ status: warning if total_outliers 0 else ok, total_outliers: total_outliers, message: f发现 {total_outliers} 个异常值 if total_outliers 0 else 无异常值 }, ensure_asciiFalse)Step 3Agent组装agent/builder.py注册5个工具def _register_tools(self) - List[BaseTool]: tools [] tool def load_data() - str: 加载传感器数据CSV文件返回数据摘要 df load_sensor_data() return get_data_summary(df) tools.append(load_data) tool def null_check() - str: 检测数据中的空值比例 return check_null(load_sensor_data()) tools.append(null_check) # duplicate_check, outlier_check, delay_check 同理... return tools运行验证执行02_test_cli.pypython 02_test_cli.py测试用例1空值检测 你: 空值检测 Entering new AgentExecutor chain... Action: load_data Action Input: {} Observation: {total_rows: 10000, columns: [timestamp, line, temperature, vibration, current]} Thought: 数据加载成功继续执行空值检测 Action: null_check Action Input: {} Observation: {status: ok, details: {temperature: {null_count: 51, null_rate: 0.0051}, vibration: {null_count: 80, null_rate: 0.008}, current: {null_count: 65, null_rate: 0.0065}}, alert_columns: [], message: 空值率正常} Final Answer: 空值检测完成空值率正常无超标列。 Agent: 空值检测完成空值率正常无超标列。测试用例2全面检测 你: 帮我检查一下数据质量 Entering new AgentExecutor chain... Action: load_data → null_check → duplicate_check → outlier_check → delay_check Observation: 空值率正常重复率0%发现199个异常值延迟正常 Final Answer: 数据质量检测完成。发现199个异常值主要集中在温度、振动和电流三个数值列中。建议进一步调查这些异常值的原因。 Agent: 数据质量检测完成。发现199个异常值主要集中在温度、振动和电流三个数值列中。建议进一步调查这些异常值的原因。Step 4定时巡检执行03_scheduled_check.pypython 03_scheduled_check.py每分钟自动执行一次质量检测报告自动保存到reports/目录按CtrlC停止生成的报告示例reports/quality_report_*.json{ timestamp: 2026-08-12T14:30:00, data_summary: {total_rows: 10000, columns: [timestamp, line, temperature, vibration, current]}, null_check: {status: ok, alert_columns: [], threshold: 0.05}, duplicate_check: {status: ok, duplicate_rate: 0.0}, outlier_check: {status: warning, total_outliers: 199}, delay_check: {status: ok, max_interval_seconds: 5.0, threshold_seconds: 60} }Step 5Web看板启动Web服务# 回到项目根目录 cd /path/to/javy21_manufacturing_ai_practice python -m cases.case04_data_quality_agent.web.app浏览器打开http://localhost:5004查看实时质量状态仪表板。Web界面功能四大质量指标卡片空值、重复、异常、延迟状态标识正常/异常每30秒自动刷新总结价值提炼本文解决了什么构建了一个数据质量智能巡检Agent能自动检测空值、重复、异常、延迟四大问题并支持人工触发和定时巡检两种模式。三个关键认知数据质量巡检不是一次性排查是常态化机制。规则明确、重复度高的巡检工作天然适合 Agent 来做人只需要处理异常告警。异常检测要分层基础规则层空值、重复 统计规则层3σ 原则 业务规则层如温度不可能超过 120℃。本文实现前两层业务规则可按工厂场景按需扩展。Agent 做数据巡检的核心价值不是 “发现问题”是 “把问题按标准流程闭环”—— 检测→告警→生成报告→跟进整改这是未来工厂数据运维的标准模式。可复用的方法论「原子化工具拆分 Agent 统一调度 结构化返回 定时 / 人工双触发」的框架可以直接复用到所有规则明确的重复性数据运维场景比如台账巡检、报表核对、主数据校验等。智能工厂国标落标本方案对应GB/T 39116-2020《智能制造能力成熟度模型》中「数据资源」能力域的「数据质量」子项实现了数据质量自动化巡检、问题自动预警、质量报告自动生成可直接用于智能制造能力成熟度三级及以上的申报技术佐证。系列导航本文属于《数据与AI工程排坑笔记》系列上一篇【热榜推荐】LangChain Agent 反复调用工具死循环结构化返回 Prompt 规则让它学会跳过下一篇《数仓分层设计助手》Case05即将发布互动与交流兄弟们你们日常处理最多的脏数据类型是哪一种A. 空值 / 缺失值B. 异常跳点 / 离谱数值C. 重复数据D. 时间戳错乱 / 延迟评论区扣字母我选 3 位兄弟直接发完整工程源码包 数据质量检测通用模板。关于作者制造业数据与AI践行者老蒋23年IT老兵。聚焦制造业数据架构与AI融合落地。全流程实战全源码开源。标签#LangChain#AI Agent#数据质量#智能制造#传感器数据#异常检测#3σ原则#巡检Agent#制造业数字化