agent replay testing

Agent 回放测试:确定性验证与回归测试

为什么 Agent 需要回放测试 Agent 系统的测试比传统软件复杂得多。传统软件的函数调用是确定性的——同样的输入产生同样的输出。但 Agent 涉及 LLM 推理、工具调用、环境交互,每一步都可能引入非确定性。当你修改了 Prompt、升级了模型、或调整了工具参数,如何确保 Agent 的行为没有退化? 回放测试(Replay Testing)是解决这个问题的核心方法:录制 Agent 的真实执行轨迹,在变更后回放这些轨迹,比较行为差异。 回放测试的核心价值: 回归保护:确保修改不会破坏已有的正确行为 行为可追溯:每次变更后的行为差异可量化、可审查 非确定性管理:在不确定的系统中建立确定性的验证基线 成本控制:无需重新执行真实环境操作,降低测试成本 回放测试架构 ┌──────────────────────────────────────────────────────┐ │ 回放测试系统 │ │ │ │ ┌──────────┐ 录制 ┌──────────┐ 存储 ┌────────┐│ │ │ Agent │ ──────> │ Recorder │ ─────> │ Trace ││ │ │ 执行环境 │ │ 记录器 │ │ Store ││ │ └──────────┘ └──────────┘ └───┬────┘│ │ │ │ │ ┌──────────┐ 回放 ┌──────────┐ 比对 ┌───v────┐│ │ │ Agent │ <────── │ Replayer │ <───── │ Trace ││ │ │ 测试环境 │ │ 回放器 │ │ Store ││ │ └──────────┘ └──────────┘ └────────┘│ │ │ │ │ v │ │ ┌──────────┐ │ │ │ Diff │ │ │ │ 差异分析 │ │ │ └──────────┘ │ └──────────────────────────────────────────────────────┘ 一、轨迹录制 轨迹数据结构 from dataclasses import dataclass, field from datetime import datetime from typing import Any, Optional import json @dataclass class AgentStep: """Agent 执行的单步操作""" step_id: str step_type: str # "reasoning" | "tool_call" | "observation" | "final_answer" timestamp: str # 推理内容 reasoning: Optional[str] = None # 工具调用 tool_name: Optional[str] = None tool_input: Optional[dict] = None tool_output: Optional[Any] = None tool_duration_ms: Optional[int] = None # 环境状态快照 env_state_before: Optional[dict] = None env_state_after: Optional[dict] = None # LLM 调用详情 model_name: Optional[str] = None prompt_tokens: Optional[int] = None completion_tokens: Optional[int] = None temperature: Optional[float] = None @dataclass class AgentTrace: """完整的 Agent 执行轨迹""" trace_id: str task: str # 用户任务描述 steps: list[AgentStep] = field(default_factory=list) metadata: dict = field(default_factory=dict) # 环境信息 agent_version: str = "" model_version: str = "" tool_versions: dict = field(default_factory=dict) environment: str = "" # "production" | "staging" | "test" # 结果 success: bool = False final_answer: str = "" total_duration_ms: int = 0 total_tokens: int = 0 def to_dict(self) -> dict: return { "trace_id": self.trace_id, "task": self.task, "steps": [ {k: v for k, v in step.__dict__.items() if v is not None} for step in self.steps ], "metadata": self.metadata, "agent_version": self.agent_version, "model_version": self.model_version, "tool_versions": self.tool_versions, "success": self.success, "final_answer": self.final_answer, "total_duration_ms": self.total_duration_ms, "total_tokens": self.total_tokens, } def save(self, path: str): with open(path, "w", encoding="utf-8") as f: json.dump(self.to_dict(), f, ensure_ascii=False, indent=2) @classmethod def load(cls, path: str) -> "AgentTrace": with open(path, "r", encoding="utf-8") as f: data = json.load(f) steps = [ AgentStep( step_id=s["step_id"], step_type=s["step_type"], timestamp=s["timestamp"], reasoning=s.get("reasoning"), tool_name=s.get("tool_name"), tool_input=s.get("tool_input"), tool_output=s.get("tool_output"), tool_duration_ms=s.get("tool_duration_ms"), env_state_before=s.get("env_state_before"), env_state_after=s.get("env_state_after"), model_name=s.get("model_name"), prompt_tokens=s.get("prompt_tokens"), completion_tokens=s.get("completion_tokens"), temperature=s.get("temperature"), ) for s in data["steps"] ] return cls( trace_id=data["trace_id"], task=data["task"], steps=steps, metadata=data.get("metadata", {}), agent_version=data.get("agent_version", ""), model_version=data.get("model_version", ""), tool_versions=data.get("tool_versions", {}), environment=data.get("environment", ""), success=data.get("success", False), final_answer=data.get("final_answer", ""), total_duration_ms=data.get("total_duration_ms", 0), total_tokens=data.get("total_tokens", 0), ) 录制器实现 class TraceRecorder: """Agent 执行轨迹录制器""" def __init__(self, agent_version: str, model_version: str): self.agent_version = agent_version self.model_version = model_version self.traces: list[AgentTrace] = [] def record_execution(self, agent, task: str, **kwargs) -> AgentTrace: """录制 Agent 的一次完整执行""" import uuid trace = AgentTrace( trace_id=str(uuid.uuid4()), task=task, agent_version=self.agent_version, model_version=self.model_version, environment=kwargs.get("environment", "test"), ) # 包装 agent 的方法以录制每一步 original_methods = self._wrap_agent_methods(agent, trace) try: # 执行 Agent result = agent.run(task) trace.success = result.get("success", False) trace.final_answer = result.get("answer", "") trace.total_duration_ms = sum( s.tool_duration_ms or 0 for s in trace.steps ) trace.total_tokens = sum( (s.prompt_tokens or 0) + (s.completion_tokens or 0) for s in trace.steps ) finally: # 恢复原始方法 self._unwrap_agent_methods(agent, original_methods) self.traces.append(trace) return trace def _wrap_agent_methods(self, agent, trace: AgentTrace) -> dict: """包装 Agent 的关键方法以实现录制""" original = {} # 包装 LLM 调用 if hasattr(agent, "llm_call"): original["llm_call"] = agent.llm_call def wrapped_llm_call(prompt, *args, **kwargs): import time step = AgentStep( step_id=f"step_{len(trace.steps)}", step_type="reasoning", timestamp=datetime.now().isoformat(), model_name=trace.model_version, temperature=kwargs.get("temperature", 0), ) start = time.perf_counter() result = original["llm_call"](prompt, *args, **kwargs) step.tool_duration_ms = int((time.perf_counter() - start) * 1000) step.reasoning = result trace.steps.append(step) return result agent.llm_call = wrapped_llm_call # 包装工具调用 if hasattr(agent, "call_tool"): original["call_tool"] = agent.call_tool def wrapped_call_tool(tool_name, tool_input, *args, **kwargs): import time step = AgentStep( step_id=f"step_{len(trace.steps)}", step_type="tool_call", timestamp=datetime.now().isoformat(), tool_name=tool_name, tool_input=tool_input, ) # 记录调用前的环境状态 if hasattr(agent, "get_env_state"): step.env_state_before = agent.get_env_state() start = time.perf_counter() result = original["call_tool"](tool_name, tool_input, *args, **kwargs) step.tool_duration_ms = int((time.perf_counter() - start) * 1000) step.tool_output = result # 记录调用后的环境状态 if hasattr(agent, "get_env_state"): step.env_state_after = agent.get_env_state() trace.steps.append(step) return result agent.call_tool = wrapped_call_tool return original def _unwrap_agent_methods(self, agent, original: dict): """恢复原始方法""" for name, method in original.items(): setattr(agent, name, method) def save_all(self, directory: str): """保存所有轨迹""" from pathlib import Path dir_path = Path(directory) dir_path.mkdir(parents=True, exist_ok=True) for trace in self.traces: trace.save(dir_path / f"{trace.trace_id}.json") 二、回放器与确定性验证 回放器设计 class TraceReplayer: """轨迹回放器:重新执行录制的 Agent 轨迹""" def __init__(self, agent, config: dict = None): self.agent = agent self.config = config or {} # 是否模拟工具输出(不实际调用工具) self.mock_tools = self.config.get("mock_tools", False) # 是否比较推理步骤 self.compare_reasoning = self.config.get("compare_reasoning", True) # 推理相似度阈值 self.reasoning_threshold = self.config.get("reasoning_threshold", 0.85) def replay(self, trace: AgentTrace) -> dict: """回放一条轨迹并比较差异""" results = { "trace_id": trace.trace_id, "task": trace.task, "original_success": trace.success, "replay_success": None, "step_results": [], "overall_diff": {}, } # 如果 mock_tools,将工具输出预加载 mocked_outputs = {} if self.mock_tools: for step in trace.steps: if step.step_type == "tool_call": mocked_outputs[step.step_id] = step.tool_output # 逐步回放 for i, original_step in enumerate(trace.steps): replay_result = self._replay_step(original_step, i, mocked_outputs) results["step_results"].append(replay_result) # 比较最终结果 results["overall_diff"] = self._compute_overall_diff( trace, results["step_results"] ) return results def _replay_step(self, original: AgentStep, index: int, mocked_outputs: dict) -> dict: """回放单个步骤""" result = { "step_id": original.step_id, "step_type": original.step_type, "match": True, "diff": {}, } if original.step_type == "tool_call": if self.mock_tools and original.step_id in mocked_outputs: # 使用模拟输出 actual_output = mocked_outputs[original.step_id] else: # 实际调用工具 actual_output = self.agent.call_tool( original.tool_name, original.tool_input ) # 比较工具输出 output_match = self._compare_outputs( original.tool_output, actual_output ) result["match"] = output_match["exact_match"] result["diff"]["output"] = output_match # 比较环境状态变化 if original.env_state_after: current_state = self.agent.get_env_state() if hasattr(self.agent, "get_env_state") else None if current_state: state_match = self._compare_states( original.env_state_after, current_state ) result["diff"]["env_state"] = state_match if not state_match["match"]: result["match"] = False elif original.step_type == "reasoning" and self.compare_reasoning: # 比较推理输出(使用语义相似度) actual_reasoning = self.agent.llm_call( self._reconstruct_prompt(original, index) ) similarity = self._semantic_similarity( original.reasoning, actual_reasoning ) result["match"] = similarity >= self.reasoning_threshold result["diff"]["reasoning"] = { "similarity": similarity, "original_length": len(original.reasoning or ""), "actual_length": len(actual_reasoning or ""), } return result def _compare_outputs(self, expected: Any, actual: Any) -> dict: """比较工具输出""" if expected == actual: return {"exact_match": True, "semantic_match": True} # 对于字符串,尝试语义比较 if isinstance(expected, str) and isinstance(actual, str): sim = self._semantic_similarity(expected, actual) return { "exact_match": False, "semantic_match": sim > 0.9, "similarity": sim, } # 对于字典,逐键比较 if isinstance(expected, dict) and isinstance(actual, dict): diffs = {} all_keys = set(expected.keys()) | set(actual.keys()) for key in all_keys: if key not in expected: diffs[key] = {"status": "added", "value": actual[key]} elif key not in actual: diffs[key] = {"status": "removed", "value": expected[key]} elif expected[key] != actual[key]: diffs[key] = { "status": "changed", "expected": expected[key], "actual": actual[key], } return { "exact_match": False, "semantic_match": len(diffs) == 0, "field_diffs": diffs, } return {"exact_match": False, "semantic_match": False} def _compare_states(self, expected: dict, actual: dict) -> dict: """比较环境状态""" diffs = {} for key in set(expected.keys()) | set(actual.keys()): if expected.get(key) != actual.get(key): diffs[key] = { "expected": expected.get(key), "actual": actual.get(key), } return {"match": len(diffs) == 0, "diffs": diffs} def _semantic_similarity(self, text1: str, text2: str) -> float: """计算语义相似度""" from sentence_transformers import SentenceTransformer import numpy as np model = SentenceTransformer("all-MiniLM-L6-v2") emb1 = model.encode([text1]) emb2 = model.encode([text2]) return float(np.dot(emb1[0], emb2[0]) / ( np.linalg.norm(emb1[0]) * np.linalg.norm(emb2[0]) )) def _reconstruct_prompt(self, step: AgentStep, index: int) -> str: """从轨迹步骤重建 LLM prompt""" # 简化实现:实际需要根据 Agent 架构重建完整上下文 return f"Step {index}: {step.reasoning or ''}" def _compute_overall_diff(self, trace: AgentTrace, step_results: list[dict]) -> dict: """计算整体差异""" total_steps = len(step_results) matched_steps = sum(1 for r in step_results if r["match"]) return { "total_steps": total_steps, "matched_steps": matched_steps, "mismatched_steps": total_steps - matched_steps, "match_rate": matched_steps / total_steps if total_steps > 0 else 0, "success_preserved": trace.success, # 简化 } 三、回归测试框架 测试套件管理 class RegressionTestSuite: """Agent 回归测试套件""" def __init__(self, suite_name: str): self.suite_name = suite_name self.test_cases: list[dict] = [] self.baselines: dict = {} # trace_id -> baseline result def add_trace_as_baseline(self, trace: AgentTrace, expected_success: bool = True, category: str = "general"): """将一条轨迹添加为回归测试基线""" self.test_cases.append({ "trace_id": trace.trace_id, "task": trace.task, "category": category, "expected_success": expected_success, "baseline_trace": trace, }) def load_from_directory(self, dir_path: str): """从目录加载所有轨迹作为基线""" from pathlib import Path for trace_file in Path(dir_path).glob("*.json"): trace = AgentTrace.load(str(trace_file)) self.add_trace_as_baseline(trace, expected_success=trace.success) def run_regression(self, agent, config: dict = None) -> dict: """运行完整回归测试""" replayer = TraceReplayer(agent, config or {}) results = [] for tc in self.test_cases: result = replayer.replay(tc["baseline_trace"]) result["category"] = tc["category"] result["expected_success"] = tc["expected_success"] result["passed"] = self._evaluate_pass(result, tc) results.append(result) return self._summarize(results) def _evaluate_pass(self, result: dict, test_case: dict) -> bool: """判断是否通过回归""" # 1. 成功状态保持 if result["overall_diff"].get("success_preserved") != test_case["expected_success"]: return False # 2. 步骤匹配率达到阈值 match_rate = result["overall_diff"].get("match_rate", 0) if match_rate < 0.8: return False return True def _summarize(self, results: list[dict]) -> dict: from collections import defaultdict total = len(results) passed = sum(1 for r in results if r["passed"]) by_category = defaultdict(lambda: {"total": 0, "passed": 0}) for r in results: by_category[r["category"]]["total"] += 1 if r["passed"]: by_category[r["category"]]["passed"] += 1 return { "suite_name": self.suite_name, "total_tests": total, "passed": passed, "failed": total - passed, "pass_rate": passed / total if total > 0 else 0, "by_category": dict(by_category), "failures": [ { "trace_id": r["trace_id"], "task": r["task"], "category": r["category"], "match_rate": r["overall_diff"].get("match_rate", 0), "mismatched_steps": r["overall_diff"].get("mismatched_steps", 0), } for r in results if not r["passed"] ], } 四、非确定性管理 确定性策略 class DeterminismManager: """管理 Agent 测试中的非确定性""" STRATEGIES = { "temperature_zero": "LLM 温度设为 0,最大化输出确定性", "mock_llm": "模拟 LLM 输出,完全确定性", "mock_tools": "模拟工具输出,消除环境非确定性", "semantic_compare": "使用语义相似度替代精确匹配", "n_run_consensus": "多次运行取共识", } @staticmethod def n_run_consensus(agent, task: str, n: int = 5, agreement_threshold: float = 0.8) -> dict: """多次运行取共识""" results = [] for _ in range(n): result = agent.run(task) results.append(result) # 计算答案一致性 from sentence_transformers import SentenceTransformer import numpy as np model = SentenceTransformer("all-MiniLM-L6-v2") embeddings = model.encode([r.get("answer", "") for r in results]) sim_matrix = np.dot(embeddings, embeddings.T) # 平均成对相似度 n_results = len(results) upper_tri = sim_matrix[np.triu_indices(n_results, k=1)] avg_agreement = np.mean(upper_tri) return { "n_runs": n, "avg_agreement": float(avg_agreement), "consensus_reached": avg_agreement >= agreement_threshold, "results": results, "strategy": "n_run_consensus", } 快照测试 class SnapshotTester: """Agent 状态快照测试""" def __init__(self, snapshot_dir: str = "snapshots"): self.snapshot_dir = snapshot_dir from pathlib import Path Path(snapshot_dir).mkdir(parents=True, exist_ok=True) def take_snapshot(self, agent, label: str) -> str: """拍摄 Agent 当前状态快照""" import hashlib snapshot = { "label": label, "timestamp": datetime.now().isoformat(), "memory": getattr(agent, "memory", None), "state": getattr(agent, "state", None), "context": getattr(agent, "context", None), "tool_registry": list(getattr(agent, "tools", {}).keys()), } # 计算快照哈希 snapshot_str = json.dumps(snapshot, sort_keys=True, ensure_ascii=False) snapshot_hash = hashlib.sha256(snapshot_str.encode()).hexdigest()[:16] # 保存快照 from pathlib import Path filepath = Path(self.snapshot_dir) / f"{label}_{snapshot_hash}.json" with open(filepath, "w", encoding="utf-8") as f: json.dump(snapshot, f, ensure_ascii=False, indent=2) return str(filepath) def compare_snapshots(self, snapshot_path_a: str, snapshot_path_b: str) -> dict: """比较两个快照""" with open(snapshot_path_a, "r", encoding="utf-8") as f: snap_a = json.load(f) with open(snapshot_path_b, "r", encoding="utf-8") as f: snap_b = json.load(f) diffs = {} all_keys = set(snap_a.keys()) | set(snap_b.keys()) for key in all_keys: if snap_a.get(key) != snap_b.get(key): diffs[key] = { "snapshot_a": snap_a.get(key), "snapshot_b": snap_b.get(key), } return { "identical": len(diffs) == 0, "diff_count": len(diffs), "diffs": diffs, } 五、CI/CD 集成 # .github/workflows/agent-regression.yml name: Agent Regression Tests on: pull_request: paths: - "agent/**" - "prompts/**" - "tools/**" push: branches: [main] jobs: regression: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Setup Python uses: actions/setup-python@v5 with: python-version: "3.11" - name: Install dependencies run: pip install -r requirements.txt - name: Load baseline traces uses: actions/cache@v4 with: path: tests/traces/baseline key: baseline-traces-${{ hashFiles('tests/traces/baseline/**') }} - name: Run regression tests run: | python -m agent_regression_test \ --suite "main" \ --baseline-dir tests/traces/baseline \ --config tests/regression_config.yaml \ --output reports/regression.json - name: Check pass rate run: | PASS_RATE=$(python -c "import json; r=json.load(open('reports/regression.json')); print(r['pass_rate'])") echo "Pass rate: $PASS_RATE" if (( $(echo "$PASS_RATE < 0.9" | bc -l) )); then echo "::error::Regression test pass rate below 90%" exit 1 fi - name: Upload regression report uses: actions/upload-artifact@v4 with: name: regression-report path: reports/regression.json 测试策略对比 策略 确定性 覆盖率 维护成本 执行速度 适用阶段 精确回放 高 低 高 快 回归测试 Mock 工具回放 高 中 中 快 CI/CD 语义比较回放 中 高 低 中 日常验证 多次运行共识 中 高 低 慢 发布前 快照测试 高 中 中 快 状态验证 全量重放 低 高 低 慢 深度验证 最佳实践 轨迹采样策略 class TraceSampler: """从生产环境采样轨迹用于回归测试""" @staticmethod def sample_diverse(traces: list[AgentTrace], target_count: int = 100) -> list[AgentTrace]: """采样多样化的轨迹集""" from collections import defaultdict import random # 按任务类型分组 by_type = defaultdict(list) for t in traces: task_type = t.metadata.get("task_type", "unknown") by_type[task_type].append(t) # 每个类型按比例采样 total = len(traces) sampled = [] for task_type, type_traces in by_type.items(): n = max(1, int(target_count * len(type_traces) / total)) # 优先采样成功和失败的案例各一半 successes = [t for t in type_traces if t.success] failures = [t for t in type_traces if not t.success] n_success = min(n // 2, len(successes)) n_failure = min(n - n_success, len(failures)) sampled.extend(random.sample(successes, n_success) if n_success else []) sampled.extend(random.sample(failures, n_failure) if n_failure else []) return sampled[:target_count] 结语 Agent 回放测试是保障系统可靠性的关键基础设施。它通过录制真实执行轨迹、回放比较行为差异,在非确定性的 Agent 系统中建立可量化的质量基线。核心原则是:录制要全、回放要快、比较要智能。精确匹配在 LLM 时代往往过于严格,语义比较 + Mock 工具的组合能在保证检测力的同时控制测试的 flakiness。将回放测试集成到 CI/CD 流程中,才能确保每次 Agent 变更都有质量保障。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

