大模型数据飞轮实战：构建自动数据流水线驱动模型迭代

摘要

数据飞轮是大模型持续进化的核心引擎：用户反馈→数据采集→标注→训练→部署→新用户反馈
自动数据标注的3种模式：LLM自标注、LLM辅助标注、人工审核标注，成本从低到高
RLHF偏好数据的自动生成：用强模型评判弱模型输出，构建DPO训练对
数据质量门控是飞轮的刹车片：去重、去噪、毒性检测、分布检测4道关卡
本文提供从数据采集到模型迭代的完整飞轮Pipeline，含Airflow调度与质量监控

数据飞轮：大模型持续进化的引擎

飞轮原理

┌──────────────────────────────────────────────────────────────┐
│              大模型数据飞轮                                    │
│                                                                │
│          ┌──────────────┐                                     │
│     ┌──→ │ 1. 用户交互  │ ←── 新功能/新场景                  │
│     │    └──────┬───────┘                                     │
│     │           │ 采集反馈数据                                 │
│     │    ┌──────▼───────┐                                     │
│     │    │ 2. 数据采集  │ 对话日志、评分、纠错                 │
│     │    └──────┬───────┘                                     │
│     │           │ 清洗+标注                                   │
│     │    ┌──────▼───────┐                                     │
│     │    │ 3. 数据标注  │ 自动标注+人工审核                   │
│     │    └──────┬───────┘                                     │
│     │           │ 构建训练集                                   │
│     │    ┌──────▼───────┐                                     │
│     │    │ 4. 模型训练  │ LoRA/QLoRA微调+RLHF对齐             │
│     │    └──────┬───────┘                                     │
│     │           │ 评估+部署                                   │
│     │    ┌──────▼───────┐                                     │
│     │    │ 5. 模型部署  │ A/B测试+灰度发布                    │
│     │    └──────┬───────┘                                     │
│     │           │ 新模型服务用户                               │
│     └────┘      │                                             │
│          ┌──────▼───────┐                                     │
│          │ 1. 用户交互  │ ← 更好的模型 → 更多用户 → 更多数据  │
│          └──────────────┘                                     │
└──────────────────────────────────────────────────────────────┘

飞轮关键指标

指标	说明	目标值
数据采集率	每日新增可训练样本数	>1000条/天
标注吞吐	每日完成标注的样本数	>500条/天
标注质量	人工抽检通过率	>95%
训练周期	从数据就绪到模型部署	<7天
模型提升	新模型vs旧模型在核心指标上的提升	>2%

自动数据采集：从用户反馈到训练样本

3种数据采集模式

模式	数据来源	质量	数量	成本
隐式反馈	点赞/踩、复制、停留时长	低	高	低
显式反馈	评分、纠错、重写	高	中	中
主动采集	标注任务、众包	最高	低	高

隐式反馈采集

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class UserFeedback:
    session_id: str
    user_id: str
    prompt: str
    response: str
    feedback_type: str
    feedback_value: float
    timestamp: datetime = field(default_factory=datetime.now)
    metadata: dict = field(default_factory=dict)

class FeedbackCollector:
    def __init__(self, db_pool):
        self.db = db_pool

    async def collect_implicit(self, session_id: str, user_id: str, event: dict):
        feedback = UserFeedback(
            session_id=session_id,
            user_id=user_id,
            prompt=event.get("prompt", ""),
            response=event.get("response", ""),
            feedback_type=event["type"],
            feedback_value=event.get("value", 0.0),
        )

        feedback_map = {
            "thumbs_up": ("positive", 1.0),
            "thumbs_down": ("negative", -1.0),
            "copy_response": ("positive", 0.5),
            "regenerate": ("negative", -0.3),
            "long_dwell_time": ("positive", 0.2),
        }

        fb_type, fb_value = feedback_map.get(event["type"], ("neutral", 0.0))
        feedback.feedback_type = fb_type
        feedback.feedback_value = fb_value

        await self._save(feedback)

    async def collect_explicit(self, session_id: str, user_id: str,
                                prompt: str, response: str,
                                corrected_response: str = None,
                                rating: int = None):
        feedback = UserFeedback(
            session_id=session_id,
            user_id=user_id,
            prompt=prompt,
            response=response,
            feedback_type="explicit",
            feedback_value=rating if rating else 0.0,
            metadata={"corrected_response": corrected_response} if corrected_response else {},
        )
        await self._save(feedback)

