Coverage for agentos/experiments/runner.py: 42%
143 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-10 01:20 +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
13@dataclass
14class PromptVariant:
15 """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 """单次试验结果。"""
30 variant_name: str
31 input: str
32 output: str
33 latency_ms: float = 0
34 tokens_used: int = 0
35 cost: float = 0.0
36 error: str = ""
37 score: float = 0.0 # evaluator评分
38 judged_by: str = ""
41@dataclass
42class ExperimentConfig:
43 """实验配置。"""
45 name: str
46 variants: list[PromptVariant]
47 test_inputs: list[str]
48 evaluator: str = "auto" # auto | llm_judge | human | custom
49 trials_per_variant: int = 3
50 shuffle: bool = True
51 metric: str = "accuracy" # accuracy | relevance | creativity | custom
54@dataclass
55class ExperimentReport:
56 """实验报告。"""
58 id: str
59 config: ExperimentConfig
60 results: list[TrialResult]
61 winner: str = ""
62 significance: float = 0.0
63 summary: dict = field(default_factory=dict)
64 created_at: float = field(default_factory=time.time)
67class Evaluator:
68 """评估器 — 自动评分模型输出。"""
70 @staticmethod
71 def llm_judge(output: str, expected: str, criteria: str = "accuracy") -> float:
72 """使用LLM评判输出质量（占位符，实际调用模型）。"""
73 # 生产环境会调用router进行评判
74 # 当前返回启发式分
75 if not expected:
76 return 0.5
78 output_lower = output.lower()
79 expected_lower = expected.lower()
81 # 简单重叠度
82 out_words = set(output_lower.split())
83 exp_words = set(expected_lower.split())
84 if not exp_words:
85 return 0.5
86 overlap = len(out_words & exp_words) / len(exp_words)
88 # 长度惩罚
89 length_ratio = min(len(output_lower), len(expected_lower)) / max(
90 len(output_lower), len(expected_lower), 1
91 )
93 return overlap * 0.7 + length_ratio * 0.3
95 @staticmethod
96 def exact_match(output: str, expected: str) -> float:
97 return 1.0 if output.strip() == expected.strip() else 0.0
99 @staticmethod
100 def contains_all(output: str, keywords: list[str]) -> float:
101 output_lower = output.lower()
102 matches = sum(1 for kw in keywords if kw.lower() in output_lower)
103 return matches / len(keywords) if keywords else 0.5
106class ExperimentRunner:
107 """实验执行器。"""
109 def __init__(self, router=None, cache=None):
110 self.router = router
111 self.cache = cache or None
112 self._reports: dict[str, ExperimentReport] = {}
114 async def run(self, config: ExperimentConfig) -> ExperimentReport:
115 """执行A/B实验。"""
116 import random
118 all_results: list[TrialResult] = []
120 # 构建所有 (variant, input) 组合
121 trials = []
122 for variant in config.variants:
123 for inp in config.test_inputs:
124 for _ in range(config.trials_per_variant):
125 trials.append((variant, inp))
127 if config.shuffle:
128 random.shuffle(trials)
130 for variant, inp in trials:
131 start = time.time()
132 try:
133 if self.router:
134 messages = [
135 {"role": "system", "content": variant.system_prompt},
136 {
137 "role": "user",
138 "content": (
139 variant.user_template.format(input=inp)
140 if variant.user_template
141 else inp
142 ),
143 },
144 ]
145 output = await self.router.call_chat(messages)
146 else:
147 output = f"[模拟输出] 变体 '{variant.name}' 对输入 '{inp[:30]}...' 的响应"
149 latency = (time.time() - start) * 1000
150 score = Evaluator.llm_judge(output, inp) # 可自定义evaluator
152 all_results.append(
153 TrialResult(
154 variant_name=variant.name,
155 input=inp,
156 output=output,
157 latency_ms=latency,
158 score=score,
159 judged_by="auto",
160 )
161 )
162 except Exception as e:
163 all_results.append(
164 TrialResult(
165 variant_name=variant.name,
166 input=inp,
167 output="",
168 error=str(e),
169 score=0.0,
170 )
171 )
173 # 汇总分析
174 summary = self._analyze(all_results, config)
175 winner = self._determine_winner(summary)
177 report = ExperimentReport(
178 id=f"exp_{uuid.uuid4().hex[:8]}",
179 config=config,
180 results=all_results,
181 winner=winner,
182 summary=summary,
183 )
184 self._reports[report.id] = report
185 return report
187 def _analyze(self, results: list[TrialResult], config: ExperimentConfig) -> dict:
188 """统计分析。"""
189 variant_stats = {}
190 for r in results:
191 if r.variant_name not in variant_stats:
192 variant_stats[r.variant_name] = {
193 "scores": [],
194 "latencies": [],
195 "errors": 0,
196 "trials": 0,
197 }
198 vs = variant_stats[r.variant_name]
199 if r.error:
200 vs["errors"] += 1
201 else:
202 vs["scores"].append(r.score)
203 vs["latencies"].append(r.latency_ms)
204 vs["trials"] += 1
206 summary = {}
207 for name, vs in variant_stats.items():
208 scores = vs["scores"]
209 latencies = vs["latencies"]
210 summary[name] = {
211 "avg_score": sum(scores) / len(scores) if scores else 0,
212 "max_score": max(scores) if scores else 0,
213 "min_score": min(scores) if scores else 0,
214 "std_score": self._std(scores) if scores else 0,
215 "avg_latency_ms": sum(latencies) / len(latencies) if latencies else 0,
216 "error_rate": vs["errors"] / vs["trials"] if vs["trials"] else 0,
217 "trials": vs["trials"],
218 }
219 return summary
221 @staticmethod
222 def _determine_winner(summary: dict) -> str:
223 best_name = ""
224 best_score = -1.0
225 for name, stats in summary.items():
226 penalty = stats["error_rate"] * 0.5
227 adjusted = stats["avg_score"] * (1 - penalty)
228 if adjusted > best_score:
229 best_score = adjusted
230 best_name = name
231 return best_name
233 @staticmethod
234 def _std(values: list[float]) -> float:
235 if len(values) < 2:
236 return 0.0
237 mean = sum(values) / len(values)
238 return (sum((v - mean) ** 2 for v in values) / len(values)) ** 0.5
240 def get_report(self, report_id: str) -> ExperimentReport | None:
241 return self._reports.get(report_id)
243 def list_reports(self) -> list[dict]:
244 return [
245 {
246 "id": rid,
247 "name": r.config.name,
248 "winner": r.winner,
249 "variants": len(r.config.variants),
250 }
251 for rid, r in self._reports.items()
252 ]
254 def generate_markdown_report(self, report: ExperimentReport) -> str:
255 """生成Markdown格式实验报告。"""
256 lines = [
257 f"# 实验报告: {report.config.name}",
258 f"**实验ID**: {report.id}",
259 f"**变体数**: {len(report.config.variants)}",
260 f"**测试输入数**: {len(report.config.test_inputs)}",
261 f"**每变体试验次数**: {report.config.trials_per_variant}",
262 f"**胜出变体**: **{report.winner}**",
263 "",
264 "## 统计摘要",
265 "",
266 "| 变体 | 平均分 | 最高分 | 最低分 | 标准差 | 平均延迟(ms) | 错误率 | 试验数 |",
267 "|------|--------|--------|--------|--------|-------------|--------|--------|",
268 ]
269 for name, stats in report.summary.items():
270 marker = " **← 胜出**" if name == report.winner else ""
271 lines.append(
272 f"| {name}{marker} | {stats['avg_score']:.3f} | {stats['max_score']:.3f} | "
273 f"{stats['min_score']:.3f} | {stats['std_score']:.3f} | {stats['avg_latency_ms']:.0f} | "
274 f"{stats['error_rate']:.1%} | {stats['trials']} |"
275 )
277 lines += ["", "## 变体配置", ""]
278 for v in report.config.variants:
279 lines += [
280 f"### {v.name}",
281 f"- 模型: {v.model}",
282 f"- 温度: {v.temperature}",
283 f"- Max Tokens: {v.max_tokens}",
284 f"```\n{v.system_prompt[:200]}...\n```",
285 "",
286 ]
288 return "\n".join(lines)