从工程实战到训练数据：AI工程化自动产出SFT数据的系统化方法

版权声明与免责声明 本文基于InstructGPT、Flan等SFT训练研究，结合工程实践经验进行综合解读。

原创性质 本文提出的反向数据生成框架、数据质量评估标准和构建流程为作者原创。

---

开头：从“有很多记录”到“有可训练样本”

在第 7 篇里，我们解决的是闭环入口问题：哪些 AI 协作轨迹值得留下，哪些必须丢弃，哪些应该进入 eval。接下来真正会卡住团队的，是另一个更具体的问题：

同样是“AI 辅助交付留下的记录”，为什么有些只能用于复盘，有些可以进入知识库，只有少数能成为训练样本？

这篇文章只做一件事：把这个筛选与加工过程讲清楚。重点不是“怎么把日志导出成 JSONL”，而是如何在工程语境里定义可训练样本的质量边界、数据契约和路由规则。

你可以把本文当作一条中段流水线：

1. 输入端是第 7 篇已经治理过的工程资产（任务契约、错误类型、评审反馈、验证证据）。
2. 处理中段是样本构建、质量门控、脱敏、分桶、导出与版本化。
3. 输出端是可进入 SFT 的候选样本，以及可回归验证的评估资产。

所以本文不讨论“是否应该训练模型”的战略问题，而是回答“当团队决定要训练时，怎样避免把低质量工程噪音训练成模型默认行为”。

---

整体架构：从开发到训练数据的自动化流水线

在深入细节之前，先来看整体架构。理想的SFT数据生成不是手动整理，而是在开发过程中自动产出：

这个流水线将BMAD-Speckit-SDD-Flow的开发流程与SFT数据提取无缝集成，实现**“开发即训练”**。

核心概念：什么是SFT训练数据

SFT的基本概念

SFT（监督微调）是让预训练模型学习特定任务的方法。

基本形式：输入（指令/问题）→ 输出（期望回答）

编程场景的SFT数据：

{
    "instruction": "实现一个支持LRU淘汰策略的缓存类",
    "input": "要求：\n- 支持get和put操作\n- 时间复杂度O(1)\n- 容量可配置",
    "output": "class LRUCache:\n    def __init__(self, capacity: int):\n        ...",
    "metadata": {
        "difficulty": "medium",
        "source": "engineering_project",
        "verified": true,
        "quality_score": 0.92
    }
}

为什么工程产出适合转化为SFT数据

特点1：真实性强

• 来自实际项目需求
• 经过实践检验
• 不是理论假设

特点2：质量可控

• 经过人工审查
• 有明确的对错标准
• 有改进轨迹记录

特点3：领域相关

• 符合团队技术栈
• 符合业务场景
• 符合编码规范

特点4：持续产生

• 每个项目都在产生新数据
• 数据量随时间增长
• 可以持续更新模型

---

可转化的工程资产类型

资产类型1：评估题目集

来源：给AI出题时积累的题目

转化方式：

原始题目（问题描述 + 测试用例）
    ↓
优质答案（人工编写或精选AI答案）
    ↓
SFT数据（指令-输入-输出三元组）

示例：

{
    "instruction": "编写一个函数，找出数组中第K大的元素",
    "input": "要求：\n- 函数签名：find_kth_largest(nums: List[int], k: int) -> int\n- 时间复杂度：O(n)或O(n log n)\n- 不使用内置排序",
    "output": "def find_kth_largest(nums: List[int], k: int) -> int:\n    def partition(left, right, pivot_idx):\n        ...",
    "metadata": {
        "algorithm": "quickselect",
        "difficulty": "medium",
        "source": "evaluation_suite",
        "pass_rate": 0.75
    }
}

资产类型2：代码审查反馈

来源：对AI生成代码的审查意见

转化方式：

问题代码（AI生成的不良代码）
    ↓
审查反馈（指出问题+改进建议）
    ↓
修正代码（人工修正后的优质代码）
    ↓
SFT数据（用于训练代码修复能力）

示例：

{
    "instruction": "修复以下代码中的安全问题",
    "input": "```python\ndef query_user(username):\n    sql = f\"SELECT * FROM users WHERE name = '{username}'\"\n    return db.execute(sql)\n```",
    "output": "```python\ndef query_user(username):\n    sql = \"SELECT * FROM users WHERE name = %s\"\n    return db.execute(sql, (username,))\n```\n\n修复说明：\n1. 使用参数化查询防止SQL注入\n2. 永远不要拼接用户输入到SQL中",
    "metadata": {
        "issue_type": "security",
        "severity": "high",
        "source": "code_review",
        " CWE_id": "CWE-89"
    }
}