2026-06-25 · 10 min · 1945 words · AI 实战派
agent benchmark suite

Agent 基准测试套件:SWE-bench vs WebArena vs GAIA

为什么 Agent 需要专属基准 传统 LLM 基准(MMLU、HumanEval)评估的是单轮输入输出的能力。但 Agent 的工作模式截然不同:它需要多步推理、工具调用、环境交互和状态管理。一个在 MMLU 上得高分的模型,未必能完成"在 GitHub 上修复一个 issue"这样的复杂任务。 Agent 基准测试需要回答的问题: 模型能否将复杂目标拆解为可执行的子任务? 模型能否正确调用工具并解析返回结果? 模型能否在失败后调整策略并重试? 模型能否在长上下文中保持目标一致性? 三大基准套件总览 维度 SWE-bench WebArena GAIA 领域 软件工程 Web 交互 通用助手 任务来源 真实 GitHub Issue 自建 Web 环境 人工设计 环境 Docker 容器 浏览器 + Web 服务 文件 + Web + 工具 评估方式 单元测试通过率 端到端功能验证 最终答案精确匹配 任务数量 2,298 (Lite: 300) 812 466 (Level 1-3) 最高分(2026初) ~35% (SWE-agent) ~42% (GPT-4o) ~25% (Level 3) 开源协议 MIT MIT Apache 2.0 SWE-bench:软件工程能力试金石 设计理念 SWE-bench 从 12 个流行 Python 开源仓库中收集真实的 GitHub Issue 及对应 Pull Request,要求 Agent 在给定代码仓库中修改代码以解决 Issue。评估标准是对应 PR 中的单元测试是否通过。 ...

