Coverage for agentos/swarm/result_fusion.py: 16%

1""" 

2v1.9.4: LLM-as-Judge Result Fusion engine. 

3 

4Aggregates multiple agent outputs with weighted fusion, 

5confidence scoring, and LLM-as-Judge quality arbitration. 

6""" 

7 

8from __future__ import annotations 

9 

10from dataclasses import dataclass, field 

11from typing import Any 

12 

13import json as _json 

14 

15 

16_JUDGE_PROMPT = """You are an expert quality judge. Given a task and multiple candidate results, 

17select the best result or synthesize a combined result. 

18 

19Task: {task} 

20 

21Candidates: 

22{candidates} 

23 

24Instructions: 

251. Evaluate each candidate for correctness, completeness, and clarity 

262. If one candidate is clearly best, output: {{"action": "select", "best_index": N, "reason": "..."}} 

273. If candidates complement each other, output: {{"action": "merge", "merged": "...", "reason": "..."}} 

284. If all candidates are poor, output: {{"action": "reject", "reason": "..."}} 

29 

30Output ONLY the JSON object, no other text. 

31JSON:""" 

32 

33 

34@dataclass 

35class FusedResult: 

36 """Result of fusion operation.""" 

37 

38 merged: Any = None 

39 best_index: int = -1 

40 confidence: float = 0.0 

41 action: str = "none" # select | merge | reject 

42 reason: str = "" 

43 individual_scores: dict[str, float] = field(default_factory=dict) 

44 all_outputs: dict[str, Any] = field(default_factory=dict) 

45 

46 

47class ResultFusion: 

48 """LLM-as-Judge result aggregation engine. 

49 

50 Combines outputs from multiple agents with: 

51 - Weighted-vote aggregation 

52 - LLM-as-Judge for quality arbitration 

53 - Confidence scoring 

54 """ 

55 

56 def __init__( 

57 self, 

58 strategy: str = "auto", 

59 llm_model: str = "gpt-4o-mini", 

60 ): 

61 self._strategy = strategy 

62 self._llm_model = llm_model 

63 

64 def fuse( 

65 self, 

66 task: str, 

67 outputs: dict[str, Any], 

68 weights: dict[str, float] | None = None, 

69 ) -> FusedResult: 

70 """Fuse multiple agent outputs into a single result. 

71 

72 Args: 

73 task: Original task description 

74 outputs: Dict of agent_name -> agent_output 

75 weights: Optional dict of agent_name -> weight (default: equal) 

76 

77 Returns: 

78 FusedResult with merged output and confidence 

79 """ 

80 if not outputs: 

81 result = FusedResult(action="reject", reason="No outputs to fuse") 

82 result.individual_scores = {} 

83 result.all_outputs = {} 

84 return result 

85 

86 if len(outputs) == 1: 

87 name, value = next(iter(outputs.items())) 

88 result = FusedResult( 

89 merged=value, 

90 best_index=0, 

91 confidence=0.7, 

92 action="select", 

93 reason="Single output", 

94 individual_scores={name: 0.7}, 

95 all_outputs=outputs, 

96 ) 

97 return result 

98 

99 weights = weights or {k: 1.0 for k in outputs} 

100 

101 # Step 1: Compute individual scores 

102 scores = self._compute_scores(outputs, weights) 

103 

104 # Step 2: Try LLM judge for arbitration 

105 llm_result = self._llm_judge(task, outputs) 

106 if llm_result: 

107 return llm_result 

108 

109 # Step 3: Fallback — weighted aggregation 

110 return self._weighted_aggregate(outputs, scores) 

111 

112 def _compute_scores( 

113 self, 

114 outputs: dict[str, Any], 

115 weights: dict[str, float], 

116 ) -> dict[str, float]: 

117 """Score each output for quality heuristics.""" 

118 scores: dict[str, float] = {} 

119 for name, output in outputs.items(): 

120 base = float(weights.get(name, 1.0)) 

121 quality = self._quality_heuristic(output) 

122 scores[name] = round(base * quality, 3) 

123 return scores 

124 

125 def _quality_heuristic(self, output: Any) -> float: 

126 """Heuristic quality score based on output characteristics.""" 

127 score = 0.5 # baseline 

128 

129 text = str(output) if output is not None else "" 

130 

131 if not text: 

132 return 0.1 

133 

134 # Length heuristic: too short is suspicious, reasonable length is good 

135 length = len(text) 

136 if 100 < length < 2000: 

137 score += 0.15 

138 elif 50 <= length <= 100: 

139 score += 0.05 

140 elif length > 5000: 

141 score += 0.05 

142 

143 # Error patterns 

144 error_keywords = ["error", "exception", "traceback", "failed", "错误", "失败"] 

145 for kw in error_keywords: 

146 if kw in text: 

147 score -= 0.15 

148 break 

149 

150 # Structure bonus 

151 if any(marker in text for marker in ("```", "##", "# ", "**", "<table")): 

152 score += 0.1 

153 

154 # Confidence keywords 

