# 10.9 Agentic 数据飞轮：让 Agent 自我进化

> 🔄 *"最好的训练数据不是人工标注的，而是 Agent 在真实环境中成功、失败、修正和完成任务时留下的轨迹。"*

第 10.8 节解决了“如何构建第一批 Agent 训练数据”。但这只是起点。

真正强大的 Agent 系统遵循一个循环：**Agent 运行 → 产生轨迹 → 评估结果与过程 → 提取成功和失败信号 → 训练更强模型 → 更强模型产生更好轨迹 → ...**

这个闭环就是 **Agentic 数据飞轮（Agentic Data Flywheel）**——也是 DeepSeek、OpenAI、Anthropic 等顶尖团队能持续迭代的核心秘密之一。

从训练范式上看，数据飞轮是 Agentic-RL 的工程化形态：SFT 提供第一版可用策略，线上环境提供真实反馈，奖励与过滤系统把反馈变成训练信号，下一轮模型再进入环境继续试错。

---

## 为什么数据飞轮是 Agentic-RL 的关键？

如果只做一次 SFT，模型能力会停留在训练集覆盖范围内；如果有数据飞轮，模型会不断遇到新任务、新工具、新错误和新边界，并把这些经验转化为下一轮能力。

| 问题 | 一次性 SFT 的做法 | 数据飞轮的做法 |
|------|------------------|---------------|
| **普通 SFT 不足以训练 Agent** | 只学习已有专家轨迹 | 持续收集真实轨迹，补齐未覆盖场景 |
| **工具调用需要环境反馈** | 离线判断工具调用是否“看起来正确” | 记录工具是否真的执行成功、返回是否可用 |
| **多步任务需要轨迹级奖励** | 逐条学习 assistant 输出 | 按整条轨迹评估任务完成率、恢复率和成本 |
| **结果正确不等于过程可靠** | 只保留最终答对的样本 | 同时检查中间步骤、异常处理和重复调用 |
| **失败轨迹也有训练价值** | 失败样本通常被丢弃 | 失败被分类、修正、转化为偏好对和专项数据 |
| **Agent 越用越强** | 训练完成后能力冻结 | 使用越多，轨迹越多，薄弱点越清晰，迭代越快 |

这就是 Agentic-RL 与数据飞轮的关系：**RL 提供“如何从反馈中学习”的算法，飞轮提供“持续产生反馈”的系统**。

---

## 飞轮的整体架构

![Agentic 数据飞轮架构图](../svg/chapter_agentic_rl_09_flywheel_loop.svg)

每一圈飞轮转动，模型能力提升一个台阶；台阶越高，产生的轨迹质量越好，下一圈的原材料更优质。

---

## 第一层：轨迹收集

飞轮的原材料是 Agent 运行时产生的**完整交互记录**（Trajectory）。

### 什么值得收集？

```python
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class AgentTrajectory:
    """一条完整的 Agent 运行轨迹"""
    
    # 基本元数据
    trajectory_id: str
    timestamp: datetime
    session_id: str
    
    # 完整交互记录
    system_prompt: str
    messages: list[dict]         # 完整的 system/user/assistant/tool 序列
    tool_definitions: list[dict] # 本次会话的工具定义
    
    # 结果信息
    final_response: str
    task_completed: bool
    
    # 成本信息（用于后续筛选）
    total_tokens: int
    total_tool_calls: int
    tool_call_failures: int
    latency_ms: float
    
    # 用户反馈（如有）
    user_rating: float | None = None
    user_followup: bool = False   # 用户是否继续追问（隐式满意信号）
    
    # 自动标注（异步计算）
    auto_quality_score: float = 0.0
    is_in_training_set: bool = False


class TrajectoryCollector:
    """线上轨迹收集系统"""
    
    def __init__(self, storage_backend, sampling_rate: float = 1.0):
        self.storage = storage_backend
        self.sampling_rate = sampling_rate  # 采样率（高流量时可<1.0）
    
    async def record(self, trajectory: AgentTrajectory):
        """记录一条轨迹（异步，不阻塞主流程）"""
        import random
        
        # 采样：不是每条都存
        if random.random() > self.sampling_rate:
            return
        
        # 脱敏处理
        cleaned = await self._anonymize(trajectory)
        
        # 异步存储（不影响线上延迟）
        await self.storage.save(cleaned)
        
        # 触发异步质量评分
        asyncio.create_task(self._score_trajectory(cleaned))
    
    async def _anonymize(self, traj: AgentTrajectory) -> AgentTrajectory:
        """脱敏：删除 PII"""
        import re
        for msg in traj.messages:
            content = str(msg.get("content", ""))
            content = re.sub(r'\b1[3-9]\d{9}\b', '[PHONE]', content)
            content = re.sub(r'\b[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\b', '[EMAIL]', content)
            msg["content"] = content
        return traj
```