2026-06-25 · 6 min · 1196 words · AI 实战派
user satisfaction metrics

AI 用户满意度指标体系:从 CSAT 到 AI-NPS

为什么传统满意度指标不够用 CSAT(Customer Satisfaction)、NPS(Net Promoter Score)和 CES(Customer Effort Score)是传统软件产品的三大满意度指标。但当产品核心引擎变成 AI 时,这些指标暴露出明显不足: CSAT 无法捕捉"惊喜时刻":AI 产品的满意度不是线性的,一次精彩的回答可能抵消十次平庸的表现 NPS 不反映信任问题:用户可能推荐你的产品,但自己在关键决策时不敢依赖它 CES 忽略心智负担:AI 产品的"effort"不只是点击次数,还包括用户验证结果正确性的认知负担 我们需要一套为 AI 量身定制的满意度指标体系。 指标体系全景 AI 用户满意度指标体系 ├── 基础指标(继承自传统软件) │ ├── CSAT-AI(会话级满意度) │ ├── NPS-AI(推荐意愿) │ └── CES-AI(任务完成努力度) ├── AI 特有指标 │ ├── 信任度(Trust Score) │ ├── 可控性感知(Controllability) │ ├── 透明度感知(Transparency) │ └── 惊喜指数(Delight Index) └── 行为指标(隐式测量) ├── 重试率(Retry Rate) ├── 人工求助率(Escalation Rate) ├── 会话深度(Conversation Depth) └── 回访留存(Return Retention) 一、CSAT-AI:会话级满意度 传统 CSAT 的问题 传统 CSAT 通常在任务完成后弹出:“您对本次服务满意吗?1-5 分”。但 AI 产品的会话可能包含多个子任务,用户可能对某些方面满意但对其他方面不满意。 ...

2026-06-25 · 9 min · 1753 words · AI 实战派
a b testing for llm

LLM A/B 测试实践:统计显著性与业务指标

