Coverage for src\learning_loop.py: 0%

1""" 

2Learning Loop Module - 学习循环系统 

3 

4提供反馈收集、自适应优化、学习闭环功能 

5""" 

6 

7import asyncio 

8import time 

9import uuid 

10from typing import Dict, List, Optional, Any, Callable 

11from dataclasses import dataclass, field 

12from enum import Enum 

13from collections import defaultdict 

14import logging 

15 

16logger = logging.getLogger(__name__) 

17 

18 

19class FeedbackType(Enum): 

20 """反馈类型""" 

21 EXPLICIT = "explicit" # 显式反馈（用户明确评价） 

22 IMPLICIT = "implicit" # 隐式反馈（行为推断） 

23 OUTCOME = "outcome" # 结果反馈（任务完成情况） 

24 

25 

26class FeedbackSentiment(Enum): 

27 """反馈情感""" 

28 POSITIVE = "positive" 

29 NEUTRAL = "neutral" 

30 NEGATIVE = "negative" 

31 

32 

33@dataclass 

34class Feedback: 

35 """反馈数据""" 

36 feedback_id: str 

37 source: FeedbackType 

38 skill_id: str 

39 context: Dict[str, Any] 

40 sentiment: FeedbackSentiment 

41 score: float # 0-10 

42 content: str = "" 

43 timestamp: float = field(default_factory=time.time) 

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

45 

46 

47@dataclass 

48class LearningInsight: 

49 """学习洞察""" 

50 insight_id: str 

51 skill_id: str 

52 insight_type: str # "improvement", "pattern", "recommendation" 

53 title: str 

54 description: str 

55 confidence: float # 0-1 

56 evidence: List[Dict] = field(default_factory=list) 

57 created_at: float = field(default_factory=time.time) 

58 applied: bool = False 

59 

60 

61@dataclass 

62class AdaptationRule: 

63 """自适应规则""" 

64 rule_id: str 

65 trigger_conditions: Dict[str, Any] # 触发条件 

66 adaptation_action: str # 适配动作 

67 priority: int = 0 

68 enabled: bool = True 

69 hit_count: int = 0 

70 success_count: int = 0 

71 

72 

73class FeedbackCollector: 

74 """反馈收集器""" 

75 

76 def __init__(self): 

77 self._feedbacks: List[Feedback] = [] 

78 self._callbacks: List[Callable] = [] 

79 self._lock = asyncio.Lock() 

80 

81 async def collect( 

82 self, 

83 feedback_type: FeedbackType, 

84 skill_id: str, 

85 context: Dict[str, Any], 

86 score: float, 

87 content: str = "", 

88 sentiment: Optional[FeedbackSentiment] = None, 

89 metadata: Optional[Dict[str, Any]] = None 

90 ) -> Feedback: 

91 """收集反馈""" 

92 # 自动判断情感 

93 if sentiment is None: 

94 if score >= 7: 

95 sentiment = FeedbackSentiment.POSITIVE 

96 elif score <= 4: 

97 sentiment = FeedbackSentiment.NEGATIVE 

98 else: 

99 sentiment = FeedbackSentiment.NEUTRAL 

100 

101 feedback = Feedback( 

102 feedback_id=str(uuid.uuid4()), 

103 source=feedback_type, 

104 skill_id=skill_id, 

105 context=context, 

106 sentiment=sentiment, 

107 score=score, 

108 content=content, 

109 metadata=metadata or {} 

110 ) 

111 

112 async with self._lock: 

113 self._feedbacks.append(feedback) 

114 

115 # 触发回调 

116 for callback in self._callbacks: 

117 try: 

118 if asyncio.iscoroutinefunction(callback): 

119 await callback(feedback) 

120 else: 

121 callback(feedback) 

122 except Exception as e: 

123 logger.error(f"反馈回调执行失败: {e}") 

124 

125 logger.info(f"反馈已收集: {skill_id}, 类型: {feedback_type.value}, 得分: {score}") 

126 return feedback 

127 

128 def register_callback(self, callback: Callable) -> None: 

129 """注册反馈回调""" 

130 self._callbacks.append(callback) 

131 

132 async def get_feedbacks( 

133 self, 

134 skill_id: Optional[str] = None, 

135 feedback_type: Optional[FeedbackType] = None, 

136 limit: int = 100 

137 ) -> List[Feedback]: 

138 """获取反馈列表""" 

139 async with self._lock: 

140 feedbacks = self._feedbacks 

141 

142 if skill_id: 

143 feedbacks = [f for f in feedbacks if f.skill_id == skill_id] 

144 if feedback_type: 

145 feedbacks = [f for f in feedbacks if f.source == feedback_type] 

146 

147 return feedbacks[-limit:] 

148 

149 async def get_feedback_stats(self, skill_id: str) -> Dict[str, Any]: 

150 """获取反馈统计""" 

