★ FEATURED ARTICLE 大家在日常开发或者数据测试中肯定遇到过这种场景业务提了一个新需求底层表结构改了一版测试环境的数据质量没人敢保证需要写一堆检测SQL去验证数据是否正确。这类SQL检测脚本往往逻辑重复度高、规则琐碎但手写起来又非常耗时。我最近在实际项目中把这块工作做了一个“Skill化”改造将编写SQL检测脚本的过程封装成了一套可复用的技能包在典型场景下脚本产出速度提升了数倍测试数据排查的效率也跟着上去了。本文就把这套思路、完整代码和落地经验整理出来希望能给同样被SQL脚本折磨的测试开发、数据开发同学一些参考。1. 为什么要做“SQL检测脚本生成 Skill”1.1 手工编写检测脚本的痛点日常工作中我们经常要写这样几类SQL脚本表结构变更后的数据校验脚本。测试环境脏数据排查脚本。接口联调前的数据准备脚本。数据仓库字段口径核对脚本。这些脚本看起来不难但写起来很烦琐。比如要排查一张订单表的异常数据你需要考虑主键是否重复、金额是否为负数、状态字段是否合法、时间字段是否为空、关联维度表是否有缺失等等。一个经验丰富的开发写一套完整的检测脚本往往也需要十几分钟而且写完还要反复检查有没有漏掉规则。更大的问题是这类脚本的“经验”通常只存在于个人脑子里。新人接手时要么不知道要检测哪些字段要么写的SQL不严谨导致测试环境数据问题没有提前暴露直接带上联调甚至上线阶段补救成本成倍增加。1.2 Skill 在这里意味着什么项目标题里提到的“Skill”你可以先把它理解成一个结构化的“技能包”。它不是某个框架也不是某个中间件而是一套组合规则库沉淀常用的SQL质量检测规则比如空值检测、重复检测、枚举合法性检测、边界值检测。提示词模板把用户的检测需求转换成大模型能理解的任务描述让大模型生成对应SQL脚本。工具函数用Python实现SQL文本的解析、规则追加、结果整理让生成出来的脚本可以直接执行、快速定位问题数据。示例库放一批典型业务场景的输入输出示例例如订单表、用户表、商品表让Skill的回答更稳定。简单来说就是把“写检测SQL”这件事从纯手工变成“描述需求 - 自动生成脚本 - 快速执行 - 输出报告”从而把重复劳动压缩掉。1.3 适用场景与读者本文内容比较适合下面几类读者经常需要在测试环境写SQL排查数据的测试开发、后端开发。想要把大模型能力接入日常工作流的效率工具爱好者。正在做数据质量平台、SQL审核工具相关工作的研发同学。接下来我会先从整体架构讲起然后给出Skill的配置方法、Python实现代码以及一个完整的实战案例。最后会补充常见问题和排查思路帮助大家在实际落地时少走弯路。2. 整体设计与工作流程2.1 系统模块划分整个“SQL检测脚本生成 Skill”可以拆成四个模块模块负责内容关键点规则库模块定义SQL检测规则常见规则沉淀为模板片段技能配置模块存放系统提示词、示例、输出格式让大模型清楚自己的角色脚本生成模块调用大模型API生成SQL脚本核心代码支持OpenAI兼容接口执行与报告模块执行生成的SQL输出问题数据强调只读、安全、可回滚这四个模块组合起来就构成一个完整的“生成 - 执行 - 排查”闭环。2.2 工作流程说明整个流程大致如下用户在入口处描述需求例如“请生成一张订单明细表的检测脚本重点排查订单金额异常、状态非法、会员ID不存在的数据”。Skill先对需求做解析匹配规则库中的模板。大模型根据系统提示词和用户需求生成一组SQL检测脚本同时输出每个检测点的说明。Python脚本自动整理输出将每个检测SQL单独拆分成可执行的文件或代码片段。开发或测试人员拿到脚本后在测试环境执行根据输出结果定位问题数据。这样设计的好处是人工只需负责描述业务场景和最终确认重复的SQL编写过程由Skill完成。而且因为规则库是持续沉淀的后续同类场景的生成质量和速度会越来越高。3. 环境准备与版本说明3.1 运行环境本文的示例代码以Python为主建议使用Python 3.9及以上版本。我本地的开发环境如下你可以根据自己的项目实际情况调整操作系统Windows 10 / macOS / Linux均可。Python版本3.10。依赖库requests用于调用大模型API。大模型服务任意支持OpenAI Chat Completions兼容接口的模型服务不管你用的是哪家只要接口格式兼容即可。如果还没有安装requests库可以执行下面这个安装命令pip install requests3.2 模型API准备因为不同项目和公司的模型服务地址、API Key都不一样这里不写死某个具体平台。你需要准备以下信息API地址例如https://your-api-endpoint/v1/chat/completions。API Key用于鉴权。模型名称例如gpt-4o-mini、deepseek-chat、qwen-plus等具体以你实际使用的服务为准。如果你是在本地用私有化模型比如通过vLLM、Ollama部署的模型只要提供了OpenAI兼容接口也可以直接沿用下面的代码。这个兼容性设计可以让Skill在不同团队之间迁移时更灵活。3.3 项目目录结构为了让代码清晰可维护我建议按下面的结构组织项目sql_skill/ ├── skill_config.py # Skill提示词与配置 ├── rule_lib.py # 检测规则库 ├── script_generator.py # 调用大模型生成脚本 ├── script_executor.py # 执行SQL脚本并输出结果 ├── run_skill.py # 主入口 └── output/ # 生成脚本的输出目录如果你只是练手也可以把全部代码写到一个文件里。但实际项目中按模块拆分会让后续扩展规则库、增加检测类型更加简单。4. 核心原理与实现代码4.1 Skill配置模块Skill配置是整个方案的“大脑”。我们需要在系统提示词中明确告诉大模型你的角色是什么要输出什么格式有哪些注意事项。下面是我的skill_config.py完整内容# 文件路径sql_skill/skill_config.py DETECT_SKILL_SYSTEM_PROMPT 你是一名资深的数据测试开发工程师擅长编写SQL检测脚本帮助测试人员在测试环境中快速发现数据质量问题。 你的任务是根据用户提供的表结构和检测需求生成一组MySQL兼容的SQL检测脚本。 要求 1. 每个检测点单独输出一条SELECT语句不要合并成复杂的嵌套查询。 2. 每条SQL必须包含注释说明检测目的。 3. 检测规则必须覆盖空值、重复值、边界值、枚举合法性、外键关联完整性。 4. 生成的SQL必须是只读查询禁止生成INSERT、UPDATE、DELETE语句。 5. 如果用户需求不明确先列出你理解的检测点再生成SQL。 6. 输出格式使用JSON包含detect_points和sql_list两个字段。 输出格式示例 { detect_points: [订单金额为空, 订单状态非法, 会员ID不存在], sql_list: [ { point: 订单金额为空, sql: SELECT order_id, order_amount FROM ods_order_info WHERE order_amount IS NULL;, explain: 查询订单金额为空的记录 } ] } 这里有几个关键点要求大模型“单独输出每条SELECT语句”是为了让后续Python脚本拆解和按行执行更简单。明确禁止生成INSERT、UPDATE、DELETE是为了避免在测试环境误操作数据这也是安全底线。使用JSON格式输出方便Python解析。如果大模型输出不稳定可以在后续代码里做兼容处理。除了系统提示词我还会在配置里补充一个“行业经验提示”让大模型生成的检测逻辑更完整# 文件路径sql_skill/skill_config.py BUSINESS_RULES_HINT 补充业务经验 1. 对于金额字段除了检查NULL还要检查是否小于0或大于业务上限。 2. 对于日期字段要检查格式是否为yyyy-MM-dd以及是否出现1900-01-01或2099-12-31这类边界值。 3. 对于状态字段要检查是否出现了枚举值以外的数据。 4. 对于关联ID字段要检查是否存在悬空引用即关联表查不到对应记录。 5. 对于文本字段要检查是否包含空字符串、空格串、乱码字符。 4.2 规则库模块规则库的作用有两个一是给大模型提供参考模板二是作为本地兜底检查。当大模型生成的SQL有遗漏时我们可以用本地规则库做二次补充。下面是rule_lib.py的简化版代码# 文件路径sql_skill/rule_lib.py COMMON_RULES [ { rule_name: 非空检查, template: SELECT * FROM {table} WHERE {column} IS NULL;, description: 检查指定字段是否存在空值 }, { rule_name: 唯一性检查, template: SELECT {column}, COUNT(*) AS cnt FROM {table} GROUP BY {column} HAVING COUNT(*) 1;, description: 检查指定字段是否存在重复值 }, { rule_name: 边界值检查, template: SELECT * FROM {table} WHERE {column} {min_value} OR {column} {max_value};, description: 检查字段值是否超出业务边界 }, { rule_name: 枚举值检查, template: SELECT DISTINCT {column} FROM {table} WHERE {column} NOT IN ({enum_values});, description: 检查字段是否出现非法枚举值 }, { rule_name: 关联完整性检查, template: SELECT t1.* FROM {table} t1 LEFT JOIN {ref_table} t2 ON t1.{fk_column} t2.{pk_column} WHERE t2.{pk_column} IS NULL;, description: 检查外键关联是否存在悬空引用 } ] def build_rule_prompt(table_name, columns_info): 根据表名和字段信息生成规则提示供大模型参考 lines [] lines.append(f针对表 {table_name}请参考以下检测规则) for rule in COMMON_RULES: lines.append(f- {rule[rule_name]}{rule[description]}) if columns_info: lines.append(字段信息如下) for col in columns_info: lines.append(f- 字段名{col[name]}类型{col[type]}说明{col.get(comment, )}) return \n.join(lines)这个模块的价值在于把团队里沉淀的检查经验显性化。哪怕大模型没有见过你这张表只要给它字段信息它也能结合规则库生成比较完整的检测SQL。4.3 脚本生成模块脚本生成模块负责调用大模型的API并解析返回结果。这里我们用requests直接调用OpenAI兼容接口这样不依赖特定SDK方便换模型服务。# 文件路径sql_skill/script_generator.py import json import requests from skill_config import DETECT_SKILL_SYSTEM_PROMPT, BUSINESS_RULES_HINT from rule_lib import build_rule_prompt class SQLScriptGenerator: def __init__(self, api_key, api_url, model_name): self.api_key api_key self.api_url api_url self.model_name model_name def build_user_message(self, table_name, columns_info, requires): 构造用户消息包含表结构、需求和规则提示 sections [] sections.append(f请针对业务表 {table_name} 生成SQL检测脚本。) if columns_info: sections.append(表结构字段信息如下) for col in columns_info: sections.append(f- {col[name]} ({col[type]}): {col.get(comment, )}) if requires: sections.append(检测需求如下) for req in requires: sections.append(f- {req}) sections.append(build_rule_prompt(table_name, columns_info)) sections.append(BUSINESS_RULES_HINT) return \n.join(sections) def generate(self, table_name, columns_info, requires): 调用大模型生成检测SQL返回解析后的结果 user_message self.build_user_message(table_name, columns_info, requires) payload { model: self.model_name, messages: [ {role: system, content: DETECT_SKILL_SYSTEM_PROMPT}, {role: user, content: user_message} ], temperature: 0.2, response_format: {type: json_object} } headers { Content-Type: application/json, Authorization: fBearer {self.api_key} } try: resp requests.post(self.api_url, jsonpayload, headersheaders, timeout60) resp.raise_for_status() data resp.json() content data[choices][0][message][content] return self._parse_response(content) except Exception as e: print(f[生成失败] {e}) return None def _parse_response(self, content): 解析大模型返回的JSON内容同时做容错 # 如果返回内容包含多余的Markdown代码块标记先去掉 content content.strip() if content.startswith(): lines content.splitlines() lines lines[1:] if lines and lines[-1].strip() : lines lines[:-1] content \n.join(lines) try: return json.loads(content) except json.JSONDecodeError: # 尝试提取JSON片段 start content.find({) end content.rfind(}) if start ! -1 and end ! -1: return json.loads(content[start:end1]) raise ValueError(大模型返回内容无法解析为JSON)这段代码里要说几个点temperature设置为0.2是为了让输出更稳定减少随机性。生成SQL脚本不需要太多创造性稳定优先。response_format字段用来要求JSON输出但不是所有模型服务都支持。如果报错可以去掉这一行靠系统提示词约束。_parse_response做了Markdown代码块的清理因为有些模型会在JSON外面包一层标记。4.4 脚本执行模块生成脚本后需要把它落到具体文件里方便后续在数据库客户端执行。这一层也可以对接你的数据库连接工具但生产环境中更推荐“生成文件 - 人工审阅 - 执行”的流程。# 文件路径sql_skill/script_executor.py import os import datetime class SQLScriptWriter: def __init__(self, output_diroutput): self.output_dir output_dir os.makedirs(output_dir, exist_okTrue) def write_scripts(self, table_name, result): 将大模型生成的SQL检测脚本写入文件并生成说明文档 if not result: print(没有可写入的脚本) return None timestamp datetime.datetime.now().strftime(%Y%m%d_%H%M%S) detect_points result.get(detect_points, []) sql_list result.get(sql_list, []) # 生成一个汇总SQL文件 sql_file os.path.join(self.output_dir, fdetect_{table_name}_{timestamp}.sql) with open(sql_file, w, encodingutf-8) as f: f.write(f-- 表的检测脚本生成时间: {timestamp}\n) f.write(f-- 检测点数量: {len(detect_points)}\n\n) for item in sql_list: point item.get(point, unknown) sql item.get(sql, ) explain item.get(explain, ) f.write(f-- 检测点: {point}\n) f.write(f-- 说明: {explain}\n) f.write(f{sql}\n\n) # 生成一个Markdown报告方便在工单中粘贴 md_file os.path.join(self.output_dir, fdetect_{table_name}_{timestamp}.md) with open(md_file, w, encodingutf-8) as f: f.write(f# {table_name} 数据质量检测报告\n\n) f.write(f- 生成时间{timestamp}\n) f.write(f- 检测点数量{len(detect_points)}\n\n) f.write(## 检测结果说明\n\n) for item in sql_list: point item.get(point, unknown) sql item.get(sql, ) explain item.get(explain, ) f.write(f### {point}\n\n) f.write(f- 说明{explain}\n) f.write(f- SQL脚本\n\nsql\n{sql}\n\n\n) print(f已生成SQL文件: {sql_file}) print(f已生成报告文件: {md_file}) return sql_file这个模块把模型返回的结果落成两种文件SQL文件用于执行Markdown报告用于展示和归档。在执行之前一定要经过人眼确认不要直接拿生成脚本就往生产库跑。4.5 主入口最后写一个入口文件把上面的模块串起来。# 文件路径sql_skill/run_skill.py from skill_config import DETECT_SKILL_SYSTEM_PROMPT, BUSINESS_RULES_HINT from script_generator import SQLScriptGenerator from script_executor import SQLScriptWriter def main(): # 配置模型服务按实际环境填写 API_KEY your-api-key API_URL https://your-api-endpoint/v1/chat/completions MODEL_NAME your-model-name generator SQLScriptGenerator(API_KEY, API_URL, MODEL_NAME) writer SQLScriptWriter() # 示例表信息 table_name ods_order_info columns_info [ {name: order_id, type: bigint, comment: 订单ID}, {name: member_id, type: bigint, comment: 会员ID}, {name: order_amount, type: decimal(10,2), comment: 订单金额}, {name: order_status, type: varchar(20), comment: 订单状态}, {name: create_time, type: datetime, comment: 下单时间} ] requires [ 排查订单金额为空或小于0的数据, 排查订单状态不是合法枚举值的数据, 排查member_id在会员表中不存在的数据, 排查下单时间异常的数据 ] result generator.generate(table_name, columns_info, requires) if result: writer.write_scripts(table_name, result) else: print(脚本生成失败) if __name__ __main__: main()运行方式很简单python run_skill.py运行成功后output目录下会多出两个文件分别是SQL脚本文件和Markdown报告文件。5. 实战案例生成一份订单表数据检测SQL5.1 业务背景假设我们负责一个电商项目的测试环境。业务最近调整了订单流程新增了一个“预约单”状态。我们需要在测试环境验证历史订单数据是否正常重点排查订单金额、订单状态、会员ID关联和下单时间这几个维度。按照传统方式我们需要打开数据库客户端写下多条查询SQL还得一条条复制执行。现在有了Skill只需要把自己的需求描述清楚然后运行脚本即可。5.2 生成结果参考下面是我用类似配置跑出来的一份示例输出内容和字段会根据模型能力有差异这里展示核心结构{ detect_points: [ 订单金额为空, 订单金额小于0, 订单状态非法, 会员ID不存在, 下单时间为空, 下单时间早于2020年 ], sql_list: [ { point: 订单金额为空, sql: SELECT order_id, member_id, order_amount FROM ods_order_info WHERE order_amount IS NULL;, explain: 查询订单金额字段为空的记录金额字段不允许为空 }, { point: 订单金额小于0, sql: SELECT order_id, member_id, order_amount FROM ods_order_info WHERE order_amount 0;, explain: 查询订单金额为负数的记录属于异常业务数据 }, { point: 订单状态非法, sql: SELECT order_id, order_status FROM ods_order_info WHERE order_status NOT IN (待支付, 已支付, 已发货, 已完成, 已取消, 预约单);, explain: 查询订单状态不在枚举范围内的记录 }, { point: 会员ID不存在, sql: SELECT t1.order_id, t1.member_id FROM ods_order_info t1 LEFT JOIN dim_member_info t2 ON t1.member_id t2.member_id WHERE t2.member_id IS NULL;, explain: 查询会员ID在会员维度表中不存在的悬空引用记录 }, { point: 下单时间为空, sql: SELECT order_id, create_time FROM ods_order_info WHERE create_time IS NULL;, explain: 查询下单时间为空的记录 }, { point: 下单时间异常, sql: SELECT order_id, create_time FROM ods_order_info WHERE create_time 2020-01-01 00:00:00;, explain: 查询下单时间早于业务起始时间的记录 } ] }可以看到生成结果基本覆盖了需求中的几个检测点还补充了“下单时间为空”和“下单时间异常”两个检测点。这些检测点对于测试人员来说都是有效信息可以直接拿去执行。5.3 执行与排查示例实际执行时假设“订单金额为空”这个检测点查出5条记录那么我们可以直接定位到这些订单ID然后用下面这串SQL去关联查询业务日志或者接口调用记录找到产生脏数据的原因SELECT a.order_id, a.order_amount, a.order_status, b.operation_type, b.operation_desc FROM ods_order_info a LEFT JOIN ods_order_operation_log b ON a.order_id b.order_id WHERE a.order_id IN ( SELECT order_id FROM ods_order_info WHERE order_amount IS NULL ) ORDER BY a.order_id;这种“检测SQL 定位SQL”的组合方式就是测试数据排查的标准动作。以前写这部分脚本可能要花不少时间现在检测SQL由Skill生成我们只需要把精力集中在分析业务原因上效率自然就上来了。6. 常见问题与排查思路在实际落地过程中你可能也会遇到一些坑。下面我整理了几个高频问题以及对应的排查方案。问题现象常见原因解决思路大模型返回的不是合法JSON模型能力较弱或者参数少了response_format在代码里增加JSON片段提取逻辑同时提高温度参数可靠性生成的SQL表名或字段名不匹配用户提供的字段信息不完整把表结构信息尽量按“字段名类型注释”完整传入生成脚本包含DELETE或UPDATE系统提示词约束不严格强化只读要求并在写出文件前做SQL黑名单校验执行生成的SQL报语法错误数据库方言不一致比如MySQL和Oracle在系统提示词中明确指定数据库类型例如“MySQL 8.0”输出脚本量太大检测点太冗余没有控制检测范围在需求描述中明确检测维度减少空泛描述本地规则库没有生效规则提示没有拼接进用户消息检查build_user_message中是否调用了build_rule_prompt这里重点说一下SQL黑名单校验。在设计上哪怕大模型已经很稳定写文件前也建议加一层本地安全校验。下面是一个简单的过滤函数可以集成到script_executor.py里# 文件路径sql_skill/script_executor.py FORBIDDEN_KEYWORDS [insert, update, delete, drop, truncate, alter, create] def check_sql_safe(sql_text): 简单过滤常见写操作SQL返回是否安全 sql_lower sql_text.lower() for keyword in FORBIDDEN_KEYWORDS: if keyword in sql_lower: return False return True在写入文件之前对每条SQL调用check_sql_safe如果包含危险关键字则跳过并告警。这个动作虽然简单但能挡住大多数误操作风险。另外还有一类问题是“字段区分大小写导致的检测结果不稳定”。例如订单状态在MySQL里是varchar(20)但大模型生成的枚举值大小写和库里不一致查询结果就会不准。针对这种情况建议在业务规则提示里补充一句“枚举值请严格匹配数据库实际存储格式注意大小写”。也可以在生成SQL时把枚举值作为参数传入由本地代码动态拼接这样更可控。7. 最佳实践与工程建议7.1 规则库要持续沉淀Skill真正有价值的地方不是“调用一次大模型”而是“规则库不断生长”。每完成一次数据质量问题排查就可以把新的检测点补充到规则库中。比如这次发现订单表存在“同一条订单重复记录的关联表数据”下次生成脚本时这条经验就会自动出现。建议团队维护一个共享的规则库而不是只放在个人项目里。7.2 生成脚本必须人工审阅大模型生成的SQL并不保证100%正确尤其是涉及业务语义的检测点。有的字段表面上非空就可以但业务上可能要求必须是当天的数据。所以凡是生成的脚本执行前一定要经过人工审阅至少要看一眼检测点和SQL条件是否符合预期。可以要求大模型生成SQL时附带“检测点说明”方便审阅。7.3 安全边界要写死在测试环境执行生成脚本时要建立边界意识优先使用只读账号连接测试库。禁止使用生产环境账号执行检测脚本。检测SQL生成时默认只允许SELECT。涉及批量造数时只允许插入临时表不允许直接修改业务表。生成的所有文件要保留时间戳方便追溯版本。这几点不是技术限制而是流程保障。数据测试本身就是要在“发现问题”和“不制造新问题”之间找平衡安全边界一定要提前划好。7.4 输出结果要面向排查生成SQL只是第一步更重要的是让排查过程变得快速。建议把生成结果输出成报告形式例如Markdown或HTML包含检测点、SQL、说明三项。这样你在和研发、产品沟通时可以直接把报告发到群里对方一眼就能看出问题在哪里不需要自己再打开数据库客户端看半天。7.5 性能考虑如果被检测的表数据量很大生成的检测SQL不能在测试环境直接裸跑。建议先在主键时间字段上加筛选条件例如限制最近7天的数据。让大模型生成SQL时带上WHERE create_time DATE_SUB(NOW(), INTERVAL 7 DAY)这类条件。或者拆成多个小任务分批检测避免一次全表扫描。在提示词中可以加一句“如果表数据量较大每条检测SQL尽量带上时间范围过滤条件”这样生成出来的脚本相对更友好。7.6 与现有测试平台集成如果你所在的团队已经有数据测试平台可以把这段生成代码封装成HTTP接口再在前端页面上增加一个输入框和按钮。用户只需要选择表名、填写检测需求后端调用Skill生成SQL再返回给前端展示。这样就能把能力开放给整个测试团队而不是某一个人本地运行。8. 总结与后续优化方向本文分享了一套SQL检测脚本生成Skill的完整实现思路从规则库、提示词配置到大模型调用、脚本落地全部代码都是可以直接复制到项目中运行的。核心收获可以总结为三点第一把SQL检测脚本从纯手工编写变成“规则库 大模型 人审”的结构化流程减少重复劳动提升排查效率。第二规则库是Skill的灵魂。团队的经验如果只存在于个人脑子里效率提升永远有限只有把经验变成模板和规则后续生成质量才会越来越稳定。第三安全边界必须前置。不管生成速度提得多快都不能跳过人工审阅和写操作拦截这是测试数据排查的基本纪律。后续可以优化的方向包括把规则库拆成YAML或JSON配置文件让业务同学也能维护。增加数据库元信息自动获取功能直接从表结构注释中读取字段说明减少手写字段信息的成本。把Skill接入自动化回归流程每天定时对核心业务表做一次数据质量巡检。增加对Oracle、PostgreSQL等多数据库方言的支持让Skill在不同技术栈之间复用。如果你也经常被SQL检测脚本淹没可以照着本文的思路先搭一个最小版本从一张业务表开始试用再逐步补充团队自己的规则库。这并不需要太多额外成本投入半天到一天时间之后每次写检测SQL都会轻松很多。希望这篇文章对你有所启发。如果觉得内容有用可以收藏备用也欢迎动手改造一套适合自己团队的“SQL检测脚本生成 Skill”。
阅读完成 · 觉得有帮助？