LLM A/B 测试的特殊性 传统 Web 产品的 A/B 测试已经相当成熟——改个按钮颜色、调整文案、优化布局,通过 CTR 和转化率就能快速判断优劣。但 LLM 产品的 A/B 测试面临独特挑战: 输出非确定性:同一个输入,两次调用可能得到不同输出 质量维度多元:没有单一指标能衡量"回答好不好" 长尾效应:大部分对话可能表现类似,但少数关键场景差异巨大 学习效应:用户可能需要时间适应新模型的行为风格 A/B 测试完整流程 ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ 假设构建 │ -> │ 实验设计 │ -> │ 执行与监控 │ -> │ 分析与决策 │ │ Hypothesis │ │ Design │ │ Execute │ │ Analyze │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ 一、假设构建 每个 A/B 测试都应始于一个清晰的假设: from dataclasses import dataclass from typing import Optional @dataclass class ABTestHypothesis: """A/B 测试假设""" # 假设描述 statement: str # "将模型从 GPT-4o-mini 升级到 GPT-4o 后, # 在复杂推理任务上的用户满意度将提升 10%" # 自变量 treatment_description: str # "使用 GPT-4o 替代 GPT-4o-mini" control_description: str # "继续使用 GPT-4o-mini" # 因变量(核心指标) primary_metric: str # "CSAT 评分" expected_effect: float # 0.10 (提升10%) expected_direction: str # "increase" or "decrease" # 次要指标 secondary_metrics: list # ["重试率", "平均对话轮次", "人工求助率"] # 护栏指标(不应恶化的指标) guardrail_metrics: list # ["延迟 P95", "成本/请求", "安全违规率"] # 目标人群 target_segment: str # "all_users" 或 "power_users" 等 # 最小可检测效应 (MDE) mde: float # 0.05 (最小可检测 5% 变化) def validate(self) -> list[str]: issues = [] if not self.statement or len(self.statement) < 10: issues.append("假设描述太短") if not self.primary_metric: issues.append("缺少主要指标") if self.expected_effect <= 0: issues.append("预期效应应为正数") if not self.guardrail_metrics: issues.append("缺少护栏指标——可能导致意外回退") return issues 二、样本量计算 import math from scipy import stats class SampleSizeCalculator: """A/B 测试样本量计算器""" @staticmethod def for_proportion( baseline_rate: float, mde: float, # 最小可检测效应(绝对值) alpha: float = 0.05, power: float = 0.80, two_sided: bool = True, ) -> dict: """ 比例类指标的样本量计算(如 CSAT 满意率、重试率) 参数: baseline_rate: 基线比例(如当前 CSAT = 0.75) mde: 最小可检测效应(如 0.05 表示检测 5% 绝对变化) alpha: 显著性水平 power: 统计功效 """ z_alpha = stats.norm.ppf(1 - alpha / 2 if two_sided else 1 - alpha) z_beta = stats.norm.ppf(power) p1 = baseline_rate p2 = baseline_rate + mde p_avg = (p1 + p2) / 2 n = ((z_alpha * math.sqrt(2 * p_avg * (1 - p_avg)) + z_beta * math.sqrt(p1 * (1 - p1) + p2 * (1 - p2))) ** 2) / (p2 - p1) ** 2 return { "sample_per_group": math.ceil(n), "total_sample": math.ceil(n * 2), "baseline_rate": baseline_rate, "mde": mde, "alpha": alpha, "power": power, "expected_duration_days": math.ceil(n * 2 / 1000), # 假设每天1000用户 } @staticmethod def for_continuous( baseline_mean: float, baseline_std: float, mde: float, alpha: float = 0.05, power: float = 0.80, ) -> dict: """ 连续型指标的样本量计算(如评分均值、延迟) 参数: baseline_mean: 基线均值 baseline_std: 基线标准差 mde: 最小可检测效应(绝对变化量) """ z_alpha = stats.norm.ppf(1 - alpha / 2) z_beta = stats.norm.ppf(power) n = 2 * ((z_alpha + z_beta) * baseline_std / mde) ** 2 return { "sample_per_group": math.ceil(n), "total_sample": math.ceil(n * 2), "baseline_mean": baseline_mean, "baseline_std": baseline_std, "mde": mde, } # 示例:计算 CSAT 从 75% 提升到 80% 所需的样本量 calc = SampleSizeCalculator() result = calc.for_proportion(baseline_rate=0.75, mde=0.05) # 输出: sample_per_group ≈ 2554, total_sample ≈ 5108 LLM 特有的样本量考量 因素 影响 调整策略 输出非确定性 增加方差,需要更多样本 对同一输入运行多次取均值 用户异质性 不同用户群体效应不同 分层随机化 冷启动效应 新模型初期表现可能不稳定 设置预热期 时段效应 不同时间段对话质量不同 确保两组同时段运行 三、实验设计 随机化策略 import hashlib import random class ABTestRouter: """A/B 测试流量分配器""" def __init__(self, experiment_id: str, traffic_split: dict = None): self.experiment_id = experiment_id # 默认 50/50 分配 self.traffic_split = traffic_split or {"control": 0.5, "treatment": 0.5} self.salt = experiment_id # 用于哈希的盐值 def assign(self, user_id: str) -> str: """ 基于用户 ID 的确定性分配 同一用户始终分到同一组 """ # 使用哈希确保确定性 hash_input = f"{self.salt}:{user_id}" hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16) bucket = (hash_value % 10000) / 10000.0 cumulative = 0.0 for group, ratio in self.traffic_split.items(): cumulative += ratio if bucket < cumulative: return group return list(self.traffic_split.keys())[-1] # fallback def assign_with_layering(self, user_id: str, existing_experiments: list[str]) -> str: """ 分层分配:避免与正在运行的其他实验冲突 """ # 将已有实验纳入哈希 hash_input = f"{self.salt}:{user_id}:{':'.join(sorted(existing_experiments))}" hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16) bucket = (hash_value % 10000) / 10000.0 cumulative = 0.0 for group, ratio in self.traffic_split.items(): cumulative += ratio if bucket < cumulative: return group return list(self.traffic_split.keys())[-1] class StratifiedRouter: """分层随机化:确保关键维度均衡""" STRATA = ["user_type", "region", "device", "usage_level"] def __init__(self, experiment_id: str, strata_weights: dict = None): self.experiment_id = experiment_id self.strata_weights = strata_weights or {} self.assignments = {} # 缓存分配结果 def assign(self, user_id: str, user_attributes: dict) -> str: # 计算分层 key strata_key = "|".join( str(user_attributes.get(s, "unknown")) for s in self.STRATA ) # 分层内随机分配 hash_input = f"{self.experiment_id}:{strata_key}:{user_id}" hash_value = int(hashlib.sha256(hash_input.encode()).hexdigest(), 16) return "treatment" if hash_value % 2 == 0 else "control" 实验配置 @dataclass class ABTestConfig: experiment_id: str name: str hypothesis: ABTestHypothesis # 流量分配 traffic_split: dict = field(default_factory=lambda: {"control": 0.5, "treatment": 0.5}) total_traffic_pct: float = 1.0 # 使用多少比例的总流量 # 时间设置 start_date: str = "" end_date: str = "" min_duration_days: int = 14 # 最短运行天数 warmup_days: int = 2 # 预热天数(数据不计入分析) # 指标配置 primary_metric: str = "" secondary_metrics: list = field(default_factory=list) guardrail_metrics: list = field(default_factory=list) # 统计参数 alpha: float = 0.05 power: float = 0.80 mde: float = 0.05 # 停止规则 early_stop_on_guardrail: bool = True guardrail_thresholds: dict = field(default_factory=lambda: { "latency_p95_ms": 5000, # P95 延迟不超过 5s "safety_violation_rate": 0.001, # 安全违规率不超过 0.1% "cost_per_session": 0.15, # 每次会话成本不超过 $0.15 }) 四、统计显著性检验 from scipy import stats import numpy as np class SignificanceTester: """A/B 测试显著性检验""" def test_proportion(self, control_success: int, control_total: int, treatment_success: int, treatment_total: int, alpha: float = 0.05) -> dict: """ 比例类指标的双比例 Z 检验 """ p_control = control_success / control_total p_treatment = treatment_success / treatment_total p_pooled = (control_success + treatment_success) / (control_total + treatment_total) se = np.sqrt(p_pooled * (1 - p_pooled) * (1/control_total + 1/treatment_total)) z_stat = (p_treatment - p_control) / se if se > 0 else 0 p_value = 2 * (1 - stats.norm.cdf(abs(z_stat))) # 置信区间 diff = p_treatment - p_control ci_lower = diff - stats.norm.ppf(1 - alpha/2) * se ci_upper = diff + stats.norm.ppf(1 - alpha/2) * se # 效应量 relative_lift = diff / p_control if p_control > 0 else 0 return { "test_type": "two_proportion_z_test", "control_rate": round(p_control, 4), "treatment_rate": round(p_treatment, 4), "absolute_diff": round(diff, 4), "relative_lift": f"{relative_lift*100:.2f}%", "z_statistic": round(z_stat, 4), "p_value": round(p_value, 6), "significant": p_value < alpha, "ci_95": [round(ci_lower, 4), round(ci_upper, 4)], "winner": "treatment" if (p_value < alpha and diff > 0) else ("control" if (p_value < alpha and diff < 0) else "no_significant_diff"), } def test_continuous(self, control_values: list[float], treatment_values: list[float], alpha: float = 0.05) -> dict: """ 连续型指标的检验(t 检验或 Mann-Whitney U 检验) """ control = np.array(control_values) treatment = np.array(treatment_values) # 正态性检验决定使用哪种检验 if len(control) >= 30 and len(treatment) >= 30: # 大样本使用 Welch t 检验 t_stat, p_value = stats.ttest_ind(treatment, control, equal_var=False) test_name = "welch_t_test" else: # 小样本使用 Mann-Whitney U u_stat, p_value = stats.mannwhitneyu(treatment, control, alternative="two-sided") t_stat = u_stat test_name = "mann_whitney_u" diff = treatment.mean() - control.mean() # 置信区间(基于 t 分布) pooled_se = np.sqrt(treatment.var(ddof=1)/len(treatment) + control.var(ddof=1)/len(control)) df = len(treatment) + len(control) - 2 ci_lower = diff - stats.t.ppf(1-alpha/2, df) * pooled_se ci_upper = diff + stats.t.ppf(1-alpha/2, df) * pooled_se return { "test_type": test_name, "control_mean": round(control.mean(), 4), "control_std": round(control.std(), 4), "treatment_mean": round(treatment.mean(), 4), "treatment_std": round(treatment.std(), 4), "absolute_diff": round(diff, 4), "relative_lift": f"{(diff/control.mean())*100:.2f}%" if control.mean() != 0 else "N/A", "statistic": round(t_stat, 4), "p_value": round(p_value, 6), "significant": p_value < alpha, "ci_95": [round(ci_lower, 4), round(ci_upper, 4)], } def sequential_test(self, daily_data: list[dict], alpha: float = 0.05) -> dict: """ 序贯检验:每天检查是否可以提前停止 使用 Bonferroni 校正控制总犯错误率 """ num_checks = len(daily_data) adjusted_alpha = alpha / num_checks # Bonferroni 校正 results = [] for i, day_data in enumerate(daily_data): result = self.test_proportion( day_data["control_success"], day_data["control_total"], day_data["treatment_success"], day_data["treatment_total"], alpha=adjusted_alpha ) result["day"] = i + 1 result["adjusted_alpha"] = adjusted_alpha results.append(result) # 如果已显著或护栏指标触发,可以停止 if result["significant"]: return { "stopped_early": True, "stopped_at_day": i + 1, "final_result": result, "all_daily_results": results, } return { "stopped_early": False, "final_result": results[-1] if results else None, "all_daily_results": results, } 五、业务指标选择 指标分层框架 class MetricFramework: """A/B 测试指标分层框架""" METRIC_TREE = { "北极星指标": { "description": "最能反映产品价值的单一指标", "llm_examples": ["每周活跃对话用户数 (WAC)", "人均每日对话轮次"], "sensitivity": "低 — 需要较长时间观察", }, "主要指标": { "description": "直接反映假设是否成立的指标", "llm_examples": ["CSAT", "任务完成率", "回答准确率"], "sensitivity": "中 — 通常 2-4 周可检测", }, "次要指标": { "description": "帮助理解主要指标变化原因", "llm_examples": ["重试率", "对话深度", "人工求助率"], "sensitivity": "高 — 快速响应变化", }, "护栏指标": { "description": "确保不出现严重回退", "llm_examples": ["延迟 P95", "安全违规率", "每次会话成本"], "sensitivity": "高 — 需要实时监控", }, "调试指标": { "description": "用于诊断问题,不用于决策", "llm_examples": ["Token 使用量", "API 错误率", "特定类别表现"], "sensitivity": "高", } } @staticmethod def recommend_metrics(scenario: str) -> dict: recommendations = { "model_upgrade": { "primary": "回答准确率(基于评估集)", "secondary": ["CSAT", "重试率", "对话深度"], "guardrails": ["延迟 P95", "安全违规率", "成本/请求"], }, "prompt_change": { "primary": "任务完成率", "secondary": ["用户重述率", "回答长度分布"], "guardrails": ["安全违规率", "Token 使用量"], }, "feature_addition": { "primary": "功能采纳率", "secondary": ["CSAT", "会话深度", "NPS"], "guardrails": ["延迟 P95", "错误率"], }, "ui_change": { "primary": "任务完成率", "secondary": ["CSAT", "CES", "会话深度"], "guardrails": ["页面加载时间", "错误率"], }, } return recommendations.get(scenario, "Unknown scenario") 六、常见陷阱与解决方案 陷阱一:Peeking(偷看) class PeekingWarning: """偷看陷阱演示与解决方案""" def demonstrate_peeking_problem(self): """ 如果每天检查 p 值并在 p<0.05 时停止, 实际犯错误率远超 5% """ # 模拟 1000 次实验(A 和 B 实际无差异) false_positive_count = 0 for _ in range(1000): control = np.random.binomial(1, 0.10, 500) # 转化率 10% treatment = np.random.binomial(1, 0.10, 500) # 每天检查(假设每天 50 个样本) for day in range(1, 11): c_slice = control[:day*50] t_slice = treatment[:day*50] # 简化:使用卡方检验 _, p_value = stats.chi2_contingency([ [c_slice.sum(), len(c_slice) - c_slice.sum()], [t_slice.sum(), len(t_slice) - t_slice.sum()] ]) if p_value < 0.05: false_positive_count += 1 break actual_fpr = false_positive_count / 1000 return { "nominal_alpha": 0.05, "actual_false_positive_rate": actual_fpr, "inflation_factor": actual_fpr / 0.05, "solution": "使用序贯检验或 Alpha Spending 函数" } 陷阱二:辛普森悖论 class SimpsonParadoxCheck: """辛普森悖论检测""" def check(self, data: pd.DataFrame, group_col: str, metric_col: str, strata_col: str) -> dict: """ 检测是否存在辛普森悖论: 整体趋势与分层趋势相反 """ # 整体差异 overall = data.groupby(group_col)[metric_col].mean() overall_diff = overall.get("treatment", 0) - overall.get("control", 0) # 分层差异 strata_results = {} reversals = [] for stratum, stratum_data in data.groupby(strata_col): stratum_overall = stratum_data.groupby(group_col)[metric_col].mean() stratum_diff = stratum_overall.get("treatment", 0) - stratum_overall.get("control", 0) strata_results[stratum] = { "control_n": len(stratum_data[stratum_data[group_col] == "control"]), "treatment_n": len(stratum_data[stratum_data[group_col] == "treatment"]), "control_mean": stratum_overall.get("control", 0), "treatment_mean": stratum_overall.get("treatment", 0), "diff": stratum_diff, } # 检测方向反转 if (overall_diff > 0 and stratum_diff < 0) or \ (overall_diff < 0 and stratum_diff > 0): reversals.append(stratum) return { "overall_diff": overall_diff, "strata": strata_results, "reversals": reversals, "simpson_paradox": len(reversals) > 0, "recommendation": "分层分析而非整体分析" if reversals else "整体分析可行", } 七、决策框架 class ABTestDecision: """A/B 测试决策框架""" @staticmethod def decide(primary_result: dict, guardrail_results: dict, secondary_results: dict, config: ABTestConfig) -> dict: # 1. 检查护栏指标 guardrail_breaches = [] for metric, result in guardrail_results.items(): threshold = config.guardrail_thresholds.get(metric) if threshold and result.get("treatment_mean", 0) > threshold: guardrail_breaches.append({ "metric": metric, "treatment_value": result["treatment_mean"], "threshold": threshold, }) if guardrail_breaches: return { "decision": "STOP", "reason": "护栏指标被突破", "breaches": guardrail_breaches, } # 2. 检查主要指标 if not primary_result["significant"]: return { "decision": "CONTINUE" if primary_result.get("p_value", 1) > 0.3 else "INCONCLUSIVE", "reason": "主要指标未达统计显著性", "p_value": primary_result["p_value"], "recommendation": "继续收集数据或增大样本量", } # 3. 主要指标显著 winner = primary_result["winner"] if winner == "treatment": # 检查次要指标是否支持 secondary_support = all( r.get("relative_lift", "").startswith("-") is False for r in secondary_results.values() ) return { "decision": "SHIP", "reason": "主要指标显著提升,护栏指标未突破", "confidence": "high" if secondary_support else "medium", "primary_lift": primary_result["relative_lift"], "secondary_support": secondary_support, } else: return { "decision": "HOLD", "reason": "主要指标显著下降", "primary_drop": primary_result["relative_lift"], } 决策速查表 情况 主要指标 护栏指标 决策 主要指标显著提升 ✅ ✅ 未突破 全量发布 主要指标显著提升 ✅ ❌ 突破 不发布,分析权衡 主要指标显著下降 ❌ ✅ 保持对照组 主要指标无显著差异 - ✅ 视成本决定 主要指标无显著差异 - ❌ 保持对照组 结语 LLM A/B 测试比传统 Web A/B 测试复杂得多,但核心原则不变:清晰的假设、足够的样本量、正确的统计检验、严格的护栏监控。最大的陷阱不是统计方法不对,而是没有想清楚要测什么就开始测。在 LLM 时代,一个好的 A/B 测试框架是产品迭代的基础设施——没有它,所有的"优化"都只是猜测。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