151 feedbacks = await self.get_feedbacks(skill_id=skill_id) 

152 

153 if not feedbacks: 

154 return {"count": 0} 

155 

156 scores = [f.score for f in feedbacks] 

157 sentiments = [f.sentiment for f in feedbacks] 

158 

159 return { 

160 "count": len(feedbacks), 

161 "avg_score": sum(scores) / len(scores), 

162 "min_score": min(scores), 

163 "max_score": max(scores), 

164 "positive_count": sum(1 for s in sentiments if s == FeedbackSentiment.POSITIVE), 

165 "neutral_count": sum(1 for s in sentiments if s == FeedbackSentiment.NEUTRAL), 

166 "negative_count": sum(1 for s in sentiments if s == FeedbackSentiment.NEGATIVE), 

167 "recent_trend": self._calculate_trend(scores[-10:]) if len(scores) >= 10 else "insufficient_data" 

168 } 

169 

170 def _calculate_trend(self, scores: List[float]) -> str: 

171 """计算趋势""" 

172 if len(scores) < 2: 

173 return "stable" 

174 

175 first_half = scores[:len(scores)//2] 

176 second_half = scores[len(scores)//2:] 

177 

178 avg1 = sum(first_half) / len(first_half) 

179 avg2 = sum(second_half) / len(second_half) 

180 

181 diff = avg2 - avg1 

182 if diff > 0.5: 

183 return "improving" 

184 elif diff < -0.5: 

185 return "declining" 

186 return "stable" 

187 

188 

189class InsightGenerator: 

190 """洞察生成器""" 

191 

192 def __init__(self): 

193 self._insights: List[LearningInsight] = [] 

194 self._lock = asyncio.Lock() 

195 

196 async def generate_from_feedbacks(self, feedbacks: List[Feedback]) -> List[LearningInsight]: 

197 """从反馈生成洞察""" 

198 insights = [] 

199 

200 # 按技能分组 

201 by_skill = defaultdict(list) 

202 for fb in feedbacks: 

203 by_skill[fb.skill_id].append(fb) 

204 

205 for skill_id, skill_feedbacks in by_skill.items(): 

206 # 检测改进机会 

207 recent_scores = [f.score for f in skill_feedbacks[-5:]] 

208 avg_score = sum(recent_scores) / len(recent_scores) 

209 

210 if avg_score < 6: 

211 insight = LearningInsight( 

212 insight_id=str(uuid.uuid4()), 

213 skill_id=skill_id, 

214 insight_type="improvement", 

215 title=f"{skill_id}需要改进", 

216 description=f"最近平均得分{avg_score:.1f}，低于预期", 

217 confidence=0.8, 

218 evidence=[{"type": "low_score", "avg": avg_score}] 

219 ) 

220 insights.append(insight) 

221 

222 # 检测模式 

223 if len(skill_feedbacks) >= 10: 

224 pattern = self._detect_pattern(skill_feedbacks) 

225 if pattern: 

226 insight = LearningInsight( 

227 insight_id=str(uuid.uuid4()), 

228 skill_id=skill_id, 

229 insight_type="pattern", 

230 title=f"{skill_id}使用模式", 

231 description=pattern["description"], 

232 confidence=pattern["confidence"], 

233 evidence=pattern["evidence"] 

234 ) 

235 insights.append(insight) 

236 

237 # 保存洞察 

238 async with self._lock: 

239 self._insights.extend(insights) 

240 

241 return insights 

242 

243 def _detect_pattern(self, feedbacks: List[Feedback]) -> Optional[Dict]: 

244 """检测模式""" 

245 # 检测时间模式 

246 hours = [f.timestamp % 86400 // 3600 for f in feedbacks] 

247 morning = sum(1 for h in hours if 6 <= h < 12) 

248 afternoon = sum(1 for h in hours if 12 <= h < 18) 

249 

250 if morning > afternoon * 1.5: 

251 return { 

252 "description": "用户倾向于在上午使用此技能", 

253 "confidence": 0.7, 

254 "evidence": [{"morning_uses": morning, "afternoon_uses": afternoon}] 

255 } 

256 

257 return None 

258 

259 async def get_insights(self, skill_id: Optional[str] = None) -> List[LearningInsight]: 

260 """获取洞察""" 

261 async with self._lock: 

262 if skill_id: 

263 return [i for i in self._insights if i.skill_id == skill_id and not i.applied] 

264 return [i for i in self._insights if not i.applied] 

265 

266 

267class AdaptationEngine: 

268 """自适应引擎""" 

269 

270 def __init__(self): 

271 self._rules: Dict[str, AdaptationRule] = {} 

272 self._lock = asyncio.Lock() 

273 

274 async def add_rule(self, rule: AdaptationRule) -> None: 

275 """添加规则""" 

276 async with self._lock: 

277 self._rules[rule.rule_id] = rule 

278 logger.info(f"自适应规则已添加: {rule.rule_id}") 

279 

280 async def evaluate_rules(self, context: Dict[str, Any]) -> List[AdaptationRule]: 

281 """评估规则""" 

282 matched_rules = [] 

283 

284 async with self._lock: 

285 for rule in self._rules.values(): 

286 if not rule.enabled: 

287 continue 

288 

289 if self._match_conditions(rule.trigger_conditions, context): 

290 matched_rules.append(rule) 

291 rule.hit_count += 1 

292 

293 # 按优先级排序 

294 matched_rules.sort(key=lambda r: r.priority, reverse=True) 

295 return matched_rules 

296 

297 def _match_conditions(self, conditions: Dict[str, Any], context: Dict[str, Any]) -> bool: 

298 """匹配条件""" 

299 for key, expected in conditions.items(): 

300 actual = context.get(key) 

301 if actual is None: 

302 return False 

303 if isinstance(expected, list): 

304 if actual not in expected: 

305 return False 

306 elif actual != expected: 

307 return False 

308 return True 

309 

310 async def record_rule_success(self, rule_id: str) -> None: 

311 """记录规则成功""" 

312 async with self._lock: 

313 if rule_id in self._rules: 

314 self._rules[rule_id].success_count += 1 

315 

316 async def get_rule_stats(self) -> Dict[str, Any]: 

317 """获取规则统计""" 

318 async with self._lock: 

319 total = len(self._rules) 

320 enabled = sum(1 for r in self._rules.values() if r.enabled) 

321 

322 return { 

323 "total_rules": total, 

324 "enabled_rules": enabled, 

325 "total_hits": sum(r.hit_count for r in self._rules.values()), 

326 "total_successes": sum(r.success_count for r in self._rules.values()) 

327 } 

328 

329 

330class LearningLoop: 

331 """学习循环系统（整合模块）""" 

332 

333 def __init__(self): 

334 self.feedback_collector = FeedbackCollector() 

335 self.insight_generator = InsightGenerator() 

336 self.adaptation_engine = AdaptationEngine() 

337 

338 self._running = False 

339 self._learning_interval = 300 # 5分钟 

340 

341 async def start(self) -> None: 

342 """启动学习循环""" 

343 self._running = True 

344 logger.info("学习循环系统已启动") 

345 

346 # 注册反馈回调 

347 self.feedback_collector.register_callback(self._on_feedback) 

348 

349 # 添加默认规则 

350 await self._add_default_rules() 

351 

352 async def stop(self) -> None: 

353 """停止学习循环""" 

354 self._running = False 

355 logger.info("学习循环系统已停止") 

356 

357 async def _on_feedback(self, feedback: Feedback) -> None: 

358 """反馈回调""" 

359 # 每次收集反馈时，尝试生成洞察 

360 if feedback.source == FeedbackType.EXPLICIT: 

361 feedbacks = await self.feedback_collector.get_feedbacks(skill_id=feedback.skill_id) 

362 await self.insight_generator.generate_from_feedbacks(feedbacks) 

363 

364 async def _add_default_rules(self) -> None: 

365 """添加默认规则""" 

366 rules = [ 

367 AdaptationRule( 

368 rule_id="low_score_alert", 

369 trigger_conditions={"score_threshold": 5}, 

370 adaptation_action="降低技能使用优先级", 

371 priority=10 

372 ), 

373 AdaptationRule( 

374 rule_id="high_score_boost", 

375 trigger_conditions={"score_threshold": 8}, 

376 adaptation_action="提高技能使用优先级", 

377 priority=5 

378 ), 

379 ] 

380 

381 for rule in rules: 

382 await self.adaptation_engine.add_rule(rule) 

383 

384 async def submit_feedback( 

385 self, 

386 feedback_type: FeedbackType, 

387 skill_id: str, 

388 context: Dict[str, Any], 

389 score: float, 

390 content: str = "" 

391 ) -> Feedback: 

392 """提交反馈""" 

393 feedback = await self.feedback_collector.collect( 

394 feedback_type=feedback_type, 

395 skill_id=skill_id, 

396 context=context, 

397 score=score, 

398 content=content 

399 ) 

400 

401 # 评估自适应规则 

402 rules = await self.adaptation_engine.evaluate_rules({ 

403 "skill_id": skill_id, 

404 "score": score, 

405 "context": context 

406 }) 

407 

408 for rule in rules: 

409 logger.info(f"触发自适应规则: {rule.rule_id}") 

410 await self.adaptation_engine.record_rule_success(rule.rule_id) 

411 

412 return feedback 

413 

414 async def get_learning_status(self) -> Dict[str, Any]: 

415 """获取学习状态""" 

416 insights = await self.insight_generator.get_insights() 

417 rule_stats = await self.adaptation_engine.get_rule_stats() 

418 

419 return { 

420 "running": self._running, 

421 "pending_insights": len(insights), 

422 "rule_stats": rule_stats 

423 }