---

## 第二层：质量过滤与标注

原始轨迹质量参差不齐，必须经过严格筛选才能进入训练集。

### 双维度过滤：结果 × 过程

```python
class TrajectoryFilter:
    """
    双维度过滤：
    - 结果维度：任务完成了吗？用户满意吗？
    - 过程维度：推理合理吗？工具使用规范吗？
    """
    
    def __init__(self, reward_model=None):
        self.reward_model = reward_model
    
    def compute_quality_score(self, traj: AgentTrajectory) -> float:
        """综合质量分（0-1）"""
        scores = {}
        
        # ── 结果维度（占 50%）──────────────────────────────
        # 1a. 显式用户满意度
        if traj.user_rating is not None:
            scores["explicit_satisfaction"] = traj.user_rating / 5.0
        else:
            scores["explicit_satisfaction"] = 0.5  # 中性默认值
        
        # 1b. 隐式满意度（用户继续对话 = 满意）
        scores["implicit_satisfaction"] = 0.8 if traj.user_followup else 0.4
        
        # 1c. 任务是否有实质性完成
        scores["task_completion"] = 1.0 if traj.task_completed else 0.2
        
        # ── 过程维度（占 50%）──────────────────────────────
        # 2a. 工具调用成功率
        if traj.total_tool_calls > 0:
            success_rate = 1 - traj.tool_call_failures / traj.total_tool_calls
            scores["tool_success_rate"] = success_rate
        else:
            scores["tool_success_rate"] = 1.0  # 没调用工具也算正常
        
        # 2b. 效率（轨迹长度惩罚，避免空转）
        optimal_calls = max(1, traj.total_tool_calls)
        efficiency = min(1.0, 5 / max(traj.total_tool_calls, 1))
        scores["efficiency"] = efficiency if traj.total_tool_calls <= 5 else 0.5
        
        # 2c. 格式合规性（工具调用都是合法 JSON）
        scores["format_compliance"] = self._check_format_compliance(traj.messages)
        
        # 2d. 奖励模型打分（如果有）
        if self.reward_model:
            rm_score = self.reward_model.score(traj)
            scores["reward_model"] = rm_score
        
        # 加权平均
        weights = {
            "explicit_satisfaction": 0.2,
            "implicit_satisfaction": 0.1,
            "task_completion": 0.2,
            "tool_success_rate": 0.2,
            "efficiency": 0.1,
            "format_compliance": 0.1,
            "reward_model": 0.1 if self.reward_model else 0,
        }
        
        total_weight = sum(weights[k] for k in scores)
        final_score = sum(scores[k] * weights[k] for k in scores) / total_weight
        return final_score
    
    def _check_format_compliance(self, messages: list[dict]) -> float:
        """检查工具调用格式合规率"""
        tool_call_messages = [m for m in messages 
                              if m.get("role") == "assistant" and "tool_calls" in m]
        if not tool_call_messages:
            return 1.0
        
        valid = 0
        for msg in tool_call_messages:
            try:
                for call in msg["tool_calls"]:
                    assert "name" in call and "arguments" in call
                    assert isinstance(call["arguments"], dict)
                valid += 1
            except (AssertionError, KeyError, TypeError):
                pass
        
        return valid / len(tool_call_messages)
    
    def filter_batch(self, 
                     trajectories: list[AgentTrajectory],
                     min_score: float = 0.7) -> list[AgentTrajectory]:
        """批量过滤，返回高质量样本"""
        filtered = []
        score_distribution = []
        
        for traj in trajectories:
            score = self.compute_quality_score(traj)
            traj.auto_quality_score = score
            score_distribution.append(score)
            
            if score >= min_score:
                filtered.append(traj)
        
        # 打印统计信息
        import statistics
        print(f"过滤结果：{len(filtered)}/{len(trajectories)} 通过")
        print(f"质量分布：mean={statistics.mean(score_distribution):.2f}, "
              f"median={statistics.median(score_distribution):.2f}")
        return filtered
```

