企业官网建设流程全解析

Agno 数据集清洗实战：基于 LLM 质量门控、MinHash 去重与 N-gram 基准去污的三层数据过滤管线

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno

在把数据集交给大模型训练之前，先做一次「能上桌」的筛选是生产级 SFT 管线的标准动作。本指南以 agno 仓库 cookbook/data_labeling/_22_dataset_curation 的官方测试日志与实现为准，系统拆解其三层过滤设计：LLM 质量门控（basic.py）、纯标准库 MinHash 近重复去重（dedup.py）、13-gram 基准去污（decontamination.py）。读完你将掌握每一层的评分规则、关键参数、判定阈值与输出格式，以及如何把它们接入指令生成、拒绝采样之后的真实训练数据流程。

目录

• 一、整体设计：三道关卡各司其职
• 二、第一关：LLM 质量门控（basic.py）
• 三、第二关：MinHash 近重复去重（dedup.py）
• 四、第三关：13-gram 基准去污（decontamination.py）
• 五、测试日志验证：可复现的判定结果
• 六、运行方式与环境依赖
• 七、在数据管线中的位置与扩展建议

一、整体设计：三道关卡各司其职

该示例位于 cookbook/data_labeling/_22_dataset_curation/，对应目录下的三个脚本恰好组成一套训练前数据清洗流水线：

设计上有两个刻意为之的原则（见 README.md）：

1. 只有质量门控使用 LLM。去重与去污是"纯标准库数学"实现，因为生产环境里它们就是这样跑的，且输出的数字必须能逐次复现。
2. 保留溯源（provenance）。被保留的行会携带打分与判定理由，被剔除的行会打印剔除原因，每个决策都可回溯。

这一层处理的是语料级别的整行取舍；如果你需要检查并修复单条标注（label-level），应使用 cookbook/data_labeling/_18_quality_review/；如果只是需要一个打分原语，可参考 cookbook/data_labeling/_17_llm_as_judge/。

二、第一关：LLM 质量门控（basic.py）

2.1 输入格式与评分目标

脚本读取固定样例 data/sample_rows.jsonl，期望每行是{"instruction", "response"}结构，共 12 行：7 行高质量 + 5 行人为植入的缺陷样本。缺陷样本覆盖了 SFT 数据最常见的五类问题：

• 含糊："Just try to sleep better and see what works for you. Everyone is different, so do whatever feels right."（睡眠建议，无实质内容）；
• 事实错误："Water boils at 50 degrees Celsius at sea level."（应为 100 度）；
• 非自包含："Summarize the passage above in two sentences."（依赖缺失的上下文"the passage above"）；
• 截断："First, make sure Python 3 is installed. Then open a terminal, change into your project directory, and run python -m venv"（句子戛然而止）；
• 指令回声："Describe the water cycle." → "Describe the water cycle."（把指令原样弹回）。

若要把input_path指向其他生成器的输出，需先把字段映射成{"instruction", "response"}：_20_instruction_generation/只产出指令、没有响应；_21_rejection_sampling/的行使用prompt/reasoning键（见 README.md）。

2.2 评分 Schema：结构化输出

质量门的核心是一个 Pydantic 结构化输出模型（basic.py#L27-L39）：

class RowVerdict(BaseModel): score: int = Field( ..., ge=1, le=5, description="Row quality on a 1-5 scale where 5 is excellent", ) verdict: Literal["keep", "drop"] = Field( ..., description="keep if score >= 4, otherwise drop" ) reason: str = Field( ..., description="One short sentence naming the deciding criterion" )

score限定在 1–5 的整数区间，verdict只能取keep/drop两个字面值，reason要求一句话点名决定性判据——这为后续溯源提供了结构化的证据字段。

2.3 评分指令：三条准则与五档刻度

打分依据三准则（basic.py#L45-L63）：

• Clarity（清晰度）：响应具体、组织良好，不是含糊的填充物；
• Factual correctness（事实正确性）：响应中每个可核查的论断都为真；
• Self-containedness（自包含性）：行本身自洽，不依赖缺失上下文（如"the passage above"）、不中途截断、不单纯回显指令。

五档刻度：

判定规则写死为：仅当 score >= 4 才保留，reason 保持为一句话。

2.4 Agent 构造与门限逻辑

judge = Agent( model=Gemini(id="gemini-3.5-flash", temperature=0), instructions=instructions, output_schema=RowVerdict, )

要点：temperature=0保证打分确定性；output_schema强制结构化输出。主循环（basic.py#L83-L131）的关键逻辑：

1. 重试而非强转：每个样本最多尝试 3 次，run.content通过isinstance(run.content, RowVerdict)检查后才接受；3 次都失败则直接抛RuntimeError，绝不静默强转坏 schema。
2. 门限是分数，verdict 只是溯源：代码中判定条件为v.verdict == "keep" and v.score >= 4（basic.py#L112）——即使 LLM 输出verdict="keep"但score < 4，该行依然被丢弃。这是防御 LLM 自相矛盾输出的关键设计。
3. 保留行写入溯源字段：通过的行以{"instruction", "response", "score", "reason"}写回data/generated/curated.jsonl（目录自动创建）。