资产类型3：多轮对话记录

来源：与AI协作解决问题的完整对话

转化方式：

完整对话（问题→AI回答→反馈→改进→...）
    ↓
抽取高质量轮次
    ↓
SFT数据（用于训练对话能力）

示例：

{
    "messages": [
        {
            "role": "user",
            "content": "请实现一个线程安全的单例模式"
        },
        {
            "role": "assistant",
            "content": "class Singleton:\n    _instance = None\n    \n    def __new__(cls):\n        if cls._instance is None:\n            cls._instance = super().__new__(cls)\n        return cls._instance"
        },
        {
            "role": "user",
            "content": "这个实现在多线程环境下有问题。如果两个线程同时判断_instance为None，会创建两个实例。"
        },
        {
            "role": "assistant",
            "content": "class Singleton:\n    _instance = None\n    _lock = threading.Lock()\n    \n    def __new__(cls):\n        if cls._instance is None:\n            with cls._lock:\n                if cls._instance is None:\n                    cls._instance = super().__new__(cls)\n        return cls._instance"
        }
    ],
    "metadata": {
        "conversation_type": "iterative_refinement",
        "topic": "concurrency",
        "source": "chat_logs"
    }
}

资产类型4：成功项目案例

来源：AI辅助完成的完整项目代码

转化方式：

项目代码（模块化、带注释、高质量）
    ↓
拆解为教学案例
    ↓
SFT数据（用于训练复杂任务分解）

示例：

{
    "instruction": "设计并实现一个JWT认证系统",
    "input": "需求：\n1. 支持token生成和验证\n2. 支持token刷新\n3. 支持token吊销\n4. 使用Redis存储会话\n\n请按以下步骤实现：\n1. 设计数据模型\n2. 实现核心功能\n3. 添加错误处理\n4. 编写测试用例",
    "output": "## 步骤1：数据模型设计\n...\n\n## 步骤2：核心功能实现\n...\n\n## 步骤3：错误处理\n...\n\n## 步骤4：测试用例\n...",
    "metadata": {
        "project_type": "authentication_system",
        "complexity": "high",
        "source": "successful_project",
        "verified_in_production": true
    }
}

---

BMAD-Speckit-SDD-Flow架构：自动化SFT数据生成系统

在前面的章节中，我们讨论了手动从工程资产中提取SFT数据的方法。但手动方式效率低下，难以规模化。本节将深入介绍BMAD-Speckit-SDD-Flow架构——一个能够在AI工程化开发过程中自动产出SFT训练数据的系统化方案。

架构设计理念

BMAD-Speckit-SDD-Flow结合了BMAD方法（多智能体敏捷开发）与Spec-Driven Development（规范驱动开发），在以下环节自动捕获和转化训练数据：

核心数据模型：CanonicalSftSample

系统的核心是一个标准化的数据模型CanonicalSftSample，它将所有来源的训练数据统一为规范格式：

interface CanonicalSftSample {
  // 样本唯一标识
  sample_id: string;
  sample_version: 'v1';

  // 数据来源追踪
  source: {
    run_id: string;              // 执行运行ID
    stage: string;               // 开发阶段
    flow: string;                // 工作流类型
    epic_id?: string;            // 所属Epic
    story_id?: string;           // 所属Story
    artifact_refs: Array<{       // 原始工件引用
      path: string;
      content_hash: string;
      kind: string;
    }>;
  };

  // 对话消息（兼容OpenAI格式）
  messages: Array<{
    role: 'system' | 'user' | 'assistant' | 'tool';
    content: string;
    tool_calls?: ToolCall[];
    tool_call_id?: string;
    weight?: 0 | 1;
  }>;

  // 工具定义（用于tool calling训练）
  tools?: Tool[];

  // 元数据
  metadata: {
    schema_targets: string[];    // 目标格式
    language: string;            // 语言
    tags?: string[];
    notes?: string[];
  };

  // 质量评估
  quality: {
    acceptance_decision: 'accepted' | 'rejected' | 'downgraded';
    phase_score: number | null;
    dimension_scores?: Record<string, number>;
    veto_triggered: boolean;
    iteration_count: number;
    has_code_pair: boolean;
    token_estimate: number;
    rejection_reasons: string[];
    warnings: string[];
  };

  // 数据来源追溯
  provenance: {
    base_commit_hash: string | null;
    content_hash: string | null;
    source_path: string | null;
    patch_ref: string | null;
    lineage: string[];
    generated_at: string;
  };