2026-06-25 · 8 min · 1692 words · AI 实战派
llm creative eval

LLM 创意评估方法:从主观评分到自动化指标

创意评估的困境 评估代码生成的正确性很简单——跑测试就行。评估事实问答的准确性也不难——对比标准答案。但当你让 LLM 写一首诗、创作一个短篇故事、或生成一个创意广告时,“好不好” suddenly becomes a very hard question。 创意评估面临三个核心挑战: 主观性:同一篇作品,不同读者评价可能天差地别 多维性:创意好不好涉及语言、结构、新颖度、情感共鸣等多个维度 缺乏标准答案:创意任务没有唯一正确答案,甚至没有"参考答案" 评估方法分类 创意评估方法 ├── 人工评估 │ ├── 整体评分(Holistic Scoring) │ ├── 多维 Rubric 评分 │ └── 排序比较(Pairwise Ranking) ├── 自动化指标 │ ├── 多样性指标(Distinct-n, Self-BLEU) │ ├── 新颖度指标(Semantic Novelty) │ └── 连贯性指标(Coherence Score) └── LLM-as-Judge ├── 单维度评分 ├── 多维度 Rubric └── 对比较(Pairwise Comparison) 一、人工评估体系 整体评分法 最直接的方式:让评分者对作品给出一个综合分数(1-5 分或 1-10 分)。 class HolisticScoring: """整体评分体系""" SCALE_5 = { 5: "杰出 — 创意独特,语言精炼,情感打动人心", 4: "优秀 — 有明确创意点,语言流畅,有一定感染力", 3: "合格 — 表达清晰,但创意普通,缺乏亮点", 2: "较差 — 表达生硬,创意陈旧,难以读完", 1: "不合格 — 逻辑混乱,语言不通,无法理解" } def __init__(self, num_raters: int = 3): self.num_raters = num_raters def aggregate(self, scores: list[int]) -> dict: import numpy as np scores = sorted(scores) return { "mean": np.mean(scores), "median": np.median(scores), "std": np.std(scores), "min": min(scores), "max": max(scores), "agreement": self._inter_rater_agreement(scores), } def _inter_rater_agreement(self, scores: list[int]) -> float: """计算评分者间一致性""" if len(scores) < 2: return 1.0 mean = sum(scores) / len(scores) variance = sum((s - mean) ** 2 for s in scores) / len(scores) max_variance = (max(scores) - min(scores)) ** 2 / 4 return 1 - (variance / max_variance) if max_variance > 0 else 1.0 多维 Rubric 评分 整体评分太粗,更精细的方法是定义多个评分维度: ...

2026-06-25 · 7 min · 1481 words · AI 实战派
hallucination measurement

LLM 幻觉测量方法:从人工标注到自动检测

