Coverage for src\skill_manager.py: 0%
222 statements
« prev ^ index » next coverage.py v7.3.4, created at 2026-04-21 14:54 +0800
1"""
2Skill Manager Module - 技能管理系统
4提供技能定义、注册、发现、评分和进化功能
5"""
7import asyncio
8import time
9import uuid
10from typing import Dict, List, Optional, Set, Any
11from dataclasses import dataclass, field
12from enum import Enum
13from collections import defaultdict
14import logging
16logger = logging.getLogger(__name__)
19class SkillCategory(Enum):
20 """技能分类"""
21 CODING = "coding" # 编程相关
22 WRITING = "writing" # 写作相关
23 ANALYSIS = "analysis" # 分析相关
24 CREATIVE = "creative" # 创意相关
25 RESEARCH = "research" # 调研相关
26 COMMUNICATION = "communication" # 沟通相关
27 GENERAL = "general" # 通用技能
30class SkillLevel(Enum):
31 """技能等级"""
32 NOVICE = 1 # 初学者
33 BEGINNER = 2 # 入门
34 INTERMEDIATE = 3 # 中级
35 ADVANCED = 4 # 高级
36 EXPERT = 5 # 专家
39@dataclass
40class SkillDefinition:
41 """技能定义"""
42 skill_id: str
43 name: str
44 description: str
45 category: SkillCategory
46 keywords: List[str] = field(default_factory=list)
47 related_skills: List[str] = field(default_factory=list)
48 usage_count: int = 0
49 success_count: int = 0
50 total_score: float = 0.0
51 created_at: float = field(default_factory=time.time)
52 last_used_at: float = 0
53 level: SkillLevel = SkillLevel.NOVICE
54 metadata: Dict[str, Any] = field(default_factory=dict)
57@dataclass
58class SkillUsage:
59 """技能使用记录"""
60 usage_id: str
61 skill_id: str
62 context: Dict[str, Any]
63 score: float # 0-10
64 feedback: str = ""
65 timestamp: float = field(default_factory=time.time)
66 duration_ms: float = 0
69@dataclass
70class SkillEvolution:
71 """技能进化记录"""
72 skill_id: str
73 from_level: SkillLevel
74 to_level: SkillLevel
75 reason: str
76 timestamp: float = field(default_factory=time.time)
77 evidence: Dict[str, Any] = field(default_factory=dict)
80class SkillRegistry:
81 """技能注册表"""
83 def __init__(self):
84 self._skills: Dict[str, SkillDefinition] = {}
85 self._skills_by_category: Dict[SkillCategory, Set[str]] = defaultdict(set)
86 self._skills_by_keyword: Dict[str, Set[str]] = defaultdict(set)
87 self._lock = asyncio.Lock()
89 async def register(self, skill: SkillDefinition) -> bool:
90 """注册技能"""
91 async with self._lock:
92 if skill.skill_id in self._skills:
93 logger.warning(f"技能已存在: {skill.skill_id}")
94 return False
96 self._skills[skill.skill_id] = skill
97 self._skills_by_category[skill.category].add(skill.skill_id)
99 for keyword in skill.keywords:
100 self._skills_by_keyword[keyword.lower()].add(skill.skill_id)
102 logger.info(f"技能已注册: {skill.name} ({skill.skill_id})")
103 return True
105 async def unregister(self, skill_id: str) -> bool:
106 """注销技能"""
107 async with self._lock:
108 if skill_id not in self._skills:
109 return False
111 skill = self._skills[skill_id]
112 self._skills_by_category[skill.category].discard(skill_id)
114 for keyword in skill.keywords:
115 self._skills_by_keyword[keyword.lower()].discard(skill_id)
117 del self._skills[skill_id]
118 return True
120 async def get(self, skill_id: str) -> Optional[SkillDefinition]:
121 """获取技能"""
122 return self._skills.get(skill_id)
124 async def find_by_keyword(self, keyword: str) -> List[SkillDefinition]:
125 """通过关键词查找技能"""
126 keyword = keyword.lower()
127 skill_ids = self._skills_by_keyword.get(keyword, set())
128 return [self._skills[sid] for sid in skill_ids if sid in self._skills]
130 async def find_by_category(self, category: SkillCategory) -> List[SkillDefinition]:
131 """通过分类查找技能"""
132 skill_ids = self._skills_by_category.get(category, set())
133 return [self._skills[sid] for sid in skill_ids if sid in self._skills]
135 async def find_similar(self, skill_id: str) -> List[SkillDefinition]:
136 """查找相似技能"""
137 skill = await self.get(skill_id)
138 if not skill:
139 return []
141 similar_ids = set(skill.related_skills)
142 for keyword in skill.keywords:
143 similar_ids.update(self._skills_by_keyword.get(keyword.lower(), set()))
145 similar_ids.discard(skill_id)
146 return [self._skills[sid] for sid in similar_ids if sid in self._skills]
148 async def list_all(self) -> List[SkillDefinition]:
149 """列出所有技能"""
150 return list(self._skills.values())
152 async def search(self, query: str) -> List[SkillDefinition]:
153 """搜索技能"""
154 query = query.lower()
155 results = []
157 for skill in self._skills.values():
158 if query in skill.name.lower() or query in skill.description.lower():
159 results.append(skill)
160 elif any(query in kw.lower() for kw in skill.keywords):
161 results.append(skill)
163 return results
166class SkillEvolutionEngine:
167 """技能进化引擎"""
169 # 升级所需的最少使用次数
170 MIN_USAGE_FOR_UPGRADE = 5
172 # 升级所需的最低平均分
173 MIN_AVG_SCORE_FOR_UPGRADE = 7.0
175 # 降级阈值
176 DOWNGRADE_SCORE_THRESHOLD = 4.0
178 def __init__(self, registry: SkillRegistry):
179 self._registry = registry
180 self._usage_history: Dict[str, List[SkillUsage]] = defaultdict(list)
181 self._evolution_history: List[SkillEvolution] = []
182 self._lock = asyncio.Lock()
184 async def record_usage(self, skill_id: str, context: Dict[str, Any], score: float, feedback: str = "", duration_ms: float = 0) -> None:
185 """记录技能使用"""
186 async with self._lock:
187 usage = SkillUsage(
188 usage_id=str(uuid.uuid4()),
189 skill_id=skill_id,
190 context=context,
191 score=score,
192 feedback=feedback,
193 duration_ms=duration_ms
194 )
195 self._usage_history[skill_id].append(usage)
197 skill = await self._registry.get(skill_id)
198 if skill:
199 skill.usage_count += 1
200 skill.total_score += score
201 skill.last_used_at = time.time()
203 logger.info(f"技能使用记录: {skill_id}, 得分: {score}")
205 async def should_upgrade(self, skill_id: str) -> bool:
206 """判断是否应该升级"""
207 skill = await self._registry.get(skill_id)
208 if not skill:
209 return False
211 if skill.usage_count < self.MIN_USAGE_FOR_UPGRADE:
212 return False
214 recent_usages = self._usage_history[skill_id][-self.MIN_USAGE_FOR_UPGRADE:]
215 avg_score = sum(u.score for u in recent_usages) / len(recent_usages)
217 return avg_score >= self.MIN_AVG_SCORE_FOR_UPGRADE and skill.level.value < SkillLevel.EXPERT.value
219 async def should_downgrade(self, skill_id: str) -> bool:
220 """判断是否应该降级"""
221 skill = await self._registry.get(skill_id)
222 if not skill or skill.level == SkillLevel.NOVICE:
223 return False
225 recent_usages = self._usage_history[skill_id][-3:]
226 if not recent_usages:
227 return False
229 avg_score = sum(u.score for u in recent_usages) / len(recent_usages)
230 return avg_score < self.DOWNGRADE_SCORE_THRESHOLD
232 async def evolve_skill(self, skill_id: str) -> Optional[SkillEvolution]:
233 """执行技能进化"""
234 async with self._lock:
235 skill = await self._registry.get(skill_id)
236 if not skill:
237 return None
239 evolution = None
241 if await self.should_upgrade(skill_id):
242 old_level = skill.level
243 skill.level = SkillLevel(skill.level.value + 1)
244 evolution = SkillEvolution(
245 skill_id=skill_id,
246 from_level=old_level,
247 to_level=skill.level,
248 reason="连续使用表现优秀",
249 evidence={"avg_score": await self._get_recent_avg_score(skill_id)}
250 )
251 logger.info(f"技能升级: {skill_id} {old_level.name} -> {skill.level.name}")
253 elif await self.should_downgrade(skill_id):
254 old_level = skill.level
255 skill.level = SkillLevel(skill.level.value - 1)
256 evolution = SkillEvolution(
257 skill_id=skill_id,
258 from_level=old_level,
259 to_level=skill.level,
260 reason="近期表现下降",
261 evidence={"avg_score": await self._get_recent_avg_score(skill_id)}
262 )
263 logger.warning(f"技能降级: {skill_id} {old_level.name} -> {skill.level.name}")
265 if evolution:
266 self._evolution_history.append(evolution)
267 skill.success_count = sum(1 for u in self._usage_history[skill_id] if u.score >= 7.0)
269 return evolution
271 async def _get_recent_avg_score(self, skill_id: str) -> float:
272 """获取近期平均分"""
273 recent = self._usage_history[skill_id][-self.MIN_USAGE_FOR_UPGRADE:]
274 if not recent:
275 return 0.0
276 return sum(u.score for u in recent) / len(recent)
278 async def get_skill_stats(self, skill_id: str) -> Dict[str, Any]:
279 """获取技能统计"""
280 usages = self._usage_history.get(skill_id, [])
281 if not usages:
282 return {"usage_count": 0}
284 scores = [u.score for u in usages]
285 return {
286 "usage_count": len(usages),
287 "avg_score": sum(scores) / len(scores),
288 "min_score": min(scores),
289 "max_score": max(scores),
290 "recent_trend": self._calculate_trend(scores[-10:]) if len(scores) >= 10 else "insufficient_data"
291 }
293 def _calculate_trend(self, scores: List[float]) -> str:
294 """计算趋势"""
295 if len(scores) < 2:
296 return "stable"
298 first_half = scores[:len(scores)//2]
299 second_half = scores[len(scores)//2:]
301 avg1 = sum(first_half) / len(first_half)
302 avg2 = sum(second_half) / len(second_half)
304 diff = avg2 - avg1
305 if diff > 0.5:
306 return "improving"
307 elif diff < -0.5:
308 return "declining"
309 return "stable"
312class SkillManager:
313 """技能管理器（整合模块）"""
315 def __init__(self):
316 self.registry = SkillRegistry()
317 self.evolution_engine = SkillEvolutionEngine(self.registry)
318 self._initialized = False
320 async def initialize(self) -> None:
321 """初始化技能管理器"""
322 if self._initialized:
323 return
325 # 注册预定义技能
326 await self._register_default_skills()
327 self._initialized = True
328 logger.info("技能管理器已初始化")
330 async def _register_default_skills(self) -> None:
331 """注册预定义技能"""
332 default_skills = [
333 SkillDefinition(
334 skill_id="coding_python",
335 name="Python编程",
336 description="Python语言编程能力",
337 category=SkillCategory.CODING,
338 keywords=["python", "编程", "代码", "函数", "类"]
339 ),
340 SkillDefinition(
341 skill_id="coding_javascript",
342 name="JavaScript编程",
343 description="JavaScript语言编程能力",
344 category=SkillCategory.CODING,
345 keywords=["javascript", "js", "前端", "web"]
346 ),
347 SkillDefinition(
348 skill_id="writing_code_doc",
349 name="代码文档编写",
350 description="编写代码注释和文档的能力",
351 category=SkillCategory.WRITING,
352 keywords=["文档", "注释", "readme", "doc"]
353 ),
354 SkillDefinition(
355 skill_id="analysis_data",
356 name="数据分析",
357 description="分析和处理数据的能力",
358 category=SkillCategory.ANALYSIS,
359 keywords=["分析", "数据", "统计", "图表"]
360 ),
361 SkillDefinition(
362 skill_id="research_web",
363 name="网络调研",
364 description="在互联网上查找和整理信息的能力",
365 category=SkillCategory.RESEARCH,
366 keywords=["调研", "搜索", "查找", "资料"]
367 ),
368 ]
370 for skill in default_skills:
371 await self.registry.register(skill)
373 async def apply_skill(self, skill_id: str, context: Dict[str, Any], score: float, feedback: str = "", duration_ms: float = 0) -> None:
374 """应用技能并记录"""
375 await self.evolution_engine.record_usage(skill_id, context, score, feedback, duration_ms)
377 # 检查是否需要进化
378 evolution = await self.evolution_engine.evolve_skill(skill_id)
379 if evolution:
380 logger.info(f"技能进化: {skill_id}, {evolution.from_level.name} -> {evolution.to_level.name}")
382 async def get_skill(self, skill_id: str) -> Optional[SkillDefinition]:
383 """获取技能"""
384 return await self.registry.get(skill_id)
386 async def search_skills(self, query: str) -> List[SkillDefinition]:
387 """搜索技能"""
388 return await self.registry.search(query)
390 async def get_skill_stats(self, skill_id: str) -> Dict[str, Any]:
391 """获取技能统计"""
392 return await self.evolution_engine.get_skill_stats(skill_id)
394 async def list_skills(self, category: Optional[SkillCategory] = None) -> List[SkillDefinition]:
395 """列出技能"""
396 if category:
397 return await self.registry.find_by_category(category)
398 return await self.registry.list_all()