Coverage for agentos/feedback/learner.py: 38%

1""" 

2AgentOS v0.30 反馈学习系统 — Human-in-the-loop + RLHF hooks。 

3支持人工评分、偏好学习、持续改进。 

4""" 

5 

6from dataclasses import dataclass, field 

7from datetime import datetime 

8from enum import Enum 

9import json 

10import os 

11 

12 

13class FeedbackType(str, Enum): 

14 

15 """反馈类型枚举。""" 

16 

17 THUMB = "thumb" # 点赞/踩 

18 RATING = "rating" # 1-5星 

19 CORRECTIVE = "corrective" # 纠正指令 

20 PREFERENCE = "preference" # A/B偏好 

21 DETAILED = "detailed" # 详细评价 

22 

23 

24@dataclass 

25class FeedbackRecord: 

26 """反馈记录。""" 

27 session_id: str 

28 iteration: int 

29 feedback_type: FeedbackType 

30 content: str # 反馈内容或评分 

31 original_output: str = "" 

32 corrected_output: str = "" 

33 timestamp: str = field(default_factory=lambda: datetime.now().isoformat()) 

34 metadata: dict = field(default_factory=dict) 

35 

36 

37class FeedbackCollector: 

38 """反馈收集器 — HITL反馈入口。""" 

39 

40 def __init__(self, storage_path: str = "./feedback_data.jsonl"): 

41 self.storage_path = storage_path 

42 self._records: list[FeedbackRecord] = [] 

43 self._callbacks: list[callable] = [] 

44 if storage_path and os.path.exists(storage_path): 

45 self._load() 

46 

47 def collect(self, record: FeedbackRecord): 

48 self._records.append(record) 

49 self._save() 

50 for cb in self._callbacks: 

51 cb(record) 

52 

53 def collect_thumbs(self, session_id: str, iteration: int, up: bool): 

54 self.collect(FeedbackRecord( 

55 session_id=session_id, 

56 iteration=iteration, 

57 feedback_type=FeedbackType.THUMB, 

58 content="up" if up else "down", 

59 )) 

60 

61 def collect_rating(self, session_id: str, iteration: int, rating: int, comment: str = ""): 

62 self.collect(FeedbackRecord( 

63 session_id=session_id, 

64 iteration=iteration, 

65 feedback_type=FeedbackType.RATING, 

66 content=str(rating), 

67 metadata={"comment": comment}, 

68 )) 

69 

70 def collect_corrective(self, session_id: str, iteration: int, correction: str, original: str = ""): 

71 self.collect(FeedbackRecord( 

72 session_id=session_id, 

73 iteration=iteration, 

74 feedback_type=FeedbackType.CORRECTIVE, 

75 content=correction, 

76 original_output=original, 

77 )) 

78 

79 def on_feedback(self, callback): 

80 self._callbacks.append(callback) 

81 

82 def stats(self) -> dict: 

83 thumbs = {"up": 0, "down": 0} 

84 ratings = [] 

85 corrective = 0 

86 for r in self._records: 

87 if r.feedback_type == FeedbackType.THUMB: 

88 if r.content == "up": 

89 thumbs["up"] += 1 

90 else: 

91 thumbs["down"] += 1 

92 elif r.feedback_type == FeedbackType.RATING: 

93 ratings.append(int(r.content)) 

94 elif r.feedback_type == FeedbackType.CORRECTIVE: 

95 corrective += 1 

96 return { 

97 "total": len(self._records), 

98 "thumbs_up": thumbs["up"], 

99 "thumbs_down": thumbs["down"], 

100 "avg_rating": sum(ratings) / len(ratings) if ratings else 0.0, 

101 "corrective_count": corrective, 

102 "satisfaction": thumbs["up"] / max(thumbs["up"] + thumbs["down"], 1), 

103 } 

104 

105 def _save(self): 

106 if not self.storage_path: 

107 return 

108 os.makedirs(os.path.dirname(self.storage_path) or ".", exist_ok=True) 

109 with open(self.storage_path, "a") as f: 

110 for r in self._records[-1:]: 

111 f.write(json.dumps({ 

112 "session_id": r.session_id, 

113 "iteration": r.iteration, 

114 "feedback_type": r.feedback_type.value, 

115 "content": r.content, 

116 "original_output": r.original_output, 

117 "corrected_output": r.corrected_output, 

118 "timestamp": r.timestamp, 

119 "metadata": r.metadata, 

120 }, ensure_ascii=False) + "\n") 

121 

122 def _load(self): 

123 with open(self.storage_path) as f: 

124 for line in f: 

125 line = line.strip() 

126 if not line: 

127 continue 

128 d = json.loads(line) 

129 self._records.append(FeedbackRecord( 

130 session_id=d["session_id"], 

131 iteration=d["iteration"], 

132 feedback_type=FeedbackType(d["feedback_type"]), 

133 content=d["content"], 

134 original_output=d.get("original_output", ""), 

135 corrected_output=d.get("corrected_output", ""), 

136 timestamp=d.get("timestamp", ""), 

137 metadata=d.get("metadata", {}), 

138 )) 

139 

140 

141class PreferenceLearner: 

142 """偏好学习器 — 从反馈中提取改进信号。""" 

143 

144 def __init__(self, window_size: int = 100): 

145 self.window_size = window_size 

146 self._recent_patterns: list[dict] = [] 

147 

148 def learn_from_feedback(self, record: FeedbackRecord): 

149 """从单条反馈中学习。""" 

150 pattern = { 

151 "type": record.feedback_type.value, 

152 "content": record.content[:200], 

153 "session": record.session_id, 

154 } 

155 self._recent_patterns.append(pattern) 

156 if len(self._recent_patterns) > self.window_size: 

157 self._recent_patterns = self._recent_patterns[-self.window_size:] 

158 

159 def get_improvement_hints(self) -> list[str]: 

160 """获取改进建议。""" 

161 hints = [] 

162 corrections = [r for r in self._recent_patterns if r["type"] == "corrective"] 

163 if corrections: 

164 hints.append(f"最近 {len(corrections)} 条纠正反馈，建议调整输出风格") 

165 thumbs_down = sum(1 for r in self._recent_patterns if r["type"] == "thumb" and r["content"] == "down") 

166 thumbs_up = sum(1 for r in self._recent_patterns if r["type"] == "thumb" and r["content"] == "up") 

167 if thumbs_down > thumbs_up: 

168 hints.append("近期满意度下降，建议优化响应质量") 

169 return hints 

170 

171 def should_retrain(self, threshold: float = 0.3) -> bool: 

172 """判断是否应该触发模型微调。""" 

173 if not self._recent_patterns: 

174 return False 

175 negative = sum(1 for r in self._recent_patterns if r["type"] in ("thumb", "corrective")) 

176 return negative / len(self._recent_patterns) > threshold