Coverage for agentos/experiments/runner.py: 42%
144 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +0800
1"""
2AgentOS v0.40 Experiments — A/B测试与Prompt实验框架。
3支持：Prompt变体对比、A/B/n测试、结果统计显著性分析、实验报告生成。
4"""
6from __future__ import annotations
8import time
9import uuid
10from dataclasses import dataclass, field
11from typing import Optional
14@dataclass
15class PromptVariant:
16 """Prompt变体。"""
17 name: str
18 system_prompt: str
19 user_template: str = ""
20 model: str = "auto"
21 temperature: float = 0.7
22 max_tokens: int = 2048
23 metadata: dict = field(default_factory=dict)
26@dataclass
27class TrialResult:
28 """单次试验结果。"""
29 variant_name: str
30 input: str
31 output: str
32 latency_ms: float = 0
33 tokens_used: int = 0
34 cost: float = 0.0
35 error: str = ""
36 score: float = 0.0 # evaluator评分
37 judged_by: str = ""
40@dataclass
41class ExperimentConfig:
42 """实验配置。"""
43 name: str
44 variants: list[PromptVariant]
45 test_inputs: list[str]
46 evaluator: str = "auto" # auto | llm_judge | human | custom
47 trials_per_variant: int = 3
48 shuffle: bool = True
49 metric: str = "accuracy" # accuracy | relevance | creativity | custom
52@dataclass
53class ExperimentReport:
54 """实验报告。"""
55 id: str
56 config: ExperimentConfig
57 results: list[TrialResult]
58 winner: str = ""
59 significance: float = 0.0
60 summary: dict = field(default_factory=dict)
61 created_at: float = field(default_factory=time.time)
64class Evaluator:
65 """评估器 — 自动评分模型输出。"""
67 @staticmethod
68 def llm_judge(output: str, expected: str, criteria: str = "accuracy") -> float:
69 """使用LLM评判输出质量（占位符，实际调用模型）。"""
70 # 生产环境会调用router进行评判
71 # 当前返回启发式分
72 if not expected:
73 return 0.5
75 output_lower = output.lower()
76 expected_lower = expected.lower()
78 # 简单重叠度
79 out_words = set(output_lower.split())
80 exp_words = set(expected_lower.split())
81 if not exp_words:
82 return 0.5
83 overlap = len(out_words & exp_words) / len(exp_words)
85 # 长度惩罚
86 length_ratio = min(len(output_lower), len(expected_lower)) / max(len(output_lower), len(expected_lower), 1)
88 return overlap * 0.7 + length_ratio * 0.3
90 @staticmethod
91 def exact_match(output: str, expected: str) -> float:
92 return 1.0 if output.strip() == expected.strip() else 0.0
94 @staticmethod
95 def contains_all(output: str, keywords: list[str]) -> float:
96 output_lower = output.lower()
97 matches = sum(1 for kw in keywords if kw.lower() in output_lower)
98 return matches / len(keywords) if keywords else 0.5
101class ExperimentRunner:
102 """实验执行器。"""
104 def __init__(self, router=None, cache=None):
105 self.router = router
106 self.cache = cache or None
107 self._reports: dict[str, ExperimentReport] = {}
109 async def run(self, config: ExperimentConfig) -> ExperimentReport:
110 """执行A/B实验。"""
111 import random
112 all_results: list[TrialResult] = []
114 # 构建所有 (variant, input) 组合
115 trials = []
116 for variant in config.variants:
117 for inp in config.test_inputs:
118 for _ in range(config.trials_per_variant):
119 trials.append((variant, inp))
121 if config.shuffle:
122 random.shuffle(trials)
124 for variant, inp in trials:
125 start = time.time()
126 try:
127 if self.router:
128 messages = [
129 {"role": "system", "content": variant.system_prompt},
130 {"role": "user", "content": variant.user_template.format(input=inp) if variant.user_template else inp},
131 ]
132 output = await self.router.call_chat(messages)
133 else:
134 output = f"[模拟输出] 变体 '{variant.name}' 对输入 '{inp[:30]}...' 的响应"
136 latency = (time.time() - start) * 1000
137 score = Evaluator.llm_judge(output, inp) # 可自定义evaluator
139 all_results.append(TrialResult(
140 variant_name=variant.name,
141 input=inp,
142 output=output,
143 latency_ms=latency,
144 score=score,
145 judged_by="auto",
146 ))
147 except Exception as e:
148 all_results.append(TrialResult(
149 variant_name=variant.name, input=inp, output="",
150 error=str(e), score=0.0,
151 ))
153 # 汇总分析
154 summary = self._analyze(all_results, config)
155 winner = self._determine_winner(summary)
157 report = ExperimentReport(
158 id=f"exp_{uuid.uuid4().hex[:8]}",
159 config=config,
160 results=all_results,
161 winner=winner,
162 summary=summary,
163 )
164 self._reports[report.id] = report
165 return report
167 def _analyze(self, results: list[TrialResult], config: ExperimentConfig) -> dict:
168 """统计分析。"""
169 variant_stats = {}
170 for r in results:
171 if r.variant_name not in variant_stats:
172 variant_stats[r.variant_name] = {"scores": [], "latencies": [], "errors": 0, "trials": 0}
173 vs = variant_stats[r.variant_name]
174 if r.error:
175 vs["errors"] += 1
176 else:
177 vs["scores"].append(r.score)
178 vs["latencies"].append(r.latency_ms)
179 vs["trials"] += 1
181 summary = {}
182 for name, vs in variant_stats.items():
183 scores = vs["scores"]
184 latencies = vs["latencies"]
185 summary[name] = {
186 "avg_score": sum(scores) / len(scores) if scores else 0,
187 "max_score": max(scores) if scores else 0,
188 "min_score": min(scores) if scores else 0,
189 "std_score": self._std(scores) if scores else 0,
190 "avg_latency_ms": sum(latencies) / len(latencies) if latencies else 0,
191 "error_rate": vs["errors"] / vs["trials"] if vs["trials"] else 0,
192 "trials": vs["trials"],
193 }
194 return summary
196 @staticmethod
197 def _determine_winner(summary: dict) -> str:
198 best_name = ""
199 best_score = -1.0
200 for name, stats in summary.items():
201 penalty = stats["error_rate"] * 0.5
202 adjusted = stats["avg_score"] * (1 - penalty)
203 if adjusted > best_score:
204 best_score = adjusted
205 best_name = name
206 return best_name
208 @staticmethod
209 def _std(values: list[float]) -> float:
210 if len(values) < 2:
211 return 0.0
212 mean = sum(values) / len(values)
213 return (sum((v - mean) ** 2 for v in values) / len(values)) ** 0.5
215 def get_report(self, report_id: str) -> Optional[ExperimentReport]:
216 return self._reports.get(report_id)
218 def list_reports(self) -> list[dict]:
219 return [{"id": rid, "name": r.config.name, "winner": r.winner, "variants": len(r.config.variants)}
220 for rid, r in self._reports.items()]
222 def generate_markdown_report(self, report: ExperimentReport) -> str:
223 """生成Markdown格式实验报告。"""
224 lines = [
225 f"# 实验报告: {report.config.name}",
226 f"**实验ID**: {report.id}",
227 f"**变体数**: {len(report.config.variants)}",
228 f"**测试输入数**: {len(report.config.test_inputs)}",
229 f"**每变体试验次数**: {report.config.trials_per_variant}",
230 f"**胜出变体**: **{report.winner}**",
231 "",
232 "## 统计摘要",
233 "",
234 "| 变体 | 平均分 | 最高分 | 最低分 | 标准差 | 平均延迟(ms) | 错误率 | 试验数 |",
235 "|------|--------|--------|--------|--------|-------------|--------|--------|",
236 ]
237 for name, stats in report.summary.items():
238 marker = " **← 胜出**" if name == report.winner else ""
239 lines.append(
240 f"| {name}{marker} | {stats['avg_score']:.3f} | {stats['max_score']:.3f} | "
241 f"{stats['min_score']:.3f} | {stats['std_score']:.3f} | {stats['avg_latency_ms']:.0f} | "
242 f"{stats['error_rate']:.1%} | {stats['trials']} |"
243 )
245 lines += ["", "## 变体配置", ""]
246 for v in report.config.variants:
247 lines += [
248 f"### {v.name}",
249 f"- 模型: {v.model}",
250 f"- 温度: {v.temperature}",
251 f"- Max Tokens: {v.max_tokens}",
252 f"```\n{v.system_prompt[:200]}...\n```",
253 "",
254 ]
256 return "\n".join(lines)