如果你的工作里也有这几件“重复但必须做”的事：业务部门隔三差五报“这个数不对”，你得翻几张表和一段段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重点说明几个参数。temperature=0是必须的，数据治理场景要求可复现、可审计，同一个输入应该得到稳定输出，所以模型温度必须归零。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_path="config.yaml"): with open(config_path, "r", encoding="utf-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(encoding="utf-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"] = f"Bearer {self.api_key}" resp = requests.post(url, json=payload, headers=headers, timeout=self.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(parents=True, exist_ok=True) (out_dir / f"{task_id}_raw.txt").write_text(raw, encoding="utf-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, }, timeout=30) 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_ascii=False), "抽样数据": json.dumps(samples, ensure_ascii=False)[:8000], "业务规则": json.dumps(rules, ensure_ascii=False), }, task_id="quality_check_order_table_20250101", ) # 按严重级别排序，先处理高风险 issues.sort(key=lambda x: {"high": 0, "medium": 1, "low": 2}[x["severity"]]) print(json.dumps(issues, ensure_ascii=False, indent=2))输出落地我建议做成Markdown表格，直接贴到团队协作平台上，方便大家认领处理。转换逻辑很简单：with open("output/quality_report.md", "w", encoding="utf-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输出稳定如磐石数据治理场景几乎没有“开放式问答”，所有产出都需要被下游程序消费，所以我最关心的永远是输出格式稳定性。第一招是temperature=0，并且全链路统一，绝不因为某个任务“想让它更有创意”就调高。第二招是输出包裹处理，我见过很多模型会在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只是让规则的完善速度变快了好几倍。如果你也想把这套东西落地，我建议从质量巡检这个场景入手，成本最低、见效最快，跑顺了再往里扩。
如果你的工作里也有这几件“重复但必须做”的事：业务部门隔三差五报“这个数不对”，你得翻几张表和一段段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重点说明几个参数。temperature=0是必须的，数据治理场景要求可复现、可审计，同一个输入应该得到稳定输出，所以模型温度必须归零。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_path="config.yaml"): with open(config_path, "r", encoding="utf-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(encoding="utf-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"] = f"Bearer {self.api_key}" resp = requests.post(url, json=payload, headers=headers, timeout=self.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(parents=True, exist_ok=True) (out_dir / f"{task_id}_raw.txt").write_text(raw, encoding="utf-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, }, timeout=30) 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_ascii=False), "抽样数据": json.dumps(samples, ensure_ascii=False)[:8000], "业务规则": json.dumps(rules, ensure_ascii=False), }, task_id="quality_check_order_table_20250101", ) # 按严重级别排序，先处理高风险 issues.sort(key=lambda x: {"high": 0, "medium": 1, "low": 2}[x["severity"]]) print(json.dumps(issues, ensure_ascii=False, indent=2))输出落地我建议做成Markdown表格，直接贴到团队协作平台上，方便大家认领处理。转换逻辑很简单：with open("output/quality_report.md", "w", encoding="utf-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输出稳定如磐石数据治理场景几乎没有“开放式问答”，所有产出都需要被下游程序消费，所以我最关心的永远是输出格式稳定性。第一招是temperature=0，并且全链路统一，绝不因为某个任务“想让它更有创意”就调高。第二招是输出包裹处理，我见过很多模型会在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只是让规则的完善速度变快了好几倍。如果你也想把这套东西落地，我建议从质量巡检这个场景入手，成本最低、见效最快，跑顺了再往里扩。