Coverage for agentos/evolution/signals.py: 40%

1""" 

2Behavior Signal Detection — User behavior signal collection and analysis. 

3 

4Signals collected: 

5 - Tool usage frequency and patterns 

6 - Explicit feedback (thumbs up/down, ratings) 

7 - Implicit feedback (corrections, re-prompts, "no/stop/undo") 

8 - Conversation context (topic shifts, depth, sentiment) 

9 - Timing signals (response latency, session duration) 

10 - Preference signals (format preferences, language, tone) 

11 

12These signals feed into the Learner to generate evolution proposals. 

13""" 

14 

15from __future__ import annotations 

16 

17import json 

18import time 

19import uuid 

20from collections import defaultdict 

21from collections.abc import Callable 

22from dataclasses import dataclass, field 

23from enum import StrEnum 

24from pathlib import Path 

25from typing import Any 

26 

27# ── Signal Types ── 

28 

29 

30class SignalType(StrEnum): 

31 TOOL_USAGE = "tool_usage" # Tool was invoked 

32 EXPLICIT_FEEDBACK = "explicit_feedback" # User gave explicit rating 

33 CORRECTION = "correction" # User corrected agent 

34 RE_PROMPT = "re_prompt" # User re-asked the same thing 

35 UNDO = "undo" # User undid an action 

36 SESSION_LENGTH = "session_length" # Session duration 

37 TOPIC_SWITCH = "topic_switch" # User changed topic abruptly 

38 FORMAT_PREFERENCE = "format_preference" # Output format preference 

39 RESPONSE_LATENCY = "response_latency" # How fast agent responded 

40 ERROR_RECOVERY = "error_recovery" # Error occurred and agent recovered 

41 PATTERN_MATCH = "pattern_match" # Recognized repeated pattern 

42 

43 

44class FeedbackPolarity(StrEnum): 

45 POSITIVE = "positive" 

46 NEGATIVE = "negative" 

47 NEUTRAL = "neutral" 

48 

49 

50# ── Signal Data ── 

51 

52 

53@dataclass 

54class BehaviorSignal: 

55 """A single observed user behavior signal.""" 

56 

57 id: str = field(default_factory=lambda: uuid.uuid4().hex[:8]) 

58 type_: SignalType = SignalType.TOOL_USAGE 

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

60 user_id: str = "default" 

61 session_id: str = "" 

62 

63 # Payload 

64 tool_name: str = "" 

65 tool_args: dict = field(default_factory=dict) 

66 tool_success: bool = True 

67 tool_duration_ms: float = 0.0 

68 

69 feedback_type: str = "" # "thumbs_up", "thumbs_down", "rating:4" 

70 feedback_text: str = "" 

71 polarity: FeedbackPolarity = FeedbackPolarity.NEUTRAL 

72 

73 context_before: str = "" # What happened before this signal 

74 context_after: str = "" # Result after this signal 

75 

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

77 

78 def to_dict(self) -> dict: 

79 return { 

80 "id": self.id, 

81 "type": self.type_.value, 

82 "timestamp": self.timestamp, 

83 "user_id": self.user_id, 

84 "session_id": self.session_id, 

85 "tool_name": self.tool_name, 

86 "tool_args": self.tool_args, 

87 "tool_success": self.tool_success, 

88 "tool_duration_ms": self.tool_duration_ms, 

89 "feedback_type": self.feedback_type, 

90 "feedback_text": self.feedback_text, 

91 "polarity": self.polarity.value, 

92 "context_before": self.context_before[:500], 

93 "context_after": self.context_after[:500], 

94 "metadata": self.metadata, 

95 } 

96 

97 

98@dataclass 

99class SignalSummary: 

100 """Aggregated signal analysis over a time window.""" 

101 

102 window_start: float = 0.0 

103 window_end: float = field(default_factory=time.time) 

104 total_signals: int = 0 

105 

106 # Tool usage 

107 top_tools: list[tuple[str, int]] = field(default_factory=list) # [(tool_name, count)] 