155 confidence_keywords = ["recommend", "建议", "recommendation", "conclusion"] 

156 for kw in confidence_keywords: 

157 if kw in text: 

158 score += 0.05 

159 

160 return max(0.0, min(1.0, score)) 

161 

162 def _llm_judge( 

163 self, task: str, outputs: dict[str, Any] 

164 ) -> FusedResult | None: 

165 """Use LLM to judge and fuse results. Returns None on failure.""" 

166 try: 

167 import os 

168 api_key = os.environ.get("OPENAI_API_KEY", "") 

169 if not api_key: 

170 return None 

171 

172 candidates_str = "\n".join( 

173 f"[{i}] {name}: {str(output)[:300]}" 

174 for i, (name, output) in enumerate(outputs.items()) 

175 ) 

176 

177 prompt = _JUDGE_PROMPT.format( 

178 task=task, 

179 candidates=candidates_str, 

180 ) 

181 

182 import requests 

183 resp = requests.post( 

184 "https://api.openai.com/v1/chat/completions", 

185 headers={"Authorization": f"Bearer {api_key}", 

186 "Content-Type": "application/json"}, 

187 json={ 

188 "model": self._llm_model, 

189 "messages": [{"role": "user", "content": prompt}], 

190 "temperature": 0.0, 

191 "max_tokens": 500, 

192 }, 

193 timeout=30, 

194 ) 

195 if resp.status_code != 200: 

196 return None 

197 

198 text = resp.json()["choices"][0]["message"]["content"] 

199 

200 start = text.find("{") 

201 end = text.rfind("}") + 1 

202 if start == -1 or end == 0: 

203 return None 

204 

205 data = _json.loads(text[start:end]) 

206 action = data.get("action", "reject") 

207 

208 agent_names = list(outputs.keys()) 

209 

210 result = FusedResult() 

211 result.action = action 

212 result.reason = data.get("reason", "") 

213 result.all_outputs = { 

214 k: str(v)[:200] for k, v in outputs.items() 

215 } 

216 result.individual_scores = { 

217 k: 0.5 for k in outputs 

218 } 

219 

220 if action == "select": 

221 idx = int(data.get("best_index", 0)) 

222 idx = max(0, min(idx, len(agent_names) - 1)) 

223 result.best_index = idx 

224 result.merged = outputs[agent_names[idx]] 

225 result.confidence = 0.8 

226 elif action == "merge": 

227 result.merged = data.get("merged", "") 

228 result.confidence = 0.75 

229 result.best_index = -1 

230 else: # reject 

231 result.merged = None 

232 result.confidence = 0.0 

233 

234 return result 

235 except Exception: 

236 return None 

237 

238 def _weighted_aggregate( 

239 self, 

240 outputs: dict[str, Any], 

241 scores: dict[str, float], 

242 ) -> FusedResult: 

243 """Weighted vote aggregation fallback.""" 

244 max_score = max(scores.values()) if scores else 0.0 

245 if max_score == 0: 

246 return FusedResult(action="reject", reason="All outputs scored zero") 

247 

248 # Find best candidate 

249 best_name = max(scores, key=scores.get) # type: ignore[arg-type] 

250 

251 # Normalize scores to confidence 

252 total = sum(scores.values()) 

253 confidence = max_score / total if total > 0 else 0.3 

254 

255 # Check for consensus: if all string outputs are similar, merge them 

256 all_str = [str(v) for v in outputs.values()] 

257 consensus = self._check_consensus(all_str) 

258 

259 if consensus: 

260 return FusedResult( 

261 merged=outputs[best_name], 

262 best_index=list(outputs.keys()).index(best_name), 

263 confidence=confidence, 

264 action="select", 

265 reason="Consensus among outputs", 

266 individual_scores=scores, 

267 all_outputs={k: str(v)[:200] for k, v in outputs.items()}, 

268 ) 

269 

270 return FusedResult( 

271 merged=outputs[best_name], 

272 best_index=list(outputs.keys()).index(best_name), 

273 confidence=confidence, 

274 action="select", 

275 reason="Weighted voting (no consensus)", 

276 individual_scores=scores, 

277 all_outputs={k: str(v)[:200] for k, v in outputs.items()}, 

278 ) 

279 

280 def _check_consensus(self, outputs: list[str]) -> bool: 

281 """Check if string outputs are similar enough for consensus.""" 

282 if len(outputs) < 2: 

283 return True 

284 

285 # Simple overlap ratio 

286 words = [set(o.lower().split()) for o in outputs] 

287 if any(len(w) == 0 for w in words): 

288 return False 

289 

290 overlaps = [] 

291 for i, wi in enumerate(words): 

292 for j, wj in enumerate(words): 

293 if i >= j: 

294 continue 

295 if len(wi | wj) == 0: 

296 overlaps.append(0.0) 

297 else: 

298 overlaps.append(len(wi & wj) / len(wi | wj)) 

299 

300 if not overlaps: 

301 return False 

302 

303 avg_overlap = sum(overlaps) / len(overlaps) 

304 return avg_overlap > 0.4