引言：知识蒸馏的范式革新

2026年，大模型知识蒸馏（Knowledge Distillation）正在经历一场深刻的方法论变革。传统的 logits 蒸馏——让小模型模仿大模型的输出概率分布——在简单分类任务上依然有效，但在大模型时代面临根本性挑战：大模型的能力不是体现在单 token 预测上，而是体现在推理链路、工具使用和长程规划上。新一代蒸馏方法不再追求"输出分布对齐"，而是转向"能力迁移"——将大模型的推理模式、工具调用策略和问题分解能力系统性地转移到小模型中。这种转变使得一个 3B 的学生模型在特定任务上达到 70B 教师模型的 85% 以上能力，而推理成本仅为后者的 1/20。## 传统蒸馏方法的局限性### Logits 蒸馏的失效场景pythonclass TraditionalLogitsDistillation: """传统 logits 蒸馏""" def __init__(self, teacher, student, temperature=2.0): self.teacher = teacher self.student = student self.T = temperature def distillation_loss(self, student_logits, teacher_logits, labels): # 软标签损失 soft_loss = F.kl_div( F.log_softmax(student_logits / self.T, dim=-1), F.softmax(teacher_logits / self.T, dim=-1), reduction="batchmean" ) * (self.T ** 2) # 硬标签损失 hard_loss = F.cross_entropy(student_logits, labels) return soft_loss + 0.5 * hard_losstext局限一：Token 级模仿无法迁移推理能力当教师模型通过 Chain-of-Thought（CoT）推理得出答案时，学生模型通过 logits 蒸馏学会的只是"在给定前缀下预测下一个 token"，而非"如何生成推理链"。学生模型可能在训练分布上完美模仿，但在新问题上完全无法推理。局限二：能力不对齐导致蒸馏失效教师模型的某些能力（如多步规划）依赖于大规模参数带来的涌现能力。小模型缺乏这种涌现基础，单纯模仿输出分布无法弥补参数差距。局限三：领域偏移问题通用语料蒸馏出来的小模型在专业领域（医疗、法律、代码）表现不佳，因为教师模型在这些领域的输出分布与通用领域差异巨大。## 新范式一：推理链蒸馏（Chain-of-Thought Distillation）### 核心思想不再蒸馏单 token 的概率分布，而是蒸馏完整的推理过程。教师模型生成包含推理步## 引言：Agent评估的工程困境
2026年，AI Agent 已广泛应用于代码生成、数据分析、客服自动化等场景。但一个尖锐的问题始终困扰着工程团队：**如何评估一个 Agent 系统的质量？**传统的 LLM 评估方法（Benchmark、困惑度）对 Agent 几乎无效。Agent 的行为是多步骤、非确定性的——同一个输入可能产生完全不同的执行路径，最终结果也可能"殊途同归"。更复杂的是，Agent 的失败模式往往是隐蔽的：它可能完成了任务但使用了低效的路径，或者得到了正确答案但基于错误的推理。本文将系统性地构建一个从开发到生产的 Agent 评估体系，涵盖离线 Benchmark、在线监控和持续改进的完整闭环。## Agent 评估与传统 LLM 评估的本质差异### 评估维度对比| 维度 | 传统 LLM 评估 | Agent 评估 ||------|-------------|-----------|| 输入 | 单次提示 | 多轮交互 + 工具调用 || 输出 | 文本生成 | 动作序列 + 最终结果 || 正确性 | 与标准答案对比 | 目标达成 + 过程质量 || 确定性 | 相对确定（温度=0） | 高度非确定 || 评估粒度 | 输出级别 | 步骤级别 + 路径级别 + 结果级别 || 评估成本 | 低（自动对比） | 高（需要模拟环境或人工） || 失败模式 | 输出错误 | 路径低效、工具误用、循环、超时 |### Agent 的六种失败模式pythonfrom enum import Enumclass AgentFailureMode(Enum): """Agent 失败模式分类""" WRONG_ANSWER = "wrong_answer" # 最终答案错误 RIGHT_ANSWER_WRONG_REASONING = "rawr" # 答案对但推理错 EFFICIENT_PATH_VIOLATION = "inefficient" # 正确但路径低效 TOOL_MISUSE = "tool_misuse" # 工具调用错误 INFINITE_LOOP = "loop" # 陷入循环 TIMEOUT = "timeout" # 超时未完成 PARTIAL_COMPLETION = "partial" # 部分完成 SAFETY_VIOLATION = "safety" # 安全违规# 各失败模式的检测难度和影响failure_analysis = { AgentFailureMode.WRONG_ANSWER: { "detect_difficulty": "低", "impact": "高", "auto_detectable": True, }, AgentFailureMode.RIGHT_ANSWER_WRONG_REASONING: { "detect_difficulty": "高", "impact": "高（隐患）", "auto_detectable": False, # 需要 LLM-as-Judge }, AgentFailureMode.EFFICIENT_PATH_VIOLATION: { "detect_difficulty": "中", "impact": "中（成本+延迟）", "auto_detectable": True, }, AgentFailureMode.TOOL_MISUSE: { "detect_difficulty": "中", "impact": "高", "auto_detectable": True, }, AgentFailureMode.INFINITE_LOOP: { "detect_difficulty": "低", "impact": "高（资源浪费）", "auto_detectable": True, },}text## 第一层：离线评估框架### 1.1 任务完成度评估pythonclass AgentEvaluator: """Agent 综合评估器""" def __init__(self, llm_judge, simulation_env): self.judge = llm_judge self.env = simulation_env def evaluate(self, agent, test_cases: List[dict]) -> dict: """评估 Agent 在测试集上的表现""" results = [] for case in test_cases: result = self._evaluate_single(agent, case) results.append(result) return self._aggregate(results) def _evaluate_single(self, agent, case: dict) -> dict: """评估单个测试用例""" trace = agent.execute( task=case["task"], env=self.env.clone(), # 隔离环境 max_steps=case.get("max_steps", 20), ) return { "case_id": case["id"], "task": case["task"], "trace": trace, "metrics": { # 1. 结果正确性 "task_success": self._check_success( trace, case["expected"] ), # 2. 步骤效率 "step_efficiency": self._step_efficiency( trace, case.get("optimal_steps", 5) ), # 3. 工具使用准确率 "tool_accuracy": self._tool_accuracy(trace), # 4. 路径质量 "path_quality": self._path_quality( trace, case.get("reference_path") ), # 5. 推理质量（LLM-as-Judge） "reasoning_quality": self._reasoning_quality( trace, case["task"] ), # 6. 安全性 "safety_score": self._safety_check(trace), # 7. 成本 "token_usage": trace.total_tokens, "api_calls": trace.total_api_calls, # 8. 延迟 "latency_ms": trace.total_latency_ms, } }text### 1.2 LLM-as-Judge 评估pythonclass LLMAsJudge: """LLM 评审员""" def __init__(self, judge_model, rubric: str): self.judge = judge_model self.rubric = rubric # 评分标准 def evaluate_reasoning(self, trace, task: str) -> dict: """评估推理质量""" prompt = f""" 你是一个 AI Agent 评估专家。请评估以下 Agent 执行轨迹。 任务：{task} Agent 执行轨迹： {self._format_trace(trace)} 评分标准（1-5分）： {self.rubric} 请从以下维度评估： 1. 理解准确性：Agent 是否正确理解了任务？ 2. 规划合理性：Agent 的步骤规划是否合理？ 3. 工具选择：Agent 是否选择了正确的工具？ 4. 错误处理：Agent 遇到错误时的恢复能力如何？ 5. 推理深度：Agent 的推理是否足够深入？ 输出 JSON 格式： {{ "understanding": {{"score": 1-5, "reason": "..."}}, "planning": {{"score": 1-5, "reason": "..."}}, "tool_selection": {{"score": 1-5, "reason": "..."}}, "error_handling": {{"score": 1-5, "reason": "..."}}, "reasoning_depth": {{"score": 1-5, "reason": "..."}}, "overall": {{"score": 1-5, "summary": "..."}} }} """ response = self.judge.generate(prompt, temperature=0.0) return self._parse_judgment(response) def evaluate_correctness(self, agent_answer, expected_answer, task: str) -> dict: """评估答案正确性（支持开放式答案）""" prompt = f""" 任务：{task} 标准答案：{expected_answer} Agent 答案：{agent_answer} 请判断 Agent 答案是否正确。对于开放式任务，判断是否 包含标准答案的关键信息点。 输出 JSON： {{ "is_correct": true/false, "confidence": 0.0-1.0, "missing_points": ["未覆盖的关键点"], "extra_points": ["多余的错误信息"], "reasoning": "判断理由" }} """ response = self.judge.generate(prompt, temperature=0.0) return self._parse_judgment(response)text### 1.3 AgentBench：多维度 Benchmark 套件pythonclass AgentBenchmark: """Agent 多维度评估套件""" BENCHMARKS = { # 代码生成与执行 "swe_bench": { "description": "真实 GitHub Issue 修复", "metrics": ["resolved", "partial", "failed"], "evaluation": "test_based", "difficulty": "hard", }, "human_eval": { "description": "函数级代码生成", "metrics": ["pass@1", "pass@10"], "evaluation": "test_based", "difficulty": "medium", }, # 工具使用 "tool_bench": { "description": "多工具编排任务", "metrics": ["success_rate", "tool_accuracy", "avg_steps", "recovery_rate"], "evaluation": "simulation", "difficulty": "medium", }, # 推理 "gaia_benchmark": { "description": "通用 AI 助手能力", "metrics": ["accuracy", "efficiency"], "evaluation": "answer_based", "difficulty": "hard", }, # 多轮对话 "mt_bench_agentic": { "description": "多轮 Agent 对话", "metrics": ["task_completion", "conversation_quality"], "evaluation": "llm_judge", "difficulty": "medium", }, # 安全性 "safety_bench": { "description": "Agent 安全行为", "metrics": ["safety_violation_rate", "refusal_accuracy"], "evaluation": "rule_based", "difficulty": "easy", }, } def run_full_evaluation(self, agent) -> dict: """运行完整评估套件""" results = {} for name, config in self.BENCHMARKS.items(): print(f"运行 {name}...") test_data = self._load_benchmark(name) if config["evaluation"] == "test_based": results[name] = self._eval_test_based(agent, test_data) elif config["evaluation"] == "simulation": results[name] = self._eval_simulation(agent, test_data) elif config["evaluation"] == "llm_judge": results[name] = self._eval_llm_judge(agent, test_data) elif config["evaluation"] == "rule_based": results[name] = self._eval_rule_based(agent, test_data) return resultstext## 第二层：在线监控体系### 2.1 实时质量监控pythonclass AgentMonitor: """Agent 生产环境监控""" def __init__(self, config): self.metrics = MetricsCollector() self.alerting = AlertManager(config.alerts) self.llm_judge = LLMJudge(config.judge_model) def monitor_execution(self, trace: AgentTrace): """监控单次 Agent 执行""" # 1. 实时指标采集 self._collect_realtime_metrics(trace) # 2. 异常检测 anomalies = self._detect_anomalies(trace) if anomalies: self.alerting.alert(anomalies) # 3. 异步质量评估（不阻塞用户） asyncio.create_task( self._async_quality_check(trace) ) def _collect_realtime_metrics(self, trace: AgentTrace): """采集实时指标""" metrics = { # 执行指标 "agent.steps.count": len(trace.steps), "agent.duration_ms": trace.duration_ms, "agent.tokens.total": trace.total_tokens, "agent.api.calls": trace.api_call_count, # 工具指标 "agent.tool.calls": trace.tool_call_count, "agent.tool.errors": trace.tool_error_count, "agent.tool.types": [s.tool_name for s in trace.steps if s.type == "tool_call"], # 质量指标 "agent.task.success": trace.success, "agent.task.partial": trace.partial_success, "agent.task.timeout": trace.timed_out, "agent.task.loop_detected": trace.loop_detected, } for key, value in metrics.items(): self.metrics.gauge(key, value) def _detect_anomalies(self, trace: AgentTrace) -> List[dict]: """异常检测""" anomalies = [] # 1. 步骤数异常（过多可能陷入循环） if len(trace.steps) > 15: anomalies.append({ "type": "excessive_steps", "severity": "warning", "detail": f"Agent 执行了 {len(trace.steps)} 步", }) # 2. 工具调用失败率 if trace.tool_call_count > 0: error_rate = trace.tool_error_count / trace.tool_call_count if error_rate > 0.3: anomalies.append({ "type": "high_tool_error_rate", "severity": "critical", "detail": f"工具错误率 {error_rate:.0%}", }) # 3. Token 消耗异常 if trace.total_tokens > 50000: anomalies.append({ "type": "high_token_usage", "severity": "warning", "detail": f"消耗 {trace.total_tokens} tokens", }) # 4. 循环检测 if self._detect_loop(trace): anomalies.append({ "type": "loop_detected", "severity": "critical", "detail": "检测到 Agent 陷入循环", }) # 5. 延迟异常 if trace.duration_ms > 60000: anomalies.append({ "type": "high_latency", "severity": "warning", "detail": f"执行耗时 {trace.duration_ms/1000:.1f}s", }) return anomalies def _detect_loop(self, trace: AgentTrace) -> bool: """循环检测算法""" actions = [s.action_signature for s in trace.steps] # 滑动窗口检测重复模式 for window_size in [2, 3, 4]: for i in range(len(actions) - window_size * 2): pattern = actions[i:i+window_size] next_pattern = actions[i+window_size:i+window_size*2] if pattern == next_pattern: return True return Falsetext### 2.2 用户反馈采集pythonclass FeedbackCollector: """用户反馈采集器""" def __init__(self): self.feedback_store = FeedbackStore() async def collect_implicit_feedback(self, trace: AgentTrace): """采集隐式反馈""" feedback = { "trace_id": trace.id, "implicit_signals": {}, } # 1. 用户是否采纳了 Agent 的建议 if trace.result_type == "suggestion": adopted = await self._check_adoption(trace) feedback["implicit_signals"]["adopted"] = adopted # 2. 用户是否重新提问（暗示不满意） if trace.result_type == "answer": reasked = await self._check_reask(trace.session_id) feedback["implicit_signals"]["reasked"] = reasked # 3. 用户是否手动修改了 Agent 的输出 if trace.result_type == "code": modified = await self._check_modification(trace) feedback["implicit_signals"]["modified"] = modified # 4. 会话是否在 Agent 回答后很快结束 if trace.result_type == "answer": quick_exit = await self._check_quick_exit(trace) feedback["implicit_signals"]["quick_exit"] = quick_exit self.feedback_store.add(feedback) async def collect_explicit_feedback(self, trace_id: str): """采集显式反馈""" # 在 Agent 回答后展示反馈入口 # "这个回答有帮助吗？" 👍 👎 # "有什么可以改进的？" [文本输入] passtext## 第三层：持续改进闭环### 3.1 基于评估的自动优化pythonclass ContinuousImprovement: """持续改进系统""" def __init__(self, agent, evaluator, monitor): self.agent = agent self.evaluator = evaluator self.monitor = monitor async def improvement_loop(self): """持续改进循环""" while True: # 1. 收集近期生产数据 traces = await self.monitor.get_recent_traces( days=7, sample_size=500 ) # 2. 识别失败模式 failure_patterns = self._analyze_failures(traces) # 3. 生成改进策略 improvements = self._generate_improvements( failure_patterns ) # 4. 离线评估改进效果 for improvement in improvements: score_before = self.evaluator.evaluate( self.agent, self.test_set ) # 应用改进 improved_agent = self._apply_improvement( self.agent, improvement ) score_after = self.evaluator.evaluate( improved_agent, self.test_set ) # 5. A/B 测试 if score_after > score_before: await self._start_ab_test( self.agent, improved_agent, traffic_split=0.1 ) await asyncio.sleep(86400) # 每天运行一次text## 评估指标体系总览textAgent 评估指标体系├── 结果指标（What）│ ├── 任务完成率│ ├── 答案正确率│ ├── 部分完成率│ └── 安全合规率├── 过程指标（How）│ ├── 步骤效率（实际/最优步数比）│ ├── 工具选择准确率│ ├── 错误恢复率│ ├── 循环检测率│ └── 推理质量评分├── 效率指标（Cost）│ ├── Token 消耗│ ├── API 调用次数│ ├── 端到端延迟│ └── 每任务成本└── 用户体验指标（Experience） ├── 采纳率 ├── 重问率 ├── 修改率 └── 满意度评分text## 结语Agent 评估是一个系统工程，不是单一的 Benchmark 分数可以概括的。离线评估保证基线质量，在线监控捕捉真实分布的异常，持续改进闭环让系统越用越好。对于工程团队，建议从"任务完成率 + 步骤效率 + 工具准确率"三个核心指标起步，逐步引入 LLM-as-Judge 和用户反馈。最重要的是：评估体系本身也需要持续迭代——随着 Agent 能力的提升，评估标准和难度也必须同步升级。骤的答案，学生模型学习生成类似的推理链。pythonclass CoTDistillation: """推理链蒸馏""" def __init__(self, teacher, student, llm_judge): self.teacher = teacher self.student = student self.judge = llm_judge # 用于评估推理质量 def generate_training_data(self, questions: List[str]): """使用教师模型生成高质量推理链""" training_data = [] for question in questions: # 1. 教师生成多个推理链（多样性采样） teacher_cots = [] for _ in range(4): # 生成4个候选 cot = self.teacher.generate( f"问题：{question}\n" f"请逐步推理并给出答案。\n" f"格式：\n推理：...\n答案：...", temperature=0.7 ) teacher_cots.append(cot) # 2. 评估并选择最佳推理链 best_cot = self._select_best(question, teacher_cots) # 3. 生成负样本（错误推理链，用于对比学习） wrong_cots = self._generate_wrong_cots(question, best_cot) training_data.append({ "question": question, "best_cot": best_cot, "wrong_cots": wrong_cots, }) return training_data def _select_best(self, question, candidates): """选择最佳推理链""" scores = [] for cot in candidates: # 使用 LLM 评估推理质量 score = self.judge.evaluate( f"评估以下推理链的质量（1-10分）：\n" f"问题：{question}\n推理链：{cot}\n" f"评估维度：正确性、清晰度、完整性" ) scores.append(score) return candidates[scores.index(max(scores))] def train_step(self, batch): """训练步骤""" total_loss = 0 for item in batch: question = item["question"] best_cot = item["best_cot"] wrong_cots = item["wrong_cots"] # 1. 正样本：学生生成最佳推理链 student_output = self.student.generate( f"问题：{question}\n推理：", return_logits=True ) positive_loss = self._sequence_loss( student_output, best_cot ) # 2. 负样本：降低错误推理链概率 for wrong_cot in wrong_cots: negative_loss = -self._sequence_loss( student_output, wrong_cot ) * 0.1 # 负样本权重较低 positive_loss += negative_loss total_loss += positive_loss return total_loss / len(batch)text### 推理链蒸馏效果| 方法 | GSM8K | MATH | HumanEval | BBH ||------|-------|------|-----------|-----|| 学生模型（7B，基线） | 42.3% | 18.5% | 45.0% | 55.2% || + Logits 蒸馏 | 45.1% | 19.8% | 47.2% | 56.8% || + CoT 蒸馏 | 58.7% | 28.3% | 52.1% | 64.5% || + CoT + 负样本对比 | 62.1% | 31.2% | 55.8% | 68.3% || 教师模型（70B） | 82.1% | 45.6% | 68.5% | 78.9% |关键发现：CoT 蒸馏在推理任务上的提升远超 logits 蒸馏，但与教师模型仍有差距。负样本对比学习进一步缩小了差距。## 新范式二：工具使用能力蒸馏### 问题背景大模型的工具调用能力（Function Calling）是 Agent 系统的基础。但这种能力难以通过传统蒸馏迁移，因为它涉及多轮交互和状态管理。pythonclass ToolUseDistillation: """工具使用能力蒸馏""" def __init__(self, teacher, student, tools): self.teacher = teacher self.student = student self.tools = tools # 可用工具列表 def generate_tool_use_traces(self, tasks: List[str]): """生成教师模型的工具使用轨迹""" traces = [] for task in tasks: # 教师模型执行任务，记录完整轨迹 trace = self._execute_with_trace(self.teacher, task) traces.append(trace) return traces def _execute_with_trace(self, model, task: str) -> dict: """执行任务并记录详细轨迹""" trace = { "task": task, "steps": [], "final_answer": None, } messages = [{"role": "user", "content": task}] for step in range(10): # 最多10步 # 模型决策 response = model.generate( messages=messages, tools=self.tools, temperature=0.0 # 贪心解码 ) if response.tool_call: # 执行工具 tool_result = self._execute_tool( response.tool_call ) trace["steps"].append({ "thought": response.thought, "tool_name": response.tool_call.name, "tool_args": response.tool_call.arguments, "tool_result": tool_result, "is_correct": self._verify_step( task, response.tool_call, tool_result ) }) messages.extend([ {"role": "assistant", "content": response.thought, "tool_call": response.tool_call}, {"role": "tool", "content": tool_result} ]) else: trace["final_answer"] = response.content break return trace def train_student(self, traces: List[dict]): """训练学生模型的工具使用能力""" for trace in traces: # 1. 思维过程蒸馏 for step in trace["steps"]: if not step["is_correct"]: continue # 跳过错误步骤 # 学生学习：给定当前状态，生成正确的工具调用 context = self._build_context(trace, step) target = { "thought": step["thought"], "tool_name": step["tool_name"], "tool_args": step["tool_args"], } student_output = self.student.generate(context) loss = self._tool_call_loss(student_output, target) loss.backward() # 2. 错误纠正学习 for step in trace["steps"]: if step["is_correct"]: continue # 学生学习：识别并纠正错误工具调用 correction = self.teacher.generate( f"以下工具调用是错误的：\n" f"调用：{step['tool_name']}({step['tool_args']})\n" f"结果：{step['tool_result']}\n" f"请分析错误原因并给出正确调用。" ) # 学生学习纠正模式 self._train_correction(student, step, correction)text### 工具使用蒸馏效果| 能力 | 学生（7B基线） | +工具蒸馏 | 教师（70B） ||------|--------------|---------|-----------|| 正确工具选择 | 65% | 89% | 95% || 参数填充准确率 | 58% | 84% | 93% || 多步工具编排 | 32% | 71% | 88% || 错误恢复 | 15% | 52% | 78% |## 新范式三：任务分解能力蒸馏### 思路分解蒸馏（Plan Distillation）大模型的另一个核心能力是将复杂问题分解为可执行的子任务。这种"规划能力"可以通过蒸馏迁移。pythonclass PlanDistillation: """任务规划能力蒸馏""" def __init__(self, teacher, student): self.teacher = teacher self.student = student def generate_planning_data(self, complex_tasks: List[str]): """生成规划训练数据""" data = [] for task in complex_tasks: # 1. 教师生成分解计划 teacher_plan = self.teacher.generate(f""" 将以下复杂任务分解为5-10个子任务： 任务：{task} 输出格式： 1. [子任务描述] → [预期输出] → [依赖的前置子任务] 2. ... """) # 2. 验证计划的可行性 plan_steps = self._parse_plan(teacher_plan) is_valid = self._validate_plan(task, plan_steps) if is_valid: # 3. 生成多种分解方式（增强泛化） alternative_plans = [] for _ in range(3): alt_plan = self.teacher.generate( f"用不同的方式分解任务：{task}\n" f"已有方案：{teacher_plan}\n" f"请给出另一种分解方式。" ) alternative_plans.append(alt_plan) data.append({ "task": task, "plans": [teacher_plan] + alternative_plans, "best_plan_idx": 0, }) return data def train(self, training_data): """训练学生的规划能力""" for item in training_data: task = item["task"] plans = item["plans"] best_idx = item["best_plan_idx"] # 1. 计划生成训练 student_plan = self.student.generate( f"分解任务：{task}" ) # 与最佳计划对齐 loss = self._plan_similarity_loss( student_plan, plans[best_idx] ) # 2. 计划评估训练（学习区分好坏计划） for i, plan in enumerate(plans): label = 1.0 if i == best_idx else 0.0 score = self.student.score( f"任务：{task}\n计划：{plan}\n评分(0-1)：" ) loss += F.mse_loss(score, torch.tensor(label)) loss.backward()text## 新范式四：渐进式蒸馏（Progressive Distillation）### 分阶段蒸馏策略一次性从 70B 蒸馏到 7B 跨度过大，信息损失严重。渐进式蒸馏通过中间模型逐步过渡。pythonclass ProgressiveDistillation: """渐进式蒸馏：70B → 30B → 13B → 7B""" def __init__(self, model_chain: List[str]): """ model_chain: ["70b", "30b", "13b", "7b"] """ self.chain = model_chain def run(self, training_data): """逐级蒸馏""" for i in range(len(self.chain) - 1): teacher_size = self.chain[i] student_size = self.chain[i + 1] print(f"蒸馏阶段 {i+1}: {teacher_size} → {student_size}") # 1. 加载教师和学生 teacher = self._load_model(teacher_size) student = self._load_model(student_size) # 2. 阶段特定蒸馏策略 if i == 0: # 第一阶段：能力蒸馏（推理+工具+规划） self._distill_capabilities(teacher, student, training_data) elif i == len(self.chain) - 2: # 最后阶段：精调（logits + CoT） self._distill_logits_and_cot(teacher, student, training_data) else: # 中间阶段：混合蒸馏 self._distill_mixed(teacher, student, training_data) # 3. 阶段评估 metrics = self._evaluate(student) print(f"阶段 {i+1} 结果: {metrics}") # 4. 如果效果下降太多，回退 if metrics["avg_score"] < self._threshold(i): print(f"阶段 {i+1} 未达标，调整策略") self._adjust_and_retry(teacher, student, training_data, i)text### 渐进式 vs 直接蒸馏对比| 方法 | 最终模型 | GSM8K | MATH | 工具调用 | 训练成本 ||------|---------|-------|------|---------|---------|| 直接蒸馏 | 7B | 52.3% | 24.1% | 71% | 1x || 渐进式(3阶段) | 7B | 61.5% | 29.8% | 84% | 2.8x || 渐进式(4阶段) | 7B | 63.2% | 31.0% | 86% | 3.5x || 教师模型 | 70B | 82.1% | 45.6% | 95% | - |## 生产级蒸馏 Pipelinepythonclass ProductionDistillationPipeline: """生产级蒸馏流水线""" def __init__(self, config): self.config = config self.stages = [ ("data_preparation", self._prepare_data), ("capability_distillation", self._distill_capabilities), ("domain_adaptation", self._domain_adapt), ("alignment_finetuning", self._align), ("evaluation", self._evaluate), ] def _prepare_data(self): """准备多源训练数据""" data = { "reasoning": self._collect_reasoning_data(50000), "tool_use": self._collect_tool_use_data(20000), "planning": self._collect_planning_data(10000), "general": self._collect_general_data(100000), } # 质量过滤 for category, items in data.items(): data[category] = [ item for item in items if self._quality_filter(item) ] return data def _domain_adapt(self, model, data): """领域适应微调""" domains = ["code", "math", "science", "medical"] for domain in domains: domain_data = data.get(domain, []) if not domain_data: continue # LoRA 微调（避免全量更新导致遗忘） lora_config = { "r": 16, "alpha": 32, "target_modules": ["q_proj", "v_proj", "k_proj"], } model = self._lora_finetune(model, domain_data, lora_config) return model def _align(self, model, data): """对齐微调（DPO/RLHF）""" # 使用 DPO 进行偏好对齐 preference_data = self._generate_preferences(model, data) model = self._dpo_train(model, preference_data) return modeltext## 蒸馏的伦理与合规### 蒸馏的边界模型蒸馏涉及知识产权和合规问题：| 蒸馏来源 | 合规风险 | 建议 ||---------|---------|------|| 开源模型（Llama, Qwen） | 低 | 遵循模型许可证 || API 蒸馏（GPT, Claude） | 高 | 违反 ToS，禁止 || 自有模型 | 无 | 完全合规 || 多模型混合蒸馏 | 中 | 需逐一检查许可证 |## 结语知识蒸馏正从"输出模仿"走向"能力迁移"。CoT 蒸馏迁移推理能力，工具蒸馏迁移 Agent 能力，规划蒸馏迁移分解能力，渐进式蒸馏减小信息损失。对于工程团队，建议采用"渐进式 + 多能力蒸馏"的组合策略，在成本和质量之间找到最优平衡。同时必须注意蒸馏的合规边界——API 蒸馏虽然技术可行，但存在严重的法律和道德风险。