108 tool_success_rate: float = 0.0 

109 

110 # Feedback 

111 positive_feedback: int = 0 

112 negative_feedback: int = 0 

113 correction_count: int = 0 

114 undo_count: int = 0 

115 re_prompt_count: int = 0 

116 

117 # Patterns 

118 detected_patterns: list[str] = field(default_factory=list) 

119 

120 def to_dict(self) -> dict: 

121 return { 

122 "window_start": self.window_start, 

123 "window_end": self.window_end, 

124 "total_signals": self.total_signals, 

125 "top_tools": self.top_tools, 

126 "tool_success_rate": self.tool_success_rate, 

127 "positive_feedback": self.positive_feedback, 

128 "negative_feedback": self.negative_feedback, 

129 "correction_count": self.correction_count, 

130 "undo_count": self.undo_count, 

131 "re_prompt_count": self.re_prompt_count, 

132 "detected_patterns": self.detected_patterns, 

133 } 

134 

135 

136# ── Signal Collector ── 

137 

138 

139class SignalCollector: 

140 """Collects and persists user behavior signals. 

141 

142 Features: 

143 - In-memory ring buffer (last N signals) 

144 - Optional disk persistence 

145 - Signal hooks for real-time processing 

146 - Aggregation windows for analysis 

147 

148 Usage: 

149 collector = SignalCollector(buffer_size=1000) 

150 

151 # Record tool usage 

152 collector.record_tool_usage("web_search", {"query": "..."}, success=True) 

153 

154 # Record explicit feedback 

155 collector.record_feedback("thumbs_up", "Great answer!") 

156 

157 # Record correction 

158 collector.record_correction("No, use Python not JS") 

159 

160 # Get summary 

161 summary = collector.summarize(hours=24) 

162 """ 

163 

164 def __init__( 

165 self, 

166 buffer_size: int = 2000, 

167 persist_path: str | None = None, 

168 ): 

169 self._buffer: list[BehaviorSignal] = [] 

170 self._buffer_size = buffer_size 

171 self._persist_path = Path(persist_path) if persist_path else None 

172 self._hooks: list[Callable[[BehaviorSignal], None]] = [] 

173 self._tool_counter: dict[str, int] = defaultdict(int) 

174 self._session_id: str = "" 

175 

176 if self._persist_path: 

177 self._persist_path.parent.mkdir(parents=True, exist_ok=True) 

178 self._load_from_disk() 

179 

180 def set_session(self, session_id: str) -> None: 

181 self._session_id = session_id 

182 

183 # ── Recording API ── 

184 

185 def record_tool_usage( 

186 self, 

187 tool_name: str, 

188 tool_args: dict = None, 

189 success: bool = True, 

190 duration_ms: float = 0.0, 

191 ) -> BehaviorSignal: 

192 """Record a tool invocation.""" 

193 signal = BehaviorSignal( 

194 type_=SignalType.TOOL_USAGE, 

195 tool_name=tool_name, 

196 tool_args=tool_args or {}, 

197 tool_success=success, 

198 tool_duration_ms=duration_ms, 

199 session_id=self._session_id, 

200 ) 

201 self._tool_counter[tool_name] += 1 

202 self._append(signal) 

203 return signal 

204 

205 def record_feedback( 

206 self, 

207 feedback_type: str, 

208 feedback_text: str = "", 

209 ) -> BehaviorSignal: 

210 """Record explicit user feedback.""" 

211 polarity = FeedbackPolarity.NEUTRAL 

212 if feedback_type in ("thumbs_up", "positive", "5", "4"): 

213 polarity = FeedbackPolarity.POSITIVE 

214 elif feedback_type in ("thumbs_down", "negative", "1", "2"): 

215 polarity = FeedbackPolarity.NEGATIVE 

216 

217 signal = BehaviorSignal( 

218 type_=SignalType.EXPLICIT_FEEDBACK, 

219 feedback_type=feedback_type, 

220 feedback_text=feedback_text, 

221 polarity=polarity, 

222 session_id=self._session_id, 

223 ) 