幻觉问题的严重性 LLM 幻觉——模型生成看似合理但实际不正确的信息——是当前大语言模型最严重的问题之一。在闲聊场景中,幻觉可能只是闹个笑话;但在医疗、法律、金融等高风险场景中,幻觉可能造成严重后果。 根据 2025 年的研究统计: 场景 典型幻觉率 后果严重程度 事实问答 5-15% 中 代码生成 10-25% 中高 医疗咨询 8-20% 极高 法律引用 15-30% 极高 历史事件 10-20% 高 数学推理 15-30% 中 人物传记 20-40% 高 幻觉分类体系 LLM 幻觉类型 ├── 事实性幻觉(Factual Hallucination) │ ├── 实体幻觉:编造不存在的人名/地名/机构 │ ├── 关系幻觉:编造人物之间的关系 │ ├── 数字幻觉:编造统计数据或日期 │ └── 引用幻觉:编造论文/法律/新闻报道 ├── 逻辑性幻觉(Logical Hallucination) │ ├── 推理跳跃:跳过关键推理步骤 │ ├── 循环论证:用结论证明结论 │ └── 因果倒置:混淆原因和结果 ├── 上下文幻觉(Contextual Hallucination) │ ├── 矛盾输出:与之前回答自相矛盾 │ ├── 忽略约束:不遵守 prompt 中的约束 │ └── 过度推断:超出给定信息范围 └── 格式幻觉(Format Hallucination) ├── 结构错误:输出格式不符合要求 └── 引用伪造:伪造可验证的引用来源 一、人工标注体系 幻觉标注框架 from dataclasses import dataclass, field from enum import Enum class HallucinationType(Enum): ENTITY = "entity" # 实体幻觉 RELATION = "relation" # 关系幻觉 NUMERIC = "numeric" # 数字幻觉 CITATION = "citation" # 引用幻觉 LOGICAL = "logical" # 逻辑幻觉 CONTEXTUAL = "contextual" # 上下文幻觉 FORMAT = "format" # 格式幻觉 NONE = "none" # 无幻觉 class HallucinationSeverity(Enum): NONE = 0 # 无幻觉 MINOR = 1 # 轻微:不影响核心信息 MODERATE = 2 # 中等:部分信息不准确 SEVERE = 3 # 严重:核心信息完全错误 CRITICAL = 4 # 致命:可能造成实际危害 @dataclass class HallucinationAnnotation: """单条幻觉标注""" span_start: int # 幻觉文本起始位置 span_end: int # 幻觉文本结束位置 hallucinated_text: str # 幻觉文本 hallucination_type: HallucinationType severity: HallucinationSeverity correct_info: str # 正确信息 source: str # 正确信息来源 annotator_id: str confidence: float # 标注者置信度 0-1 @dataclass class HallucinationDocument: """一份完整文档的幻觉标注""" doc_id: str prompt: str response: str annotations: list[HallucinationAnnotation] = field(default_factory=list) @property def hallucination_rate(self) -> float: """幻觉率:有幻觉的句子占比""" if not self.response: return 0.0 sentences = self.response.split("。") hallucinated_sentences = set() for ann in self.annotations: for i, sent in enumerate(sentences): if ann.hallucinated_text in sent: hallucinated_sentences.add(i) return len(hallucinated_sentences) / len(sentences) if sentences else 0 @property def severity_score(self) -> float: """严重度评分:加权幻觉得分""" weights = {0: 0, 1: 0.25, 2: 0.5, 3: 0.75, 4: 1.0} if not self.annotations: return 0.0 return sum(weights[a.severity.value] for a in self.annotations) / len(self.annotations) 标注一致性度量 class AnnotationAgreement: """标注者间一致性计算""" @staticmethod def cohen_kappa(annotator1: list[str], annotator2: list[str]) -> float: """Cohen's Kappa:两个标注者的一致性""" from collections import Counter n = len(annotator1) labels = sorted(set(annotator1 + annotator2)) # 观察一致率 observed = sum(1 for a, b in zip(annotator1, annotator2) if a == b) / n # 期望一致率 c1 = Counter(annotator1) c2 = Counter(annotator2) expected = sum((c1[l] / n) * (c2[l] / n) for l in labels) if expected == 1.0: return 1.0 return (observed - expected) / (1 - expected) @staticmethod def fleiss_kappa(annotations: list[list[str]]) -> float: """Fleiss' Kappa:多标注者一致性""" import numpy as np n = len(annotations[0]) # 样本数 k = len(annotations) # 标注者数 labels = sorted(set(l for ann in annotations for l in ann)) m = len(labels) # 构建计数矩阵 counts = np.zeros((n, m)) label_idx = {l: i for i, l in enumerate(labels)} for annotator_labels in annotations: for i, label in enumerate(annotator_labels): counts[i][label_idx[label]] += 1 # 观察一致率 P_i = (np.sum(counts**2, axis=1) - k) / (k * (k - 1)) P_bar = np.mean(P_i) # 期望一致率 p_j = np.sum(counts, axis=0) / (n * k) P_e = np.sum(p_j**2) if P_e == 1.0: return 1.0 return (P_bar - P_e) / (1 - P_e) 二、自动检测算法 基于检索的幻觉检测 class RetrievalBasedDetector: """基于检索的幻觉检测:将生成内容与知识库比对""" def __init__(self, knowledge_base, embedding_model="all-MiniLM-L6-v2"): from sentence_transformers import SentenceTransformer import faiss import numpy as np self.model = SentenceTransformer(embedding_model) self.kb_texts = knowledge_base embeddings = self.model.encode(knowledge_base) # 构建 FAISS 索引 dim = embeddings.shape[1] self.index = faiss.IndexFlatIP(dim) self.index.add(embeddings.astype('float32')) def detect(self, response: str, top_k: int = 5, threshold: float = 0.7) -> dict: """检测回答中的幻觉""" sentences = self._split_sentences(response) hallucinated = [] verified = [] for sent in sentences: # 检索最相关的知识库条目 sent_emb = self.model.encode([sent]).astype('float32') scores, indices = self.index.search(sent_emb, top_k) max_score = scores[0][0] best_match = self.kb_texts[indices[0][0]] if max_score < threshold: # 无法在知识库中找到支持 → 可能是幻觉 hallucinated.append({ "text": sent, "max_similarity": float(max_score), "best_match": best_match, "verdict": "unsupported", }) else: verified.append({ "text": sent, "similarity": float(max_score), "source": best_match, "verdict": "supported", }) return { "total_sentences": len(sentences), "hallucinated_count": len(hallucinated), "verified_count": len(verified), "hallucination_rate": len(hallucinated) / len(sentences) if sentences else 0, "details": {"hallucinated": hallucinated, "verified": verified}, } def _split_sentences(self, text: str) -> list[str]: import re # 按中英文标点分句 sentences = re.split(r'[。!?.!?\n]+', text) return [s.strip() for s in sentences if s.strip()] 基于 NLI 的幻觉检测 class NLIDetector: """基于自然语言推理(NLI)的幻觉检测""" def __init__(self, model_name="moritzlaurer/DeBERTa-v3-base-mnli-fever-nli"): from transformers import AutoTokenizer, AutoModelForSequenceClassification import torch self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModelForSequenceClassification.from_pretrained(model_name) self.model.eval() # 标签: 0=entailment, 1=neutral, 2=contradiction def detect(self, response: str, reference: str) -> dict: """ 判断 response 是否被 reference 支持 entailment: reference 支持 response contradiction: reference 与 response 矛盾 neutral: 无法判断 """ import torch sentences = self._split_sentences(response) results = [] for sent in sentences: inputs = self.tokenizer(reference, sent, return_tensors="pt", truncation=True, max_length=512) with torch.no_grad(): logits = self.model(**inputs).logits probs = torch.softmax(logits, dim=0) results.append({ "sentence": sent, "entailment_prob": probs[0].item(), "neutral_prob": probs[1].item(), "contradiction_prob": probs[2].item(), "verdict": self._classify(probs), }) hallucinated = [r for r in results if r["verdict"] == "contradiction"] unsupported = [r for r in results if r["verdict"] == "neutral"] return { "total_sentences": len(results), "supported": len(results) - len(hallucinated) - len(unsupported), "contradicted": len(hallucinated), "unsupported": len(unsupported), "hallucination_rate": (len(hallucinated) + len(unsupported)) / len(results) if results else 0, "details": results, } def _classify(self, probs): labels = ["entailment", "neutral", "contradiction"] idx = probs.argmax().item() return labels[idx] def _split_sentences(self, text): import re return [s.strip() for s in re.split(r'[。!?.!?\n]+', text) if s.strip()] 基于 LLM 的幻觉检测 class LLMHallucinationDetector: """使用强力 LLM 检测幻觉""" def __init__(self, judge_model: str = "gpt-4o"): self.judge_model = judge_model def detect(self, prompt: str, response: str, reference_context: str = None) -> dict: """检测回答中的幻觉""" judge_prompt = f"""你是事实核查专家。请检查以下 AI 回答中是否存在幻觉(与事实不符的内容)。 用户问题:{prompt} AI 回答:{response} """ if reference_context: judge_prompt += f"\n参考信息(权威来源):\n{reference_context}\n" judge_prompt += """ 请逐句检查,对每句话标注: - "supported":有事实依据支持 - "contradicted":与已知事实矛盾 - "unsupported":无法验证,可能是编造 输出 JSON: { "sentences": [ {"text": "...", "verdict": "supported/contradicted/unsupported", "reason": "..."} ], "overall_hallucination": "none/minor/moderate/severe", "hallucination_rate": 0.0-1.0, "key_issues": ["问题1", "问题2"] }""" import json result = call_llm(self.judge_model, judge_prompt) try: return json.loads(result) except json.JSONDecodeError: import re match = re.search(r'\{.*\}', result, re.DOTALL) if match: return json.loads(match.group()) return {"error": "parse failed", "raw": result} def detect_with_search(self, prompt: str, response: str) -> dict: """结合搜索引擎的幻觉检测""" # 1. 提取需要验证的关键陈述 claims = self._extract_claims(response) # 2. 对每个陈述进行搜索验证 results = [] for claim in claims: search_results = web_search(claim) verification = self._verify_claim(claim, search_results) results.append(verification) return { "total_claims": len(claims), "verified": sum(1 for r in results if r["verdict"] == "supported"), "hallucinated": sum(1 for r in results if r["verdict"] == "contradicted"), "unverifiable": sum(1 for r in results if r["verdict"] == "unsupported"), "details": results, } def _extract_claims(self, text: str) -> list[str]: """提取需要验证的事实陈述""" prompt = f"""从以下文本中提取需要验证的事实性陈述,每行一个: {text} 只输出陈述列表。""" result = call_llm(self.judge_model, prompt) return [line.strip() for line in result.strip().split("\n") if line.strip()] 三、专门化幻觉检测工具 SelfCheckGPT class SelfCheckGPT: """ SelfCheckGPT: 通过多次采样检测幻觉 核心思想:如果模型对同一问题多次生成的回答一致,则可信度高; 如果不一致,则可能存在幻觉 """ def __init__(self, model_name: str, num_samples: int = 5): self.model_name = model_name self.num_samples = num_samples def check(self, prompt: str, response: str) -> dict: # 1. 生成多个样本 samples = [] for _ in range(self.num_samples): # 使用较高温度增加多样性 sample = call_llm(self.model_name, prompt, temperature=0.7) samples.append(sample) # 2. 计算一致性 consistency_scores = [] for i, sample in enumerate(samples): if i == 0: continue score = self._sentence_level_consistency(response, sample) consistency_scores.append(score) avg_consistency = sum(consistency_scores) / len(consistency_scores) if consistency_scores else 1.0 return { "original_response": response, "num_samples": self.num_samples, "avg_consistency": avg_consistency, "hallucination_score": 1 - avg_consistency, # 不一致 = 幻觉概率 "verdict": "likely_hallucinated" if avg_consistency < 0.6 else "likely_reliable", } def _sentence_level_consistency(self, response: str, sample: str) -> float: """计算句子级一致性""" from sentence_transformers import SentenceTransformer import numpy as np model = SentenceTransformer("all-MiniLM-L6-v2") resp_sents = self._split_sentences(response) sample_sents = self._split_sentences(sample) resp_emb = model.encode(resp_sents) sample_emb = model.encode(sample_sents) # 对每个原句,找到样本中最相似的句子 sim_matrix = resp_emb @ sample_emb.T max_sims = sim_matrix.max(axis=1) return float(np.mean(max_sims)) def _split_sentences(self, text): import re return [s.strip() for s in re.split(r'[。!?.!?\n]+', text) if s.strip()] 四、评估指标体系 class HallucinationMetrics: """幻觉评估指标集合""" @staticmethod def precision_recall_f1(annotations: list[dict]) -> dict: """ 计算幻觉检测的 Precision/Recall/F1 annotations: [{"pred_hallucinated": bool, "gt_hallucinated": bool}] """ tp = sum(1 for a in annotations if a["pred_hallucinated"] and a["gt_hallucinated"]) fp = sum(1 for a in annotations if a["pred_hallucinated"] and not a["gt_hallucinated"]) fn = sum(1 for a in annotations if not a["pred_hallucinated"] and a["gt_hallucinated"]) tn = sum(1 for a in annotations if not a["pred_hallucinated"] and not a["gt_hallucinated"]) precision = tp / (tp + fp) if (tp + fp) > 0 else 0 recall = tp / (tp + fn) if (tp + fn) > 0 else 0 f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0 accuracy = (tp + tn) / len(annotations) if annotations else 0 return { "precision": round(precision, 4), "recall": round(recall, 4), "f1": round(f1, 4), "accuracy": round(accuracy, 4), "tp": tp, "fp": fp, "fn": fn, "tn": tn, } @staticmethod def hallucination_rate(docs: list[HallucinationDocument]) -> dict: """计算总体幻觉率""" rates = [d.hallucination_rate for d in docs] severities = [d.severity_score for d in docs] import numpy as np return { "mean_hallucination_rate": round(np.mean(rates), 4), "median_hallucination_rate": round(np.median(rates), 4), "p95_hallucination_rate": round(np.percentile(rates, 95), 4), "mean_severity": round(np.mean(severities), 4), "docs_with_hallucination": sum(1 for d in docs if d.annotations), "total_docs": len(docs), } @staticmethod def by_type_breakdown(docs: list[HallucinationDocument]) -> dict: """按幻觉类型分解""" from collections import defaultdict by_type = defaultdict(int) for doc in docs: for ann in doc.annotations: by_type[ann.hallucination_type.value] += 1 total = sum(by_type.values()) return { t: {"count": c, "percentage": round(c / total * 100, 1) if total > 0 else 0} for t, c in sorted(by_type.items(), key=lambda x: -x[1]) } 五、检测方法对比 方法 准确率 召回率 成本 实时性 适用场景 人工标注 95%+ 90%+ 极高 慢 基线建立 检索比对 80% 70% 中 中 有知识库时 NLI 模型 85% 75% 低 快 有参考文本时 LLM-as-Judge 88% 82% 中高 中 通用检测 SelfCheckGPT 78% 85% 高(多次采样) 慢 无参考文本时 搜索验证 82% 68% 中 慢 事实性声明 实践建议 分层检测策略 class LayeredHallucinationDetection: """分层幻觉检测策略""" def __init__(self): self.fast_checker = NLIDetector() # 快速初筛 self.deep_checker = LLMHallucinationDetector() # 深度检测 self.search_checker = None # 搜索验证(按需启用) def check(self, prompt: str, response: str, reference: str = None) -> dict: # 层 1:NLI 快速检测(<100ms) if reference: nli_result = self.fast_checker.detect(response, reference) if nli_result["hallucination_rate"] < 0.1: # NLI 认为基本无幻觉,直接返回 return {"layer": "nli", "result": nli_result, "confidence": "high"} # 层 2:LLM 深度检测(1-3s) llm_result = self.deep_checker.detect(prompt, response, reference) if llm_result.get("overall_hallucination") in ["none", "minor"]: return {"layer": "llm", "result": llm_result, "confidence": "high"} # 层 3:搜索验证(5-10s,仅对高风险内容) if llm_result.get("overall_hallucination") in ["severe", "moderate"]: search_result = self.deep_checker.detect_with_search(prompt, response) return {"layer": "search", "result": search_result, "confidence": "highest"} return {"layer": "llm", "result": llm_result, "confidence": "medium"} 结语 幻觉检测是 LLM 可靠性的最后一道防线。没有单一方法能完美检测所有类型的幻觉——检索方法依赖知识库的覆盖度,NLI 方法需要参考文本,LLM-as-Judge 本身也可能产生幻觉。最佳实践是分层检测:快速方法做初筛,深度方法做验证,搜索方法做兜底。同时,定期进行人工标注作为基线,校准自动检测系统的准确率。记住:降低幻觉的根本在于模型训练和 RAG 增强,检测只是发现问题的手段,而非解决方案本身。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

