大模型資料飛輪實戰：構建自動資料流水線驅動模型迭代

摘要

資料飛輪是大模型持續進化的核心引擎：使用者回饋→資料採集→標註→訓練→部署→新使用者回饋
自動資料標註的3種模式：LLM自標註、LLM輔助標註、人工審核標註，成本從低到高
RLHF偏好資料的自動生成：用強模型評判弱模型輸出，構建DPO訓練對
資料品質門控是飛輪的煞車片：去重、去噪、毒性偵測、分佈偵測4道關卡
本文提供從資料採集到模型迭代的完整飛輪Pipeline，含Airflow排程與品質監控

資料飛輪：大模型持續進化的引擎

飛輪原理

┌──────────────────────────────────────────────────────────────┐
│              大模型資料飛輪                                    │
│                                                                │
│          ┌──────────────┐                                     │
│     ┌──→ │ 1. 使用者    │ ←── 新功能/新場景                  │
│     │    │ 互動         │                                     │
│     │    └──────┬───────┘                                     │
│     │           │ 採集回饋資料                                 │
│     │    ┌──────▼───────┐                                     │
│     │    │ 2. 資料採集  │ 對話日誌、評分、糾錯                 │
│     │    └──────┬───────┘                                     │
│     │           │ 清洗+標註                                   │
│     │    ┌──────▼───────┐                                     │
│     │    │ 3. 資料標註  │ 自動標註+人工審核                   │
│     │    └──────┬───────┘                                     │
│     │           │ 構建訓練集                                   │
│     │    ┌──────▼───────┐                                     │
│     │    │ 4. 模型訓練  │ LoRA/QLoRA微調+RLHF對齊             │
│     │    └──────┬───────┘                                     │
│     │           │ 評估+部署                                   │
│     │    ┌──────▼───────┐                                     │
│     │    │ 5. 模型部署  │ A/B測試+灰度發布                    │
│     │    └──────┬───────┘                                     │
│     │           │ 新模型服務使用者                             │
│     └────┘      │                                             │
│          ┌──────▼───────┐                                     │
│          │ 1. 使用者    │ ← 更好的模型 → 更多使用者 → 更多資料│
│          │ 互動         │                                     │
│          └──────────────┘                                     │
└──────────────────────────────────────────────────────────────┘

飛輪關鍵指標

指標	說明	目標值
資料採集率	每日新增可訓練樣本數	>1,000條/天
標註吞吐	每日完成標註的樣本數	>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自標註	1,000條/小時	¥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

總結與延伸閱讀