自动数据标注：3种模式与实现

模式1：LLM自标注

class LLMSelfAnnotator:
    def __init__(self, llm_client, judge_model: str = "Qwen/Qwen2.5-72B-Instruct"):
        self.llm = llm_client
        self.judge_model = judge_model

    async def annotate(self, prompt: str, response: str) -> dict:
        judge_prompt = f"""评估以下AI回答的质量。

问题：{prompt}
回答：{response}

请从以下维度评分(1-5)：
1. 准确性：回答是否正确
2. 完整性：回答是否完整
3. 清晰度：回答是否清晰易懂
4. 安全性：回答是否安全无害

输出JSON格式：{{"accuracy": N, "completeness": N, "clarity": N, "safety": N, "overall": N, "reason": "..."}}"""

        result = self.llm.chat.completions.create(
            model=self.judge_model,
            messages=[{"role": "user", "content": judge_prompt}],
            temperature=0.0,
            max_tokens=512,
            response_format={"type": "json_object"},
        )

        return json.loads(result.choices[0].message.content)

模式2：LLM辅助+人工审核

class HumanInLoopAnnotator:
    def __init__(self, llm_annotator: LLMSelfAnnotator, review_threshold: float = 3.0):
        self.llm_annotator = llm_annotator
        self.review_threshold = review_threshold

    async def annotate_batch(self, samples: list[dict]) -> list[dict]:
        results = []
        for sample in samples:
            auto_result = await self.llm_annotator.annotate(sample["prompt"], sample["response"])
            sample["auto_annotation"] = auto_result

            if auto_result.get("overall", 0) < self.review_threshold:
                sample["needs_review"] = True
            else:
                sample["needs_review"] = False

            results.append(sample)
        return results

标注模式对比

模式	吞吐	成本/条	精度	适用场景
LLM自标注	1000条/小时	¥0.01	80%	大规模初筛
LLM辅助+人工	200条/小时	¥0.5	95%	生产推荐
纯人工标注	50条/小时	¥5.0	99%	高价值数据

RLHF偏好数据自动生成

Constitutional AI偏好对生成

class PreferenceDataGenerator:
    def __init__(self, target_model, judge_model, llm_client):
        self.target = target_model
        self.judge = judge_model
        self.llm = llm_client

    async def generate_preference_pairs(self, prompts: list[str]) -> list[dict]:
        pairs = []
        for prompt in prompts:
            response_a = await self._generate_with_params(prompt, temperature=0.3)
            response_b = await self._generate_with_params(prompt, temperature=0.9)

            winner = await self._judge_preference(prompt, response_a, response_b)

            pairs.append({
                "prompt": prompt,
                "chosen": response_a if winner == "A" else response_b,
                "rejected": response_b if winner == "A" else response_a,
            })

        return pairs

    async def _judge_preference(self, prompt: str, response_a: str, response_b: str) -> str:
        judge_prompt = f"""你是回答质量评判专家。哪个回答更好？

问题：{prompt}

回答A：{response_a}

回答B：{response_b}

只输出 "A" 或 "B"。"""

        result = self.llm.chat.completions.create(
            model=self.judge,
            messages=[{"role": "user", "content": judge_prompt}],
            temperature=0.0,
            max_tokens=1,
        )
        return result.choices[0].message.content.strip()

数据质量门控：飞轮的刹车片

4道质量关卡

┌──────────────────────────────────────────────────────────┐
│              数据质量4道关卡                                │
│                                                            │
│  关卡1: 去重                                              │
│  ┌──────────────────────────────────────────┐             │
│  │ MinHash/SimHash 语义去重                  │             │
│  │ 相似度>0.85 → 丢弃重复样本                │             │
│  └──────────────────────────────────────────┘             │
│                    ↓                                       │
│  关卡2: 去噪                                              │
│  ┌──────────────────────────────────────────┐             │
│  │ 规则过滤 + LLM质量评估                    │             │
│  │ 空回答/乱码/过短 → 丢弃                   │             │
│  └──────────────────────────────────────────┘             │
│                    ↓                                       │
│  关卡3: 毒性检测                                          │
│  ┌──────────────────────────────────────────┐             │
│  │ 毒性分类器 + 敏感词过滤                    │             │
│  │ 有害内容 → 丢弃或脱敏                      │             │
│  └──────────────────────────────────────────┘             │
│                    ↓                                       │
│  关卡4: 分布检测                                          │
│  ┌──────────────────────────────────────────┐             │
│  │ 新数据 vs 训练集分布对比                   │             │
│  │ 分布偏移过大 → 人工审核                    │             │
│  └──────────────────────────────────────────┘             │
└──────────────────────────────────────────────────────────┘