224 self._append(signal) 

225 return signal 

226 

227 def record_correction(self, correction_text: str, context: str = "") -> BehaviorSignal: 

228 """Record a user correction.""" 

229 signal = BehaviorSignal( 

230 type_=SignalType.CORRECTION, 

231 feedback_text=correction_text, 

232 context_before=context, 

233 polarity=FeedbackPolarity.NEGATIVE, 

234 session_id=self._session_id, 

235 ) 

236 self._append(signal) 

237 return signal 

238 

239 def record_undo(self, action: str = "") -> BehaviorSignal: 

240 """Record an undo action.""" 

241 signal = BehaviorSignal( 

242 type_=SignalType.UNDO, 

243 tool_name=action, 

244 polarity=FeedbackPolarity.NEGATIVE, 

245 session_id=self._session_id, 

246 ) 

247 self._append(signal) 

248 return signal 

249 

250 def record_re_prompt(self, original_query: str = "") -> BehaviorSignal: 

251 """Record a re-prompt (user asked again differently).""" 

252 signal = BehaviorSignal( 

253 type_=SignalType.RE_PROMPT, 

254 context_before=original_query, 

255 polarity=FeedbackPolarity.NEGATIVE, 

256 session_id=self._session_id, 

257 ) 

258 self._append(signal) 

259 return signal 

260 

261 def record_format_preference(self, format_type: str) -> BehaviorSignal: 

262 """Record output format preference.""" 

263 signal = BehaviorSignal( 

264 type_=SignalType.FORMAT_PREFERENCE, 

265 feedback_type=format_type, 

266 session_id=self._session_id, 

267 ) 

268 self._append(signal) 

269 return signal 

270 

271 def record_error_recovery(self, error: str, recovered: bool = True) -> BehaviorSignal: 

272 """Record an error that the agent recovered from.""" 

273 signal = BehaviorSignal( 

274 type_=SignalType.ERROR_RECOVERY, 

275 context_before=error, 

276 tool_success=recovered, 

277 session_id=self._session_id, 

278 ) 

279 self._append(signal) 

280 return signal 

281 

282 # ── Analysis ── 

283 

284 def summarize(self, hours: float = 24) -> SignalSummary: 

285 """Generate a summary of signals over the last N hours.""" 

286 now = time.time() 

287 cutoff = now - hours * 3600 

288 

289 signals = [s for s in self._buffer if s.timestamp >= cutoff] 

290 

291 summary = SignalSummary( 

292 window_start=cutoff, 

293 window_end=now, 

294 total_signals=len(signals), 

295 ) 

296 

297 tool_counts: dict[str, int] = defaultdict(int) 

298 total_tools = 0 

299 successful_tools = 0 

300 

301 for s in signals: 

302 if s.type_ == SignalType.TOOL_USAGE: 

303 tool_counts[s.tool_name] += 1 

304 total_tools += 1 

305 if s.tool_success: 

306 successful_tools += 1 

307 

308 elif s.type_ == SignalType.EXPLICIT_FEEDBACK: 

309 if s.polarity == FeedbackPolarity.POSITIVE: 

310 summary.positive_feedback += 1 

311 elif s.polarity == FeedbackPolarity.NEGATIVE: 

312 summary.negative_feedback += 1 

313 

314 elif s.type_ == SignalType.CORRECTION: 

315 summary.correction_count += 1 

316 

317 elif s.type_ == SignalType.UNDO: 

318 summary.undo_count += 1 

319 

320 elif s.type_ == SignalType.RE_PROMPT: 

321 summary.re_prompt_count += 1 

322 

323 summary.top_tools = sorted(tool_counts.items(), key=lambda x: -x[1])[:10] 

324 summary.tool_success_rate = successful_tools / max(total_tools, 1) 

325 

326 # Detect patterns 

327 summary.detected_patterns = self._detect_patterns(signals) 

328 

329 return summary 

