1. AI Agent为什么能在数据治理里真正落地做数据治理做到第三年我最深的感受就一个字烦。规则规则写不清楚异常异常清不完业务方天天催口径统一技术侧又总说人力不够。身边不少团队开始聊“AI改造数据治理”但真正落地的很少多数停在PPT层面。直到我自己拿了套轻量化环境把AI Agent塞进数据治理的日常流程跑了三个真实场景才意识到过去那些重复劳动确实可以让模型替我们干很大一部分。先明确一个前提这里说的AI Agent不是那种能自己写诗聊天的机器人而是一个能感知输入、调用工具、按固定流程产出结构化结果的智能执行体。在数据治理领域它典型的工作方式是接入数据样本理解质量问题生成修复建议再调Python代码执行校验和清洗最后输出一份人话版报告。整个过程里人是审核者Agent是执行助理效率提升非常明显。我见过太多团队一上来就买重平台、搭大集群、引全套商业工具结果治理工作流反而更重了。其实数据治理的高频动作就三类质量稽查、异常清洗、元数据梳理。这三类动作共性是重复度高、规则模糊、依赖经验判断恰好是LLM最擅长处理的。用轻量化方式把这三件事交给AI Agent启动成本低见效快后续也能平滑扩展。这篇文章分享的就是我实际跑通的方案完整的环境搭建、三大场景的Prompt库和配套Python代码以及我在真实项目中踩过的坑。我会尽量把能直接抄的都写出来你拿到手改改路径就能跑。2. 轻量化环境搭建不烧卡也能跑的治理工作台2.1 整体架构与选型逻辑做AI Agent数据治理最怕环境没搭好就陷入基础架构泥潭。我见过有人为了跑一个字段校验先拉了Spark集群再上了统一权限系统折腾两周还没开始写业务。这个思路完全反了。轻量化环境的核心约束是一台日常开发机能跑、数据量在几千到几万条规模内不卡、模型调用成本趋近于零。基于这三点我选择了单机容器化架构模型推理本地部署一个开源小参数模型7B~14B量化版足够处理字段理解、规则生成、文本标准化这类任务。量化后显存占用约6~8GB16GB内存的电脑完全带得动。应用层纯Python脚本按场景拆成agent_quality.py、agent_clean.py、agent_lineage.py三个模块不引入重型框架方便按需调试。数据层文件落到CSV/Parquet结果存SQLite血缘关系用JSON保存后续要扩展再换图数据库现在先用最便宜的方案验证效果。为什么不直接调商业大模型API两个原因数据安全是第一红线业务数据特别是客户明细不能出域其次是高频调用的成本会随数据量线性上升而数据治理又是周期性任务长期算下来本地部署更划算。当然你只是想快速验证流程也可以先用API跑通逻辑完全一致。2.2 Docker单机部署与模型加载我这里把环境搭建分为两步第一步起推理服务第二步装Python依赖。推理服务我用Docker来跑核心命令长这样# 拉取推理镜像此处不指定具体镜像名用通用术语inference_image代替 docker run -d \ --name llm-server \ -p 8080:8080 \ -v /opt/models:/models \ inference_image容器起来后需要把模型文件放到/opt/models目录下。社区惯例是下载GGUF格式的量化模型文件我实际选的是7B的Q4_K_M量化版文件大小约4GB既能保证理解能力又不会让普通开发机内存爆炸。应用层依赖非常少我用requirements.txt固定pandas2.1.4 requests2.31.0 openpyxl3.1.2 tabulate0.9.0安装命令pip install -r requirements.txt装完先验证一下推理服务是否正常。写个最简单的连通性测试import requests resp requests.post( http://localhost:8080/v1/chat/completions, json{ model: local-model, messages: [{role: user, content: 用一句话说明什么是数据质量}], temperature: 0.2, max_tokens: 200, }, timeout60, ) print(resp.json()[choices][0][message][content])能正常返回就说明环境通了一半。这里有个小经验temperature参数在数据治理场景要压得很低我一般固定在0.1~0.2之间。数据治理要的是稳定和一致不是创造性发挥。如果temperature太高同样的输入两次返回的规则可能不一样后面就全乱套了。2.3 环境验证与测试样本准备服务起好之后先拿测试样本整体走一遍流程。我找了一份脱敏的订单数据试跑字段包括订单号、客户ID、订单金额、下单时间、支付状态大概几千行刻意混入了一些脏数据订单号里有空值和重复值订单金额出现负数和“未知”这样明显不合法的值下单时间的格式有的是2024-01-05有的是2024/01/05还有的是2024.01.05先让Agent对这份样本做一次无监督自检看看它能发现哪些问题。这一步验证很有价值能让你在写正式Prompt之前摸清模型输出的稳定性。我在环境搭建阶段通常这么做直接把样本灌给模型问“请指出这张表中可能存在的质量问题字段和原因”然后看它输出的字段是否和真实问题基本吻合。第一次跑的时候可能会有输出不稳定的情况比如返回的字段顺序每次不一致、JSON格式偶尔崩掉。所以在代码里我统一加了JSON解析兜底逻辑解析失败就自动重试一次仍失败则降级返回规则引擎结果。这个容错设计后面会讲但环境验证阶段就能暴露出来值得提前处理。3. 核心思路Agent解耦数据治理的四种执行模式3.1 把“规则写死”改成“规则动态生成”传统数据治理的实现方式是把业务口径翻译成代码里的if-else例如“订单金额不可为负数”“订单号不为空且唯一”。这样做的好处是确定性强坏处是规则一多就爆炸几十张表、上百个字段、上千条规则维护成本高到没人愿意碰。AI Agent不一样。它会把规则生成变成一个动态动作输入字段名和样本值模型自动列出可能的质量规则。这等于把“运维静态规则”变成了“按需推理规则”新增字段和表时不用再手写规则让Agent先给一版初稿人再审核调整。我在实践中的体会是这个模式让规则上线时间从以天计缩短到以分钟计而且规则遗漏率明显下降。3.2 从“全量清洗”变为“采样理解定向执行”传统清洗脚本通常全量跑遇到格式不统一就写正则一个个适配。驱动质量问题千奇百怪今天来一个“金额单位不一致”明天来一个“时间戳带了时区后缀”写正则写到崩溃。我的Agent方案改成两层先采样让模型“理解”问题类型再根据理解结果选择对应的清洗函数执行。正则还是要写但写的范围更聚焦了并且可以由模型给出正则建议人来最终确认。这有点像医生先看病再开药而不是不管什么病都先来一套全身体检。3.3 用“规则引擎LLM”双轨制保障安全底限这里必须强调一个原则AI Agent在数据治理里不能单打独斗。LLM有幻觉如果让模型直接改数据可能把好数据改成坏数据。所以我所有场景的设计都是“双轨制”模型负责理解和生成建议确定性代码负责校验和执行。比如模型判断“此字段有身份证号格式问题”但实际执行格式校验的是Python的正则函数不是模型自己改数据。这个边界很重要后面所有场景都基于这个思路。3.4 以“结构化输出”衔接Agent与后端Agent的产出必须是结构化数据不能是一段人话。我在所有Prompt里都要求模型输出JSON格式包含字段名、问题类型、严重级别、具体证据、修复建议。这样后续无论接报告系统、数据表还是调度平台都很顺。在代码里用一个统一的parse_llm_json函数做解析配置json.loads加异常处理确保即使模型偶尔抽风流程也能走完。4. 场景一实战用Agent做数据质量智能稽查4.1 场景痛点与Agent设计质量稽查是数据治理里最烦的任务。传统做法是配置调度任务跑规则第二天看报告看到告警再手工排查。问题是规则覆盖不全、告警误报率高、多个规则冲突时看不出优先级。这个场景里Agent的作用是“质量初审员”给它一批数据样本它告诉你哪里可能有问题、为什么、严重程度如何、建议怎么修。注意是“可能有问题”的辅助判断最终结论还要人来确认。实际效果是原来一个数据质量检查任务从“写规则跑批人工复核”三小时压缩到“Agent首查人工复核”四十分钟。4.2 Prompt库质量核查Prompt模板这是我经过多轮迭代后稳定使用的Prompt模板你直接复制就能用。核心设计是明确角色、限定输入输出、给出样例、要求JSON返回。你是一名数据质量分析专家。我将给你一张数据表的字段信息和抽样样本请帮我完成以下任务 1. 识别每个字段可能存在的质量问题如缺失、重复、格式错误、非法值、常识性矛盾。 2. 按严重程度分级严重导致数据不可用、中等影响部分分析、轻微规范性问题。 3. 给出每个问题的一条样例证据并给出修复建议。 字段结构如下 {字段名: 类型, 说明, 示例} 抽样数据如下最多展示50行 {样本} 请以JSON数组格式返回每个元素结构如下 { 字段名: 字段名称, 问题类型: 缺失/重复/格式错误/非法值/矛盾/其他, 严重级别: 严重/中等/轻微, 证据: 具体的异常样例, 修复建议: 一句话可落地的修复方式 } 只需要返回JSON不要输出其他解释。这个Prompt有几个设计细节我踩过坑才总结出来加“最多展示50行”这句话非常重要。模型上下文窗口有限超过一定行数不仅变慢输出质量还会断崖式下降。我试过几千行的样本硬灌模型开始编造不存在的错。“只需要返回JSON”可以大幅提高输出结构化率但不是100%保证所以代码里还要做解析兜底。给字段结构而不是只给字段名能让模型更准确识别类型问题。比如字段类型是“float”但出现“未知”字符串模型一看就知道类型冲突。4.3 Agent主控代码实现质量稽查Agent的主控逻辑分四步读取文件、构造Prompt、调用模型、解析与汇总。下面是精简版代码我加了详细注释import pandas as pd import requests import json import random LLM_URL http://localhost:8080/v1/chat/completions MODEL_NAME local-model def parse_llm_json(text: str): 解析模型返回的JSON失败时尝试提取json块再失败返回空列表 try: return json.loads(text) except json.JSONDecodeError: start text.find(json) if start ! -1: end text.find(, start 7) if end ! -1: try: return json.loads(text[start7:end]) except json.JSONDecodeError: return [] return [] def build_prompt(df: pd.DataFrame, fields_desc: dict) - str: # 采样最多50行 sample_df df.sample(nmin(50, len(df)), random_state42) sample_str sample_df.to_string() field_str json.dumps(fields_desc, ensure_asciiFalse) prompt f你是一名数据质量分析专家...完整Prompt见上文... 字段结构如下 {field_str} 抽样数据如下最多展示50行 {sample_str} return prompt def call_llm(prompt: str, temperature: float 0.1): resp requests.post( LLM_URL, json{ model: MODEL_NAME, messages: [{role: user, content: prompt}], temperature: temperature, max_tokens: 1500, }, timeout120, ) return resp.json()[choices][0][message][content] def quality_check(df: pd.DataFrame, fields_desc: dict) - list: prompt build_prompt(df, fields_desc) result_text call_llm(prompt) issues parse_llm_json(result_text) # 简单兜底如果列表为空重试一次 if not issues: issues parse_llm_json(call_llm(prompt, temperature0.0)) return issues if __name__ __main__: df pd.read_csv(orders_sample.csv) fields_desc { 订单号: {类型: string, 说明: 订单唯一编号}, 订单金额: {类型: float, 说明: 订单总金额单位元}, 下单时间: {类型: datetime, 说明: 用户下单时间}, 支付状态: {类型: string, 说明: 支付状态枚举}, } issues quality_check(df, fields_desc) print(json.dumps(issues, ensure_asciiFalse, indent2))我用这份代码在测试数据上跑了一次模型输出的问题和预设的脏数据基本吻合。它识别出了订单号的缺失和重复、订单金额的负数与非法值、下单时间的格式不统一还额外指出了支付状态中一个很少见的枚举值这是规则引擎没覆盖到的盲区。Agent的另一个附加价值在这里体现它能发现你没预期到的新问题。4.4 质量报告落地与人工复核Agent输出的JSON可以直接转成Markdown报告或Excel。我习惯转成一张质量问题汇总表包含“字段名|问题类型|严重级别|证据|修复建议”然后人工逐条复核能自动修的进清洗流程需要业务确认的发给对应负责人。这一步最重要的是别让Agent直接改数据。它给出的修复建议是参考执行前必须过一遍自己的判断。我在实际项目里遇到过一次模型建议“删除所有订单时间为空的记录”这明显有问题大量空记录可能是上游接口漏传应该去修上游而不是删数据。人没有审核就直接执行损失会很大。所以流程上我强制加了一道人审环节哪怕只是扫一眼汇总表。5. 场景二实战异常数据自动清洗与标准化5.1 清洗流程的Agent化设计如果说质量稽查是“找出问题”清洗就是“解决问题”。清洗的难点永远不在写代码而在判断“这个字段该按什么规则清洗”。过去怎么做人工打开数据看几行凭经验写正则、写映射、写替换逻辑。表一多每个字段都得手工适配而且要反复调。Agent在这里的角色是“清洗规则生成器”负责三步识别问题类型、给出清洗策略和正则表示、由Python函数执行清洗。输出结构我设计成一个清洗规则列表每条包含字段名、清洗类型、参数、说明。Python代码按这个规则列表依次执行。5.2 Prompt库清洗规则生成Prompt模板这个Prompt是场景二的核心。关键点是要求模型输出可执行的Python正则和清洗参数同时明确约束“模型不直接修改数据只输出规则”。你是一名资深数据清洗工程师。下面是一个字段的样本数据以及该字段的问题描述请为这个字段生成清洗规则。 字段名{字段名} 字段类型{字段类型} 问题描述{问题描述} 样本数据最多30个唯一值 {样本值列表} 请根据样本给出清洗规则。清洗类型可选 - regex_replace用正则替换特定模式 - strip_whitespace去除前后空白 - to_datetime解析为统一日期格式给定输入格式和输出格式 - map_value枚举值映射 - filter_invalid过滤非法值 - fill_null空值填充 请以JSON格式返回{{ 字段名: {字段名}, 清洗类型: regex_replace, 参数: {{ pattern: 正则表达式, replacement: 替换为, 说明: 为什么这样做 }} }} 只需要返回JSON。在订单时间字段的清洗任务里我用这个Prompt得到的结果是{ 字段名: 下单时间, 清洗类型: to_datetime, 参数: { input_formats: [%Y-%m-%d, %Y/%m/%d, %Y.%m.%d], output_format: %Y-%m-%d, 说明: 样本中存在三种日期分隔符统一为YYYY-MM-DD } }这个结果和我想的一致。关键价值在于模型能从样本值里自动识别出格式差异省去人工查看数据样式再写格式串的功夫。对数据仓库里几十个时间字段来说这个能力非常实用。5.3 清洗引擎执行代码拿到清洗规则后代码执行部分很直接。我封装了一个apply_clean_rules函数按规则类型分派到不同的清洗逻辑def apply_clean_rules(df: pd.DataFrame, rules: list) - pd.DataFrame: 按Agent生成的规则列表清洗数据 import re from datetime import datetime result_df df.copy() clean_log [] for rule in rules: field rule.get(字段名) clean_type rule.get(清洗类型) params rule.get(参数, {}) if field not in result_df.columns: continue if clean_type regex_replace: pattern params.get(pattern) replacement params.get(replacement, ) result_df[field] result_df[field].astype(str).str.replace( pattern, replacement, regexTrue ) clean_log.append({字段: field, 操作: f正则替换: {pattern} - {replacement}}) elif clean_type strip_whitespace: result_df[field] result_df[field].astype(str).str.strip() clean_log.append({字段: field, 操作: 去除前后空白}) elif clean_type to_datetime: input_formats params.get(input_formats, []) output_format params.get(output_format, %Y-%m-%d) def parse_date(val): val str(val).strip() for fmt in input_formats: try: return datetime.strptime(val, fmt).strftime(output_format) except ValueError: continue return val # 无法解析则保留原值 result_df[field] result_df[field].apply(parse_date) clean_log.append({字段: field, 操作: f日期格式化 - {output_format}}) elif clean_type map_value: mapping params.get(mapping, {}) result_df[field] result_df[field].map(lambda x: mapping.get(x, x)) clean_log.append({字段: field, 操作: f枚举映射: {mapping}}) elif clean_type fill_null: fill_value params.get(fill_value, ) null_mask result_df[field].isna() result_df.loc[null_mask, field] fill_value clean_log.append({字段: field, 操作: f空值填充: {fill_value}}) elif clean_type filter_invalid: valid_values params.get(valid_values, []) if valid_values: result_df result_df[result_df[field].isin(valid_values)] clean_log.append({字段: field, 操作: 过滤非法值}) return result_df, clean_log我在实际清洗订单表时跑下来清洗前有效记录率约87%清洗后提升到99%以上。有一个坑要说一下日期解析失败的行会保留原值这会导致后续分析时报错。所以我在清洗后总会加一条检查语句把无法解析的日期单独抽出来看一眼确认是保留还是转人工处理。5.4 清洗效果评估与回填策略清洗完成不等于结束。我习惯生成三样东西清洗前后数据对比统计、清洗日志明细、被过滤/修改记录的备份。尤其是备份被修改前的原始数据这是必须养成的职业习惯一旦发现问题可以随时回溯否则被人问“你凭什么改我的字段”时会非常被动。对比统计怎么做最常见的是“清洗前有效值数量”和“清洗后有效值数量”的对比表外加每个规则生效的行数。我还会加一个“异常值分布”的简单透视比如清洗前订单金额的负数分布在一个特定区间说明可能是某个支付渠道的反向退款没有正确标识这有助于反推上游修复而不仅仅是下游洗数据。6. 场景三实战元数据与血缘的辅助管理6.1 业务价值与Agent介入点元数据和血缘管理是数据治理里最不受重视但最救命的部分。出了数据问题第一件事永远是“这个字段哪里来的上游怎么算的”没有血缘关系记录排查全靠问人。但建设数据资产目录又是一个大工程很多小团队根本无力投入专人维护。Agent在这里的定位是“被动式文档生成器”把SQL脚本、数据字典、建表语句喂给模型让它自动抽取字段信息、表间依赖关系和业务含义说明。它不能替代完整的数据资产平台但能快速生产一份“够用”的元数据文档和血缘草稿缓解最痛的信息孤岛问题。6.2 Prompt库血缘解析Prompt模板血缘解析的Prompt我设计的重点在于引导模型理解SQL的依赖关系你是一名数据仓库工程师。请你分析下面这段SQL脚本提取数据血缘关系。 SQL脚本 {sql_text} 请输出 1. 脚本中出现的所有表名和字段名 2. 每个输出字段来自哪些输入表的哪些字段字段级血缘 3. 主要的加工逻辑过滤、关联、聚合、计算 4. 表间依赖关系哪个表依赖哪个表 SQL中可能存在多张表、多次CTE临时表请务必理清依赖链条。 请以JSON格式返回 { 输入表: [表A, 表B], 输出表: 表C, 字段血缘: [ {输出字段: 字段X, 来源: 表A.字段A, 加工逻辑: 直接映射} ], 表依赖: [{上游表: 表A, 下游表: 表C, 依赖方式: JOIN}] } 只需要返回JSON。模型对清晰结构的SQL理解得不错但遇到几百行的大脚本时输出字段会开始丢三落四。我实测的经验是脚本超过200行拆成多个子片段分别抽取再合并结果准确率提升非常明显。这也符合Agent工具的一个通用原则一件干不了就拆件干而不是让单次推理硬扛。6.3 字段血缘抽取代码实现血缘抽取的Python代码核心就是构造Prompt、调用模型、解析JSON、合并结果def extract_lineage(sql_path: str) - dict: with open(sql_path, r, encodingutf-8) as f: sql_text f.read() prompt f完整血缘Prompt见上文 SQL脚本 {sql_text} resp_text call_llm(prompt, temperature0.1) lineage parse_llm_json(resp_text) return lineage def extract_lineage_chunked(sql_path: str, chunk_size: int 150): 将长SQL分段解析再合并血缘结果 with open(sql_path, r, encodingutf-8) as f: sql_text f.read() # 按分号简单拆分成子句 stmts [s for s in sql_text.split(;) if s.strip()] all_lineages [] current_stmt for stmt in stmts: current_stmt stmt ; if len(current_stmt.split(\n)) chunk_size: all_lineages.append(extract_lineage_from_text(current_stmt)) current_stmt if current_stmt.strip(): all_lineages.append(extract_lineage_from_text(current_stmt)) return merge_lineages(all_lineages) def extract_lineage_from_text(sql_fragment: str): prompt f血缘Prompt用{fragment}替换SQL脚本位置 return parse_llm_json(call_llm(prompt))这段实现里merge_lineages要处理重复表和重复字段血缘问题我的处理策略是“来源相同则合并来源冲突则并列保留”把这个规则写死在代码里。这样至少不会把两条冲突血缘直接覆盖掉为后续人工审核保留了信息。6.4 数据字典自动补全与维护建议除了血缘Agent还可以顺手把数据字典的“字段说明”和“业务含义”补全。很多系统数据库里的字段名是col_001这种无意义命名看到名字完全不知道是什么。让模型根据字段值样本来猜含义准确率大约七到八成虽然不能直接对外发布但作为初稿让业务方核对修改能节省大量时间。数据字典补全的Prompt我附在这里你是数据资产管理员。下面是一个数据表的字段信息请为每个字段生成一份业务含义说明和命名建议。 字段信息 {字段信息} 返回JSON数组 [{原字段名: col_001, 推断含义: 订单创建时间, 说明: ..., 建议命名: order_create_time}]我的经验是模型对“时间”“金额”“状态”这类通用字段的推断准确度很高但涉及行业特有概念时准确率会下降所以生成结果只能作为辅助参考发布前必须经业务方确认。不过这正是速度上的胜利原来请业务专家逐字段解释一百个字段要一整天现在模型先拟一版专家改错补充一小时内搞定。7. 常见问题与避坑指南实操记录7.1 环境部署类问题问题本地模型服务启动后调用超时这是最常遇到的。原因多数是模型加载到内存后推理速度慢尤其第一轮调用要加载权重耗时可能几十秒。我的处理办法是在正式跑任务前先“预热”模型用一个简单请求让它把权重加载好再进行实际调用。问题容器内存不足模型OOM7B量化模型通常需要6~8GB内存但如果机器本身只有8GB内存再叠加Python应用基本就满了。解决思路是换更小尺寸模型或者更激进的量化等级或者把应用和模型分开部署一个跑在Docker里一个跑在宿主机上。7.2 Prompt与Agent调优类问题问题模型输出偶尔不是合法JSON我用“只需要返回JSON不要输出其他解释”来约束成功率达到九成以上但偶尔还是会输出Markdown代码块或附带说明。我的兜底方案是第一个parse_llm_json函数加Markdown块提取逻辑再失败就重试一次temperature取0。双重兜底后成功率基本100%。问题同一份数据跑两次结果不一致这就是temperature过高导致的。数据治理要求输出确定性我在投产场景里全部使用temperature0.1甚至0.0。有些推理服务支持seed参数固定随机种子能进一步提高一致性。如果业务要求绝对可复现建议设置seed。7.3 数据安全与流程规范类问题问题样本数据包含敏感信息我的原则是任何样本送进模型前先做脱敏处理把姓名、手机号、身份证号替换成脱敏模板。脱敏方式可以简单粗暴固定字符串、加盐哈希、或者用假名库替换。本地部署还有一个好处是模型在本地跑数据不出域安全合规压力小很多。问题Agent误判或漏判导致治理遗漏这个不能完全避免但可以降低。我的做法是维护一个“人工审核规则清单”对Agent识别的高风险直接跳过、中风险人审、低风险自动处理。另外定期用标注过的验证集评估Agent输出质量如果准确率低于设定阈值就调整Prompt或补充示例。Agent是提效工具不是免责工具这个思路要先立住。8. 最终实践心得与后续扩展方向跑完这三个场景我最大的体会是AI Agent数据治理的价值不在于替代人而在于把“找人问、翻文档、试规则”的时间压缩到近乎为零。过去我做大半天的数据质量检查现在半小时内能完成初筛剩下的时间全部放在人工复核和业务沟通上这才是真正的效率改善。最后分享一个使用心得别一上来就追求Agent全自动先让它做“副驾驶”你来做“驾驶员”。从质量稽查、清洗规则生成、血缘梳理这些单个动作开始跑熟之后再逐步把动作串成自动化工作流。我在实际项目中也是先三个场景单独跑稳定之后再考虑串成一条“从质检到清洗再到出报告”的流水线。如果你后续想扩展我建议按这三个方向走一是接入定时调度让Agent周期性巡检数据质量二是把血缘结果迁移到图数据库做可视化分析三是把Prompt库版本化按表类型积累沉淀成组织的治理知识库。等这三步都走完你手里的AI Agent数据治理体系就真正成型了而且全程不依赖重型基础设施。 
长河君 · 2026/10/21 3:25:40 · 阅读 1.2k AI Agent实战：轻量化数据治理的三大核心场景与落地指南
1. AI Agent为什么能在数据治理里真正落地做数据治理做到第三年我最深的感受就一个字烦。规则规则写不清楚异常异常清不完业务方天天催口径统一技术侧又总说人力不够。身边不少团队开始聊“AI改造数据治理”但真正落地的很少多数停在PPT层面。直到我自己拿了套轻量化环境把AI Agent塞进数据治理的日常流程跑了三个真实场景才意识到过去那些重复劳动确实可以让模型替我们干很大一部分。先明确一个前提这里说的AI Agent不是那种能自己写诗聊天的机器人而是一个能感知输入、调用工具、按固定流程产出结构化结果的智能执行体。在数据治理领域它典型的工作方式是接入数据样本理解质量问题生成修复建议再调Python代码执行校验和清洗最后输出一份人话版报告。整个过程里人是审核者Agent是执行助理效率提升非常明显。我见过太多团队一上来就买重平台、搭大集群、引全套商业工具结果治理工作流反而更重了。其实数据治理的高频动作就三类质量稽查、异常清洗、元数据梳理。这三类动作共性是重复度高、规则模糊、依赖经验判断恰好是LLM最擅长处理的。用轻量化方式把这三件事交给AI Agent启动成本低见效快后续也能平滑扩展。这篇文章分享的就是我实际跑通的方案完整的环境搭建、三大场景的Prompt库和配套Python代码以及我在真实项目中踩过的坑。我会尽量把能直接抄的都写出来你拿到手改改路径就能跑。2. 轻量化环境搭建不烧卡也能跑的治理工作台2.1 整体架构与选型逻辑做AI Agent数据治理最怕环境没搭好就陷入基础架构泥潭。我见过有人为了跑一个字段校验先拉了Spark集群再上了统一权限系统折腾两周还没开始写业务。这个思路完全反了。轻量化环境的核心约束是一台日常开发机能跑、数据量在几千到几万条规模内不卡、模型调用成本趋近于零。基于这三点我选择了单机容器化架构模型推理本地部署一个开源小参数模型7B~14B量化版足够处理字段理解、规则生成、文本标准化这类任务。量化后显存占用约6~8GB16GB内存的电脑完全带得动。应用层纯Python脚本按场景拆成agent_quality.py、agent_clean.py、agent_lineage.py三个模块不引入重型框架方便按需调试。数据层文件落到CSV/Parquet结果存SQLite血缘关系用JSON保存后续要扩展再换图数据库现在先用最便宜的方案验证效果。为什么不直接调商业大模型API两个原因数据安全是第一红线业务数据特别是客户明细不能出域其次是高频调用的成本会随数据量线性上升而数据治理又是周期性任务长期算下来本地部署更划算。当然你只是想快速验证流程也可以先用API跑通逻辑完全一致。2.2 Docker单机部署与模型加载我这里把环境搭建分为两步第一步起推理服务第二步装Python依赖。推理服务我用Docker来跑核心命令长这样# 拉取推理镜像此处不指定具体镜像名用通用术语inference_image代替 docker run -d \ --name llm-server \ -p 8080:8080 \ -v /opt/models:/models \ inference_image容器起来后需要把模型文件放到/opt/models目录下。社区惯例是下载GGUF格式的量化模型文件我实际选的是7B的Q4_K_M量化版文件大小约4GB既能保证理解能力又不会让普通开发机内存爆炸。应用层依赖非常少我用requirements.txt固定pandas2.1.4 requests2.31.0 openpyxl3.1.2 tabulate0.9.0安装命令pip install -r requirements.txt装完先验证一下推理服务是否正常。写个最简单的连通性测试import requests resp requests.post( http://localhost:8080/v1/chat/completions, json{ model: local-model, messages: [{role: user, content: 用一句话说明什么是数据质量}], temperature: 0.2, max_tokens: 200, }, timeout60, ) print(resp.json()[choices][0][message][content])能正常返回就说明环境通了一半。这里有个小经验temperature参数在数据治理场景要压得很低我一般固定在0.1~0.2之间。数据治理要的是稳定和一致不是创造性发挥。如果temperature太高同样的输入两次返回的规则可能不一样后面就全乱套了。2.3 环境验证与测试样本准备服务起好之后先拿测试样本整体走一遍流程。我找了一份脱敏的订单数据试跑字段包括订单号、客户ID、订单金额、下单时间、支付状态大概几千行刻意混入了一些脏数据订单号里有空值和重复值订单金额出现负数和“未知”这样明显不合法的值下单时间的格式有的是2024-01-05有的是2024/01/05还有的是2024.01.05先让Agent对这份样本做一次无监督自检看看它能发现哪些问题。这一步验证很有价值能让你在写正式Prompt之前摸清模型输出的稳定性。我在环境搭建阶段通常这么做直接把样本灌给模型问“请指出这张表中可能存在的质量问题字段和原因”然后看它输出的字段是否和真实问题基本吻合。第一次跑的时候可能会有输出不稳定的情况比如返回的字段顺序每次不一致、JSON格式偶尔崩掉。所以在代码里我统一加了JSON解析兜底逻辑解析失败就自动重试一次仍失败则降级返回规则引擎结果。这个容错设计后面会讲但环境验证阶段就能暴露出来值得提前处理。3. 核心思路Agent解耦数据治理的四种执行模式3.1 把“规则写死”改成“规则动态生成”传统数据治理的实现方式是把业务口径翻译成代码里的if-else例如“订单金额不可为负数”“订单号不为空且唯一”。这样做的好处是确定性强坏处是规则一多就爆炸几十张表、上百个字段、上千条规则维护成本高到没人愿意碰。AI Agent不一样。它会把规则生成变成一个动态动作输入字段名和样本值模型自动列出可能的质量规则。这等于把“运维静态规则”变成了“按需推理规则”新增字段和表时不用再手写规则让Agent先给一版初稿人再审核调整。我在实践中的体会是这个模式让规则上线时间从以天计缩短到以分钟计而且规则遗漏率明显下降。3.2 从“全量清洗”变为“采样理解定向执行”传统清洗脚本通常全量跑遇到格式不统一就写正则一个个适配。驱动质量问题千奇百怪今天来一个“金额单位不一致”明天来一个“时间戳带了时区后缀”写正则写到崩溃。我的Agent方案改成两层先采样让模型“理解”问题类型再根据理解结果选择对应的清洗函数执行。正则还是要写但写的范围更聚焦了并且可以由模型给出正则建议人来最终确认。这有点像医生先看病再开药而不是不管什么病都先来一套全身体检。3.3 用“规则引擎LLM”双轨制保障安全底限这里必须强调一个原则AI Agent在数据治理里不能单打独斗。LLM有幻觉如果让模型直接改数据可能把好数据改成坏数据。所以我所有场景的设计都是“双轨制”模型负责理解和生成建议确定性代码负责校验和执行。比如模型判断“此字段有身份证号格式问题”但实际执行格式校验的是Python的正则函数不是模型自己改数据。这个边界很重要后面所有场景都基于这个思路。3.4 以“结构化输出”衔接Agent与后端Agent的产出必须是结构化数据不能是一段人话。我在所有Prompt里都要求模型输出JSON格式包含字段名、问题类型、严重级别、具体证据、修复建议。这样后续无论接报告系统、数据表还是调度平台都很顺。在代码里用一个统一的parse_llm_json函数做解析配置json.loads加异常处理确保即使模型偶尔抽风流程也能走完。4. 场景一实战用Agent做数据质量智能稽查4.1 场景痛点与Agent设计质量稽查是数据治理里最烦的任务。传统做法是配置调度任务跑规则第二天看报告看到告警再手工排查。问题是规则覆盖不全、告警误报率高、多个规则冲突时看不出优先级。这个场景里Agent的作用是“质量初审员”给它一批数据样本它告诉你哪里可能有问题、为什么、严重程度如何、建议怎么修。注意是“可能有问题”的辅助判断最终结论还要人来确认。实际效果是原来一个数据质量检查任务从“写规则跑批人工复核”三小时压缩到“Agent首查人工复核”四十分钟。4.2 Prompt库质量核查Prompt模板这是我经过多轮迭代后稳定使用的Prompt模板你直接复制就能用。核心设计是明确角色、限定输入输出、给出样例、要求JSON返回。你是一名数据质量分析专家。我将给你一张数据表的字段信息和抽样样本请帮我完成以下任务 1. 识别每个字段可能存在的质量问题如缺失、重复、格式错误、非法值、常识性矛盾。 2. 按严重程度分级严重导致数据不可用、中等影响部分分析、轻微规范性问题。 3. 给出每个问题的一条样例证据并给出修复建议。 字段结构如下 {字段名: 类型, 说明, 示例} 抽样数据如下最多展示50行 {样本} 请以JSON数组格式返回每个元素结构如下 { 字段名: 字段名称, 问题类型: 缺失/重复/格式错误/非法值/矛盾/其他, 严重级别: 严重/中等/轻微, 证据: 具体的异常样例, 修复建议: 一句话可落地的修复方式 } 只需要返回JSON不要输出其他解释。这个Prompt有几个设计细节我踩过坑才总结出来加“最多展示50行”这句话非常重要。模型上下文窗口有限超过一定行数不仅变慢输出质量还会断崖式下降。我试过几千行的样本硬灌模型开始编造不存在的错。“只需要返回JSON”可以大幅提高输出结构化率但不是100%保证所以代码里还要做解析兜底。给字段结构而不是只给字段名能让模型更准确识别类型问题。比如字段类型是“float”但出现“未知”字符串模型一看就知道类型冲突。4.3 Agent主控代码实现质量稽查Agent的主控逻辑分四步读取文件、构造Prompt、调用模型、解析与汇总。下面是精简版代码我加了详细注释import pandas as pd import requests import json import random LLM_URL http://localhost:8080/v1/chat/completions MODEL_NAME local-model def parse_llm_json(text: str): 解析模型返回的JSON失败时尝试提取json块再失败返回空列表 try: return json.loads(text) except json.JSONDecodeError: start text.find(json) if start ! -1: end text.find(, start 7) if end ! -1: try: return json.loads(text[start7:end]) except json.JSONDecodeError: return [] return [] def build_prompt(df: pd.DataFrame, fields_desc: dict) - str: # 采样最多50行 sample_df df.sample(nmin(50, len(df)), random_state42) sample_str sample_df.to_string() field_str json.dumps(fields_desc, ensure_asciiFalse) prompt f你是一名数据质量分析专家...完整Prompt见上文... 字段结构如下 {field_str} 抽样数据如下最多展示50行 {sample_str} return prompt def call_llm(prompt: str, temperature: float 0.1): resp requests.post( LLM_URL, json{ model: MODEL_NAME, messages: [{role: user, content: prompt}], temperature: temperature, max_tokens: 1500, }, timeout120, ) return resp.json()[choices][0][message][content] def quality_check(df: pd.DataFrame, fields_desc: dict) - list: prompt build_prompt(df, fields_desc) result_text call_llm(prompt) issues parse_llm_json(result_text) # 简单兜底如果列表为空重试一次 if not issues: issues parse_llm_json(call_llm(prompt, temperature0.0)) return issues if __name__ __main__: df pd.read_csv(orders_sample.csv) fields_desc { 订单号: {类型: string, 说明: 订单唯一编号}, 订单金额: {类型: float, 说明: 订单总金额单位元}, 下单时间: {类型: datetime, 说明: 用户下单时间}, 支付状态: {类型: string, 说明: 支付状态枚举}, } issues quality_check(df, fields_desc) print(json.dumps(issues, ensure_asciiFalse, indent2))我用这份代码在测试数据上跑了一次模型输出的问题和预设的脏数据基本吻合。它识别出了订单号的缺失和重复、订单金额的负数与非法值、下单时间的格式不统一还额外指出了支付状态中一个很少见的枚举值这是规则引擎没覆盖到的盲区。Agent的另一个附加价值在这里体现它能发现你没预期到的新问题。4.4 质量报告落地与人工复核Agent输出的JSON可以直接转成Markdown报告或Excel。我习惯转成一张质量问题汇总表包含“字段名|问题类型|严重级别|证据|修复建议”然后人工逐条复核能自动修的进清洗流程需要业务确认的发给对应负责人。这一步最重要的是别让Agent直接改数据。它给出的修复建议是参考执行前必须过一遍自己的判断。我在实际项目里遇到过一次模型建议“删除所有订单时间为空的记录”这明显有问题大量空记录可能是上游接口漏传应该去修上游而不是删数据。人没有审核就直接执行损失会很大。所以流程上我强制加了一道人审环节哪怕只是扫一眼汇总表。5. 场景二实战异常数据自动清洗与标准化5.1 清洗流程的Agent化设计如果说质量稽查是“找出问题”清洗就是“解决问题”。清洗的难点永远不在写代码而在判断“这个字段该按什么规则清洗”。过去怎么做人工打开数据看几行凭经验写正则、写映射、写替换逻辑。表一多每个字段都得手工适配而且要反复调。Agent在这里的角色是“清洗规则生成器”负责三步识别问题类型、给出清洗策略和正则表示、由Python函数执行清洗。输出结构我设计成一个清洗规则列表每条包含字段名、清洗类型、参数、说明。Python代码按这个规则列表依次执行。5.2 Prompt库清洗规则生成Prompt模板这个Prompt是场景二的核心。关键点是要求模型输出可执行的Python正则和清洗参数同时明确约束“模型不直接修改数据只输出规则”。你是一名资深数据清洗工程师。下面是一个字段的样本数据以及该字段的问题描述请为这个字段生成清洗规则。 字段名{字段名} 字段类型{字段类型} 问题描述{问题描述} 样本数据最多30个唯一值 {样本值列表} 请根据样本给出清洗规则。清洗类型可选 - regex_replace用正则替换特定模式 - strip_whitespace去除前后空白 - to_datetime解析为统一日期格式给定输入格式和输出格式 - map_value枚举值映射 - filter_invalid过滤非法值 - fill_null空值填充 请以JSON格式返回{{ 字段名: {字段名}, 清洗类型: regex_replace, 参数: {{ pattern: 正则表达式, replacement: 替换为, 说明: 为什么这样做 }} }} 只需要返回JSON。在订单时间字段的清洗任务里我用这个Prompt得到的结果是{ 字段名: 下单时间, 清洗类型: to_datetime, 参数: { input_formats: [%Y-%m-%d, %Y/%m/%d, %Y.%m.%d], output_format: %Y-%m-%d, 说明: 样本中存在三种日期分隔符统一为YYYY-MM-DD } }这个结果和我想的一致。关键价值在于模型能从样本值里自动识别出格式差异省去人工查看数据样式再写格式串的功夫。对数据仓库里几十个时间字段来说这个能力非常实用。5.3 清洗引擎执行代码拿到清洗规则后代码执行部分很直接。我封装了一个apply_clean_rules函数按规则类型分派到不同的清洗逻辑def apply_clean_rules(df: pd.DataFrame, rules: list) - pd.DataFrame: 按Agent生成的规则列表清洗数据 import re from datetime import datetime result_df df.copy() clean_log [] for rule in rules: field rule.get(字段名) clean_type rule.get(清洗类型) params rule.get(参数, {}) if field not in result_df.columns: continue if clean_type regex_replace: pattern params.get(pattern) replacement params.get(replacement, ) result_df[field] result_df[field].astype(str).str.replace( pattern, replacement, regexTrue ) clean_log.append({字段: field, 操作: f正则替换: {pattern} - {replacement}}) elif clean_type strip_whitespace: result_df[field] result_df[field].astype(str).str.strip() clean_log.append({字段: field, 操作: 去除前后空白}) elif clean_type to_datetime: input_formats params.get(input_formats, []) output_format params.get(output_format, %Y-%m-%d) def parse_date(val): val str(val).strip() for fmt in input_formats: try: return datetime.strptime(val, fmt).strftime(output_format) except ValueError: continue return val # 无法解析则保留原值 result_df[field] result_df[field].apply(parse_date) clean_log.append({字段: field, 操作: f日期格式化 - {output_format}}) elif clean_type map_value: mapping params.get(mapping, {}) result_df[field] result_df[field].map(lambda x: mapping.get(x, x)) clean_log.append({字段: field, 操作: f枚举映射: {mapping}}) elif clean_type fill_null: fill_value params.get(fill_value, ) null_mask result_df[field].isna() result_df.loc[null_mask, field] fill_value clean_log.append({字段: field, 操作: f空值填充: {fill_value}}) elif clean_type filter_invalid: valid_values params.get(valid_values, []) if valid_values: result_df result_df[result_df[field].isin(valid_values)] clean_log.append({字段: field, 操作: 过滤非法值}) return result_df, clean_log我在实际清洗订单表时跑下来清洗前有效记录率约87%清洗后提升到99%以上。有一个坑要说一下日期解析失败的行会保留原值这会导致后续分析时报错。所以我在清洗后总会加一条检查语句把无法解析的日期单独抽出来看一眼确认是保留还是转人工处理。5.4 清洗效果评估与回填策略清洗完成不等于结束。我习惯生成三样东西清洗前后数据对比统计、清洗日志明细、被过滤/修改记录的备份。尤其是备份被修改前的原始数据这是必须养成的职业习惯一旦发现问题可以随时回溯否则被人问“你凭什么改我的字段”时会非常被动。对比统计怎么做最常见的是“清洗前有效值数量”和“清洗后有效值数量”的对比表外加每个规则生效的行数。我还会加一个“异常值分布”的简单透视比如清洗前订单金额的负数分布在一个特定区间说明可能是某个支付渠道的反向退款没有正确标识这有助于反推上游修复而不仅仅是下游洗数据。6. 场景三实战元数据与血缘的辅助管理6.1 业务价值与Agent介入点元数据和血缘管理是数据治理里最不受重视但最救命的部分。出了数据问题第一件事永远是“这个字段哪里来的上游怎么算的”没有血缘关系记录排查全靠问人。但建设数据资产目录又是一个大工程很多小团队根本无力投入专人维护。Agent在这里的定位是“被动式文档生成器”把SQL脚本、数据字典、建表语句喂给模型让它自动抽取字段信息、表间依赖关系和业务含义说明。它不能替代完整的数据资产平台但能快速生产一份“够用”的元数据文档和血缘草稿缓解最痛的信息孤岛问题。6.2 Prompt库血缘解析Prompt模板血缘解析的Prompt我设计的重点在于引导模型理解SQL的依赖关系你是一名数据仓库工程师。请你分析下面这段SQL脚本提取数据血缘关系。 SQL脚本 {sql_text} 请输出 1. 脚本中出现的所有表名和字段名 2. 每个输出字段来自哪些输入表的哪些字段字段级血缘 3. 主要的加工逻辑过滤、关联、聚合、计算 4. 表间依赖关系哪个表依赖哪个表 SQL中可能存在多张表、多次CTE临时表请务必理清依赖链条。 请以JSON格式返回 { 输入表: [表A, 表B], 输出表: 表C, 字段血缘: [ {输出字段: 字段X, 来源: 表A.字段A, 加工逻辑: 直接映射} ], 表依赖: [{上游表: 表A, 下游表: 表C, 依赖方式: JOIN}] } 只需要返回JSON。模型对清晰结构的SQL理解得不错但遇到几百行的大脚本时输出字段会开始丢三落四。我实测的经验是脚本超过200行拆成多个子片段分别抽取再合并结果准确率提升非常明显。这也符合Agent工具的一个通用原则一件干不了就拆件干而不是让单次推理硬扛。6.3 字段血缘抽取代码实现血缘抽取的Python代码核心就是构造Prompt、调用模型、解析JSON、合并结果def extract_lineage(sql_path: str) - dict: with open(sql_path, r, encodingutf-8) as f: sql_text f.read() prompt f完整血缘Prompt见上文 SQL脚本 {sql_text} resp_text call_llm(prompt, temperature0.1) lineage parse_llm_json(resp_text) return lineage def extract_lineage_chunked(sql_path: str, chunk_size: int 150): 将长SQL分段解析再合并血缘结果 with open(sql_path, r, encodingutf-8) as f: sql_text f.read() # 按分号简单拆分成子句 stmts [s for s in sql_text.split(;) if s.strip()] all_lineages [] current_stmt for stmt in stmts: current_stmt stmt ; if len(current_stmt.split(\n)) chunk_size: all_lineages.append(extract_lineage_from_text(current_stmt)) current_stmt if current_stmt.strip(): all_lineages.append(extract_lineage_from_text(current_stmt)) return merge_lineages(all_lineages) def extract_lineage_from_text(sql_fragment: str): prompt f血缘Prompt用{fragment}替换SQL脚本位置 return parse_llm_json(call_llm(prompt))这段实现里merge_lineages要处理重复表和重复字段血缘问题我的处理策略是“来源相同则合并来源冲突则并列保留”把这个规则写死在代码里。这样至少不会把两条冲突血缘直接覆盖掉为后续人工审核保留了信息。6.4 数据字典自动补全与维护建议除了血缘Agent还可以顺手把数据字典的“字段说明”和“业务含义”补全。很多系统数据库里的字段名是col_001这种无意义命名看到名字完全不知道是什么。让模型根据字段值样本来猜含义准确率大约七到八成虽然不能直接对外发布但作为初稿让业务方核对修改能节省大量时间。数据字典补全的Prompt我附在这里你是数据资产管理员。下面是一个数据表的字段信息请为每个字段生成一份业务含义说明和命名建议。 字段信息 {字段信息} 返回JSON数组 [{原字段名: col_001, 推断含义: 订单创建时间, 说明: ..., 建议命名: order_create_time}]我的经验是模型对“时间”“金额”“状态”这类通用字段的推断准确度很高但涉及行业特有概念时准确率会下降所以生成结果只能作为辅助参考发布前必须经业务方确认。不过这正是速度上的胜利原来请业务专家逐字段解释一百个字段要一整天现在模型先拟一版专家改错补充一小时内搞定。7. 常见问题与避坑指南实操记录7.1 环境部署类问题问题本地模型服务启动后调用超时这是最常遇到的。原因多数是模型加载到内存后推理速度慢尤其第一轮调用要加载权重耗时可能几十秒。我的处理办法是在正式跑任务前先“预热”模型用一个简单请求让它把权重加载好再进行实际调用。问题容器内存不足模型OOM7B量化模型通常需要6~8GB内存但如果机器本身只有8GB内存再叠加Python应用基本就满了。解决思路是换更小尺寸模型或者更激进的量化等级或者把应用和模型分开部署一个跑在Docker里一个跑在宿主机上。7.2 Prompt与Agent调优类问题问题模型输出偶尔不是合法JSON我用“只需要返回JSON不要输出其他解释”来约束成功率达到九成以上但偶尔还是会输出Markdown代码块或附带说明。我的兜底方案是第一个parse_llm_json函数加Markdown块提取逻辑再失败就重试一次temperature取0。双重兜底后成功率基本100%。问题同一份数据跑两次结果不一致这就是temperature过高导致的。数据治理要求输出确定性我在投产场景里全部使用temperature0.1甚至0.0。有些推理服务支持seed参数固定随机种子能进一步提高一致性。如果业务要求绝对可复现建议设置seed。7.3 数据安全与流程规范类问题问题样本数据包含敏感信息我的原则是任何样本送进模型前先做脱敏处理把姓名、手机号、身份证号替换成脱敏模板。脱敏方式可以简单粗暴固定字符串、加盐哈希、或者用假名库替换。本地部署还有一个好处是模型在本地跑数据不出域安全合规压力小很多。问题Agent误判或漏判导致治理遗漏这个不能完全避免但可以降低。我的做法是维护一个“人工审核规则清单”对Agent识别的高风险直接跳过、中风险人审、低风险自动处理。另外定期用标注过的验证集评估Agent输出质量如果准确率低于设定阈值就调整Prompt或补充示例。Agent是提效工具不是免责工具这个思路要先立住。8. 最终实践心得与后续扩展方向跑完这三个场景我最大的体会是AI Agent数据治理的价值不在于替代人而在于把“找人问、翻文档、试规则”的时间压缩到近乎为零。过去我做大半天的数据质量检查现在半小时内能完成初筛剩下的时间全部放在人工复核和业务沟通上这才是真正的效率改善。最后分享一个使用心得别一上来就追求Agent全自动先让它做“副驾驶”你来做“驾驶员”。从质量稽查、清洗规则生成、血缘梳理这些单个动作开始跑熟之后再逐步把动作串成自动化工作流。我在实际项目中也是先三个场景单独跑稳定之后再考虑串成一条“从质检到清洗再到出报告”的流水线。如果你后续想扩展我建议按这三个方向走一是接入定时调度让Agent周期性巡检数据质量二是把血缘结果迁移到图数据库做可视化分析三是把Prompt库版本化按表类型积累沉淀成组织的治理知识库。等这三步都走完你手里的AI Agent数据治理体系就真正成型了而且全程不依赖重型基础设施。