
🧠 记忆锚点:把真实失败变成高质量样本,经评测灰度回到产品;飞轮靠质量与闭环,不靠堆日志。
💡 答案要点
数据飞轮 = 产品使用 → 收集数据 → 改进模型 → 产品更好 → 更多使用 的正向循环
┌─────────────────────────────────────────────────────┐
│ 数据飞轮循环 │
│ │
│ 用户使用 Agent ──→ 收集交互日志 │
│ ↑ ↓ │
│ 产品更好 数据清洗/标注 │
│ ↑ ↓ │
│ 模型更新 ←── 微调/RAG更新/规则优化 │
└─────────────────────────────────────────────────────┘四步构建数据飞轮:
展开 Python 代码示例(77 行)
python
class DataFlywheel:
"""Agent 数据飞轮实现"""
# Step 1: 全量日志采集
def collect_interaction_logs(self, interaction: dict):
"""每次 Agent 交互后记录"""
log = {
"timestamp": datetime.now().isoformat(),
"session_id": interaction["session_id"],
"user_input": interaction["user_input"],
"agent_output": interaction["agent_output"],
"tools_called": interaction["tools_called"],
"latency_ms": interaction["latency_ms"],
"tokens_used": interaction["tokens_used"],
# 关键:收集隐式反馈
"user_continued": interaction.get("user_continued", False),
"user_thumbs_up": interaction.get("feedback"),
"task_completed": interaction.get("task_completed"),
}
self.data_store.append(log)
# Step 2: 自动质量评估(减少人工标注成本)
def auto_label(self, log: dict) -> dict:
"""用规则+LLM-as-Judge 自动打标"""
score = 0.5 # 默认中等
# 规则信号
if log["user_thumbs_up"] == "up":
score = 1.0
elif log["user_thumbs_up"] == "down":
score = 0.0
elif log["user_continued"]:
score = 0.7 # 用户继续对话=满意
elif log["task_completed"]:
score = 0.8
# LLM-as-Judge 补充评估(仅对 score=0.5 的模糊样本)
if score == 0.5:
judge_result = self.llm_judge(
question=log["user_input"],
answer=log["agent_output"]
)
score = judge_result["score"]
return {**log, "quality_score": score}
# Step 3: 高质量数据筛选
def select_training_data(self, logs: list) -> dict:
"""从日志中筛选训练数据"""
labeled = [self.auto_label(log) for log in logs]
return {
# 高分样本 → SFT 正样本
"positive": [l for l in labeled if l["quality_score"] >= 0.8],
# 低分样本 → SFT 负样本 / DPO 对比样本
"negative": [l for l in labeled if l["quality_score"] <= 0.3],
# 中等样本 → 人工审核队列
"review": [l for l in labeled if 0.3 < l["quality_score"] < 0.8],
}
# Step 4: 定期更新模型
def update_cycle(self):
"""每周/每月触发更新循环"""
# 收集上周数据
recent_logs = self.get_recent_logs(days=7)
training_data = self.select_training_data(recent_logs)
# 高频错误模式 → 更新 RAG 知识库
error_patterns = self.extract_error_patterns(training_data["negative"])
self.update_knowledge_base(error_patterns)
# 积累足够正负样本 → 触发微调
if len(training_data["positive"]) > 1000:
self.trigger_finetuning(
positives=training_data["positive"],
negatives=training_data["negative"]
)数据飞轮的三大价值:
| 价值 | 说明 |
|---|---|
| 持续改进 | 每周迭代,模型越用越好 |
| 降低标注成本 | 隐式反馈 + LLM-as-Judge 替代大部分人工标注 |
| 构建竞争壁垒 | 数据积累越多,后来者越难追上 |
面试话术:
示例表达(仅在能用本人经历或可复现实验佐证时使用): "数据飞轮是 AI 产品的护城河。我设计的飞轮是四步:全量日志→自动打标(规则+LLM-as-Judge)→筛选高质量训练数据→定期触发 RAG 更新或微调。关键是'隐式反馈'的利用——用户是否继续对话、是否完成任务,这些比显式点赞更真实且量大。我们用这套机制让客服 Agent 在3个月内成功率从 72% 提到 89%,完全数据驱动,不需要手动写规则。"