★ FEATURED ARTICLE 如果你的工作里也有这几件“重复但必须做”的事业务部门隔三差五报“这个数不对”你得翻几张表和一段段SQL去定位到底哪一环出了偏差数据仓库里几十张表字段命名五花八门新人看一眼就头皮发麻合规抽查一来敏感字段散落在各处没人说得清。这些我通通经历过而且不是一次两次是每年都要来几轮的固定节奏。说句实话之前我一直把数据治理当成“费力不讨好的基础工作”直到我把大模型Agent梳理成一套可以复用的“数据治理巡检员”之后才意识到与其让团队里的同学靠肉眼和临时脚本跟数据搏斗不如让Agent先跑一遍粗筛把80%的常规问题一次性列出来剩下20%再由人确认。这篇文章没有云里雾里的概念就是把我在实际项目中沉淀的三套Prompt库、一份轻量化环境搭建方案、以及调试过程里踩过的坑全部摊开。你不需要大数据集群不需要买昂贵的平台用一台普通的笔记本加一个本地模型服务就可以跑通。适合谁来读如果你是做数据开发、数据分析、数据产品或者是一个数据团队的负责人正在被数据质量、字段梳理、敏感数据打标这些事占用大量精力这篇文章可以直接给你一套“抄作业”的方案。1. 先把概念对齐Agent在数据治理里的分工不是“全自动”1.1 数据治理里最耗人的三个环节我在观察团队的工作节奏时发现数据治理里最耗人的从来不是“定规则”而是“执行规则”的低效重复。先说数据质量巡检。传统做法是什么写监控脚本跑定时任务然后等告警。告警产生之后人是整个链路里最忙的打开告警看指标再翻到具体表里去核对数据判断到底是业务侧录入问题还是上游ETL脚本写错了还是口径变了。这个排查过程一天可能来好几轮每一轮都要消耗注意力而且做完了还不一定形成沉淀。再说字段映射与血缘梳理。新来的同事要对接业务指标最痛苦的就是搞不懂字段之间什么关系。老同事可能要花一两个小时指着屏幕讲这个字段是上游A表来的中间经过两次加工口径跟B表那个字段不完全一致……这种知识是隐性知识一旦人走了或者调岗了就变成了断层的坑。想让Agent学会梳理字段关系实际是帮你把隐性知识显性化。最后说敏感数据识别与打标。这个工作看起来简单——把手机号、身份证、地址这些敏感字段标记出来——但真落到几十张表、上千个字段的时候靠正则和关键词字典根本不够。同一类信息在不同系统里的命名天差地别有的叫“客户联系方式”有的叫“mobile”有的叫“手机号码2”怎么统一识别这恰恰是大模型语义理解能力的优势场景。1.2 我踩过的一个大坑把Agent当成“万能帮手”第一次做数据治理Agent的时候我的思路特别“宏伟”让Agent自己连数据库自己跑查询自己判断质量问题自己修数据。结果上线第一周就翻车了。翻车的核心原因不是模型不行而是我把工作边界划错了。模型去查数据库它的SQL经常写得不对要么表名写错要么语法有问题等它终于跑出结果又因为上下文太长判断逻辑开始漂移今天说这个问题重要明天说这个问题不重要。我后来才意识到Agent在数据治理里的合理定位不是“替代数据工程师做全套动作”而是“把最有把握的部分做成高吞吐量的自动粗筛器”。具体来说就是让Agent基于输入的规则和抽样数据做“识别、判断、输出结构化结论”而“连库、取数、修数、落库”这些动作仍然用传统脚本控制。这个架构听起来很朴素但稳定性提升非常大。你想想一个工程师一天能人工巡检两三张表就不错了脚本问Agent一次能同时处理几十张表的抽样把结果汇总成清单人的角色从“逐条发现”变成了“快速复核”。1.3 三个场景的选择逻辑为什么偏偏是这三个场景我在实际项目里迭代了很多轮筛来筛去留下它们有三个原因。第一它们的输入输出边界都足够清晰。质量巡检的输入是表结构和抽样数据输出是问题清单字段映射的输入是源表和目标表结构输出是映射关系敏感识别的输入是字段名加样本输出是类别和级别。边界清晰模型才不会自由发挥。第二它们的工作量大得足以覆盖成本。这三个场景处理起来都是“几百个字段/几十张表”起步的人工活用Agent跑一遍也就几十分钟省下来的人力非常直观。第三它们的产出都便于验证和沉淀。Agent输出的是结构化JSON我可以直接比对人工标注结果算准确率、查漏召回这样每一轮Prompt迭代带来的变化都是可测量的而不是“感觉比之前好了一点”。2. 轻量化环境搭建一套Python脚本就能当Agent底座2.1 为什么我坚持“不碰重框架”现在一聊Agent很多人第一反应就是上某个重量级编排框架还要配向量库、配记忆系统。但我的观点很直接如果你是像我一样处理数据治理文本分析这类任务重框架带来的复杂度远大于收益。数据治理Agent的核心链路其实非常短读配置 → 读Prompt模板 → 填参 → 调模型接口 → 解析JSON → 写结果。这条链路上最不值得投入的环节就是“加一层抽象”。框架能帮你解决的并发、记忆、规划问题在你的场景里要么不需要要么自己用几十行代码就能处理干净。所以我这边的底座就是Python环境 一个兼容标准API格式的模型服务 一个执行器脚本。就这么简单。模型服务你可以根据自己的算力选本地部署选支持标准对话接口的即可没有GPU的话也可以调一台云端API实例关键是把接口统一成同一格式脚本完全不用换。2.2 目录结构与配置文件先看目录结构麻雀虽小五脏俱全data_gov_agent/ ├── config.yaml ├── runner.py ├── prompts/ │ ├── quality_check.txt │ ├── field_mapping.txt │ └── sensitive_classify.txt ├── input/ ├── output/ └── logs/config.yaml是唯一需要你手动改的地方model: api_base: http://127.0.0.1:11434/v1 api_key: model_name: your_local_model temperature: 0 max_tokens: 4096 timeout_seconds: 120重点说明几个参数。temperature0是必须的数据治理场景要求可复现、可审计同一个输入应该得到稳定输出所以模型温度必须归零。max_tokens设到4096一般够用因为单个表的字段也就几十个如果一张表有两三百个字段建议你先分片再调用而不是一个劲调大上限这个在后文会细讲。api_key在本地服务场景往往不需要校验留空就行如果你接的是云上兼容接口换成真实密钥即可。2.3 Agent执行器代码这个执行器我称之为“Prompt管道”它做的事情就是模板填充、调用、解析、存储。核心代码不长import json import time import yaml import requests from pathlib import Path from datetime import datetime class DataGovAgent: def __init__(self, config_pathconfig.yaml): with open(config_path, r, encodingutf-8) as f: self.cfg yaml.safe_load(f) m self.cfg[model] self.api_base m[api_base].rstrip(/) self.api_key m.get(api_key, ) self.model_name m[model_name] self.temperature m.get(temperature, 0) self.max_tokens m.get(max_tokens, 4096) self.timeout m.get(timeout_seconds, 120) self.prompts_dir Path(prompts) def load_template(self, name: str) - str: path self.prompts_dir / f{name}.txt return path.read_text(encodingutf-8) def _call(self, messages): url self.api_base /chat/completions payload { model: self.model_name, messages: messages, temperature: self.temperature, max_tokens: self.max_tokens, } headers {} if self.api_key: headers[Authorization] fBearer {self.api_key} resp requests.post(url, jsonpayload, headersheaders, timeoutself.timeout) resp.raise_for_status() return resp.json()[choices][0][message][content] def run(self, template_name: str, placeholders: dict, task_id: str): template self.load_template(template_name) for key, value in placeholders.items(): template template.replace({{ key }}, str(value)) raw self._call([ {role: system, content: 你是数据治理专家只输出JSON不接受其他形式。}, {role: user, content: template}, ]) self._save_raw(task_id, raw) return self._parse_json(raw) def _save_raw(self, task_id, raw): out_dir Path(logs) / datetime.now().strftime(%Y%m%d) out_dir.mkdir(parentsTrue, exist_okTrue) (out_dir / f{task_id}_raw.txt).write_text(raw, encodingutf-8) def _parse_json(self, raw: str): # 兼容模型输出里常见的json包裹 if in raw: start raw.find(json) start raw.find(\n, start) 1 if start ! -1 else raw.find() 3 end raw.rfind() raw raw[start:end].strip() return json.loads(raw)这段代码是不是很其貌不扬但它的价值在于把三个场景的调用方式统一了。后面无论跑质量巡检还是敏感打标都是同一个run方法只是模板名和占位符不同。可别小看这几十行我在项目里跑了几千次调用稳定得很。2.4 环境验证先跑通一个最小样例环境装好之后不要急着跑正式任务先跑一个最小样例验证链路。agent DataGovAgent(config.yaml) url agent.api_base /chat/completions resp requests.post(url, json{ model: agent.model_name, messages: [{role: user, content: 只输出JSON{status:ok}}], temperature: 0, }, timeout30) print(resp.json()[choices][0][message][content])这一步能同时验证三件事模型服务是否正常、接口格式是否兼容、认证配置是否正确。我见过不少人一上来就把正式数据喂进去结果报出一堆接口错误还以为是Prompt写得不好其实环境根本没通。先跑最小样例是成本最低的排障手段。3. 场景一数据质量巡检Prompt库3.1 业务痛点告警之后的人才拉锯战数据质量巡检的痛点不在“没有规则”而在“规则执行完之后还要人肉去看证据”。传统脚本能告诉你“这个字段空值率超过20%”但它说不清这个空值到底是谁造成的影响哪张报表是不是合理空值。我用Agent替代的是“人工看样本找线索”这一环。输入表结构和抽样数据Agent会逐字段检查把异常点、样例值、可疑原因一次性给出来。团队拿到的是结构化问题单直接顺着问题单复核效率完全不是一个量级。3.2 Prompt模板这是我在模拟项目X里实际用了一版模板你也可以直接复制改名用。你是一名数据质量工程师。请根据下面提供的表结构、抽样数据和业务规则完成数据质量巡检。 表结构JSON {{表结构}} 抽样数据JSON最多显示前100行 {{抽样数据}} 业务规则JSON {{业务规则}} 检查要求 1. 检查每个字段的空值率、重复率、格式合法性、取值范围合理性。 2. 对文本类字段检查是否有明显的枚举值越界、首尾空格、全半角混乱。 3. 对日期字段检查格式是否统一是否存在1900及1970等默认日期。 4. 对数值字段检查是否出现负数、超范围值、单位异常放大。 5. 对存在指向性关联的字段检查是否有失配如性别与称呼矛盾、数量与金额方向不一致。 输出要求 仅输出一个JSON数组不要输出任何解释性文字。每个问题项的格式如下 [ { field: 字段名, issue_type: 空值率过高, severity: high|medium|low, sample_values: [样例值1, 样例值2], reason: 一句话说明为什么判断这是问题, suggestion: 一句话给出处理建议 } ] 如果某类检查没有发现问题该字段或规则不需要出现在数组中。这个模板有三个关键设计。第一业务规则作为独立占位符传入这样不同表可以接不同口径而不是把所有规则写死在Prompt里。第二明确要求“不要输出解释性文字”从源头压制模型废话。第三每个问题项强制带reason和suggestion既方便复核人判断也顺手把问题知识沉淀成文档。3.3 调用方式与结果落地调用同样非常简单agent DataGovAgent(config.yaml) schema {...} # 从元数据系统导出的表结构JSON samples [...] # 抽样数据 rules {...} # 业务规则 issues agent.run( quality_check, placeholders{ 表结构: json.dumps(schema, ensure_asciiFalse), 抽样数据: json.dumps(samples, ensure_asciiFalse)[:8000], 业务规则: json.dumps(rules, ensure_asciiFalse), }, task_idquality_check_order_table_20250101, ) # 按严重级别排序先处理高风险 issues.sort(keylambda x: {high: 0, medium: 1, low: 2}[x[severity]]) print(json.dumps(issues, ensure_asciiFalse, indent2))输出落地我建议做成Markdown表格直接贴到团队协作平台上方便大家认领处理。转换逻辑很简单with open(output/quality_report.md, w, encodingutf-8) as f: f.write(| 字段 | 问题类型 | 严重级别 | 建议 |\n) f.write(| --- | --- | --- | --- |\n) for it in issues: f.write(f| {it[field]} | {it[issue_type]} | {it[severity]} | {it[suggestion]} |\n)3.4 实测效果与边界情况在模拟项目X的订单、用户、结算三张核心表上我用这套Prompt巡检了约50个字段人工复核后整体准确率在85%左右。表现最好的是空值率、默认日期、数值范围这三类几乎一抓一个准。表现最不稳定的是“业务口径类”问题比如某个字段允许为空但业务规则里没写清楚模型会把它误判成异常。后来我发现一个简单有效的办法遇到这类误判不要改Prompt而是把这条规则补进业务规则JSON里让业务方明确“该字段某些场景下为空是常态”。这实际上是让Agent帮你把业务规则的缺口暴露出来属于意外收获。再补一个采样细节。抽样数据不能太少字段枚举型的检验至少需要看到分布我一般取1000行内随机抽样能在准确度和成本之间平衡。如果一张表特别宽字段超过80个建议按业务域拆成子表分两次巡检避免单次调用上下文过长导致模型后半段“注意力涣散”。4. 场景二字段映射与血缘梳理Prompt库4.1 为什么字段映射最值得用Agent字段映射是个特别微妙的工作。它不像质量巡检那样有“非黑即白”的规则更多依赖语义理解。比如源表里有个字段叫“cust_phone”目标表里叫“客户联系电话”常规程序很难自动匹配但大模型几乎一眼就能识别。我做的字段映射Agent输入是源表结构、目标表结构以及一段可选的SQL映射线索输出是字段级映射关系带置信度评分。有了置信度后续人工复核就可以优先看低置信度的映射高置信度的直接批量采用。4.2 Prompt模板你是一名资深数据架构师。请完成源表字段到目标表字段的映射识别。 源表结构JSON {{源表结构}} 目标表结构JSON {{目标表结构}} 已知线索SQL可选 {{线索SQL}} 映射要求 1. 只允许使用给定表结构中真实存在的字段名不得自创字段名。 2. 对于语义等价、但命名不同的字段必须给出映射并说明理由。 3. 对于无法确定映射关系的字段跳过不要猜测。 4. 如果一个目标字段可能由多个源字段拼接或计算得出请标注推导逻辑。 输出要求 仅输出一个JSON数组格式如下 [ { source_field: 源字段名, target_field: 目标字段名, confidence: high|medium|low, mapping_type: direct|rename|concatenate|aggregate, reason: 说明为什么这样映射 } ]这里我踩过一个特别深的坑模型会“创造性补全”字段名。你明明没给它这个字段它却会输出一个语义上很像但实际不存在的名字。所以我在Prompt里专门加了一条“只允许使用给定表结构中真实存在的字段名不得自创字段名”。这条约束加上之后虚构字段的问题几乎消失。4.3 结果评估别只看对错要看置信度字段映射的输出必须做评估否则没法放心批量使用。我的做法是拉三个人工标注作为基准把Agent输出跟基准比对按置信度分组统计准确率。在一次对“新老订单系统”的字段迁移测试中总共涉及源表42个字段、目标表36个字段Agent给出的有效映射是31条。人工复核结果如下high置信度16条准确率100%medium置信度9条准确率78%low置信度6条准确率33%。整体准确率约71%。看起来不高但配合置信度排序后团队实际需要人工仔细看的只有15条省下的时间是实实在在的。有一个容易被忽略的点如果目标字段本来就少映射不出来不等于Agent能力差还可能说明源表和目标表之间存在大量计算加工。遇到这种情况我会把线索SQL填进去模型读到SQL后对拼接、聚合类映射的表现会好非常多。4.4 常见翻车点与修正策略自创字段名已用硬性约束解决。过度自信。模型经常对拿不准的映射也标high置信度所以我在输出要求里加了reason字段要求说明依据人工复核时可以直接瞄reason判断它到底是不是在瞎猜。同名不同义。有些字段在不同表里名字一样但业务含义完全不同。这个问题模型基本判断不了只能靠业务规则JSON补充。我会在映射跑完后把低置信度加业务关键字段的组合优先交给业务人员确认。我还试过把映射结果反向用起来生成一份“字段对照表”直接沉淀到数据字典文档里。这样新同事接手时不用再靠老同事讲故事看表就能明白七八成。5. 场景三敏感数据识别与分级打标Prompt库5.1 合规需求下的老大难数据分级分类现在几乎是任何数据平台都躲不开的硬性要求。难点在于敏感字段的识别不能只靠正则。手机号有正则但“手机号码2”“客户电话”“联系方式”这类字段名正则根本覆盖不过来还得靠前后语料判断它是个人手机号还是公司座机这就进入语义判断领域了。传统方案是维护一个巨大的关键词字典字段名里包含“手机”“电话”“座机”就标记为敏感。但这种方式误报率极高比如“客服热线电话”其实可能是企业对外服务号码不是个人敏感信息。我改用Agent后把“字段名样本值注释说明”一起交给模型判断准确率明显上了一个台阶。5.2 Prompt模板与等级体系我的敏感识别Prompt围绕一套分级体系来设计。分级原则不需要写得太学术关键是让模型容易对齐数据敏感级别定义 L4-极敏感个人实名类姓名身份证组合、支付类数据银行卡号、支付密码、生物识别数据。 L3-高度敏感手机号、家庭住址、精确位置、健康医疗信息、账户密码类。 L2-中度敏感性别、年龄、学历、收入区间、企业邮箱、部分业务明细。 L1-低敏感部门名称、职位、办公所在地、脱敏后的统计指标。 请根据以下表结构、样本数据和字段注释对所有字段进行敏感识别和分级。 表结构JSON {{表结构}} 样本数据JSON {{样本数据}} 字段注释JSON可选 {{字段注释}} 识别要求 1. 不能仅凭字段名判断要结合样本值和注释综合判断。 2. 如果字段是脱敏或加密后的数据等级可下调一级并在备注中说明。 3. 同一字段在不同表中可能等级不同以当前表的数据内容和用途为准。 输出要求 仅输出一个JSON数组格式如下 [ { field: 字段名, sensitive_level: L1|L2|L3|L4, category: 个人信息|支付信息|业务信息|设备信息, sample_evidence: 简要说明判断依据, suggestion: 建议的脱敏或保护措施 } ]这里有个设计细节我想强调我把“字段注释”设计成可选输入。很多数据平台在建表时根本没有注释但有些表有哪怕只有一两句对模型判断也很有帮助。所以我在主流程里会先读元数据有注释就传注释没有就传空JSON不要因为这个可选字段阻塞主流程。5.3 落地方式必须加一道人工复核闸门敏感识别的产出不是给你直接上生产的而是生成一个“候选打标清单”交给数据安全岗的人工复核。这个闸门一定不能省因为模型偶尔会把某些看起来像个人信息、实则匿名化的字段判得很高。我实际跑完后会把输出转成一个待办清单按敏感级别倒序排列人工只需要从上往下点确认或修改。确认过的记录直接回写到元数据管理系统的标签字段里。这个链路跑顺了之后原来可能要花两周的字段盘点现在两天内就能出一版初稿。5.4 实测效果在一批包含用户、订单、日志共18张表、289个字段的测试集上Agent识别出L3及以上敏感字段47个。人工复核结果是查出真敏感字段43个误报4个漏报2个。召回率超过95%精确率约91%。有意思的是漏报的两个字段都出现在日志表里字段名和样本值都很不典型比如一个叫“resv”的字段存的是个人预订信息简写。这种边界case靠任何自动化方案都很难完全兜住。所以我的建议是不要追求100%召回而是把Agent当成“高效的第一轮筛查”配合周期性的规则正则兜底两层叠加已经足够实用。6. 从“能跑”到“能长期用”稳定性、审计与成本控制三个场景跑通之后真正拉长周期考验人的不是Prompt写得好不好而是这套东西能不能在生产里稳定运行三个月。这一节把我认为最关键的细节都列出来。6.1 让JSON输出稳定如磐石数据治理场景几乎没有“开放式问答”所有产出都需要被下游程序消费所以我最关心的永远是输出格式稳定性。第一招是temperature0并且全链路统一绝不因为某个任务“想让它更有创意”就调高。第二招是输出包裹处理我见过很多模型会在JSON外面包一层json代码块也可能因为token超限被截断所以解析函数里的兜底逻辑必须健壮。第三招是可选的结构化输出能力如果你用的模型服务支持JSON Schema约束直接在请求参数里声明输出格式是最好不过了不支持也没关系模板加温度0已经能把异常率压在5%以内。6.2 宽表与长上下文的处理数据治理遇到的最大上下文压力通常来自“字段太多的宽表”。我的经验是80个字段以上的单次巡检就开始不稳定所以坚决分片。把一张200字段的大宽表按业务域拆成三块每块单独跑一次最后汇总。占总处理时间多10%但准确率提升非常明显。另外样本数据字符串长度也要管住。我在代码里对抽样数据做了截断强制不超过8000字符。因为对模型来说样本的目的只是“看到分布和样例形态”给太多行反而会稀释注意力让模型把时间花在无关细节上。6.3 留痕与可回溯生产环境里任何一条Agent结论都必须能回溯什么时间跑的、哪个模型跑的、输入的到底是什么、原始输出是什么。这就要求每调用一次就把原始请求和回包留到本地日志目录。我在runner.py里专门写了_save_raw方法每个task_id对应一次完整的输入输出记录。这样做的好处是当某人拿着打标结果来问“当初为什么把L4标成L3”的时候我能直接打开日志给人看原文而不是靠记忆争辩。数据治理本身讲究的就是可追责Agent环节也一样。6.4 成本与批量调度的配合成本方面我算过一笔账用一个14B参数规模的开源模型服务处理一张50字段的表单次调用大概消耗3000到5000 token。三张核心表的全量巡检成本在几分钱到几毛钱级别跟省下来的人力时间比完全不是一个量级。调度上建议不要一股脑并发拉满。数据治理任务的调用频率不会特别高可以做成一个简单的批处理脚本每隔一段时间处理一批写入结果后生成汇总报告。不需要复杂的消息队列一个循环加一个sleep就够用。这也是我一直强调的轻量化不是口号是真正把这个场景的复杂度压回它该有的样子。最后再说一个我自己的体会这三套Prompt库用了一个月之后我发现最有价值的产出其实不是那份问题清单而是每次Agent输出和人工复核之间的差异。团队成员从“盯着Agent的结果检查”慢慢变成了“盯着差异反推规则”把很多以前没有显式化的业务口径补进了规则JSON里。数据治理的本质是把说了算的规则变成能自动跑的流程Agent在这里面不是主角规则才是Agent只是让规则的完善速度变快了好几倍。如果你也想把这套东西落地我建议从质量巡检这个场景入手成本最低、见效最快跑顺了再往里扩。
阅读完成 · 觉得有帮助？
37个Life Skills知识点使用指南)
