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 

10import json as _json 

11from dataclasses import dataclass, field 

12from typing import Any 

13 

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

15select the best result or synthesize a combined result. 

16 

17Task: {task} 

18 

19Candidates: 

20{candidates} 

21 

22Instructions: 

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

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

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

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

27 

28Output ONLY the JSON object, no other text. 

29JSON:""" 

30 

31 

32@dataclass 

33class FusedResult: 

34 """Result of fusion operation.""" 

35 

36 merged: Any = None 

37 best_index: int = -1 

38 confidence: float = 0.0 

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

40 reason: str = "" 

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

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

43 

44 

45class ResultFusion: 

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

47 

48 Combines outputs from multiple agents with: 

49 - Weighted-vote aggregation 

50 - LLM-as-Judge for quality arbitration 

51 - Confidence scoring 

52 """ 

53 

54 def __init__( 

55 self, 

56 strategy: str = "auto", 

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

58 ): 

59 self._strategy = strategy 

60 self._llm_model = llm_model 

61 

62 def fuse( 

63 self, 

64 task: str, 

65 outputs: dict[str, Any], 

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

67 ) -> FusedResult: 

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

69 

70 Args: 

71 task: Original task description 

72 outputs: Dict of agent_name -> agent_output 

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

74 

75 Returns: 

76 FusedResult with merged output and confidence 

77 """ 

78 if not outputs: 

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

80 result.individual_scores = {} 

81 result.all_outputs = {} 

82 return result 

83 

84 if len(outputs) == 1: 

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

86 result = FusedResult( 

87 merged=value, 

88 best_index=0, 

89 confidence=0.7, 

90 action="select", 

91 reason="Single output", 

92 individual_scores={name: 0.7}, 

93 all_outputs=outputs, 

94 ) 

95 return result 

96 

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

98 

99 # Step 1: Compute individual scores 

100 scores = self._compute_scores(outputs, weights) 

101 

102 # Step 2: Try LLM judge for arbitration 

103 llm_result = self._llm_judge(task, outputs) 

104 if llm_result: 

105 return llm_result 

106 

107 # Step 3: Fallback — weighted aggregation 

108 return self._weighted_aggregate(outputs, scores) 

109 

110 def _compute_scores( 

111 self, 

112 outputs: dict[str, Any], 

113 weights: dict[str, float], 

114 ) -> dict[str, float]: 

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

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

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

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

119 quality = self._quality_heuristic(output) 

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

121 return scores 

122 

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

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

125 score = 0.5 # baseline 

126 

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

128 

129 if not text: 

130 return 0.1 

131 

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

133 length = len(text) 

134 if 100 < length < 2000: 

135 score += 0.15 

136 elif 50 <= length <= 100: 

137 score += 0.05 

138 elif length > 5000: 

139 score += 0.05 

140 

141 # Error patterns 

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

143 for kw in error_keywords: 

144 if kw in text: 

145 score -= 0.15 

146 break 

147 

148 # Structure bonus 

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

150 score += 0.1 

151 

152 # Confidence keywords 

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

154 for kw in confidence_keywords: 

155 if kw in text: 

156 score += 0.05 

157 

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

159 

160 def _llm_judge(self, task: str, outputs: dict[str, Any]) -> FusedResult | None: 

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

162 try: 

163 import os 

164 

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

166 if not api_key: 

167 return None 

168 

169 candidates_str = "\n".join( 

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

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

172 ) 

173 

174 prompt = _JUDGE_PROMPT.format( 

175 task=task, 

176 candidates=candidates_str, 

177 ) 

178 

179 import requests 

180 

181 resp = requests.post( 

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

183 headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, 

184 json={ 

185 "model": self._llm_model, 

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

187 "temperature": 0.0, 

188 "max_tokens": 500, 

189 }, 

190 timeout=30, 

191 ) 

192 if resp.status_code != 200: 

193 return None 

194 

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

196 

197 start = text.find("{") 

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

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

200 return None 

201 

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

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

204 

205 agent_names = list(outputs.keys()) 

206 

207 result = FusedResult() 

208 result.action = action 

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

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

211 result.individual_scores = {k: 0.5 for k in outputs} 

212 

213 if action == "select": 

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

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

216 result.best_index = idx 

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

218 result.confidence = 0.8 

219 elif action == "merge": 

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

221 result.confidence = 0.75 

222 result.best_index = -1 

223 else: # reject 

224 result.merged = None 

225 result.confidence = 0.0 

226 

227 return result 

228 except Exception: 

229 return None 

230 

231 def _weighted_aggregate( 

232 self, 

233 outputs: dict[str, Any], 

234 scores: dict[str, float], 

235 ) -> FusedResult: 

236 """Weighted vote aggregation fallback.""" 

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

238 if max_score == 0: 

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

240 

241 # Find best candidate 

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

243 

244 # Normalize scores to confidence 

245 total = sum(scores.values()) 

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

247 

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

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

250 consensus = self._check_consensus(all_str) 

251 

252 if consensus: 

253 return FusedResult( 

254 merged=outputs[best_name], 

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

256 confidence=confidence, 

257 action="select", 

258 reason="Consensus among outputs", 

259 individual_scores=scores, 

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

261 ) 

262 

263 return FusedResult( 

264 merged=outputs[best_name], 

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

266 confidence=confidence, 

267 action="select", 

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

269 individual_scores=scores, 

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

271 ) 

272 

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

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

275 if len(outputs) < 2: 

276 return True 

277 

278 # Simple overlap ratio 

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

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

281 return False 

282 

283 overlaps = [] 

284 for i, wi in enumerate(words): 

285 for j, wj in enumerate(words): 

286 if i >= j: 

287 continue 

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

289 overlaps.append(0.0) 

290 else: 

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

292 

293 if not overlaps: 

294 return False 

295 

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

297 return avg_overlap > 0.4