### 负样本的价值：失败的轨迹也有用

在 Agent 系统里，失败不是噪声，而是最密集的学习信号。成功轨迹告诉模型“这条路走通了”，失败轨迹则告诉模型“哪些动作看起来合理但实际上会导致任务失败”。

失败轨迹至少有三种用法：

1. **做偏好学习**：同一个任务下，成功轨迹作为 `chosen`，失败轨迹作为 `rejected`，用于 DPO/IPO 等偏好优化。
2. **做错误恢复 SFT**：截取到失败点之前的上下文，让强模型或人工专家补写正确的下一步。
3. **做 RL 奖励修正**：把高频失败类型转化为惩罚项，例如非法工具、重复调用、未检查结果、绕过审批。

| 失败类型 | 离线日志里的表现 | 训练价值 |
|---------|----------------|----------|
| 工具幻觉 | 调用了不存在的 API | 强化工具 schema 约束 |
| 参数错误 | 字段缺失、类型错误、单位错误 | 学习参数校验与自我修正 |
| 环境异常 | API 超时、网页元素变化、权限不足 | 学习重试、降级和求助策略 |
| 长程断链 | 忘记前面结果，重复执行同一步 | 学习状态跟踪和轨迹级规划 |
| 结果偶然正确 | 最终答对但步骤混乱、成本极高 | 学习过程奖励和成本控制 |

这也是为什么数据飞轮不能只收集“用户点赞”的样本。真正有价值的系统会同时收集高质量成功轨迹和可诊断失败轨迹，前者用于增强能力，后者用于修补边界。

```python
class NegativeSampleExtractor:
    """从失败轨迹中提取对比学习样本"""
    
    def extract_contrastive_pairs(
        self, 
        good_traj: AgentTrajectory,
        bad_traj: AgentTrajectory
    ) -> dict:
        """
        构建偏好对（用于 DPO 训练）
        相同任务，一个成功轨迹（chosen）+ 一个失败轨迹（rejected）
        """
        return {
            "prompt": good_traj.messages[1]["content"],  # 用户问题
            "chosen": self._trajectory_to_text(good_traj),
            "rejected": self._trajectory_to_text(bad_traj),
            "chosen_score": good_traj.auto_quality_score,
            "rejected_score": bad_traj.auto_quality_score,
        }
    
    def extract_error_cases(self, 
                            bad_trajectories: list[AgentTrajectory]) -> list[dict]:
        """
        从失败案例中提取训练样本：
        错误发生点 → 模型应该怎么做（人工修正或强模型修正）
        """
        error_samples = []
        
        for traj in bad_trajectories:
            # 找到第一个出错的位置
            error_point = self._find_first_error(traj.messages)
            if error_point is None:
                continue
            
            # 截取到出错点之前的上下文
            context = traj.messages[:error_point]
            
            # 用强模型生成正确的续写
            correct_continuation = self._get_correct_continuation(context, traj.tool_definitions)
            
            if correct_continuation:
                error_samples.append({
                    "context": context,
                    "wrong_response": traj.messages[error_point],
                    "correct_response": correct_continuation,
                    "error_type": self._classify_error(traj.messages[error_point]),
                })
        
        return error_samples
    
    def _classify_error(self, message: dict) -> str:
        """对错误类型分类（用于分析）"""
        content = str(message.get("content", ""))
        tool_calls = message.get("tool_calls", [])
        
        if tool_calls:
            call_name = tool_calls[0].get("name", "")
            if "not_exist" in call_name.lower() or call_name not in self.available_tools:
                return "hallucinated_tool"
            if not isinstance(tool_calls[0].get("arguments"), dict):
                return "invalid_format"
        
        if "error" in content.lower() and len(content) < 30:
            return "bare_error_propagation"  # 直接把错误消息返回给用户
        
        return "reasoning_error"
```

---

## 第三层：自动标注与奖励模型

对于没有用户反馈的轨迹，需要自动判断质量。

### 基于可验证任务的奖励（RLVR 思路）

