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

1""" 

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

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

4""" 

5 

6import json 

7import os 

8from dataclasses import dataclass, field 

9from datetime import datetime 

10from enum import StrEnum 

11 

12 

13class FeedbackType(StrEnum): 

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

15 

16 THUMB = "thumb" # 点赞/踩 

17 RATING = "rating" # 1-5星 

18 CORRECTIVE = "corrective" # 纠正指令 

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

20 DETAILED = "detailed" # 详细评价 

21 

22 

23@dataclass 

24class FeedbackRecord: 

25 """反馈记录。""" 

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( 

55 FeedbackRecord( 

56 session_id=session_id, 

57 iteration=iteration, 

58 feedback_type=FeedbackType.THUMB, 

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

60 ) 

61 ) 

62 

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

64 self.collect( 

65 FeedbackRecord( 

66 session_id=session_id, 

67 iteration=iteration, 

68 feedback_type=FeedbackType.RATING, 

69 content=str(rating), 

70 metadata={"comment": comment}, 

71 ) 

72 ) 

73 

74 def collect_corrective( 

75 self, session_id: str, iteration: int, correction: str, original: str = "" 

76 ): 

77 self.collect( 

78 FeedbackRecord( 

79 session_id=session_id, 

80 iteration=iteration, 

81 feedback_type=FeedbackType.CORRECTIVE, 

82 content=correction, 

83 original_output=original, 

84 ) 

85 ) 

86 

87 def on_feedback(self, callback): 

88 self._callbacks.append(callback) 

89 

90 def stats(self) -> dict: 

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

92 ratings = [] 

93 corrective = 0 

94 for r in self._records: 

95 if r.feedback_type == FeedbackType.THUMB: 

96 if r.content == "up": 

97 thumbs["up"] += 1 

98 else: 

99 thumbs["down"] += 1 

100 elif r.feedback_type == FeedbackType.RATING: 

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

102 elif r.feedback_type == FeedbackType.CORRECTIVE: 

103 corrective += 1 

104 return { 

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

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

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

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

109 "corrective_count": corrective, 

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

111 } 

112 

113 def _save(self): 

114 if not self.storage_path: 

115 return 

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

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

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

119 f.write( 

120 json.dumps( 

121 { 

122 "session_id": r.session_id, 

123 "iteration": r.iteration, 

124 "feedback_type": r.feedback_type.value, 

125 "content": r.content, 

126 "original_output": r.original_output, 

127 "corrected_output": r.corrected_output, 

128 "timestamp": r.timestamp, 

129 "metadata": r.metadata, 

130 }, 

131 ensure_ascii=False, 

132 ) 

133 + "\n" 

134 ) 

135 

136 def _load(self): 

137 with open(self.storage_path) as f: 

138 for line in f: 

139 line = line.strip() 

140 if not line: 

141 continue 

142 d = json.loads(line) 

143 self._records.append( 

144 FeedbackRecord( 

145 session_id=d["session_id"], 

146 iteration=d["iteration"], 

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

148 content=d["content"], 

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

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

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

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

153 ) 

154 ) 

155 

156 

157class PreferenceLearner: 

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

159 

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

161 self.window_size = window_size 

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

163 

164 def learn_from_feedback(self, record: FeedbackRecord): 

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

166 pattern = { 

167 "type": record.feedback_type.value, 

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

169 "session": record.session_id, 

170 } 

171 self._recent_patterns.append(pattern) 

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

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

174 

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

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

177 hints = [] 

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

179 if corrections: 

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

181 thumbs_down = sum( 

182 1 for r in self._recent_patterns if r["type"] == "thumb" and r["content"] == "down" 

183 ) 

184 thumbs_up = sum( 

185 1 for r in self._recent_patterns if r["type"] == "thumb" and r["content"] == "up" 

186 ) 

187 if thumbs_down > thumbs_up: 

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

189 return hints 

190 

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

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

193 if not self._recent_patterns: 

194 return False 

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

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