  // 数据集划分
  split: {
    assignment: 'train' | 'validation' | 'test' | 'holdout';
    seed: number;
    strategy: string;
    group_key: string | null;
  };

  // 数据脱敏信息
  redaction: {
    status: 'clean' | 'redacted' | 'blocked';
    applied_rules: string[];
    findings: Array<{
      kind: string;
      severity: 'low' | 'medium' | 'high' | 'critical';
      field_path: string;
      action?: string;
    }>;
    redacted_fields: string[];
  };

  // 导出兼容性
  export_compatibility: {
    openai_chat: ExportDecision;
    hf_conversational: ExportDecision;
    hf_tool_calling: ExportDecision;
  };
}

SFT数据提取管道详解

数据提取管道包含四个核心阶段：

阶段1：Candidate Builder（候选构建）

// 从运行记录构建候选样本
function buildCanonicalSample(
  record: RunScoreRecord,        // 评估运行记录
  sourceContent: string,          // 原始工件内容
  codePair: { input: string; output: string },  // 代码对比
  options: BuildOptions
): CanonicalSftSample {

  // 1. 从审计报告中提取指令（§1问题描述 + §4修复方案）
  const instruction = extractInstruction(sourceContent);

  // 2. 构建对话消息
  const messages = buildCanonicalMessages(
    instruction,
    codePair.input,    // 修改前代码
    codePair.output    // 修改后代码
  );

  // 3. 计算确定性数据集划分（基于story hash）
  const split = assignDeterministicSplit({
    seed: options.splitSeed ?? 42,
    groupKey: parsedStory
      ? `epic-${parsedStory.epicId}/story-${parsedStory.storyId}`
      : record.run_id,
  });

  // 4. 构建完整样本
  return {
    sample_id: buildCanonicalSampleId({...}),
    source: { run_id, stage, epic_id, story_id, artifact_refs },
    messages,
    quality: { phase_score, iteration_count, has_code_pair, ... },
    provenance: { base_commit_hash, content_hash, patch_ref, ... },
    split,
    // ... 其他字段
  };
}

关键特性：

• Git Diff提取：自动从base_commit到当前HEAD提取代码变更
• 指令提取：从审计报告的标准章节（§1问题、§4方案）提取训练指令
• 缓存机制：避免重复构建，提高性能

阶段2：Quality Gates（质量门控）

质量门控系统对候选样本进行多维度评估，决定是否接受：

interface QualityGateOptions {
  minScore?: number;          // 最低分数门槛（默认90）
  maxIterations?: number;     // 最大迭代次数
  maxTokens?: number;         // 最大token数
  requireCodePair?: boolean;  // 是否要求代码对比
}

function applyQualityGates(
  sample: CanonicalSftSample,
  options: QualityGateOptions
): CanonicalSftSample {
  const hardReasons: string[] = [];  // 硬性拒绝原因
  const softReasons: string[] = [];  // 软性警告原因

  // 硬性检查
  if (!sample.provenance.base_commit_hash) {
    hardReasons.push('prov_missing_hash');
  }
  if ((sample.quality.phase_score ?? 0) < minScore) {
    hardReasons.push('score_below_floor');
  }
  if (sample.quality.veto_triggered) {
    hardReasons.push('veto_triggered');  // 关键审计项否决
  }
  if (sample.redaction.status === 'blocked') {
    hardReasons.push('redaction_blocked');  // 脱敏阻断
  }

  // 软性检查
  if (sample.quality.iteration_count > maxIterations) {
    softReasons.push('too_many_iterations');
  }
  if (!sample.quality.has_code_pair) {
    softReasons.push('missing_code_pair');
  }

  // 决定：accepted / rejected / downgraded
  const acceptanceDecision =
    hardReasons.length > 0 ? 'rejected' :
    softReasons.length > 0 ? 'downgraded' : 'accepted';

  return { ...sample, quality: { ...sample.quality,
    acceptanceDecision,
    rejection_reasons: [...hardReasons, ...softReasons]
  }};
}

质量维度：

检查项	类型	说明
来源完整性	硬性	必须有commit hash、source path
分数门槛	硬性	phase_score >= 90（可配置）
否决触发	硬性	关键审计项未通过
脱敏阻断	硬性	检测到私钥等敏感信息
消息完整性	硬性	必须包含user和assistant消息
迭代次数	软性	超过maxIterations降级
代码对比	软性	无input/output代码对降级

阶段3：Redaction（数据脱敏）

自动检测和脱敏敏感信息，确保训练数据安全：