330 

331 def get_tool_ranking(self, top_n: int = 10) -> list[tuple[str, int]]: 

332 return sorted(self._tool_counter.items(), key=lambda x: -x[1])[:top_n] 

333 

334 def get_feedback_ratio(self, hours: float = 168) -> float: 

335 """Positive feedback ratio over time window.""" 

336 summary = self.summarize(hours) 

337 total = summary.positive_feedback + summary.negative_feedback 

338 if total == 0: 

339 return 0.5 

340 return summary.positive_feedback / total 

341 

342 # ── Hooks ── 

343 

344 def on_signal(self, hook: Callable[[BehaviorSignal], None]) -> None: 

345 """Register a hook called on every new signal.""" 

346 self._hooks.append(hook) 

347 

348 # ── Internal ── 

349 

350 def _append(self, signal: BehaviorSignal) -> None: 

351 self._buffer.append(signal) 

352 if len(self._buffer) > self._buffer_size: 

353 self._buffer = self._buffer[-self._buffer_size :] 

354 

355 for hook in self._hooks: 

356 try: 

357 hook(signal) 

358 except Exception: 

359 pass 

360 

361 if self._persist_path: 

362 self._save_to_disk() 

363 

364 def _detect_patterns(self, signals: list[BehaviorSignal]) -> list[str]: 

365 """Detect behavioral patterns from signals.""" 

366 patterns = [] 

367 

368 # Pattern: frequent corrections on same topic 

369 corrections = [s for s in signals if s.type_ == SignalType.CORRECTION] 

370 if len(corrections) >= 3: 

371 patterns.append(f"frequent_corrections:{len(corrections)}") 

372 

373 # Pattern: high undo rate 

374 undos = [s for s in signals if s.type_ == SignalType.UNDO] 

375 if len(undos) >= 2: 

376 patterns.append(f"high_undo_rate:{len(undos)}") 

377 

378 # Pattern: repeated tool failures 

379 failed_tools = [ 

380 s for s in signals if s.type_ == SignalType.TOOL_USAGE and not s.tool_success 

381 ] 

382 if len(failed_tools) >= 3: 

383 tools = set(s.tool_name for s in failed_tools) 

384 patterns.append(f"failing_tools:{','.join(tools)}") 

385 

386 # Pattern: positive feedback streak 

387 positive = [s for s in signals if s.polarity == FeedbackPolarity.POSITIVE] 

388 if len(positive) >= 5: 

389 patterns.append(f"positive_streak:{len(positive)}") 

390 

391 return patterns 

392 

393 def _save_to_disk(self) -> None: 

394 if not self._persist_path: 

395 return 

396 try: 

397 data = [s.to_dict() for s in self._buffer[-500:]] 

398 self._persist_path.write_text(json.dumps(data, ensure_ascii=False, indent=2)) 

399 except Exception: 

400 pass 

401 

402 def _load_from_disk(self) -> None: 

403 if not self._persist_path or not self._persist_path.exists(): 

404 return 

405 try: 

406 data = json.loads(self._persist_path.read_text()) 

407 for item in data[-500:]: 

408 signal = BehaviorSignal( 

409 id=item.get("id", ""), 

410 type_=SignalType(item.get("type", "tool_usage")), 

411 timestamp=item.get("timestamp", 0), 

412 user_id=item.get("user_id", "default"), 

413 session_id=item.get("session_id", ""), 

414 tool_name=item.get("tool_name", ""), 

415 tool_success=item.get("tool_success", True), 

416 feedback_type=item.get("feedback_type", ""), 

417 feedback_text=item.get("feedback_text", ""), 

418 polarity=FeedbackPolarity(item.get("polarity", "neutral")), 

419 context_before=item.get("context_before", ""), 

420 context_after=item.get("context_after", ""), 

421 metadata=item.get("metadata", {}), 

422 ) 

423 self._buffer.append(signal) 

424 self._tool_counter[signal.tool_name] += 1 

425 except Exception: 

426 pass