2026-06-25 · 8 min · 1579 words · AI 实战派
llm eval pipeline

LLM 评估流水线搭建:从数据集到报告

为什么需要评估流水线 大语言模型迭代速度极快,每次模型更新或 Prompt 修改都需要回答一个核心问题:新的版本到底比旧的好多少? 靠人工试几个案例远远不够,你需要一条系统化的评估流水线。 评估流水线的核心价值: 可复现:同一套数据集和指标,任何人任何时候跑都能得到一致结果 可比较:不同模型版本之间的差异被量化为具体数字 可扩展:从 100 条测试用例扩展到 10000 条只需改一个参数 可追踪:历史评估结果存档,形成模型演进的时间线 流水线架构总览 一条完整的 LLM 评估流水线包含五个核心阶段: ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ 数据集构建 │ -> │ 评估执行 │ -> │ 指标计算 │ -> │ 结果分析 │ -> │ 报告生成 │ │ Dataset │ │ Execution │ │ Metrics │ │ Analysis │ │ Report │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ 阶段一:数据集构建 数据集是评估的基石。一个高质量的评估数据集应具备以下特征: 特征 说明 示例 代表性 覆盖实际使用场景的主要类型 问答、摘要、翻译、代码生成 多样性 包含不同难度和长度的输入 简单事实题 vs 多步推理题 无泄漏 不包含训练数据中的内容 使用人工新写的题目 可验证 有标准答案或明确的评判标准 精确匹配 / 人工评分标准 可扩展 能方便地增加新类别 模块化的数据结构 数据集格式设计 推荐使用 JSONL 格式,每行一个测试样本: ...

2026-06-25 · 6 min · 1180 words · AI 实战派
multimodal eval method

多模态模型评估:视觉理解与跨模态推理