// 脱敏规则
const REDACTION_RULES = {
  // 邮箱地址 → 中等风险，脱敏处理
  email: {
    pattern: /\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b/gi,
    severity: 'medium',
    action: 'redact',  // 替换为[REDACTED_EMAIL]
  },

  // Secret Token → 高风险，脱敏处理
  secretToken: {
    pattern: /\bsk-[A-Za-z0-9]{16,}\b/g,  // OpenAI API Key等
    severity: 'high',
    action: 'redact',  // 替换为[REDACTED_SECRET]
  },

  // 私钥 → 致命风险，阻断样本
  privateKey: {
    pattern: /BEGIN [A-Z ]+ PRIVATE KEY/,
    severity: 'critical',
    action: 'block',  // 样本被拒绝
  },
};

function applyCanonicalRedaction(sample: CanonicalSftSample): CanonicalSftSample {
  let status: 'clean' | 'redacted' | 'blocked' = 'clean';
  const findings: RedactionFinding[] = [];

  const messages = sample.messages.map((message, index) => {
    let content = message.content;

    // 应用每条规则
    for (const [ruleName, rule] of Object.entries(REDACTION_RULES)) {
      if (rule.pattern.test(content)) {
        if (rule.action === 'block') {
          status = 'blocked';
          findings.push({ kind: ruleName, severity: rule.severity, ... });
        } else if (rule.action === 'redact') {
          status = status === 'blocked' ? 'blocked' : 'redacted';
          content = content.replace(rule.pattern, `[REDACTED_${ruleName.toUpperCase()}]`);
          findings.push({ kind: ruleName, severity: rule.severity, action: 'redact' });
        }
      }
    }

    return { ...message, content };
  });

  return { ...sample, messages, redaction: { status, findings } };
}

阶段4：Split Assignment（数据集划分）

使用确定性算法分配训练/验证/测试集，确保可复现：

function assignDeterministicSplit(options: {
  seed: number;
  groupKey: string | null;
}): CanonicalSplit {
  const stableKey = `${options.seed}:${options.groupKey ?? 'ungrouped'}`;
  const hash = sha256(stableKey).digest('hex');
  const bucket = parseInt(hash.slice(0, 8), 16) % 100;

  // 80%训练 / 10%验证 / 10%测试
  let assignment: 'train' | 'validation' | 'test' = 'train';
  if (bucket >= 80 && bucket < 90) assignment = 'validation';
  if (bucket >= 90) assignment = 'test';

  return { assignment, seed: options.seed, strategy: 'story_hash_v1' };
}

多格式导出系统

系统支持导出为多种流行的SFT训练格式：

OpenAI Chat格式

// 导出为OpenAI微调格式
{
  "messages": [
    { "role": "system", "content": "You are a senior coding agent." },
    { "role": "user", "content": "修复以下SQL注入问题...\n\nCurrent implementation:\ndef query(user):\n  sql = f\"SELECT * FROM users WHERE name = '{user}'\"" },
    { "role": "assistant", "content": "def query(user):\n  sql = \"SELECT * FROM users WHERE name = %s\"\n  return db.execute(sql, (user,))" }
  ],
  "tools": [...],  // 可选工具定义
  "parallel_tool_calls": false
}

HuggingFace Conversational格式

// 导出为HF对话格式
{
  "system": "You are a senior coding agent.",
  "conversations": [
    { "from": "human", "value": "修复以下SQL注入问题..." },
    { "from": "gpt", "value": "def query(user):\n  sql = \"SELECT * FROM users WHERE name = %s\"..." }
  ]
}

HuggingFace Tool Calling格式

// 导出为HF工具调用格式
{
  "system": "You are a coding assistant with tool access.",
  "conversations": [
    { "from": "human", "value": "分析这段代码的复杂度" },
    { "from": "gpt", "value": "", "tool_calls": [...] },
    { "from": "tool", "value": "{\"complexity\": \"O(n^2)\", ...}" },
    { "from": "gpt", "value": "这段代码的时间复杂度是O(n^2)..." }
  ],
  "tools": [...]
}

CLI使用示例

# 基础提取（默认phase_score >= 90）
npx ts-node scripts/sft-extract.ts

# 指定最低分数
npx ts-node scripts/sft-extract.ts --min-score 85

# 指定输出路径
npx ts-node scripts/sft-extract.ts --output ./custom-sft-data.jsonl

# 完整示例
npx ts-node scripts/sft-extract.ts \
  --min-score 90 \
  --output ./training-data/sft-v1.0.jsonl

输出摘要示例：

共提取 156 条，覆盖 12 个 Story；跳过 23 条（原因：无source_path: 10, git diff失败: 8, phase_score低于阈值: 5）

---

数据构建流程：从原始资产到训练数据

