AI 应用的数据飞轮——用用户数据持续改进模型
2026/4/30大约 9 分钟
AI 应用的数据飞轮——用用户数据持续改进模型
适读人群：正在做 AI 产品的工程师或创业者 | 阅读时长：约14分钟 | 核心价值：数据飞轮的实操路径，以及那些让飞轮转不起来的坑
2023年底我在一个 AI 写作辅助产品里尝试搭数据飞轮。那时候我对这个概念很兴奋，觉得这是一个一旦转起来就越来越厉害的正向循环：用户用产品 → 产生反馈数据 → 用数据微调模型 → 模型更好 → 更多用户用 → 产生更多数据……
理论很美，实操很残酷。
这篇文章是我这一年多折腾数据飞轮的真实经历，包括转起来的部分和根本没转起来的部分。
飞轮的理论模型
先把概念捋清楚，不然后面没法讨论。
数据飞轮的核心假设是：AI 产品天然有一个能产生有标注数据的渠道——用户的使用行为本身就是反馈信号。
一个典型的飞轮长这样：
用户使用产品
|
产生反馈
(显式: 点赞/踩, 修改AI输出)
(隐式: 停留时间, 复制, 重新生成)
|
数据清洗和标注
|
构建训练集
|
微调或强化学习
|
模型效果提升
|
用户体验更好 -> 更多使用 -> 更多反馈这个闭环能转起来的前提条件：
- 反馈信号有足够信息量（不能全是噪声）
- 数据量足够训练（通常 SFT 微调至少需要几百到几千条）
- 有能力做微调或 RLHF（有工程资源和算法资源）
- 微调后效果提升显著到用户能感知
这四个条件，同时满足比你想象的难得多。
第一步：反馈数据采集
先把反馈采集做起来。不采集数据，飞轮根本没有原料。
我在产品里做了三类反馈采集：
显式反馈（用户主动操作）
# 反馈事件记录
class FeedbackCollector:
async def record_explicit_feedback(
self,
user_id: str,
response_id: str,
feedback_type: str, # 'thumbs_up' | 'thumbs_down' | 'edited' | 'regenerated'
edited_content: str = None,
reason_code: str = None # 用户选的原因标签
):
"""记录显式反馈"""
# 取回对应的原始对话
original = await self.db.fetchrow(
"""SELECT prompt, response, model_version, prompt_version
FROM ai_responses WHERE id = $1""",
response_id
)
if not original:
return
await self.db.execute(
"""INSERT INTO feedback_records
(user_id, response_id, feedback_type,
original_prompt, original_response,
edited_content, reason_code,
model_version, prompt_version, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NOW())""",
user_id, response_id, feedback_type,
original['prompt'], original['response'],
edited_content, reason_code,
original['model_version'], original['prompt_version']
)
async def record_implicit_feedback(
self,
user_id: str,
response_id: str,
dwell_seconds: float,
copied: bool,
scroll_depth: float
):
"""记录隐式行为信号"""
await self.db.execute(
"""INSERT INTO implicit_feedback
(user_id, response_id, dwell_seconds, copied, scroll_depth, created_at)
VALUES ($1, $2, $3, $4, $5, NOW())""",
user_id, response_id, dwell_seconds, copied, scroll_depth
)我在 UI 上加了一个"这里不对，我来改"的入口，用户可以直接在 AI 的输出上修改，然后提交。这种"人工修正版本"是质量最高的训练数据，因为它包含了完整的(问题, 坏回答, 好回答)三元组。
第二步：数据清洗和质量过滤
原始反馈数据里有大量噪声，不能直接用来训练。
class DatasetBuilder:
async def build_sft_dataset(
self,
min_date: str,
quality_threshold: float = 0.7
) -> list[dict]:
"""从反馈数据构建 SFT 训练集"""
samples = []
# 1. 用户编辑版本：最高质量
edited = await self.db.fetch(
"""SELECT original_prompt, original_response, edited_content
FROM feedback_records
WHERE feedback_type = 'edited'
AND created_at > $1
AND LENGTH(edited_content) > 20 -- 过滤掉只改了标点的
AND LENGTH(edited_content) < 5000 -- 过滤超长
""",
min_date
)
for row in edited:
# 计算编辑距离，过滤掉改动太小的（可能只是错误点击）
edit_ratio = self._compute_edit_ratio(
row['original_response'], row['edited_content']
)
if edit_ratio > 0.1: # 改动超过10%才算有意义
samples.append({
"prompt": row['original_prompt'],
"chosen": row['edited_content'], # 用户改后的版本
"rejected": row['original_response'], # AI 原始输出
"source": "user_edit",
"quality": 0.9
})
# 2. 明确点赞的回答：正样本
thumbs_up = await self.db.fetch(
"""SELECT f.original_prompt, f.original_response
FROM feedback_records f
WHERE f.feedback_type = 'thumbs_up'
AND f.created_at > $1
AND NOT EXISTS (
-- 同一用户对同一 response 也有负面反馈
SELECT 1 FROM feedback_records f2
WHERE f2.response_id = f.response_id
AND f2.feedback_type = 'thumbs_down'
)""",
min_date
)
for row in thumbs_up:
# 只保留高质量的点赞（用停留时间作为质量代理指标）
dwell = await self.db.fetchval(
"""SELECT dwell_seconds FROM implicit_feedback
WHERE response_id = (
SELECT id FROM ai_responses
WHERE response = $1 LIMIT 1
)""",
row['original_response']
)
if dwell and dwell > 30: # 停留超过30秒才算认真看了
samples.append({
"prompt": row['original_prompt'],
"response": row['original_response'],
"source": "thumbs_up",
"quality": 0.7
})
return samples
def _compute_edit_ratio(self, original: str, edited: str) -> float:
"""计算编辑比例，用 Levenshtein 距离"""
import Levenshtein
distance = Levenshtein.distance(original, edited)
max_len = max(len(original), len(edited))
return distance / max_len if max_len > 0 else 0现实挑战：数据量不够
这是我遇到的最大问题。
我们产品的 MAU 在几千人级别，平均每天有几百次 AI 交互。其中：
- 主动点赞率：约 5%，每天约 15 条
- 主动点踩率：约 3%，每天约 9 条
- 用户编辑率：约 2%，每天约 6 条
按这个速度，一个月能积累的"用户编辑"数据大约 180 条。SFT 微调通常需要几百到几千条，一个月的数据勉强够做一次小规模实验。
这跟大厂讲的数据飞轮差距很大。大厂每天可能有几百万次交互，数据量根本不是问题。小产品面临的是：飞轮需要数据，但数据需要用户，用户需要好产品，好产品需要飞轮——一个先有鸡还是先有蛋的问题。
我的解法：
用数据增强弥补数量不足
class DataAugmenter:
async def augment_dataset(self, samples: list[dict]) -> list[dict]:
"""数据增强，扩充训练集"""
augmented = list(samples) # 先复制原始数据
for sample in samples:
if sample['source'] == 'user_edit' and sample['quality'] > 0.8:
# 对高质量样本做 prompt 改写（用 GPT-4o-mini 生成等价问法）
paraphrases = await self._generate_prompt_paraphrases(
sample['prompt'], n=3
)
for para in paraphrases:
augmented.append({
**sample,
"prompt": para,
"source": "augmented",
"quality": sample['quality'] * 0.8 # 增强数据质量打折
})
return augmented
async def _generate_prompt_paraphrases(self, prompt: str, n: int) -> list[str]:
"""生成同一意思的不同表达"""
resp = await llm_client.chat(
messages=[{
"role": "user",
"content": f"""将以下问题改写成{n}种不同的表达方式，保持意思完全相同：
原问题：{prompt}
直接输出{n}个改写版本，每行一个，不要编号："""
}],
model="gpt-4o-mini"
)
return [line.strip() for line in resp.content.split('\n') if line.strip()][:n]和标注团队合作
当数据量实在不够，我找了一个众包标注平台，花了大约 3000 元请人标注了 500 条数据。这个成本不低，但和微调后效果提升带来的用户留存提升比，还是合算的。
标注指南要写清楚，不然标注一致性很差：
# AI 写作助手标注指南
## 评分维度（各25分，总分100分）
1. 准确性：内容是否符合用户意图，有无明显错误
2. 流畅性：语言是否自然流畅，无语法问题
3. 实用性：输出是否能直接使用，无需大量修改
4. 风格贴合：是否符合用户指定的语气和风格要求
## 评分标准
- 90-100分：优秀，无需修改可直接使用
- 70-89分：良好，小修改后可用
- 50-69分：一般，需要较多修改
- 50分以下：差，基本不可用
## 常见错误类型（自动打低分）
- 内容与用户要求不符（-30分）
- 语气与要求不符（-20分）
- 明显的事实错误（-40分）
- 出现重复段落（-20分）微调实施：用 LoRA 降低门槛
数据集准备好之后，怎么微调？
对小团队来说，全量微调太贵。LoRA 是标准方案：
# 使用 LLaMA Factory 做 LoRA 微调（简化示例）
# 实际用命令行 + YAML 配置更方便
from transformers import AutoTokenizer, AutoModelForCausalLM
from peft import LoraConfig, get_peft_model, TaskType
import torch
def prepare_lora_model(base_model_name: str):
tokenizer = AutoTokenizer.from_pretrained(base_model_name)
model = AutoModelForCausalLM.from_pretrained(
base_model_name,
torch_dtype=torch.bfloat16,
device_map="auto"
)
lora_config = LoraConfig(
task_type=TaskType.CAUSAL_LM,
r=16, # LoRA rank，越大参数越多
lora_alpha=32, # 缩放系数，通常是 r 的 2 倍
lora_dropout=0.1,
target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
bias="none"
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters() # 看看实际训练的参数量
return model, tokenizer我用的是 A100 40GB 的云 GPU，微调 Qwen2.5-7B 大约需要 2-4 小时，成本在 30-60 元人民币左右（按量计费）。这个成本是可以接受的。
效果评估：飞轮转没转
微调完之后要评估，这步很多人忽视了，直接上线，结果翻车。
class ModelEvaluator:
async def compare_models(
self,
baseline_model: str,
new_model: str,
eval_prompts: list[str]
) -> dict:
"""用同一批 prompt 比较两个模型的输出质量"""
results = {
"win": 0, # new_model 更好
"tie": 0, # 差不多
"lose": 0, # baseline 更好
"details": []
}
for prompt in eval_prompts:
# 两个模型各跑一次
baseline_output = await self._run_model(baseline_model, prompt)
new_output = await self._run_model(new_model, prompt)
if baseline_output == new_output:
results["tie"] += 1
continue
# 用 GPT-4o 做裁判（比人工评估便宜，但也有偏差）
judge_result = await self._judge_with_gpt4(
prompt, baseline_output, new_output
)
results[judge_result] += 1
results["details"].append({
"prompt": prompt[:100],
"result": judge_result
})
total = len(eval_prompts)
results["win_rate"] = results["win"] / total
results["summary"] = f"新模型胜率: {results['win_rate']:.1%}"
return results
async def _judge_with_gpt4(
self, prompt: str, output_a: str, output_b: str
) -> str:
"""用 GPT-4 判断哪个输出更好"""
judge_prompt = f"""你是一个客观的评估者。比较以下两个 AI 回答，判断哪个更好。
用户问题：{prompt}
回答A：{output_a}
回答B：{output_b}
只输出一个字：A（A更好）、B（B更好）或 C（差不多）"""
resp = await llm_client.chat(
messages=[{"role": "user", "content": judge_prompt}],
model="gpt-4o"
)
result = resp.content.strip().upper()
if "A" in result:
return "lose" # new_model 是 B，A 赢了说明 new 输了
elif "B" in result:
return "win"
else:
return "tie"我做的第一次微调，效果评估结果是：新模型胜率 54%，基线胜率 28%，平局 18%。这个结果说明微调确实有提升，但提升幅度不是特别大。用户上线后的评分提升了 0.2 分，从 3.8 到 4.0。
飞轮目前的状态
诚实说，我的飞轮目前处于"缓慢转动"状态，没有达到想象中那种飞速转动的效果。
原因：用户量还没到大规模积累数据的量级，数据的自然流入速度不够快，需要人工介入（标注、增强）来补。
但它确实是在转的，每隔一到两个月微调一次，每次都有可见的改进，积累下来产品体验在持续变好。
对于中小团队来说，数据飞轮不是"搭好了自己转"的神器，而是一个需要持续人工投入的工程。只是这个投入会随着用户量增长逐渐减少，而效果会持续提升。
这就是为什么大厂的 AI 产品越来越强——他们的飞轮已经转得足够快，靠自然积累就能持续改进。这是规模效应，不是魔法。