多模态评估的特殊性 单模态(纯文本)评估已经相当成熟,但当我们进入多模态领域——图像、视频、音频与文本的交叉理解——评估的复杂度呈指数级增长。多模态模型不仅要理解每种模态的信息,还要在不同模态间建立语义关联。 多模态评估的核心挑战: 对齐问题:文本描述和图像内容是否语义一致? 细粒度理解:模型是否真正"看到"了图像中的关键细节? 跨模态推理:能否基于图像信息进行文本推理,或反向操作? 评估成本:人工标注图文对的成本远高于纯文本 评估维度全景 多模态评估维度 ├── 视觉感知 │ ├── 图像识别(VQA, Image Captioning) │ ├── 细粒度理解(OCR, 属性识别) │ └── 空间推理(位置关系, 3D 理解) ├── 跨模态推理 │ ├── 图文推理(图→文推理) │ ├── 文图推理(文→图检索/生成) │ └── 多模态链式推理 ├── 多模态对话 │ ├── 多轮图像对话 │ └── 视频问答 └── 生成质量 ├── 图文一致性 ├── 视觉质量 └── 创意与忠实度 一、视觉理解评估 图像问答(VQA)评估 VQA 是最基础的多模态评估形式:给模型一张图片和一个问题,要求输出答案。 class VQAEvaluator: """VQA 评估器""" def __init__(self, eval_mode: str = "vqa_accuracy"): self.eval_mode = eval_mode def evaluate(self, predictions: list[dict]) -> dict: """ predictions: [{"question_id": int, "answer": str, "gt_answers": [str, ...]}] """ if self.eval_mode == "vqa_accuracy": return self._vqa_accuracy(predictions) elif self.eval_mode == "exact_match": return self._exact_match(predictions) def _vqa_accuracy(self, predictions: list[dict]) -> dict: """ 标准 VQA 准确率: 对每个问题,如果至少 3/10 的标注者给出了相同答案,则算正确 简化版:答案出现在 GT 答案列表中即算正确 """ correct = 0 for pred in predictions: gt = [a.strip().lower() for a in pred["gt_answers"]] ans = pred["answer"].strip().lower() # VQA 标准的 soft accuracy count = gt.count(ans) min_count = 1 # 简化:至少1个匹配 if count >= min_count: correct += min(1, count / 3.0) accuracy = correct / len(predictions) if predictions else 0 return {"vqa_accuracy": accuracy, "total": len(predictions)} def _exact_match(self, predictions: list[dict]) -> dict: correct = 0 for pred in predictions: gt = [a.strip().lower() for a in pred["gt_answers"]] ans = pred["answer"].strip().lower() if ans in gt: correct += 1 return {"exact_match": correct / len(predictions) if predictions else 0} class FineGrainedVQAEvaluator(VQAEvaluator): """细粒度 VQA 评估:按问题类型分桶""" QUESTION_TYPES = { "object": "图中有什么物体?", "count": "图中有几个XX?", "color": "XX是什么颜色的?", "spatial": "XX在YY的哪个位置?", "attribute": "XX有什么特征?", "relation": "XX和YY是什么关系?", "scene": "这是什么场景?", "ocr": "图中的文字写了什么?", } def evaluate_by_type(self, predictions: list[dict]) -> dict: from collections import defaultdict by_type = defaultdict(list) for pred in predictions: by_type[pred.get("question_type", "unknown")].append(pred) results = {} for q_type, preds in by_type.items(): results[q_type] = { "count": len(preds), "accuracy": self._exact_match(preds)["exact_match"], } return results 图像描述(Captioning)评估 class CaptioningEvaluator: """图像描述评估""" def __init__(self): self.metrics = {} def evaluate(self, prediction: str, references: list[str]) -> dict: results = {} # CIDEr-D: 专为图像描述设计的指标 results["cider"] = self._cider(prediction, references) # BLEU-4 results["bleu4"] = self._bleu(prediction, references, n=4) # METEOR results["meteor"] = self._meteor(prediction, references) # ROUGE-L results["rouge_l"] = self._rouge_l(prediction, references) # CLIPScore: 基于 CLIP 的图文匹配度 results["clip_score"] = self._clip_score(prediction, references) return results def _cider(self, pred: str, refs: list[str]) -> float: """CIDEr: 共识评估,基于 TF-IDF 加权的 n-gram 重叠""" # 简化实现,实际使用 pycocoevalcap from pycocoevalcap.cider.cider import Cider cider_scorer = Cider() gts = {0: refs} res = {0: [pred]} score, _ = cider_scorer.compute_score(gts, res) return score def _bleu(self, pred: str, refs: list[str], n: int = 4) -> float: from pycocoevalcap.bleu.bleu import Bleu bleu_scorer = Bleu(n) gts = {0: refs} res = {0: [pred]} score, _ = bleu_scorer.compute_score(gts, res) return score[n-1] def _meteor(self, pred: str, refs: list[str]) -> float: from pycocoevalcap.meteor.meteor import Meteor meteor_scorer = Meteor() gts = {0: refs} res = {0: [pred]} score, _ = meteor_scorer.compute_score(gts, res) return score def _rouge_l(self, pred: str, refs: list[str]) -> float: from pycocoevalcap.rouge.rouge import Rouge rouge_scorer = Rouge() gts = {0: refs} res = {0: [pred]} score, _ = rouge_scorer.compute_score(gts, res) return score def _clip_score(self, pred: str, refs: list[str], image_path: str = None) -> float: """CLIPScore: 使用 CLIP 计算图文匹配度""" from transformers import CLIPProcessor, CLIPModel from PIL import Image import torch model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32") processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32") image = Image.open(image_path) inputs = processor(text=[pred], images=image, return_tensors="pt", padding=True) with torch.no_grad(): outputs = model(**inputs) # 归一化的图文相似度 score = outputs.logits_per_image.item() # 归一化到 0-1 return min(max(score / 100, 0), 1) OCR 评估 class OCREvaluator: """OCR 能力评估""" def evaluate(self, prediction: str, ground_truth: str) -> dict: results = {} # 字符级准确率 results["char_accuracy"] = self._char_accuracy(prediction, ground_truth) # 词级准确率 results["word_accuracy"] = self._word_accuracy(prediction, ground_truth) # 编辑距离 results["edit_distance"] = self._edit_distance(prediction, ground_truth) # 归一化编辑距离 max_len = max(len(prediction), len(ground_truth), 1) results["normalized_edit_distance"] = results["edit_distance"] / max_len # ANLS (Average Normalized Levenshtein Similarity) results["anls"] = 1 - results["normalized_edit_distance"] return results def _char_accuracy(self, pred: str, gt: str) -> float: """字符级准确率""" if not gt: return 1.0 if not pred else 0.0 correct = sum(1 for p, g in zip(pred, gt) if p == g) # 加上长度差异惩罚 correct += 0 # 多出或缺少的字符算错 return correct / len(gt) def _word_accuracy(self, pred: str, gt: str) -> float: pred_words = pred.split() gt_words = gt.split() if not gt_words: return 1.0 if not pred_words else 0.0 correct = sum(1 for p, g in zip(pred_words, gt_words) if p == g) return correct / len(gt_words) def _edit_distance(self, s1: str, s2: str) -> int: """Levenshtein 编辑距离""" if len(s1) < len(s2): return self._edit_distance(s2, s1) if len(s2) == 0: return len(s1) previous_row = range(len(s2) + 1) for i, c1 in enumerate(s1): current_row = [i + 1] for j, c2 in enumerate(s2): insertions = previous_row[j + 1] + 1 deletions = current_row[j] + 1 substitutions = previous_row[j] + (c1 != c2) current_row.append(min(insertions, deletions, substitutions)) previous_row = current_row return previous_row[-1] 二、跨模态推理评估 图文推理任务 class CrossModalReasoningEvaluator: """跨模态推理评估""" TASK_TYPES = [ "visual_entailment", # 视觉蕴含:图→文 是否支持 "visual_reasoning", # 视觉推理:基于图的逻辑推理 "image_text_matching", # 图文匹配 "visual_commonsense", # 视觉常识推理 "multimodal_cot", # 多模态链式推理 ] def evaluate_visual_entailment(self, predictions: list[dict]) -> dict: """ 视觉蕴含:判断文本假设是否被图像支持 标签: entailment / neutral / contradiction """ from sklearn.metrics import classification_report, accuracy_score labels = ["entailment", "neutral", "contradiction"] y_true = [p["gt_label"] for p in predictions] y_pred = [p["pred_label"] for p in predictions] report = classification_report(y_true, y_pred, labels=labels, output_dict=True) return { "accuracy": accuracy_score(y_true, y_pred), "per_class": {l: report[l] for l in labels}, } def evaluate_image_text_matching(self, predictions: list[dict]) -> dict: """ 图文匹配:给定图片和多个文本,选择最匹配的 """ correct = 0 for pred in predictions: if pred["pred_match"] == pred["gt_match"]: correct += 1 return {"accuracy": correct / len(predictions) if predictions else 0} def evaluate_multimodal_cot(self, predictions: list[dict]) -> dict: """ 多模态链式推理:评估推理步骤和最终答案 """ results = [] for pred in predictions: # 评估最终答案 answer_correct = self._check_answer(pred["pred_answer"], pred["gt_answer"]) # 评估推理步骤(使用 LLM-as-Judge) reasoning_score = self._evaluate_reasoning( pred["image_description"], pred["reasoning_steps"], pred["pred_answer"] ) results.append({ "answer_correct": answer_correct, "reasoning_score": reasoning_score, }) return { "answer_accuracy": sum(r["answer_correct"] for r in results) / len(results), "avg_reasoning_score": sum(r["reasoning_score"] for r in results) / len(results), } def _check_answer(self, pred: str, gt: str) -> bool: pred_clean = pred.strip().lower() gt_clean = gt.strip().lower() return gt_clean in pred_clean or pred_clean == gt_clean def _evaluate_reasoning(self, image_desc: str, reasoning: str, answer: str) -> float: """使用 LLM 评估推理质量""" prompt = f"""请评估以下多模态推理的质量(0-10分): 图像描述:{image_desc} 推理过程:{reasoning} 最终答案:{answer} 评估维度: - 推理是否基于图像信息 - 逻辑是否连贯 - 是否有跳步或错误 请输出一个数字(0-10)。""" # 调用 LLM 评估 score = call_llm("gpt-4o", prompt) try: return float(score.strip()) / 10.0 except ValueError: return 0.5 主流多模态基准对比 基准 评估能力 任务数 模态 特点 VQAv2 视觉问答 1.1M 图+文 经典 VQA 基准 GQA 场景图推理 22M 图+文 结构化推理 MMBench 综合多模态 4K+ 图+文 多维能力评估 MMMU 学科多模态 11.5K 图+文 大学级别学科 MathVista 数学视觉推理 6K+ 图+文 数学+视觉 MMMU-Health 医学多模态 1K+ 图+文 医学领域 VideoMME 视频理解 900 视频+文 长视频理解 SEED-Bench 多场景理解 19K 图/视频+文 多模态多场景 三、图文一致性评估 生成图像的文本一致性 当模型从文本生成图像(或反向)时,需要评估跨模态的一致性: ...

2026-06-25 · 8 min · 1589 words · AI 实战派
llm benchmark 2026 comprehensive

2026 LLM Benchmark 全面解读:MMLU/GPQA/SWE-Bench 谁还有效

引言 2026 年,大语言模型的能力已逼近甚至超越人类专家水平,但「如何衡量模型到底有多强」这个问题反而变得越来越棘手。MMLU 接近饱和、GPQA 区分度下降、SWE-Bench 被质疑数据泄漏——Benchmark 信任危机正在加剧。本文系统盘点主流 Benchmark 的现状、问题与新方向。 一、主流 Benchmark 盘点 1.1 MMLU / MMLU-Pro MMLU(Massive Multitask Language Understanding)涵盖 57 个学科的多选题,曾是衡量模型知识广度的金标准。2026 年现状: 模型 MMLU 得分 MMLU-Pro 得分 GPT-4o (2024) 88.7% 74.7% Claude 3.5 Sonnet 88.3% 78.0% GPT-5 (2026) 92.1% 85.3% Gemini 2.5 Ultra 91.8% 84.9% DeepSeek V3.5 90.4% 82.1% 问题: 头部模型在 MMLU 上已超过 90%,区分度严重不足。MMLU-Pro 将选项扩展到 10 个并引入更复杂的推理题,但头部模型仍在快速趋近饱和。 数据污染: 研究表明,MMLU 的部分题目出现在 Common Crawl 等公开语料中,模型可能通过记忆而非理解作答。MMLU-Pro 虽然缓解了部分问题,但仍有约 12% 的题目可在公开数据中找到高相似度匹配。 1.2 GPQA(Google-Proof Q&A) GPQA 由博士级专家出题,设计初衷是「无法通过 Google 搜索直接回答」。分为 GPQA-Main、GPQA-Extended 和 GPQA-Diamond 三个子集。 ...

2026-06-25 · 2 min · 387 words · AI 实战派
agent eval framework

Agent 评估框架:如何衡量 Agent 的真正能力

引言 2026 年,AI Agent 已从简单的对话助手演进为能自主规划、调用工具、执行多步骤任务的复杂系统。但 Agent 的评估远比 LLM 困难——同一个任务可能有无数条正确路径,中间步骤的错误可能被后续步骤弥补,而「完成任务」的定义本身也充满模糊性。本文系统梳理 Agent 评估的方法论与框架。 一、Agent 评估的核心难点 1.1 与传统 LLM 评估的本质区别 维度 LLM 评估 Agent 评估 输入 单轮/多轮 prompt 任务描述 + 环境 + 工具集 输出 文本回答 多种模态(文本、代码、文件、API 调用) 评估单元 单次回答 多步骤轨迹 正确性 可对照参考答案 路径多样,无唯一正确答案 环境依赖 无 依赖外部环境状态 成本 低 高(每步消耗 tokens + 工具调用) 可复现性 高 低(环境状态变化) 1.2 核心挑战 挑战一:路径多样性。 「订一张明天去上海的机票」这个任务,Agent 可以先查日历确认日期、再搜索航班、最后预订;也可以直接搜索航班并预订。路径不同但都正确。 挑战二:中间错误的可恢复性。 Agent 在第 2 步调错了 API,但在第 3 步意识到并纠正了——这算不算错误?如何量化? 挑战三:环境不可复现。 Agent 调用的 API 返回值可能随时间变化,同一任务在不同时间执行可能得到不同结果。 ...

2026-06-25 · 3 min · 594 words · AI 实战派
鲁ICP备2026018361号