```python
class VerifiableRewardCalculator:
    """
    对于有可验证答案的任务，直接验证结果。
    这是最精准的奖励信号，无需奖励模型。
    
    类似于 DeepSeek-R1 的做法：只训练可验证任务，
    奖励来自于答案本身的正确性。
    """
    
    def __init__(self):
        self.verifiers = {
            "math": self._verify_math,
            "code": self._verify_code_execution,
            "sql": self._verify_sql_result,
            "factual": self._verify_factual_claim,
        }
    
    def compute_reward(self, 
                       traj: AgentTrajectory, 
                       task_type: str,
                       ground_truth=None) -> float:
        """计算奖励分数"""
        if task_type not in self.verifiers:
            return None  # 无法自动验证
        
        final_response = traj.final_response
        return self.verifiers[task_type](final_response, ground_truth)
    
    def _verify_math(self, response: str, ground_truth: float) -> float:
        """验证数学答案"""
        import re
        # 提取回复中的数字
        numbers = re.findall(r'-?\d+\.?\d*', response)
        if not numbers:
            return 0.0
        
        predicted = float(numbers[-1])  # 通常最后一个数字是答案
        if abs(predicted - ground_truth) < 1e-6:
            return 1.0
        elif abs(predicted - ground_truth) / (abs(ground_truth) + 1e-9) < 0.05:
            return 0.5  # 近似正确
        return 0.0
    
    def _verify_code_execution(self, response: str, expected_output: str) -> float:
        """提取并执行代码，验证输出"""
        import subprocess, re, tempfile
        
        # 提取代码块
        code_match = re.search(r'```python\n(.*?)\n```', response, re.DOTALL)
        if not code_match:
            return 0.0
        
        code = code_match.group(1)
        
        try:
            with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
                f.write(code)
                fname = f.name
            
            result = subprocess.run(
                ['python', fname], 
                capture_output=True, text=True, timeout=5
            )
            
            actual_output = result.stdout.strip()
            return 1.0 if actual_output == expected_output.strip() else 0.0
            
        except subprocess.TimeoutExpired:
            return 0.0
        except Exception:
            return 0.0


class LLMRewardModel:
    """
    当任务不可自动验证时，用强 LLM 作为奖励模型。
    
    成本权衡：每条轨迹调用一次 GPT-4.1 评估。
    建议：只对边界分数（0.5-0.7）的样本调用奖励模型。
    """
    
    EVALUATION_PROMPT = """你是一个 Agent 轨迹质量评估专家。

任务描述：{task}

Agent 的完整轨迹：
{trajectory}

请从以下维度评估（1-5分）：
1. **推理质量**：思维过程是否清晰、合理？
2. **工具使用**：是否恰当选择并正确调用了工具？
3. **错误处理**：遇到错误时是否优雅处理？
4. **回答质量**：最终回答是否准确、完整？
5. **效率**：是否避免了不必要的步骤？

输出 JSON 格式：
{{"scores": [score1, score2, score3, score4, score5], "overall": <0-1的浮点数>, "reason": "<简短说明>"}}"""
    
    async def score(self, traj: AgentTrajectory) -> float:
        """用 GPT-4.1 对轨迹打分"""
        from openai import AsyncOpenAI
        client = AsyncOpenAI()
        
        trajectory_text = self._format_trajectory(traj)
        
        response = await client.chat.completions.create(
            model="gpt-4.1-mini",  # 用便宜的模型降低成本
            messages=[{
                "role": "user",
                "content": self.EVALUATION_PROMPT.format(
                    task=traj.messages[1]["content"],
                    trajectory=trajectory_text
                )
            }],
            response_format={"type": "json_object"},
            temperature=0,
        )
        
        result = json.loads(response.choices[0].message.content)
        return result.get("overall", 0.5)
```

---

## 第四层：训练与迭代

### 飞轮的迭代节奏

![典型迭代节奏（生产团队）](../svg/chapter_agentic_rl_09_iteration_timeline.svg)

### 从 SFT 飞轮到 RL 飞轮

数据飞轮通常不是一开始就做完整 RL，而是分阶段演进：

| 阶段 | 主要训练方式 | 适合时机 | 核心收益 |
|------|--------------|---------|---------|
| **冷启动** | 人工/合成轨迹 SFT | 没有线上数据时 | 让模型学会基本工具格式 |
| **成功轨迹飞轮** | 高质量轨迹继续 SFT | 有少量真实用户时 | 提升常见任务稳定性 |
| **偏好飞轮** | DPO/IPO/奖励模型 | 有成功与失败对比时 | 让模型区分好过程和坏过程 |
| **Agentic-RL 飞轮** | GRPO/PPO/环境 rollout | 有可执行环境和奖励时 | 优化长期任务完成率与成本 |

