Coverage for agentos/evolution/learner.py: 25%
110 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-02 09: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
19import time
20from dataclasses import dataclass, field
21from typing import Any, Optional
23from agentos.evolution.engine import EvolutionEngine, EvolutionProposal
24from agentos.evolution.signals import (
25 BehaviorSignal,
26 SignalCollector,
27 SignalSummary,
28 FeedbackPolarity,
29)
32@dataclass
33class LearningInsight:
34 """A single insight derived from behavior signals."""
36 category: str # tool_recommendation, param_tuning, format_adaptation, prompt_refinement, workflow
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) -> Optional[EvolutionProposal]:
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(self, summary: SignalSummary,
147 signals: list[BehaviorSignal]) -> list[LearningInsight]:
148 """Analyze tool usage to suggest new tools or deprecate unused ones."""
149 insights = []
151 # Low tool success rate → suggest alternatives
152 if summary.tool_success_rate < 0.7 and summary.total_signals > 10:
153 failing = [t for t, _ in summary.top_tools[:3]]
154 insights.append(LearningInsight(
155 category="tool_recommendation",
156 title="Tool Success Rate Low",
157 description=f"Tools {failing} have low success rate ({summary.tool_success_rate:.0%}). Consider alternatives or improve error handling.",
158 confidence=0.75,
159 evidence_count=summary.total_signals,
160 ))
162 # High undo rate → tool is confusing
163 if summary.undo_count >= 3:
164 insights.append(LearningInsight(
165 category="tool_recommendation",
166 title="High Undo Rate Detected",
167 description=f"Users undo actions frequently ({summary.undo_count} times). Tool UX may need improvement.",
168 confidence=0.65,
169 evidence_count=summary.undo_count,
170 ))
172 return insights
174 def _analyze_feedback_for_tuning(self, summary: SignalSummary) -> list[LearningInsight]:
175 """Analyze feedback to suggest parameter tuning."""
176 insights = []
178 total_feedback = summary.positive_feedback + summary.negative_feedback
179 if total_feedback < 5:
180 return insights
182 ratio = summary.positive_feedback / max(total_feedback, 1)
184 if ratio < 0.4:
185 insights.append(LearningInsight(
186 category="param_tuning",
187 title="Low Satisfaction Ratio",
188 description=f"Positive feedback ratio is {ratio:.0%}. Consider adjusting agent temperature or personality.",
189 confidence=0.8,
190 evidence_count=total_feedback,
191 ))
193 if ratio > 0.9:
194 insights.append(LearningInsight(
195 category="param_tuning",
196 title="High Satisfaction — Lock Settings",
197 description=f"Positive feedback ratio is {ratio:.0%}. Current settings work well; consider locking as default.",
198 confidence=0.7,
199 evidence_count=total_feedback,
200 ))
202 return insights
204 def _analyze_format_preferences(self, signals: list[BehaviorSignal]) -> list[LearningInsight]:
205 """Learn preferred output formats."""
206 preferences = [s for s in signals if s.type_.value == "format_preference"]
207 if not preferences:
208 return []
210 format_counts = {}
211 for s in preferences:
212 fmt = s.feedback_type or "unknown"
213 format_counts[fmt] = format_counts.get(fmt, 0) + 1
215 top = max(format_counts, key=format_counts.get)
216 if format_counts[top] >= 3:
217 return [LearningInsight(
218 category="format_adaptation",
219 title=f"Format Preference: {top}",
220 description=f"User prefers {top} format ({format_counts[top]} signals). Default to this format.",
221 confidence=0.85,
222 evidence_count=format_counts[top],
223 )]
225 return []
227 def _analyze_corrections(self, signals: list[BehaviorSignal]) -> list[LearningInsight]:
228 """Analyze corrections to suggest prompt refinements."""
229 corrections = [s for s in signals if s.type_.value == "correction"]
230 if len(corrections) < 2:
231 return []
233 # Group corrections by topic
234 return [LearningInsight(
235 category="prompt_refinement",
236 title="Frequent Corrections Detected",
237 description=f"User made {len(corrections)} corrections. Review agent responses for accuracy improvements.",
238 confidence=0.7,
239 evidence_count=len(corrections),
240 )]
242 def _analyze_workflow_patterns(self, signals: list[BehaviorSignal]) -> list[LearningInsight]:
243 """Detect repeated tool sequences to suggest workflow shortcuts."""
244 tool_sequences = []
245 current_seq = []
247 for s in signals:
248 if s.type_.value == "tool_usage" and s.tool_name:
249 current_seq.append(s.tool_name)
250 if len(current_seq) >= 2:
251 tool_sequences.append(tuple(current_seq[-2:]))
253 if len(tool_sequences) < 5:
254 return []
256 from collections import Counter
257 seq_counter = Counter(tool_sequences)
258 top_seq = seq_counter.most_common(1)[0]
260 if top_seq[1] >= 3:
261 return [LearningInsight(
262 category="workflow",
263 title="Repeated Tool Sequence",
264 description=f"Sequence {' → '.join(top_seq[0])} repeated {top_seq[1]} times. Consider creating a shortcut or composite tool.",
265 confidence=0.65,
266 evidence_count=top_seq[1],
267 )]
269 return []
271 def _analyze_health(self, summary: SignalSummary) -> list[LearningInsight]:
272 """General health check insights."""
273 insights = []
275 if summary.re_prompt_count >= 3:
276 insights.append(LearningInsight(
277 category="prompt_refinement",
278 title="Clarify Intent Better",
279 description=f"User re-asked {summary.re_prompt_count} times. First responses may not understand user intent.",
280 confidence=0.6,
281 evidence_count=summary.re_prompt_count,
282 ))
284 return insights
286 # ── Stats ──
288 def get_stats(self) -> dict[str, Any]:
289 return {
290 "total_insights": len(self._insights),
291 "total_applied": self._applied_count,
292 "auto_propose": self._auto_propose,
293 "min_confidence": self._min_confidence,
294 "latest_insights": [
295 {
296 "category": i.category,
297 "title": i.title,
298 "confidence": i.confidence,
299 "proposal_id": i.proposal_id,
300 }
301 for i in self._insights[-5:]
302 ],
303 }