Coverage for agentos/evaluation/scorers.py: 25%
176 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-08 01:44 +0800
1"""
2AgentOS v0.70 — 评测打分系统。
3基因来源: ROUGE/BLEU 经典算法 + 语义相似度
5评分策略:
6- ROUGE-L: 最长公共子序列召回率 (摘要质量)
7- BLEU: n-gram精确率 (翻译质量)
8- Semantic: 基于embedding的语义相似度
9- Exact: 精确匹配
10- Contains: 包含匹配
11"""
13from __future__ import annotations
15import math
16import re
17from collections import Counter
18from dataclasses import dataclass, field
19from typing import Any
21# ── ROUGE-L ─────────────────────────────────────
24def _lcs_length(x: list, y: list) -> int:
25 """最长公共子序列长度（DP优化版）。"""
26 if len(x) < len(y):
27 x, y = y, x
28 prev = [0] * (len(y) + 1)
29 for i in range(1, len(x) + 1):
30 curr = [0] * (len(y) + 1)
31 for j in range(1, len(y) + 1):
32 if x[i - 1] == y[j - 1]:
33 curr[j] = prev[j - 1] + 1
34 else:
35 curr[j] = max(prev[j], curr[j - 1])
36 prev = curr
37 return prev[len(y)]
40def rouge_l(reference: str, candidate: str) -> float:
41 """ROUGE-L F1 score (character-level)。"""
42 if not reference or not candidate:
43 return 0.0
45 ref_chars = list(reference)
46 cand_chars = list(candidate)
47 lcs = _lcs_length(ref_chars, cand_chars)
49 if len(cand_chars) == 0 or len(ref_chars) == 0:
50 return 0.0
52 recall = lcs / len(ref_chars)
53 precision = lcs / len(cand_chars)
55 if recall + precision == 0:
56 return 0.0
57 return 2 * recall * precision / (recall + precision)
60# ── BLEU ────────────────────────────────────────
63def _ngrams(tokens: list[str], n: int) -> Counter:
64 return Counter(tuple(tokens[i : i + n]) for i in range(len(tokens) - n + 1))
67def bleu(reference: str, candidate: str, max_n: int = 4, smoothing: bool = True) -> float:
68 """BLEU score (token-level, with smoothing for short texts)."""
69 ref_tokens = _tokenize(reference)
70 cand_tokens = _tokenize(candidate)
72 if not cand_tokens or not ref_tokens:
73 return 0.0
75 precisions = []
76 for n in range(1, max_n + 1):
77 ref_ngrams = _ngrams(ref_tokens, n)
78 cand_ngrams = _ngrams(cand_tokens, n)
80 if not cand_ngrams:
81 if smoothing:
82 precisions.append(1.0 / (2**n)) # Laplace-like decay
83 else:
84 precisions.append(0.0)
85 continue
87 clipped = sum(min(cand_ngrams[ng], ref_ngrams.get(ng, 0)) for ng in cand_ngrams)
88 prec = clipped / sum(cand_ngrams.values())
89 precisions.append(prec)
91 if any(p == 0 for p in precisions):
92 if smoothing:
93 # Method 1 smoothing: replace zeros with small epsilon
94 precisions = [p if p > 0 else 1.0 / (2**i) for i, p in enumerate(precisions)]
95 else:
96 return 0.0
98 # Brevity penalty
99 bp = min(1.0, math.exp(1 - len(ref_tokens) / max(len(cand_tokens), 1)))
101 # Geometric mean of n-gram precisions
102 log_sum = sum(math.log(p) for p in precisions)
103 return bp * math.exp(log_sum / max_n)
106def _tokenize(text: str) -> list[str]:
107 """英文分词 + 数字/标点分离。"""
108 # Split on whitespace, keep punctuation as separate tokens for Chinese
109 text = text.lower()
110 # For Chinese: character-level
111 if re.search(r"[\u4e00-\u9fff]", text):
112 tokens = []
113 for ch in text:
114 if ch.strip():
115 tokens.append(ch)
116 return tokens
117 # English
118 return re.findall(r"\w+|[^\w\s]", text)
121# ── Semantic Similarity ─────────────────────────
124def semantic_similarity(candidate: str, reference: str, embedder: Any = None) -> float:
125 """
126 基于embedding的语义相似度（cosine similarity）。
127 需要传入embedder实例或使用默认LocalEmbedder。
128 Falls back to character Jaccard similarity if embedder unavailable.
129 """
130 if not candidate or not reference:
131 return 0.0
133 if embedder is None:
134 from agentos.cache.embedder import LocalEmbedder
136 embedder = LocalEmbedder()
138 try:
139 emb_cand = embedder.embed(candidate)
140 emb_ref = embedder.embed(reference)
141 from agentos.cache.embedder import cosine_similarity as cos_sim
143 return float(cos_sim(emb_cand, emb_ref))
144 except Exception:
145 # Fallback: character-level Jaccard similarity
146 set_a = set(candidate.lower())
147 set_b = set(reference.lower())
148 if not set_a or not set_b:
149 return 0.0
150 intersection = set_a & set_b
151 union = set_a | set_b
152 return len(intersection) / len(union) if union else 0.0
155# ── Exact / Contains ────────────────────────────
158def exact_match(reference: str, candidate: str) -> float:
159 """精确匹配：返回 0.0 或 1.0。"""
160 return 1.0 if reference.strip() == candidate.strip() else 0.0
163def contains_match(reference: str, candidate: str) -> float:
164 """候选文本是否包含参考文本（忽略大小写）。"""
165 return 1.0 if reference.lower() in candidate.lower() else 0.0
168# ── Composite Scorer ────────────────────────────
171@dataclass
172class ScoringStrategy:
173 """评分配置策略。"""
175 name: str = "composite"
176 weights: dict[str, float] = field(
177 default_factory=lambda: {
178 "rouge_l": 0.3,
179 "bleu": 0.2,
180 "exact": 0.2,
181 "contains": 0.3,
182 }
183 )
184 pass_threshold: float = 0.6
187@dataclass
188class ScoreResult:
189 """评分结果。"""
191 reference: str
192 candidate: str
193 scores: dict[str, float] = field(default_factory=dict)
194 weighted_score: float = 0.0
195 passed: bool = False
196 details: str = ""
198 def to_dict(self) -> dict:
199 return {
200 "scores": self.scores,
201 "weighted_score": round(self.weighted_score, 4),
202 "passed": self.passed,
203 "details": self.details,
204 }
207class CompositeScorer:
208 """
209 复合评分器 — 多策略加权。
210 """
212 def __init__(self, strategy: ScoringStrategy | None = None):
213 self.strategy = strategy or ScoringStrategy()
215 def score(self, reference: str, candidate: str, embedder: Any = None) -> ScoreResult:
216 """对候选文本打分。"""
217 scores: dict[str, float] = {}
219 # ROUGE-L
220 if "rouge_l" in self.strategy.weights:
221 scores["rouge_l"] = rouge_l(reference, candidate)
223 # BLEU
224 if "bleu" in self.strategy.weights:
225 scores["bleu"] = bleu(reference, candidate)
227 # Exact
228 if "exact" in self.strategy.weights:
229 scores["exact"] = exact_match(reference, candidate)
231 # Contains
232 if "contains" in self.strategy.weights:
233 scores["contains"] = contains_match(reference, candidate)
235 # Semantic
236 if "semantic" in self.strategy.weights or self.strategy.weights.get("semantic", 0) > 0:
237 scores["semantic"] = semantic_similarity(candidate, reference, embedder)
239 # Weighted
240 weighted = sum(scores.get(k, 0) * w for k, w in self.strategy.weights.items())
242 passed = weighted >= self.strategy.pass_threshold
243 details = ", ".join(f"{k}={v:.3f}" for k, v in scores.items())
245 return ScoreResult(
246 reference=reference,
247 candidate=candidate,
248 scores=scores,
249 weighted_score=weighted,
250 passed=passed,
251 details=details,
252 )
254 def batch_score(
255 self,
256 pairs: list[tuple[str, str]],
257 embedder: Any = None,
258 ) -> list[ScoreResult]:
259 """批量评分。"""
260 return [self.score(ref, cand, embedder) for ref, cand in pairs]
263# ── Pre-built Strategies ────────────────────────
265STRATEGY_CODE_GEN = ScoringStrategy(
266 name="code_generation",
267 weights={"rouge_l": 0.1, "bleu": 0.1, "exact": 0.3, "contains": 0.5},
268 pass_threshold=0.5,
269)
271STRATEGY_QA = ScoringStrategy(
272 name="question_answering",
273 weights={"rouge_l": 0.3, "contains": 0.5, "exact": 0.2},
274 pass_threshold=0.5,
275)
277STRATEGY_SUMMARY = ScoringStrategy(
278 name="summarization",
279 weights={"rouge_l": 0.6, "bleu": 0.1, "semantic": 0.3},
280 pass_threshold=0.25,
281)
283STRATEGY_TRANSLATION = ScoringStrategy(
284 name="translation",
285 weights={"bleu": 0.6, "rouge_l": 0.2, "semantic": 0.2},
286 pass_threshold=0.30,
287)
290# ── LLM‑as‑Judge ────────────────────────────────
292_JUDGE_PROMPT = """You are an evaluation judge. Grade the following answer against the reference.
293Output ONLY a number between 0.0 and 1.0 and a one-sentence reason.
295Task: {task}
296Reference (expected): {reference}
297Candidate (actual): {candidate}
299Score (0.0-1.0):
300Reason:"""
303def llm_judge(
304 reference: str,
305 candidate: str,
306 task: str = "general",
307 model: str = "gpt-4o-mini",
308 api_key: str = "",
309) -> float:
310 """
311 LLM‑as‑Judge: 用 LLM 评估候选答案与参考答案的一致性。
312 需要 OPENAI_API_KEY (或兼容 endpoint)。
313 Returns 0.0 on any error / no key.
314 """
315 if not api_key:
316 import os
318 api_key = os.environ.get("OPENAI_API_KEY", "")
320 if not api_key:
321 return 0.0
323 prompt = _JUDGE_PROMPT.format(task=task, reference=reference, candidate=candidate)
325 try:
326 import requests
328 resp = requests.post(
329 "https://api.openai.com/v1/chat/completions",
330 headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
331 json={
332 "model": model,
333 "messages": [{"role": "user", "content": prompt}],
334 "temperature": 0.0,
335 "max_tokens": 50,
336 },
337 timeout=15,
338 )
339 if resp.status_code != 200:
340 return 0.0
342 data = resp.json()
343 text = data["choices"][0]["message"]["content"].strip()
345 # Extract first float from response
346 import re as _re
348 m = _re.search(r"(\d+\.?\d*)", text)
349 if m:
350 return max(0.0, min(1.0, float(m.group(1))))
352 return 0.0
353 except Exception:
354 return 0.0
357# ── Strategy w/ LLM‑Judge ───────────────────────
359STRATEGY_QA_JUDGE = ScoringStrategy(
360 name="qa_with_judge",
361 weights={"rouge_l": 0.2, "contains": 0.3, "exact": 0.1, "judge": 0.4},
362 pass_threshold=0.55,
363)
365STRATEGY_SUMMARY_JUDGE = ScoringStrategy(
366 name="summary_with_judge",
367 weights={"rouge_l": 0.3, "bleu": 0.1, "judge": 0.6},
368 pass_threshold=0.55,
369)
371STRATEGY_CODE_JUDGE = ScoringStrategy(
372 name="code_with_judge",
373 weights={"rouge_l": 0.05, "bleu": 0.05, "exact": 0.2, "contains": 0.3, "judge": 0.4},
374 pass_threshold=0.55,
375)
378class CompositeScorerV2(CompositeScorer):
379 """v2 scorer with optional LLM‑as‑Judge."""
381 def __init__(self, strategy: ScoringStrategy | None = None, llm_model: str = "gpt-4o-mini"):
382 super().__init__(strategy)
383 self._llm_model = llm_model
385 def score(
386 self, reference: str, candidate: str, embedder: Any = None, task: str = "general"
387 ) -> ScoreResult:
388 scores: dict[str, float] = {}
390 if "rouge_l" in self.strategy.weights:
391 scores["rouge_l"] = rouge_l(reference, candidate)
392 if "bleu" in self.strategy.weights:
393 scores["bleu"] = bleu(reference, candidate)
394 if "exact" in self.strategy.weights:
395 scores["exact"] = exact_match(reference, candidate)
396 if "contains" in self.strategy.weights:
397 scores["contains"] = contains_match(reference, candidate)
398 if "semantic" in self.strategy.weights:
399 scores["semantic"] = semantic_similarity(candidate, reference, embedder)
400 if "judge" in self.strategy.weights:
401 scores["judge"] = llm_judge(reference, candidate, task=task, model=self._llm_model)
403 weighted = sum(scores.get(k, 0) * w for k, w in self.strategy.weights.items())
404 passed = weighted >= self.strategy.pass_threshold
405 details = ", ".join(f"{k}={v:.3f}" for k, v in scores.items())
407 return ScoreResult(
408 reference=reference,
409 candidate=candidate,
410 scores=scores,
411 weighted_score=weighted,
412 passed=passed,
413 details=details,
414 )