关键判断标准是：当系统已经能稳定生成合法动作，但仍然在长任务、异常恢复和成本控制上不稳定时，就应该从“继续模仿成功样本”转向“在环境中优化轨迹收益”。

Agentic-RL 飞轮关注的不只是“这次是否完成”，而是更完整的收益函数：

```text
总奖励 = 任务完成奖励
       + 工具调用成功奖励
       + 中间检查奖励
       + 错误恢复奖励
       - 重复调用惩罚
       - 高延迟/高成本惩罚
       - 安全违规惩罚
```

这样，模型不会只追求最终答案，而会学习一条更可靠、更便宜、更安全的完成路径。

### 混合训练：新数据 + 旧数据防止灾难性遗忘

```python
class FlyWheelTrainer:
    """管理飞轮的训练循环"""
    
    def __init__(self, base_model, memory_ratio: float = 0.3):
        self.base_model = base_model
        self.memory_ratio = memory_ratio  # 旧数据占比（防止遗忘）
        self.all_historical_data = []     # 所有历史高质量数据
    
    def prepare_training_data(self, 
                               new_high_quality: list[AgentTrajectory],
                               max_samples: int = 20000) -> list[dict]:
        """
        混合新旧数据：
        - 70% 新数据（当前迭代收集的）
        - 30% 旧数据（防止灾难性遗忘）
        """
        self.all_historical_data.extend(new_high_quality)
        
        # 新数据
        new_samples = [self._to_training_format(t) for t in new_high_quality]
        
        # 历史数据（按质量分数加权采样）
        n_historical = int(max_samples * self.memory_ratio)
        historical_sample = self._weighted_sample(
            self.all_historical_data, 
            n=min(n_historical, len(self.all_historical_data)),
            weight_key="auto_quality_score"
        )
        historical_formatted = [self._to_training_format(t) for t in historical_sample]
        
        # 合并 + 打散
        import random
        combined = new_samples + historical_formatted
        random.shuffle(combined)
        
        print(f"训练数据：{len(new_samples)} 新 + {len(historical_formatted)} 旧 = {len(combined)} 总")
        return combined[:max_samples]
    
    def _weighted_sample(self, data, n, weight_key):
        """按质量分数加权采样（质量高的被选中概率更大）"""
        import random
        weights = [getattr(d, weight_key, 0.5) for d in data]
        return random.choices(data, weights=weights, k=min(n, len(data)))
    
    def run_iteration(self, 
                       new_trajectories: list[AgentTrajectory],
                       iteration: int) -> str:
        """运行一次飞轮迭代"""
        print(f"\n{'='*50}")
        print(f"飞轮第 {iteration} 轮迭代")
        print(f"{'='*50}")
        
        # 1. 过滤
        filter = TrajectoryFilter()
        high_quality = filter.filter_batch(new_trajectories, min_score=0.72)
        print(f"✅ 高质量样本：{len(high_quality)} 条")
        
        # 2. 提取失败案例（用于 DPO）
        low_quality = [t for t in new_trajectories 
                       if t.auto_quality_score < 0.4]
        print(f"📊 低质量样本（DPO 负样本）：{len(low_quality)} 条")
        
        # 3. 准备训练数据
        training_data = self.prepare_training_data(high_quality)
        
        # 4. SFT 微调
        output_path = f"./agent-v{iteration}/sft"
        self._run_sft(training_data, output_path)
        
        # 5. DPO（如果有足够的偏好对）
        if len(low_quality) >= 100:
            self._run_dpo(high_quality, low_quality, f"./agent-v{iteration}/dpo")
        
        # 6. 评估新版本
        eval_results = self._evaluate_on_benchmark()
        print(f"📈 评估结果（迭代 {iteration}）：{eval_results}")
        
        return f"./agent-v{iteration}"
    
    def _run_sft(self, data: list[dict], output_path: str):
        """执行 SFT 训练"""
        print(f"开始 SFT 训练（{len(data)} 条样本）...")
        # ... 实际调用 Trainer
    
    def _run_dpo(self, chosen_trajs, rejected_trajs, output_path: str):
        """执行 DPO 训练"""
        print("开始 DPO 对比学习...")
        # ... 实际调用 DPOTrainer
    
    def _evaluate_on_benchmark(self) -> dict:
        """在标准基准上评估"""
        return {
            "tool_accuracy": 0.0,  # 占位，实际运行评估
            "task_completion": 0.0,
            "format_validity": 0.0,
        }
```

