Skip to content
🔗 分享本题
查看我的学习进度 →

Agent 数据飞轮将获授权的真实失败清洗标注为高质量数据,经评测灰度改进后回流产品

🧠 记忆锚点:把真实失败变成高质量样本,经评测灰度回到产品;飞轮靠质量与闭环,不靠堆日志。

💡 答案要点

数据飞轮 = 产品使用 → 收集数据 → 改进模型 → 产品更好 → 更多使用 的正向循环

┌─────────────────────────────────────────────────────┐
│                   数据飞轮循环                        │
│                                                      │
│   用户使用 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%,完全数据驱动,不需要手动写规则。"