Coverage for agentos/evolution/learner.py: 24%
109 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-08 20:40 +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) # noqa: E501
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
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: (
35 str # tool_recommendation, param_tuning, format_adaptation, prompt_refinement, workflow
36 )
37 title: str
38 description: str
39 confidence: float # 0.0 - 1.0
40 evidence_count: int
41 proposal_id: str = ""
42 source_signals: list[str] = field(default_factory=list)
45class Learner:
46 """Learning engine — from signals to proposals.
48 Usage:
49 from agentos.evolution import EvolutionEngine, SignalCollector, Learner
51 collector = SignalCollector()
52 engine = EvolutionEngine()
53 learner = Learner(collector, engine)
55 # After accumulating signals...
56 insights = learner.analyze()
57 for insight in insights:
58 proposal = learner.propose_from_insight(insight)
59 print(f"Proposed: {proposal.description}")
60 """
62 def __init__(
63 self,
64 collector: SignalCollector,
65 engine: EvolutionEngine,
66 min_confidence: float = 0.6,
67 auto_propose: bool = False,
68 ):
69 self._collector = collector
70 self._engine = engine
71 self._min_confidence = min_confidence
72 self._auto_propose = auto_propose
73 self._insights: list[LearningInsight] = []
74 self._applied_count: int = 0
76 # ── Analysis ──
78 def analyze(self, hours: float = 168) -> list[LearningInsight]:
79 """Analyze recent signals and generate learning insights."""
80 summary = self._collector.summarize(hours)
81 signals: list[BehaviorSignal] = self._collector._buffer[:]
82 insights: list[LearningInsight] = []
84 # 1. Tool recommendation
85 insights.extend(self._analyze_tool_patterns(summary, signals))
87 # 2. Parameter tuning suggestions
88 insights.extend(self._analyze_feedback_for_tuning(summary))
90 # 3. Format adaptation
91 insights.extend(self._analyze_format_preferences(signals))
93 # 4. Prompt refinement from corrections
94 insights.extend(self._analyze_corrections(signals))
96 # 5. Workflow optimization
97 insights.extend(self._analyze_workflow_patterns(signals))
99 # 6. General health
100 insights.extend(self._analyze_health(summary))
102 # Filter by confidence
103 insights = [i for i in insights if i.confidence >= self._min_confidence]
104 self._insights = insights
106 # Auto-propose if enabled
107 if self._auto_propose:
108 for insight in insights:
109 self.propose_from_insight(insight)
111 return insights
113 # ── Insight → Proposal ──
115 def propose_from_insight(self, insight: LearningInsight) -> EvolutionProposal | None:
116 """Convert a learning insight into an evolution proposal."""
117 proposal = self._engine.propose(
118 agent_name="marvis",
119 change_type=insight.category,
120 description=f"{insight.title}: {insight.description}",
121 new_value={
122 "category": insight.category,
123 "title": insight.title,
124 "description": insight.description,
125 "confidence": insight.confidence,
126 },
127 confidence=insight.confidence,
128 risk_level="medium" if insight.confidence > 0.5 else "low",
129 insight_id=id(insight),
130 )
131 insight.proposal_id = proposal.id
132 return proposal
134 def approve_all(self) -> int:
135 """Approve all pending proposals above confidence threshold."""
136 count = 0
137 for proposal in self._engine.list_proposals(status="pending"):
138 self._engine.approve(proposal.id, approved_by="learner-auto")
139 self._engine.apply(proposal.id)
140 count += 1
141 self._applied_count += 1
142 return count
144 # ── Private Analyzers ──
146 def _analyze_tool_patterns(
147 self, summary: SignalSummary, signals: list[BehaviorSignal]
148 ) -> list[LearningInsight]:
149 """Analyze tool usage to suggest new tools or deprecate unused ones."""
150 insights = []
152 # Low tool success rate → suggest alternatives
153 if summary.tool_success_rate < 0.7 and summary.total_signals > 10:
154 failing = [t for t, _ in summary.top_tools[:3]]
155 insights.append(
156 LearningInsight(
157 category="tool_recommendation",
158 title="Tool Success Rate Low",
159 description=f"Tools {failing} have low success rate ({summary.tool_success_rate:.0%}). Consider alternatives or improve error handling.", # noqa: E501
160 confidence=0.75,
161 evidence_count=summary.total_signals,
162 )
163 )
165 # High undo rate → tool is confusing
166 if summary.undo_count >= 3:
167 insights.append(
168 LearningInsight(
169 category="tool_recommendation",
170 title="High Undo Rate Detected",
171 description=f"Users undo actions frequently ({summary.undo_count} times). Tool UX may need improvement.", # noqa: E501
172 confidence=0.65,
173 evidence_count=summary.undo_count,
174 )
175 )
177 return insights
179 def _analyze_feedback_for_tuning(self, summary: SignalSummary) -> list[LearningInsight]:
180 """Analyze feedback to suggest parameter tuning."""
181 insights = []
183 total_feedback = summary.positive_feedback + summary.negative_feedback
184 if total_feedback < 5:
185 return insights
187 ratio = summary.positive_feedback / max(total_feedback, 1)
189 if ratio < 0.4:
190 insights.append(
191 LearningInsight(
192 category="param_tuning",
193 title="Low Satisfaction Ratio",
194 description=f"Positive feedback ratio is {ratio:.0%}. Consider adjusting agent temperature or personality.", # noqa: E501
195 confidence=0.8,
196 evidence_count=total_feedback,
197 )
198 )
200 if ratio > 0.9:
201 insights.append(
202 LearningInsight(
203 category="param_tuning",
204 title="High Satisfaction — Lock Settings",
205 description=f"Positive feedback ratio is {ratio:.0%}. Current settings work well; consider locking as default.", # noqa: E501
206 confidence=0.7,
207 evidence_count=total_feedback,
208 )
209 )
211 return insights
213 def _analyze_format_preferences(self, signals: list[BehaviorSignal]) -> list[LearningInsight]:
214 """Learn preferred output formats."""
215 preferences = [s for s in signals if s.type_.value == "format_preference"]
216 if not preferences:
217 return []
219 format_counts = {}
220 for s in preferences:
221 fmt = s.feedback_type or "unknown"
222 format_counts[fmt] = format_counts.get(fmt, 0) + 1
224 top = max(format_counts, key=format_counts.get)
225 if format_counts[top] >= 3:
226 return [
227 LearningInsight(
228 category="format_adaptation",
229 title=f"Format Preference: {top}",
230 description=f"User prefers {top} format ({format_counts[top]} signals). Default to this format.",
231 confidence=0.85,
232 evidence_count=format_counts[top],
233 )
234 ]
236 return []
238 def _analyze_corrections(self, signals: list[BehaviorSignal]) -> list[LearningInsight]:
239 """Analyze corrections to suggest prompt refinements."""
240 corrections = [s for s in signals if s.type_.value == "correction"]
241 if len(corrections) < 2:
242 return []
244 # Group corrections by topic
245 return [
246 LearningInsight(
247 category="prompt_refinement",
248 title="Frequent Corrections Detected",
249 description=f"User made {len(corrections)} corrections. Review agent responses for accuracy improvements.", # noqa: E501
250 confidence=0.7,
251 evidence_count=len(corrections),
252 )
253 ]
255 def _analyze_workflow_patterns(self, signals: list[BehaviorSignal]) -> list[LearningInsight]:
256 """Detect repeated tool sequences to suggest workflow shortcuts."""
257 tool_sequences = []
258 current_seq = []
260 for s in signals:
261 if s.type_.value == "tool_usage" and s.tool_name:
262 current_seq.append(s.tool_name)
263 if len(current_seq) >= 2:
264 tool_sequences.append(tuple(current_seq[-2:]))
266 if len(tool_sequences) < 5:
267 return []
269 from collections import Counter
271 seq_counter = Counter(tool_sequences)
272 top_seq = seq_counter.most_common(1)[0]
274 if top_seq[1] >= 3:
275 return [
276 LearningInsight(
277 category="workflow",
278 title="Repeated Tool Sequence",
279 description=f"Sequence {' → '.join(top_seq[0])} repeated {top_seq[1]} times. Consider creating a shortcut or composite tool.", # noqa: E501
280 confidence=0.65,
281 evidence_count=top_seq[1],
282 )
283 ]
285 return []
287 def _analyze_health(self, summary: SignalSummary) -> list[LearningInsight]:
288 """General health check insights."""
289 insights = []
291 if summary.re_prompt_count >= 3:
292 insights.append(
293 LearningInsight(
294 category="prompt_refinement",
295 title="Clarify Intent Better",
296 description=f"User re-asked {summary.re_prompt_count} times. First responses may not understand user intent.", # noqa: E501
297 confidence=0.6,
298 evidence_count=summary.re_prompt_count,
299 )
300 )
302 return insights
304 # ── Stats ──
306 def get_stats(self) -> dict[str, Any]:
307 return {
308 "total_insights": len(self._insights),
309 "total_applied": self._applied_count,
310 "auto_propose": self._auto_propose,
311 "min_confidence": self._min_confidence,
312 "latest_insights": [
313 {
314 "category": i.category,
315 "title": i.title,
316 "confidence": i.confidence,
317 "proposal_id": i.proposal_id,
318 }
319 for i in self._insights[-5:]
320 ],
321 }