2.5 输出格式示例

保留行携带完整门控溯源（见 README.md）：

{"instruction": "Convert 25 degrees Celsius to Fahrenheit and show the formula.", "response": "Using F = C * 9/5 + 32: F = 25 * 9/5 + 32 = 45 + 32 = 77. So 25 degrees Celsius is 77 degrees Fahrenheit.", "score": 5, "reason": "The response is clear, factually correct, and self-contained."} {"instruction": "Explain what HTTP status code 404 means.", "response": "HTTP 404 Not Found means the server understood the request but could not find the requested resource at that URL. It indicates a client-side addressing problem (bad link or mistyped path), not a server failure; server failures use 5xx codes instead.", "score": 5, "reason": "The response is clear, factually correct, and self-contained."}

控制台逐行打印行号 / score / verdict / instruction 预览 / reason，最终输出一行汇总：wrote 7 rows to data/generated/curated.jsonl: kept 7, dropped 5 of 12。

三、第二关：MinHash 近重复去重（dedup.py）

3.1 为什么用 MinHash

去重环节刻意不调用 LLM（README.md）。实现原理在 dedup.py#L10-L23 有严谨的数学说明：一个 keyed hash 函数在 shingle 全集上施加了一个伪随机排序；对两个 shingle 集合 A、B，A∪B 上的最小哈希在 A∩B 中的概率恰好等于 Jaccard 相似度 |A∩B|/|A∪B|。用 64 个独立排序取平均，即可无偏估计该概率，标准误差约为 sqrt(J·(1−J)/64)，在 J≈0.7 处约 ±0.06。

3.2 三个可调参数

NUM_HASHES = 64 # MinHash 签名槽位数（排序样本数） SHINGLE_SIZE = 3 # 词 3-gram 粒度 SIMILARITY_THRESHOLD = 0.7 # 估计 Jaccard 的聚类阈值

• SHINGLE_SIZE = 3：词 3-gram 能抓住逐字复制、轻度编辑和近似同义；更重的改写若共享的 3-gram 太少会落在阈值以下——该场景需要基于 embedding 的去重，MinHash 力不能及（这是脚本自述的诚实边界）。
• NUM_HASHES = 64：直接决定估计精度与计算成本，64 槽在误差与开销间取得平衡。
• SIMILARITY_THRESHOLD = 0.7：估计 Jaccard ≥ 0.7 的行对进入并查集（union-find）合并。

3.3 纯标准库实现链

完整实现无需任何第三方依赖（dedup.py#L26-L157）：

1. shingles()：文本小写化后按空格分词，产出词 3-gram 集合；短于一个 shingle 的文本退化为整句单 shingle，避免空集。
2. hash_shingle()：hashlib.blake2b(shingle, key=key, digest_size=8)，以固定 key 生成 8 字节摘要转整数。key 固定 = 完全确定性。
3. minhash_signature()：第 i 个槽取"以 distinct key（perm-00…perm-63）键控的 hash 函数 i 对该集合所有 shingle 的最小值"，即每次排序的一个采样。
4. estimated_jaccard()：两个 64 槽签名对应位置相等的比例。
5. 并查集聚类：所有行对相似度 ≥ 0.7 即 union；聚类后每个簇保留第一行，其余丢弃。

3.4 测试夹具与输出

夹具为 10 行模块级数据（dedup.py#L40-L123），植入两个簇：

• 簇 A（轻编辑级）：row-00 / row-03 / row-07——回文函数题，仅个别单词替换；
• 簇 B（近似同义级）：row-02 / row-06——客户邮件题，delayed↔postponed、apologize for the inconvenience↔apologize for the trouble等措辞差异。

输出会打印每个超阈值行对的est_jaccard、每个簇的成员与保留/丢弃决定，以及汇总行kept 7 of 10 rows: dropped 3 near-duplicates across 2 clusters。由于无随机数、hash key 固定，数值可逐次精确复现。

四、第三关：13-gram 基准去污（decontamination.py）

4.1 原理与保护集构建

去污的目标是：训练数据中若包含评测集（benchmark）题目，会污染评测结果。经典做法是 n-gram 重叠检查（LLM 训练数据去污报告的标准手段），此处取13 词连续 n-gram（decontamination.py#L1-L27）。

保护集来自 data/benchmark_sample.jsonl——这是人为发明的夹具，并非真实 benchmark，包含 8 道题（火车均速、行星轴倾角、矩形面积、质数、配比、化学元素、折扣价、八面体棱数），每行{"id", "question", "answer"}。构建时仅保护question文本（夹具答案多为单词，贡献不出 13-gram）：

protected = {} for bench in benchmark_rows: for gram in ngrams(tokenize(bench["question"]), NGRAM_SIZE): protected[gram] = bench["id"] # gram -> 来源题号，用于溯源

分词用re.findall(r"[a-z0-9]+", text.lower())，即小写后仅保留字母数字 token。

4.2 判定规则与两个边界案例

对每条训练行，将其instruction + " " + response分词取 13-gram，与保护集求交集；共享至少一个保护 13-gram 即标记并丢弃。脚本在两个刻意植入的样本上演示"能抓什么、不能抓什么"：