质量门控实现

class DataQualityGate:
    def __init__(self, similarity_threshold: float = 0.85, min_length: int = 20):
        self.similarity_threshold = similarity_threshold
        self.min_length = min_length
        self.seen_hashes = set()

    def check(self, sample: dict) -> tuple[bool, str]:
        content = sample.get("prompt", "") + sample.get("response", "")

        if len(content) < self.min_length:
            return False, "内容过短"

        content_hash = hashlib.md5(content.encode()).hexdigest()
        if content_hash in self.seen_hashes:
            return False, "重复内容"
        self.seen_hashes.add(content_hash)

        if self._contains_toxic_content(sample):
            return False, "包含有害内容"

        return True, "通过"

    def _contains_toxic_content(self, sample: dict) -> bool:
        toxic_keywords = ["暴力", "自杀", "炸弹制造"]
        text = (sample.get("prompt", "") + sample.get("response", "")).lower()
        return any(kw in text for kw in toxic_keywords)

飞轮调度：Airflow+K8s闭环系统

Airflow DAG

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    "owner": "ml-team",
    "depends_on_past": False,
    "start_date": datetime(2026, 1, 1),
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    "llm_data_flywheel",
    default_args=default_args,
    schedule_interval="@weekly",
    catchup=False,
) as dag:

    collect_feedback = PythonOperator(
        task_id="collect_feedback",
        python_callable=collect_user_feedback,
    )

    quality_gate = PythonOperator(
        task_id="quality_gate",
        python_callable=run_quality_checks,
    )

    auto_annotate = PythonOperator(
        task_id="auto_annotate",
        python_callable=run_auto_annotation,
    )

    generate_preference = PythonOperator(
        task_id="generate_preference_data",
        python_callable=generate_rlhf_pairs,
    )

    train_model = PythonOperator(
        task_id="train_model",
        python_callable=run_lora_training,
    )

    evaluate = PythonOperator(
        task_id="evaluate_model",
        python_callable=evaluate_new_model,
    )

    deploy = PythonOperator(
        task_id="deploy_model",
        python_callable=deploy_canary,
    )

    collect_feedback >> quality_gate >> auto_annotate >> generate_preference >> train_model >> evaluate >> deploy

总结与引流

数据飞轮是大模型持续进化的核心引擎。从用户反馈到模型部署的闭环Pipeline，让模型每周都在变好。关键在于数据质量门控——飞轮转得越快，刹车片越重要。

开发要点回顾：

数据飞轮5步：采集→标注→训练→评估→部署→循环
LLM辅助+人工审核是标注的性价比最优解
RLHF偏好数据可用强模型评判弱模型自动生成
4道质量关卡：去重→去噪→毒性检测→分布检测
Airflow+K8s实现飞轮自动化调度

相关阅读：

大模型微调实战：LoRA、QLoRA与RLHF — 飞轮中的微调环节
AI Agent多轮记忆实战 — 用户反馈数据的记忆化
大模型推理加速基准测试 — 新模型部署的推理优化

权威参考：

摘要

目录

数据飞轮：大模型持续进化的引擎

飞轮原理

飞轮关键指标

自动数据采集：从用户反馈到训练样本

3种数据采集模式

隐式反馈采集

自动数据标注：3种模式与实现

模式1：LLM自标注

模式2：LLM辅助+人工审核

标注模式对比

RLHF偏好数据自动生成

Constitutional AI偏好对生成

数据质量门控：飞轮的刹车片

4道质量关卡

质量门控实现

飞轮调度：Airflow+K8s闭环系统

Airflow DAG

总结与引流