阶段1：资产收集与分类

收集范围：

• 评估题目及优质答案
• 代码审查记录（问题代码+反馈+修正）
• 多轮对话日志
• 项目文档和代码
• 错误案例分析

分类标准：

DATA_CATEGORIES = {
    "code_generation": {
        "description": "代码生成任务",
        "sources": ["evaluation_questions", "project_code"],
        "format": "instruction-input-output"
    },
    "code_review": {
        "description": "代码审查和修复",
        "sources": ["review_feedback", "bug_fixes"],
        "format": "problem-feedback-solution"
    },
    "conversation": {
        "description": "多轮对话",
        "sources": ["chat_logs"],
        "format": "messages"
    },
    "architecture": {
        "description": "架构设计",
        "sources": ["design_docs", "system_docs"],
        "format": "instruction-context-output"
    }
}

阶段2：数据清洗与筛选

清洗步骤：

1. 去重

   def deduplicate(data):
       # 基于代码相似度去重
       # 基于问题描述去重
       # 保留质量更高的版本

2. 质量筛选

   def quality_filter(item):
       # 代码必须通过测试
       # 代码质量评分 > 阈值
       # 有明确的问题-答案对应关系
       # 无敏感信息（密码、密钥等）

3. 格式标准化

   def normalize_format(item, target_format):
       # 统一字段命名
       # 统一代码风格
       # 添加元数据

筛选标准：

维度	标准	权重
功能正确性	通过所有测试用例	40%
代码质量	静态分析得分 > 0.8	25%
安全性	无高危漏洞	20%
可理解性	有清晰注释和说明	15%

阶段3：数据增强与扩充

增强方法1：变体生成

# 生成同一问题的多种表述
def generate_variants(original_instruction, n=5):
    variants = []
    # 使用同义词替换
    # 改变句式结构
    # 增减细节信息
    return variants

增强方法2：难度调整

# 生成同一问题的不同难度版本
def adjust_difficulty(item, target_difficulty):
    if target_difficulty == "easy":
        # 提供更多提示
        # 减少约束条件
    elif target_difficulty == "hard":
        # 增加边界条件
        # 增加性能要求

增强方法3：跨语言迁移

# 将Python题目迁移到JavaScript/Go等
def migrate_language(item, target_language):
    # 语法转换
    # 惯用法调整
    # 生态适配

阶段4：格式转换与导出

通用SFT格式：

{
    "dataset_info": {
        "name": "company_coding_sft_v1",
        "version": "1.0.0",
        "created_at": "2024-04-01",
        "total_samples": 10000,
        "categories": {
            "code_generation": 6000,
            "code_review": 2000,
            "conversation": 1500,
            "architecture": 500
        }
    },
    "data": [
        {
            "id": "cg_0001",
            "type": "code_generation",
            "instruction": "...",
            "input": "...",
            "output": "...",
            "metadata": {
                "source": "evaluation_suite",
                "difficulty": "medium",
                "language": "python",
                "quality_score": 0.92,
                "verified": true
            }
        }
    ]
}

框架特定格式：

• Alpaca格式
• ShareGPT格式
• OpenAI fine-tuning格式
• HuggingFace datasets格式

---

数据质量评估标准

评估维度

维度1：准确性

• 代码能否正确运行
• 是否符合需求描述
• 测试用例通过率

维度2：完整性

• 是否包含必要信息
• 是否包含边界情况处理
• 是否包含错误处理

维度3：一致性

• 输入输出对应关系清晰
• 风格统一
• 术语一致

维度4：多样性

• 问题类型覆盖
• 难度分布合理
• 语言/框架覆盖

维度5：安全性

• 无恶意代码
• 无敏感信息泄露
• 符合安全规范

质量评分模型

def calculate_quality_score(item):
    scores = {
        'correctness': evaluate_correctness(item),      # 40%
        'completeness': evaluate_completeness(item),    # 25%
        'consistency': evaluate_consistency(item),      # 20%
        'safety': evaluate_safety(item),                # 15%
    }

    weights = {
        'correctness': 0.40,
        'completeness': 0.25,
        'consistency': 0.20,
        'safety': 0.15
    }

    total_score = sum(scores[k] * weights[k] for k in scores)
    return total_score

人工审核要点

必审项：

• 代码功能正确性（运行验证）
• 无安全漏洞（静态扫描）
• 无敏感信息（正则匹配）
• 格式规范（自动化检查）

抽检项（20%抽样）：

• 代码质量（人工评估）
• 教学价值（专家评估）
• 场景真实性（业务方确认）

---

SFT训练基础：如何使用这些数据

训练流程概览

原始数据
    ↓
数据预处理（清洗、格式化）
    ↓
数据集构建（训练/验证/测试划分）
    ↓
模型微调（SFT训练）
    ↓
模型评估
    ↓
模型部署

简易训练示例（使用HuggingFace）

2026-03 来源口径：训练 API 和 LoRA 配置以 TRL SFTTrainer 与 PEFT LoRA 文档为准；基础模型使用占位 ID，因为实际项目必须先复核模型卡、许可证、训练数据边界和团队内部评估快照。

from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from peft import LoraConfig, get_peft_model
from trl import SFTTrainer

# 1. 加载基础模型（示例占位：按 2026-03 模型卡、许可证和内部评测选择）
base_model_id = "your-org/code-model-base"
model = AutoModelForCausalLM.from_pretrained(base_model_id)
tokenizer = AutoTokenizer.from_pretrained(base_model_id)

# 2. 配置LoRA（降低训练成本）
lora_config = LoraConfig(
    r=16,
    lora_alpha=32,
    target_modules=["q_proj", "v_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM"
)
model = get_peft_model(model, lora_config)

# 3. 准备数据集
dataset = load_dataset("json", data_files="company_coding_sft_v1.json")

# 4. 配置训练参数
training_args = TrainingArguments(
    output_dir="./results",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    learning_rate=2e-4,
    save_steps=100,
    logging_steps=10,
)

# 5. 开始训练
trainer = SFTTrainer(
    model=model,
    tokenizer=tokenizer,
    train_dataset=dataset["train"],
    eval_dataset=dataset["validation"],
    args=training_args,
)
trainer.train()

# 6. 保存模型
model.save_pretrained("./company_coding_model")

训练建议

数据量建议：

• 最小：1,000条高质量样本
• 推荐：10,000+条样本
• 理想：100,000+条样本

质量 > 数量：

• 1,000条高质量数据 > 10,000条低质量数据
• 宁可少而精，不要多而杂

持续迭代：

• 先用小数据集验证可行性
• 根据效果逐步增加数据
• 建立数据-训练-评估的闭环

---

实战案例：完整的数据构建流程

背景

某团队希望基于6个月的工程实践，构建一个专属的代码生成模型。

资产盘点

资产类型	原始数量	可用数量	备注
评估题目	300	250	筛选后
代码审查记录	500	400	去重后
对话日志	1000条	800条	筛选后
项目代码	10万行	2万行	精选模块

数据处理

Step 1: 提取与清洗

# 从评估题目提取
sft_items = []
for question in evaluation_questions:
    if question.quality_score > 0.8:
        sft_items.append({
            "instruction": question.description,
            "input": question.requirements,
            "output": question.reference_solution
        })

# 从审查记录提取
for review in code_reviews:
    if review.severity in ["high", "critical"]:
        sft_items.append({
            "instruction": f"修复以下{review.issue_type}问题",
            "input": review.problematic_code,
            "output": review.fixed_code + "\n\n" + review.explanation
        })

Step 2: 质量评分

for item in sft_items:
    item["quality_score"] = calculate_quality_score(item)

# 筛选高质量数据
high_quality_items = [i for i in sft_items if i["quality_score"] > 0.85]

Step 3: 数据增强

# 生成变体
augmented_items = []
for item in high_quality_items:
    augmented_items.append(item)
    # 生成3个变体
    for variant in generate_variants(item, n=3):
        augmented_items.append(variant)

Step 4: 人工审核

# 抽样人工审核
sample = random.sample(augmented_items, k=len(augmented_items)//5)
for item in sample:
    review_result = manual_review(item)
    if not review_result.approved:
        augmented_items.remove(item)

最终数据集

## 数据集统计

- 总样本数：5,000
- 类别分布：
  - 代码生成：60% (3,000)
  - 代码修复：25% (1,250)
  - 对话交互：10% (500)
  - 架构设计：5% (250)
- 语言分布：
  - Python：70%
  - JavaScript：20%
  - Go：10%
- 难度分布：
  - Easy：30%
  - Medium：50%
  - Hard：20%
- 平均质量分：0.88

训练效果

基线模型：团队在 2026-03 复核过模型卡、许可证和内部基准的代码基础模型 训练方式：LoRA微调 训练时间：4小时（单卡A100）

效果对比：

指标	基线模型	微调后	提升
团队内部测试集通过率	65%	82%	+17%
代码风格匹配度	60%	85%	+25%
安全漏洞率	15%	5%	-10%
团队满意度	70%	90%	+20%

---

数据资产化：长期运营建议

建立数据收集机制

自动收集点：

• 每次代码审查后，询问是否可用于训练
• 定期导出对话日志
• 项目代码归档时筛选高质量模块

数据管道：

工程实践
    ↓
自动收集
    ↓
初步筛选
    ↓
质量评分
    ↓
人工审核（抽样）
    ↓
进入训练池
    ↓
定期训练更新

版本管理

datasets/
├── v1.0.0_2024q1/
│   ├── train.jsonl
│   ├── validation.jsonl
│   └── metadata.json
├── v1.1.0_2024q2/
│   └── ...
└── v2.0.0_2024q3/
    └── ...

持续迭代

月度：收集新数据 季度：更新数据集版本 半年：重新训练模型 年度：评估整体效果，调整策略

---

筛选标准与素材选择：构建高质量训练集

在使用BMAD-Speckit-SDD-Flow自动提取SFT数据时，建立科学的筛选标准至关重要。以下是经过实践验证的筛选框架：

三层筛选体系

三层筛选确保数据质量的同时保持多样性，最终保留约50%的高质量样本。

素材选择策略

1. 基于Scenario的选择

BMAD-Speckit-SDD-Flow区分不同scenario，优先选择高质量场景：

Scenario	描述	优先级	原因
real_dev	真实开发场景	高	来自实际编码过程，质量最高
evaluation	评估场景	中	有明确的对错标准
synthetic	合成数据	低	可能缺乏真实性

2. 基于Score Pattern的选择

// 优先选择以下score pattern的记录
const HIGH_VALUE_PATTERNS = [
  // 高初始分+低迭代 = 一次做对，优秀样本
  { initialScore: '>80', iterations: '<=2' },

  // 低初始分+高最终分+适度迭代 = 有效学习样本
  { initialScore: '<60', finalScore: '>90', iterations: '3-5' },

  // 有veto但最终通过 = 关键改进样本
  { vetoTriggered: true, finalScore: '>90' },
];

// 避免以下pattern
const LOW_VALUE_PATTERNS = [
  // 多次迭代仍低分 = 质量存疑
  { iterations: '>5', finalScore: '<80' },

  // 无代码对比 = 教学价值有限
  { hasCodePair: false },
];

3. 基于Content Category的平衡

// 理想的训练集构成
const TARGET_COMPOSITION = {
  codeGeneration: 0.50,     // 50% 代码生成
  codeReview: 0.25,         // 25% 代码审查
  bugFix: 0.15,             // 15% Bug修复
  architecture: 0.10,       // 10% 架构设计
};

// 每个类别内部再细分
const CODE_GEN_SUBCATEGORIES = {
  algorithm: 0.30,          // 算法实现
  dataStructure: 0.25,      // 数据结构
  apiDesign: 0.25,          // API设计
  utility: 0.20,            // 工具函数
};

质量门控配置建议

根据不同使用场景，调整Quality Gates参数：

场景A：构建基础模型（高严格）

const STRICT_GATES = {
  minScore: 95,             // 只选最高分
  maxIterations: 2,         // 最多2次迭代
  requireCodePair: true,    // 必须有代码对比
  maxTokens: 4096,          // 限制token数
};
// 预期保留率：20-30%

场景B：构建增强数据集（中等严格）

const BALANCED_GATES = {
  minScore: 85,             // 较好分数即可
  maxIterations: 4,         // 允许更多迭代
  requireCodePair: false,   // 可以没有代码对比
  maxTokens: 8192,          // 放宽token限制
};
// 预期保留率：50-60%

场景C：构建实验数据集（宽松）

const EXPERIMENTAL_GATES = {
  minScore: 70,             // 及格即可
  maxIterations: 10,        // 不限制迭代
  requireCodePair: false,   // 不强制代码对比
  maxTokens: 16384,         // 较大token限制
};
// 预期保留率：80-90%

---

实战案例：完整的数据生成流程

背景

某团队使用BMAD-Speckit-SDD-Flow进行AI辅助开发3个月，希望构建专属的代码修复模型。他们积累了以下原始数据：

资产类型	原始数量	来源
Story实现记录	48个	BMAD-Speckit工作流
评估运行记录	1,200条	Scoring System
审计报告	48份	Audit模块
代码提交	156个commit	Git历史

Step 1: 数据提取

# 运行SFT提取命令
npx ts-node scripts/sft-extract.ts \
  --min-score 85 \
  --output ./sft-training/bugfix-v1.0.jsonl

# 提取过程日志
[INFO] Loading scoring records from packages/scoring/data...
[INFO] Found 1,200 scoring records
[INFO] Filtering by scenario: real_dev
[INFO] Filtering by phase_score >= 85
[INFO] Building candidate samples...
[INFO] Applying quality gates...
[INFO] Applying redaction rules...
[WARNING] Blocked 3 samples containing private keys
[INFO] Assigning deterministic splits...
[INFO] Exporting to JSONL...

# 输出摘要
共提取 312 条，覆盖 48 个 Story
- Accepted: 298 条
- Downgraded: 14 条
- Rejected: 888 条
  - 分数低于阈值: 720 条
  - 无source_path: 68 条
  - 缺少代码对比: 56 条
  - 脱敏阻断: 44 条（含私钥）

数据集划分：
- Train: 238 条 (76%)
- Validation: 37 条 (12%)
- Test: 37 条 (12%)

Step 2: 质量分析

// 运行质量分析报告
import { analyzeDataset } from './analytics';

const analysis = analyzeDataset('./sft-training/bugfix-v1.0.jsonl');

console.log(analysis.summary);
/*
{
  totalSamples: 312,
  avgPhaseScore: 91.5,
  avgTokenCount: 2847,
  categories: {
    securityFix: 89,      // SQL注入、XSS等
    performanceFix: 67,   // 算法优化
    styleFix: 76,         // 代码风格
    logicFix: 80          // 逻辑错误
  },
  languages: {
    typescript: 180,
    python: 89,
    go: 43
  },
  redactionSummary: {
    clean: 308,
    redacted: 4,          // 邮箱地址被脱敏
    blocked: 0            // 私钥样本已移除
  }
}
*/

Step 3: 格式转换与导出

// 转换为多种训练格式
import { exportDataset } from './export';

// 导出为OpenAI格式
await exportDataset({
  input: './sft-training/bugfix-v1.0.jsonl',
  output: './sft-training/openai-format/',
  format: 'openai_chat',
  filter: s => s.quality.acceptance_decision === 'accepted'
});

// 导出为HuggingFace格式
await exportDataset({
  input: './sft-training/bugfix-v1.0.jsonl',
  output: './sft-training/hf-format/',
  format: 'hf_conversational',
  splits: ['train', 'validation', 'test']
});

Step 4: 训练与评估

# 使用导出的数据进行训练（示例）
from transformers import AutoModelForCausalLM, TrainingArguments
from trl import SFTTrainer

# 加载基础模型：这里使用占位 ID，实际项目必须记录模型卡、许可证和评估快照
model = AutoModelForCausalLM.from_pretrained("your-org/code-model-base")

# 配置训练
training_args = TrainingArguments(
    output_dir="./bugfix-model",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    learning_rate=2e-4,
)

# 使用导出的数据集
dataset = load_dataset("json", data_files={
    "train": "./sft-training/hf-format/train.jsonl",
    "validation": "./sft-training/hf-format/validation.jsonl",
})

# 训练
trainer = SFTTrainer(
    model=model,
    train_dataset=dataset["train"],
    eval_dataset=dataset["validation"],
    args=training_args,
)
trainer.train()

训练效果对比

指标	基线模型	微调后	提升
安全漏洞修复成功率	62%	88%	+26%
性能问题识别率	58%	82%	+24%
代码风格合规率	71%	89%	+18%
平均修复建议质量	3.2/5	4.3/5	+1.1

---

结语：先把训练样本做对，再谈训练规模

把工程协作数据转成 SFT 样本，真正困难的部分从来不是“导出什么格式”，而是“哪些样本值得训练、哪些样本必须拒绝”。这背后需要任务契约、错误分类、人工反馈、验证证据和治理门控共同生效。

本文给出的核心框架可以归纳为三点：

1. 先建数据契约，再建数据量。样本必须可追溯、可验证、可解释，才能进入训练池。
2. 先做质量门控，再做自动化。没有硬门控和软门控，自动化只会放大噪音。
3. 先守住评估隔离，再追求指标提升。train/eval 混淆会让“模型进步”变成统计幻觉。

如果团队刚起步，最实用的顺序仍然是：先从小样本做通完整流程，再逐步扩展规模；先提升样本硬度，再提升训练频率。

这一篇完成的是“从交付轨迹到可训练样本”的中段工程。系列最后一篇会回到更长期的问题：当这套闭环稳定运行后，组织如何判断未来 AI 编程评估和协作范式的演进方向，避免把今天的流程当成明天的上限。

---

参考与致谢

• InstructGPT — Ouyang et al., OpenAI
• Flan Collection — Longpre et al., Google
• BMAD-Speckit-SDD-Flow — BMAD Method Team
• TRL SFTTrainer — Hugging Face
• PEFT LoRA — Hugging Face