• train-02（能抓）：指令是bench-01的逐字复制（"A train travels 180 kilometers in 2 hours and 15 minutes. What is its average speed in kilometers per hour?"），必然命中保护 13-gram；
• train-04（抓不到）：bench-03（矩形面积题）的近似改写，同一个问题换词重述后不共享任何 13 个连续词，检查按设计漏过；
• train-06（天然豁免）："What is 2 + 2?"全文不足 13 词，range()自然产出零个 13-gram，任何 n-gram 检查都无法标记此类短行。

对于漏过的改写样本，脚本会诚实地打印限制说明：paraphrase contamination needs fuzzy or embedding-based methods; exact n-gram overlap cannot see it（decontamination.py#L176-L184）。

五、测试日志验证：可复现的判定结果

TEST_LOG.md（2026-07-18，基于gemini-3.5-flash+ agno 2.7.4）完整记录了三个脚本的实测结果：

basic.py — PASS：12 行中 7 行高质量样本全部以 score 5 保留；5 个植入缺陷全部被剔除，且 reason 精准命中决定性判据：

• 含糊睡眠建议 →"extremely vague"；
• 50°C 沸点 →"factually incorrect as water boils at 100 degrees Celsius"；
• 缺失段落 →"refers to a missing passage"；
• 截断 venv 步骤 →"incomplete and ends abruptly"；
• 指令回声 →"merely echoes the instruction back"。

汇总行wrote 7 rows to data/generated/curated.jsonl: kept 7, dropped 5 of 12。在加入"分数门限强制"逻辑后共运行三次，12 行的判定结果完全一致。

dedup.py — PASS：两个植入簇均被检出，无多余误报。簇 1 = [row-00, row-03, row-07]，其中est_jaccard(row-00, row-03) = 0.797、est_jaccard(row-00, row-07) = 0.797，但est_jaccard(row-03, row-07) = 0.625低于阈值，仅经 row-00 传递合并——打印输出如实反映了这一传递性。簇 2 = [row-02, row-06]，est_jaccard = 0.797。汇总行kept 7 of 10 rows: dropped 3 near-duplicates across 2 clusters，数值跨运行精确复现。

decontamination.py — PASS：8 道 benchmark 题构建出56 个互不相同的保护 13-gram；train-02以完整匹配 13-gram（"180 kilometers in 2 hours and 15 minutes what is its average speed"）被标记并归因到bench-01；train-04未被标记，并打印了上述诚实限制行。汇总行kept 7 of 8 training rows, dropped 1 contaminated: ['train-02']，两次运行输出完全一致。

这三份日志证明了一个共同点：判定全部可复现——LLM 门控依赖 temperature=0 与固定指令，两个纯数学过滤器依赖固定 hash key 与确定性算法。

六、运行方式与环境依赖

从仓库根目录直接运行三个脚本：

python cookbook/data_labeling/_22_dataset_curation/basic.py python cookbook/data_labeling/_22_dataset_curation/dedup.py python cookbook/data_labeling/_22_dataset_curation/decontamination.py

依赖说明（见 README.md 与 cookbook/data_labeling/README.md）：

• 仅basic.py需要GOOGLE_API_KEY（Gemini 3.5 Flash 为各 cookbook 默认模型）；dedup.py与decontamination.py不发起任何 API 调用。
• basic.py依赖 agno 的Agent/RunOutput、Gemini模型以及pydantic；后两个脚本仅使用json、hashlib、itertools、re、pathlib等标准库。
• 通用环境搭建可执行./scripts/demo_setup.sh并source .venvs/demo/bin/activate后运行。

七、在数据管线中的位置与扩展建议

在官方数据标注 cookbook 的合成数据工作流中（见 cookbook/data_labeling/README.md），_22_dataset_curation的典型位置是：

_20_instruction_generation/ （自指令/进化解生成候选指令） ↓ _21_rejection_sampling/ （采样 K 个解，经校验器或 judge 保留） ↓ _22_dataset_curation/ ← 本指南：质量门控 → 去重 → 去污（针对你的 eval 集）

三个脚本均以"过滤器"角色单文件可运行，便于复制改造：

• 替换basic.py中的input_path指向你自己的生成结果，前提是按{"instruction", "response"}映射字段；
• 将dedup.py的ROWS常量替换为真实语料，按需调整SIMILARITY_THRESHOLD（想更激进可降到 0.6 左右，代价是误杀风险上升）；
• 将decontamination.py的benchmark_sample.jsonl换成你真实的评测集（保持question字段），即可对训练语料执行与公开去污报告同源的 13-gram 检查。

需要说明的适用边界：MinHash 3-gram 与 13-gram 重叠检查都是精确文本匹配的近似，对强改写、跨语言或短行（不足 13 词）无法覆盖，这些场景需引入 embedding 相似度或模糊匹配；而质量门控的判定质量与 judge 模型和评分指令直接相关，建议在换模型后按 TEST_LOG.md 的方式重跑基准夹具，确认判定稳定性后再投入生产语料。

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno