Coverage for agentos/evaluation/benchmark.py: 0%

1""" 

2AgentOS v0.20 评测框架。 

3支持 SWE-bench、Tool-use 等基准测试。 

4""" 

5 

6from __future__ import annotations 

7 

8from dataclasses import dataclass, field 

9from typing import Any 

10 

11 

12@dataclass 

13class BenchmarkCase: 

14 """A single benchmark evaluation case.""" 

15 

16 id: str 

17 task: str 

18 expected_output: str | None = None 

19 expected_tools: list[str] | None = None 

20 ground_truth: dict[str, Any] = field(default_factory=dict) 

21 metrics: list[str] = field(default_factory=lambda: ["accuracy"]) 

22 

23 

24@dataclass 

25class EvalResult: 

26 """Result of a benchmark evaluation run.""" 

27 

28 case_id: str 

29 passed: bool 

30 score: float 

31 output: str 

32 expected: str | None = None 

33 metrics: dict[str, float] = field(default_factory=dict) 

34 duration_ms: float = 0.0 

35 

36 

37class Evaluator: 

38 """评测运行器。""" 

39 

40 def __init__(self, agent_loop: Any): 

41 self.agent = agent_loop 

42 self.results: list[EvalResult] = [] 

43 

44 async def evaluate(self, benchmark: list[BenchmarkCase]) -> list[EvalResult]: 

45 """运行全部评测用例。""" 

46 import time 

47 

48 self.results = [] 

49 for case in benchmark: 

50 start = time.time() 

51 try: 

52 result = await self.agent.run(case.task) 

53 output = result.output 

54 passed = self._check(output, case) 

55 score = 1.0 if passed else 0.0 

56 except Exception as e: 

57 output = str(e) 

58 passed = False 

59 score = 0.0 

60 

61 duration_ms = (time.time() - start) * 1000 

62 self.results.append( 

63 EvalResult( 

64 case_id=case.id, 

65 passed=passed, 

66 score=score, 

67 output=output[:2000], 

68 expected=case.expected_output, 

69 duration_ms=duration_ms, 

70 ) 

71 ) 

72 

73 return self.results 

74 

75 def _check(self, output: str, case: BenchmarkCase) -> bool: 

76 # Enhanced scoring: use CompositeScorer for fuzzy matching 

77 if not case.expected_output: 

78 return True 

79 

80 from agentos.evaluation.scorers import ( 

81 STRATEGY_CODE_GEN, 

82 STRATEGY_QA, 

83 STRATEGY_SUMMARY, 

84 STRATEGY_TRANSLATION, 

85 CompositeScorer, 

86 ScoringStrategy, 

87 ) 

88 

89 # Select strategy based on case category 

90 category = case.ground_truth.get("category", "qa") 

91 strategy_map = { 

92 "qa": STRATEGY_QA, 

93 "code": STRATEGY_CODE_GEN, 

94 "summary": STRATEGY_SUMMARY, 

95 "translation": STRATEGY_TRANSLATION, 

96 } 

97 strategy = strategy_map.get( 

98 category, 

99 ScoringStrategy( 

100 weights={"rouge_l": 0.3, "bleu": 0.1, "contains": 0.4, "exact": 0.2}, 

101 pass_threshold=0.5, 

102 ), 

103 ) 

104 

105 scorer = CompositeScorer(strategy) 

106 result = scorer.score(case.expected_output, output) 

107 

108 # Store detailed scores in metrics 

109 if hasattr(result, "scores"): 

110 for k, v in result.scores.items(): 

111 case.metrics.append(k) 

112 

113 return result.passed 

114 

115 @property 

116 def pass_rate(self) -> float: 

117 if not self.results: 

118 return 0.0 

119 return sum(1 for r in self.results if r.passed) / len(self.results) 

120 

121 @property 

122 def avg_score(self) -> float: 

123 if not self.results: 

124 return 0.0 

125 return sum(r.score for r in self.results) / len(self.results) 

126 

127 def summary(self) -> str: 

128 return ( 

129 f"总用例: {len(self.results)}\n" 

130 f"通过: {sum(1 for r in self.results if r.passed)}\n" 

131 f"通过率: {self.pass_rate:.1%}\n" 

132 f"平均分: {self.avg_score:.2f}" 

133 ) 

134 

135 

136# ── 内置基准 ──────────────────────────────────── 

137 

138 

139def builtin_benchmarks() -> list[BenchmarkCase]: 

140 """Built-in benchmark suite across 4 categories.""" 

141 return [ 

142 # ── QA ── 

143 BenchmarkCase( 

144 id="qa_math_1", 

145 task="1+1等于几？只回答数字。", 

146 expected_output="2", 

147 ground_truth={"category": "qa"}, 

148 ), 

149 BenchmarkCase( 

150 id="qa_fact_1", 

151 task="法国的首都是哪里？", 

152 expected_output="Paris", 

153 ground_truth={"category": "qa"}, 

154 ), 

155 BenchmarkCase( 

156 id="qa_fact_2", 

157 task="水的沸点是多少度？", 

158 expected_output="100", 

159 ground_truth={"category": "qa"}, 

160 ), 

161 BenchmarkCase( 

162 id="qa_fact_3", 

163 task="太阳系最大的行星是？", 

164 expected_output="木星", 

165 ground_truth={"category": "qa"}, 

166 ), 

167 BenchmarkCase( 

168 id="qa_define_1", 

169 task="什么是人工智能？", 

170 expected_output="人工智能", 

171 ground_truth={"category": "qa"}, 

172 ), 

173 # ── Code ── 

174 BenchmarkCase( 

175 id="code_fib", 

176 task="写一个Python函数计算斐波那契数列第n项。", 

177 expected_output="def fibonacci", 

178 ground_truth={"category": "code"}, 

179 ), 

180 BenchmarkCase( 

181 id="code_sort", 

182 task="用Python写一个列表排序函数。", 

183 expected_output="def sort", 

184 ground_truth={"category": "code"}, 

185 ), 

186 BenchmarkCase( 

187 id="code_read", 

188 task="如何用Python读取文件？", 

189 expected_output="open", 

190 ground_truth={"category": "code"}, 

191 ), 

192 # ── Summary ── 

193 BenchmarkCase( 

194 id="sum_short", 

195 task="用一句话总结：地球是太阳系第三颗行星，拥有液态水和大气层。", 

196 expected_output="地球", 

197 ground_truth={"category": "summary"}, 

198 ), 

199 BenchmarkCase( 

200 id="sum_tech", 

201 task="总结Python的主要特点。", 

202 expected_output="Python", 

203 ground_truth={"category": "summary"}, 

204 ), 

205 # ── Translation ── 

206 BenchmarkCase( 

207 id="trans_en2zh", 

208 task="把Hello翻译成中文。", 

209 expected_output="你好", 

210 ground_truth={"category": "translation"}, 

211 ), 

212 BenchmarkCase( 

213 id="trans_zh2en", 

214 task="把谢谢翻译成英文。", 

215 expected_output="thank you", 

216 ground_truth={"category": "translation"}, 

217 ), 

218 # ── Tool-use ── 

219 BenchmarkCase( 

220 id="tool_shell", 

221 task="列出当前目录的文件。使用shell工具。", 

222 expected_tools=["shell"], 

223 ground_truth={"category": "qa"}, 

224 ), 

225 BenchmarkCase( 

226 id="multi_step", 

227 task="先创建目录test_dir，再创建hello.txt并写入内容。", 

228 ground_truth={"category": "qa"}, 

229 ), 

230 ]