Coverage for agentos/evolution/learner.py: 24%
109 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +0800
1"""
2Learning Engine — Analyzes behavior signals to generate evolution proposals.
4The Learner sits between the SignalCollector and EvolutionEngine:
5 1. SignalCollector gathers user behavior signals
6 2. Learner analyzes signals, detects patterns, and suggests improvements
7 3. EvolutionEngine manages the proposal lifecycle (pending → approved → applied)
9Learning strategies:
10 - Tool recommendation: suggest new tools based on usage patterns
11 - Parameter tuning: adjust temperature, max_tokens based on feedback
12 - Format adaptation: learn preferred output formats
13 - Prompt refinement: improve system prompts based on corrections
14 - Workflow optimization: suggest shortcuts for repeated tasks
15"""
17from __future__ import annotations
19from dataclasses import dataclass, field
20from typing import Any, Optional
22from agentos.evolution.engine import EvolutionEngine, EvolutionProposal
23from agentos.evolution.signals import (
24 BehaviorSignal,
25 SignalCollector,
26 SignalSummary,
27)
30@dataclass
31class LearningInsight:
32 """A single insight derived from behavior signals."""
34 category: str # tool_recommendation, param_tuning, format_adaptation, prompt_refinement, workflow
35 title: str
36 description: str
37 confidence: float # 0.0 - 1.0
38 evidence_count: int
39 proposal_id: str = ""
40 source_signals: list[str] = field(default_factory=list)
43class Learner:
44 """Learning engine — from signals to proposals.
46 Usage:
47 from agentos.evolution import EvolutionEngine, SignalCollector, Learner
49 collector = SignalCollector()
50 engine = EvolutionEngine()
51 learner = Learner(collector, engine)
53 # After accumulating signals...
54 insights = learner.analyze()
55 for insight in insights:
56 proposal = learner.propose_from_insight(insight)
57 print(f"Proposed: {proposal.description}")
58 """
60 def __init__(
61 self,
62 collector: SignalCollector,
63 engine: EvolutionEngine,
64 min_confidence: float = 0.6,
65 auto_propose: bool = False,
66 ):
67 self._collector = collector
68 self._engine = engine
69 self._min_confidence = min_confidence
70 self._auto_propose = auto_propose
71 self._insights: list[LearningInsight] = []
72 self._applied_count: int = 0
74 # ── Analysis ──
76 def analyze(self, hours: float = 168) -> list[LearningInsight]:
77 """Analyze recent signals and generate learning insights."""
78 summary = self._collector.summarize(hours)
79 signals: list[BehaviorSignal] = self._collector._buffer[:]
80 insights: list[LearningInsight] = []
82 # 1. Tool recommendation
83 insights.extend(self._analyze_tool_patterns(summary, signals))
85 # 2. Parameter tuning suggestions
86 insights.extend(self._analyze_feedback_for_tuning(summary))
88 # 3. Format adaptation
89 insights.extend(self._analyze_format_preferences(signals))
91 # 4. Prompt refinement from corrections
92 insights.extend(self._analyze_corrections(signals))
94 # 5. Workflow optimization
95 insights.extend(self._analyze_workflow_patterns(signals))
97 # 6. General health
98 insights.extend(self._analyze_health(summary))
100 # Filter by confidence
101 insights = [i for i in insights if i.confidence >= self._min_confidence]
102 self._insights = insights
104 # Auto-propose if enabled
105 if self._auto_propose:
106 for insight in insights:
107 self.propose_from_insight(insight)
109 return insights
111 # ── Insight → Proposal ──
113 def propose_from_insight(self, insight: LearningInsight) -> Optional[EvolutionProposal]:
114 """Convert a learning insight into an evolution proposal."""
115 proposal = self._engine.propose(
116 agent_name="marvis",
117 change_type=insight.category,
118 description=f"{insight.title}: {insight.description}",
119 new_value={
120 "category": insight.category,
121 "title": insight.title,
122 "description": insight.description,
123 "confidence": insight.confidence,
124 },
125 confidence=insight.confidence,
126 risk_level="medium" if insight.confidence > 0.5 else "low",
127 insight_id=id(insight),
128 )
129 insight.proposal_id = proposal.id
130 return proposal
132 def approve_all(self) -> int:
133 """Approve all pending proposals above confidence threshold."""
134 count = 0
135 for proposal in self._engine.list_proposals(status="pending"):
136 self._engine.approve(proposal.id, approved_by="learner-auto")
137 self._engine.apply(proposal.id)
138 count += 1
139 self._applied_count += 1
140 return count
142 # ── Private Analyzers ──
144 def _analyze_tool_patterns(self, summary: SignalSummary,
145 signals: list[BehaviorSignal]) -> list[LearningInsight]:
146 """Analyze tool usage to suggest new tools or deprecate unused ones."""
147 insights = []
149 # Low tool success rate → suggest alternatives
150 if summary.tool_success_rate < 0.7 and summary.total_signals > 10:
151 failing = [t for t, _ in summary.top_tools[:3]]
152 insights.append(LearningInsight(
153 category="tool_recommendation",
154 title="Tool Success Rate Low",
155 description=f"Tools {failing} have low success rate ({summary.tool_success_rate:.0%}). Consider alternatives or improve error handling.",
156 confidence=0.75,
157 evidence_count=summary.total_signals,
158 ))
160 # High undo rate → tool is confusing
161 if summary.undo_count >= 3:
162 insights.append(LearningInsight(
163 category="tool_recommendation",
164 title="High Undo Rate Detected",
165 description=f"Users undo actions frequently ({summary.undo_count} times). Tool UX may need improvement.",
166 confidence=0.65,
167 evidence_count=summary.undo_count,
168 ))
170 return insights
172 def _analyze_feedback_for_tuning(self, summary: SignalSummary) -> list[LearningInsight]:
173 """Analyze feedback to suggest parameter tuning."""
174 insights = []
176 total_feedback = summary.positive_feedback + summary.negative_feedback
177 if total_feedback < 5:
178 return insights
180 ratio = summary.positive_feedback / max(total_feedback, 1)
182 if ratio < 0.4:
183 insights.append(LearningInsight(
184 category="param_tuning",
185 title="Low Satisfaction Ratio",
186 description=f"Positive feedback ratio is {ratio:.0%}. Consider adjusting agent temperature or personality.",
187 confidence=0.8,
188 evidence_count=total_feedback,
189 ))
191 if ratio > 0.9:
192 insights.append(LearningInsight(
193 category="param_tuning",
194 title="High Satisfaction — Lock Settings",
195 description=f"Positive feedback ratio is {ratio:.0%}. Current settings work well; consider locking as default.",
196 confidence=0.7,
197 evidence_count=total_feedback,
198 ))
200 return insights
202 def _analyze_format_preferences(self, signals: list[BehaviorSignal]) -> list[LearningInsight]:
203 """Learn preferred output formats."""
204 preferences = [s for s in signals if s.type_.value == "format_preference"]
205 if not preferences:
206 return []
208 format_counts = {}
209 for s in preferences:
210 fmt = s.feedback_type or "unknown"
211 format_counts[fmt] = format_counts.get(fmt, 0) + 1
213 top = max(format_counts, key=format_counts.get)
214 if format_counts[top] >= 3:
215 return [LearningInsight(
216 category="format_adaptation",
217 title=f"Format Preference: {top}",
218 description=f"User prefers {top} format ({format_counts[top]} signals). Default to this format.",
219 confidence=0.85,
220 evidence_count=format_counts[top],
221 )]
223 return []
225 def _analyze_corrections(self, signals: list[BehaviorSignal]) -> list[LearningInsight]:
226 """Analyze corrections to suggest prompt refinements."""
227 corrections = [s for s in signals if s.type_.value == "correction"]
228 if len(corrections) < 2:
229 return []
231 # Group corrections by topic
232 return [LearningInsight(
233 category="prompt_refinement",
234 title="Frequent Corrections Detected",
235 description=f"User made {len(corrections)} corrections. Review agent responses for accuracy improvements.",
236 confidence=0.7,
237 evidence_count=len(corrections),
238 )]
240 def _analyze_workflow_patterns(self, signals: list[BehaviorSignal]) -> list[LearningInsight]:
241 """Detect repeated tool sequences to suggest workflow shortcuts."""
242 tool_sequences = []
243 current_seq = []
245 for s in signals:
246 if s.type_.value == "tool_usage" and s.tool_name:
247 current_seq.append(s.tool_name)
248 if len(current_seq) >= 2:
249 tool_sequences.append(tuple(current_seq[-2:]))
251 if len(tool_sequences) < 5:
252 return []
254 from collections import Counter
255 seq_counter = Counter(tool_sequences)
256 top_seq = seq_counter.most_common(1)[0]
258 if top_seq[1] >= 3:
259 return [LearningInsight(
260 category="workflow",
261 title="Repeated Tool Sequence",
262 description=f"Sequence {' → '.join(top_seq[0])} repeated {top_seq[1]} times. Consider creating a shortcut or composite tool.",
263 confidence=0.65,
264 evidence_count=top_seq[1],
265 )]
267 return []
269 def _analyze_health(self, summary: SignalSummary) -> list[LearningInsight]:
270 """General health check insights."""
271 insights = []
273 if summary.re_prompt_count >= 3:
274 insights.append(LearningInsight(
275 category="prompt_refinement",
276 title="Clarify Intent Better",
277 description=f"User re-asked {summary.re_prompt_count} times. First responses may not understand user intent.",
278 confidence=0.6,
279 evidence_count=summary.re_prompt_count,
280 ))
282 return insights
284 # ── Stats ──
286 def get_stats(self) -> dict[str, Any]:
287 return {
288 "total_insights": len(self._insights),
289 "total_applied": self._applied_count,
290 "auto_propose": self._auto_propose,
291 "min_confidence": self._min_confidence,
292 "latest_insights": [
293 {
294 "category": i.category,
295 "title": i.title,
296 "confidence": i.confidence,
297 "proposal_id": i.proposal_id,
298 }
299 for i in self._insights[-5:]
300 ],
301 }