---

## 飞轮的三个关键加速因素

### 1. 任务难度课程（Curriculum）

```python
class CurriculumManager:
    """
    从简单任务开始，逐步增加难度。
    简单任务 → 基础工具使用 → 多步推理 → 错误恢复 → 复杂长链
    """
    
    DIFFICULTY_LEVELS = {
        1: {"max_tool_calls": 1, "max_steps": 2},   # 单工具调用
        2: {"max_tool_calls": 3, "max_steps": 5},   # 顺序多工具
        3: {"max_tool_calls": 5, "max_steps": 8},   # 条件分支
        4: {"max_tool_calls": 8, "max_steps": 12},  # 复杂多步推理
        5: {"max_tool_calls": None, "max_steps": None},  # 无限制
    }
    
    def get_difficulty_for_iteration(self, iteration: int) -> int:
        """随迭代逐步提升难度"""
        return min(5, 1 + iteration // 3)  # 每3轮提升一级
    
    def filter_by_difficulty(self, 
                              trajectories: list[AgentTrajectory],
                              level: int) -> list[AgentTrajectory]:
        """只保留目标难度级别的轨迹"""
        config = self.DIFFICULTY_LEVELS[level]
        result = []
        for traj in trajectories:
            if (config["max_tool_calls"] is None or 
                    traj.total_tool_calls <= config["max_tool_calls"]):
                result.append(traj)
        return result
```

### 2. 探索性采样（增加多样性）

```python
class ExploratoryAgentRunner:
    """
    在线上 Agent 中引入探索：
    用更高的 temperature 运行一部分请求，
    发现"不常走的路径"，丰富训练数据多样性。
    """
    
    def __init__(self, base_agent, exploration_rate: float = 0.1):
        self.base_agent = base_agent
        self.exploration_rate = exploration_rate
    
    async def run(self, user_request: str) -> tuple[str, AgentTrajectory]:
        """运行 Agent，部分请求使用探索性配置"""
        import random
        
        is_exploratory = random.random() < self.exploration_rate
        
        if is_exploratory:
            # 高 temperature 激励探索
            config = {"temperature": 0.8, "tag": "exploratory"}
        else:
            # 正常生产配置
            config = {"temperature": 0.2, "tag": "production"}
        
        response, trajectory = await self.base_agent.run(
            user_request, **config
        )
        trajectory.metadata = config
        
        return response, trajectory
```

### 3. 合成数据扩充（覆盖盲区）

```python
class BlindSpotFixer:
    """
    分析现有训练数据的盲区，有针对性地合成补充。
    
    例如：发现模型在"工具调用失败后的处理"上准确率只有 40%，
    就专门合成 500 条这类场景的轨迹。
    """
    
    def identify_weaknesses(self, 
                             eval_results: dict,
                             threshold: float = 0.6) -> list[str]:
        """识别表现低于阈值的技能维度"""
        weak_skills = []
        for skill, score in eval_results.items():
            if score < threshold:
                weak_skills.append(skill)
        return weak_skills
    
    async def synthesize_for_weakness(self, 
                                       skill: str,
                                       n: int = 500) -> list[dict]:
        """针对薄弱技能合成专项训练数据"""
        
        skill_to_scenario = {
            "error_recovery": "工具调用返回错误，Agent 需要正确处理",
            "tool_selection": "多个工具可选，需要正确判断使用哪个",
            "format_validity": "需要输出严格的 JSON 格式工具调用",
            "multi_step_planning": "需要多步工具调用才能完成的复杂任务",
        }
        
        scenario = skill_to_scenario.get(skill, skill)
        
        # 批量合成
        # ... 调用合成管道
        print(f"合成 {n} 条 '{skill}' 场景训练数据...")
        return []  # 实际实现
```

---

## 飞轮的效果：现实案例参考

