Coverage for integrations / social / agent_evolution_service.py: 27.7%
101 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2HevolveSocial - Agent Evolution Service
3Agent generation progression, specialization trees, collaboration tracking.
4"""
5import json
6import logging
7from datetime import datetime
8from typing import Optional, Dict, List
10from sqlalchemy import desc, func
11from sqlalchemy.orm import Session
13from .models import User, AgentEvolution, AgentCollaboration, ResonanceWallet
14from .resonance_engine import ResonanceService
16logger = logging.getLogger('hevolve_social')
18# Generation XP requirements: 100 * 1.5^(gen-1)
19def xp_for_generation(gen: int) -> int:
20 return int(100 * (1.5 ** (gen - 1)))
22# Specialization trees (unlocked at generation 3)
23SPECIALIZATION_TREES = {
24 'analyst': {
25 'name': 'Analyst',
26 'tiers': ['Analyst', 'Data Sage', 'Oracle'],
27 'description': 'Focused on research, data analysis, and insight generation.',
28 },
29 'creator': {
30 'name': 'Creator',
31 'tiers': ['Creator', 'Artisan', 'Visionary'],
32 'description': 'Focused on content creation, design, and creative tasks.',
33 },
34 'executor': {
35 'name': 'Executor',
36 'tiers': ['Executor', 'Engineer', 'Automaton'],
37 'description': 'Focused on task execution, automation, and efficiency.',
38 },
39 'communicator': {
40 'name': 'Communicator',
41 'tiers': ['Communicator', 'Diplomat', 'Ambassador'],
42 'description': 'Focused on interaction, teaching, and community building.',
43 },
44}
47class AgentEvolutionService:
49 @staticmethod
50 def get_or_create_evolution(db: Session, agent_id: str) -> AgentEvolution:
51 """Get or create evolution profile for an agent."""
52 evo = db.query(AgentEvolution).filter_by(user_id=agent_id).first()
53 if not evo:
54 evo = AgentEvolution(
55 user_id=agent_id,
56 generation=1,
57 evolution_xp=0,
58 evolution_xp_next=xp_for_generation(2),
59 )
60 db.add(evo)
61 db.flush()
62 return evo
64 @staticmethod
65 def get_evolution(db: Session, agent_id: str) -> Optional[Dict]:
66 """Get agent evolution profile."""
67 evo = db.query(AgentEvolution).filter_by(user_id=agent_id).first()
68 if not evo:
69 return None
70 result = evo.to_dict()
71 # Add tree info if specialized
72 if evo.specialization_path and evo.specialization_path in SPECIALIZATION_TREES:
73 tree = SPECIALIZATION_TREES[evo.specialization_path]
74 result['tree_name'] = tree['name']
75 result['tree_description'] = tree['description']
76 result['tree_tiers'] = tree['tiers']
77 return result
79 @staticmethod
80 def award_evolution_xp(db: Session, agent_id: str, amount: int,
81 source: str = '') -> Dict:
82 """Award evolution XP to an agent, checking for generation advancement."""
83 evo = AgentEvolutionService.get_or_create_evolution(db, agent_id)
84 evo.evolution_xp += amount
86 advanced = False
87 while evo.evolution_xp >= evo.evolution_xp_next:
88 evo.evolution_xp -= evo.evolution_xp_next
89 evo.generation += 1
90 evo.evolution_xp_next = xp_for_generation(evo.generation + 1)
91 advanced = True
93 # Update specialization tier
94 if evo.specialization_path and evo.specialization_path in SPECIALIZATION_TREES:
95 tree = SPECIALIZATION_TREES[evo.specialization_path]
96 tiers = tree['tiers']
97 tier_idx = min((evo.generation - 3) // 5, len(tiers) - 1)
98 tier_idx = max(0, tier_idx)
99 evo.spec_tier = tiers[tier_idx]
101 return {
102 'generation': evo.generation,
103 'evolution_xp': evo.evolution_xp,
104 'evolution_xp_next': evo.evolution_xp_next,
105 'advanced': advanced,
106 'specialization': evo.specialization_path,
107 'spec_tier': evo.spec_tier,
108 }
110 @staticmethod
111 def specialize(db: Session, agent_id: str, path: str) -> Optional[Dict]:
112 """Choose a specialization path (available at generation 3+)."""
113 if path not in SPECIALIZATION_TREES:
114 return None
116 evo = AgentEvolutionService.get_or_create_evolution(db, agent_id)
117 if evo.generation < 3:
118 return None
119 if evo.specialization_path:
120 return None # Already specialized
122 evo.specialization_path = path
123 tree = SPECIALIZATION_TREES[path]
124 evo.spec_tier = tree['tiers'][0]
126 return {
127 'specialization': path,
128 'spec_tier': evo.spec_tier,
129 'tree_name': tree['name'],
130 'tree_description': tree['description'],
131 }
133 @staticmethod
134 def record_collaboration(db: Session, agent_a_id: str, agent_b_id: str,
135 task_id: str = None,
136 collaboration_type: str = 'co_task',
137 quality_score: float = 1.0) -> Dict:
138 """Record a collaboration between two agents."""
139 collab = AgentCollaboration(
140 agent_a_id=agent_a_id,
141 agent_b_id=agent_b_id,
142 task_id=task_id,
143 collaboration_type=collaboration_type,
144 quality_score=quality_score,
145 )
146 db.add(collab)
148 # Update collaboration counts and bonus
149 for aid in [agent_a_id, agent_b_id]:
150 evo = AgentEvolutionService.get_or_create_evolution(db, aid)
151 evo.total_collaborations = (evo.total_collaborations or 0) + 1
152 # Collaboration bonus grows with count, capped at 2.0x
153 evo.collaboration_bonus = min(
154 1.0 + (evo.total_collaborations * 0.02), 2.0
155 )
157 db.flush()
158 return collab.to_dict()
160 @staticmethod
161 def get_collaborations(db: Session, agent_id: str,
162 limit: int = 50, offset: int = 0) -> List[Dict]:
163 """Get collaboration history for an agent."""
164 from sqlalchemy import or_
165 collabs = db.query(AgentCollaboration).filter(
166 or_(
167 AgentCollaboration.agent_a_id == agent_id,
168 AgentCollaboration.agent_b_id == agent_id,
169 )
170 ).order_by(desc(AgentCollaboration.created_at)).offset(offset).limit(limit).all()
171 return [c.to_dict() for c in collabs]
173 @staticmethod
174 def get_agent_leaderboard(db: Session, limit: int = 50,
175 offset: int = 0) -> List[Dict]:
176 """Get agent leaderboard by generation and evolution XP."""
177 rows = db.query(AgentEvolution, User).join(
178 User, User.id == AgentEvolution.user_id
179 ).order_by(
180 desc(AgentEvolution.generation),
181 desc(AgentEvolution.evolution_xp),
182 ).offset(offset).limit(limit).all()
184 result = []
185 for i, (evo, user) in enumerate(rows, start=offset + 1):
186 entry = evo.to_dict()
187 entry['rank'] = i
188 entry['username'] = user.username
189 entry['display_name'] = user.display_name
190 entry['avatar_url'] = user.avatar_url
191 result.append(entry)
192 return result
194 @staticmethod
195 def get_showcase(db: Session, limit: int = 20) -> List[Dict]:
196 """Get top agents for showcase."""
197 rows = db.query(AgentEvolution, User, ResonanceWallet).join(
198 User, User.id == AgentEvolution.user_id
199 ).outerjoin(
200 ResonanceWallet, ResonanceWallet.user_id == AgentEvolution.user_id
201 ).order_by(
202 desc(AgentEvolution.generation),
203 ).limit(limit).all()
205 result = []
206 for evo, user, wallet in rows:
207 entry = {
208 'user_id': user.id,
209 'username': user.username,
210 'display_name': user.display_name,
211 'avatar_url': user.avatar_url,
212 'generation': evo.generation,
213 'specialization': evo.specialization_path,
214 'spec_tier': evo.spec_tier,
215 'total_tasks': evo.total_tasks,
216 'total_collaborations': evo.total_collaborations,
217 }
218 if wallet:
219 entry['pulse'] = wallet.pulse
220 entry['level'] = wallet.level
221 result.append(entry)
222 return result