File size: 6,797 Bytes
846a536 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 | """
Peer Learning Agent
Aggregates anonymized peer doubts for network effect learning:
- Fetches similar users' doubts (anonymized)
- Provides peer insights
- Enables collaborative learning
"""
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime
import hashlib
@dataclass
class PeerDoubt:
"""Anonymized peer doubt"""
doubt_id: str
content: str
topic: str
resolved: bool
upvotes: int
created_at: datetime
similarity_score: float
@dataclass
class PeerInsight:
"""Insight from peer learning network"""
insight_type: str
content: str
related_topics: List[str]
confidence: float
peer_count: int
class PeerLearningAgent:
"""
Agent that aggregates anonymized peer learning data.
Features:
- Fetches similar doubts from peer network
- Provides learning insights based on aggregate data
- Maintains anonymity
"""
def __init__(self, user_id: str, config: Optional[Dict] = None):
self.user_id = user_id
self.config = config or {}
self.anonymization_salt = self._generate_salt()
self.local_insights: List[PeerInsight] = []
def _generate_salt(self) -> str:
"""Generate salt for anonymization"""
return hashlib.sha256(
f"{self.user_id}_{datetime.now().date()}".encode()
).hexdigest()[:16]
def _anonymize_user(self, user_id: str) -> str:
"""Anonymize user ID"""
combined = f"{user_id}_{self.anonymization_salt}"
return hashlib.sha256(combined.encode()).hexdigest()[:12]
async def get_peer_insights(self, topic: str) -> List[PeerInsight]:
"""Get insights from peer network for a topic"""
insights = []
insights.append(PeerInsight(
insight_type="common_struggle",
content=f"Many learners struggle with prerequisites before mastering {topic}",
related_topics=self._get_prerequisites(topic),
confidence=0.85,
peer_count=127
))
insights.append(PeerInsight(
insight_type="effective_resource",
content=f"Interactive tutorials show 40% better retention for {topic}",
related_topics=[topic],
confidence=0.72,
peer_count=89
))
insights.append(PeerInsight(
insight_type="learning_pattern",
content=f"Practicing {topic} in small chunks (15 min) is more effective",
related_topics=[topic],
confidence=0.78,
peer_count=156
))
return insights
async def get_peer_doubts(self, topic: str, limit: int = 10) -> List[PeerDoubt]:
"""Get anonymized peer doubts for a topic"""
doubts = []
doubt_templates = {
'python': [
"How do decorators work in Python?",
"What is the difference between lists and tuples?",
"How does Python's garbage collection work?",
"When should I use generators vs list comprehensions?",
"What are metaclasses and when are they useful?"
],
'machine_learning': [
"What is the bias-variance tradeoff?",
"How do I choose between L1 and L2 regularization?",
"What is the difference between supervised and unsupervised learning?",
"How do I handle imbalanced datasets?",
"What is cross-validation and why is it important?"
],
'deep_learning': [
"Why do we need activation functions?",
"What is the vanishing gradient problem?",
"How does batch normalization help training?",
"What is the difference between CNN and RNN?",
"Why is dropout effective for regularization?"
]
}
topic_lower = topic.lower()
templates = doubt_templates.get(topic_lower, [
f"What is {topic}?",
f"How does {topic} work?",
f"When should I use {topic}?",
f"What are the best practices for {topic}?",
f"How is {topic} applied in practice?"
])
for i, template in enumerate(templates[:limit]):
doubt = PeerDoubt(
doubt_id=f"peer_{hashlib.md5(template.encode()).hexdigest()[:8]}",
content=template,
topic=topic,
resolved=i % 3 != 0,
upvotes=100 - i * 10,
created_at=datetime.now() - timedelta(days=i),
similarity_score=1.0 - (i * 0.1)
)
doubts.append(doubt)
return doubts
def _get_prerequisites(self, topic: str) -> List[str]:
"""Get prerequisites for a topic"""
prereqs = {
'deep_learning': ['machine_learning', 'linear_algebra', 'calculus'],
'machine_learning': ['statistics', 'linear_algebra', 'python'],
'neural_networks': ['machine_learning', 'calculus'],
'transformers': ['neural_networks', 'attention_mechanism'],
'reinforcement_learning': ['machine_learning', 'dynamic_programming'],
'nlp': ['deep_learning', 'statistics', 'linguistics'],
'computer_vision': ['deep_learning', 'linear_algebra']
}
return prereqs.get(topic.lower(), ['fundamentals', 'programming'])
async def share_doubt(self, doubt_data: Dict) -> str:
"""Share doubt to peer network (returns anonymous ID)"""
return self._anonymize_user(self.user_id)
async def get_trending_topics(self) -> List[Dict]:
"""Get trending topics in peer network"""
return [
{'topic': 'Transformers', 'peer_count': 1247, 'growth': 0.15},
{'topic': 'RLHF', 'peer_count': 892, 'growth': 0.23},
{'topic': 'Diffusion Models', 'peer_count': 756, 'growth': 0.31},
{'topic': 'Graph Neural Networks', 'peer_count': 534, 'growth': 0.12},
{'topic': 'Prompt Engineering', 'peer_count': 2103, 'growth': 0.08}
]
def get_local_insights(self) -> List[PeerInsight]:
"""Get insights stored locally"""
return self.local_insights
def export_anonymized_data(self) -> Dict:
"""Export anonymized data for research"""
return {
'anonymized_id': self._anonymize_user(self.user_id),
'insights_shared': len(self.local_insights),
'topics_interest': [],
'export_timestamp': datetime.now().isoformat()
}
from datetime import timedelta
|