| 团队 | 方法 | 迭代次数 | 效果 |
|------|------|---------|------|
| **DeepSeek** | GRPO + 自生成数学轨迹 | ~10 轮 | 数学推理从 GPT-4 级别追上 o1 |
| **Reflection-70B** | Self-reflection 自我批评 | ~5 轮 | Llama 70B 超越 GPT-4 (有争议) |
| **STaR / V-STaR** | 用正确推理链自举 | 5 轮 | 数学准确率 +40% |
| **AgentTuning** | 多任务 Agent 轨迹微调 | 1 轮 | 通用 Agent 能力 +30% |

> 📌 核心规律：**数据飞轮前 3 轮改善最快**（模型从"不会用工具"到"会用工具"的跨越），后续每轮收益递减，需要更精细的数据工程。

---

## 实战检查清单

开始构建你的 Agentic 数据飞轮之前，确认以下条件：

```
基础条件：
□ Agent 线上系统已稳定运行（每天 > 100 次调用）
□ 轨迹记录系统已部署（收集 system/user/tool/assistant 完整上下文）
□ 用户反馈渠道已接入（点赞/踩、满意度评分）

数据管道：
□ 脱敏流水线已就绪（GDPR/PIPL 合规）
□ 质量评分函数已实现并在 100 条样本上验证
□ 存储系统能支撑每天增量（建议用对象存储 + Parquet 格式）

训练条件：
□ 有 GPU 资源（至少 A100/H100 × 1，建议 × 4）
□ 训练代码已验证（本地小规模跑通）
□ 评估基准已定义（工具准确率、任务完成率等）

节奏规划：
□ 明确迭代周期（建议 2-4 周一次）
□ 有版本对比的 A/B 测试方案
□ 明确何时"停止迭代"的条件（边际收益低于阈值）
```

---

## 本节小结

Agentic 数据飞轮的本质是：**用 Agent 自身的运行数据来训练更强的 Agent，形成自我强化循环**。

更准确地说，它把三种信号合在一起：

- **成功信号**：哪些轨迹真的完成了任务
- **失败信号**：哪些动作导致了错误、卡住或高成本
- **过程信号**：哪些中间步骤让结果更可靠、更可复现

> 📋 **每一环的关键**
>
> - **收集**：记录完整轨迹，不只是输入输出，还要有工具调用细节
> - **过滤**：双维度（结果 × 过程），质量分 > 0.7 才入训练集
> - **标注**：可验证任务用规则奖励；不可验证用 LLM-as-Judge
> - **训练**：先 SFT，再偏好学习，最后在可执行环境中做 Agentic-RL
> - **部署**：保留 10% 探索流量（发现新场景）

飞轮启动需要初始模型和初始数据，但一旦转起来，**数据质量和模型能力会互相拉动提升**。这也是为什么“先发优势”在 Agent 领域如此重要——早一步启动飞轮，就早一步积累别人赶不上的环境反馈、失败案例和任务轨迹。

> 🔗 **与前序章节的关系**：  
> - **第 3.8 节**：训练数据的质量工程（工程基础）  
> - **第 10.2 节**：SFT + LoRA 训练方法（训练技术）  
> - **第 10.5 节**：PPO/DPO/GRPO 算法（奖励信号）  
> - **第 10.8 节**：Agent 专项数据构建（冷启动数据）  
> - **本节（10.9）**：将 SFT、环境反馈、失败轨迹和 RL 训练串成持续迭代系统

---

## 参考文献

1. Zelikman et al. "STaR: Bootstrapping Reasoning With Reasoning." NeurIPS 2022.
2. Singh et al. "Beyond Human Data: Scaling Self-Training for Problem-Solving with Language Models (V-STaR)." 2024.
3. Zeng et al. "AgentTuning: Enabling Generalized Agent Abilities for LLMs." 2023.
4. Chen et al. "Self-play Fine-tuning Converts Weak Language Models to Strong Language Models (SPIN)." ICML 2024.
5. Guo et al. "DeepSeek-R1: Incentivizing Reasoning Capability in LLMs via Reinforcement Learning." DeepSeek 2025.
6. Mitra et al. "AgentInstruct: Toward Generative Teaching with Agentic Flows." Microsoft Research 2024.

---

*下一节：[10.10 Self-Evolution Agent：从会执行到会自我改进](./10_self_evolution_agent.md)*
