""" AI自进化服务层 负责管理用户级别的进化引擎、技能数据、驾驶舱数据 """ import os import json import logging import threading import time import random from datetime import datetime, timedelta, date from typing import Optional, List, Dict from sqlalchemy.orm import Session from sqlalchemy import func, desc from collections import Counter from ..models import ( User, EvolutionSkill, EvolutionEvent, EvolutionTrajectory, UserEvolutionConfig, Conversation, Message, SkillVersion, SkillTrajectoryLink, MarketApp, SkillPublishRecord, EvolutionReport, Notification, TeamSharedSkill, Team ) from ..config import UPLOAD_DIR logger = logging.getLogger(__name__) # 能力维度定义 ABILITY_DIMENSIONS = [ {"label": "文字创作", "key": "writing"}, {"label": "数据分析", "key": "data_analysis"}, {"label": "代码能力", "key": "coding"}, {"label": "知识问答", "key": "qa"}, {"label": "流程自动化", "key": "automation"}, {"label": "创意生成", "key": "creativity"}, {"label": "沟通协作", "key": "communication"}, {"label": "逻辑推理", "key": "reasoning"}, ] # 分类到维度的映射 CATEGORY_DIMENSION_MAP = { "writing": "writing", "code": "coding", "data": "data_analysis", "automation": "automation", "creative": "creativity", "knowledge": "qa", "communication": "communication", "reasoning": "reasoning", } class PipelineScheduler: """每30秒扫描discovered轨迹，投递到xSkill流水线""" def __init__(self, user_id: str, db_session_factory, interval: int = 30): self.user_id = user_id self.db_session_factory = db_session_factory self.poll_interval = interval self._stop = threading.Event() self._thread = None def start(self): if self._thread and self._thread.is_alive(): logger.info(f"PipelineScheduler 已在运行，用户 {self.user_id}") return self._stop.clear() self._thread = threading.Thread(target=self._run_loop, daemon=True) self._thread.start() logger.info(f"PipelineScheduler 已启动，用户 {self.user_id}，间隔 {self.poll_interval}s") def stop(self): self._stop.set() if self._thread: self._thread.join(timeout=10) logger.info(f"PipelineScheduler 已停止，用户 {self.user_id}") def is_running(self) -> bool: return self._thread is not None and self._thread.is_alive() def _run_loop(self): logger.info(f"PipelineScheduler 开始运行循环，用户 {self.user_id}") while not self._stop.is_set(): try: db = self.db_session_factory() try: trajectories = db.query(EvolutionTrajectory).filter( EvolutionTrajectory.user_id == self.user_id, EvolutionTrajectory.status == "discovered" ).order_by(EvolutionTrajectory.created_at).limit(10).all() for traj in trajectories: if self._stop.is_set(): break try: self._process_trajectory(db, traj) except Exception as e: logger.error(f"处理轨迹 {traj.id} 失败: {e}") traj.error_msg = str(e)[:500] traj.retry_count = (traj.retry_count or 0) + 1 traj.status = "error" db.commit() finally: db.close() except Exception as e: logger.error(f"PipelineScheduler 扫描循环异常: {e}") self._stop.wait(self.poll_interval) def _process_trajectory(self, db, traj): """处理单条轨迹：discovered -> splitting -> split_done -> indexed -> clustering -> done""" logger.info(f"开始处理轨迹 {traj.id}，用户 {self.user_id}") # step 1: discovered -> splitting traj.status = "splitting" db.commit() # 模拟AtomTask拆分：生成2-5个模拟atom time.sleep(random.uniform(0.1, 0.5)) atom_count = random.randint(2, 5) traj.atom_count = atom_count traj.tasks_extracted = atom_count db.commit() # step 2: splitting -> split_done traj.status = "split_done" db.commit() # 为每个atom创建SkillTrajectoryLink for i in range(atom_count): link = SkillTrajectoryLink( skill_name="__pending__", user_id=self.user_id, trajectory_id=traj.id, atom_id=f"{traj.id}_atom_{i}", weightscore=random.randint(1, 10) ) db.add(link) db.commit() # step 3: split_done -> indexed (模拟向量索引) time.sleep(random.uniform(0.1, 0.3)) traj.status = "indexed" db.commit() # step 4: indexed -> clustering (模拟聚类) time.sleep(random.uniform(0.1, 0.3)) traj.status = "clustering" db.commit() # step 5: clustering -> done # 检查是否有同名skill，有则归属，无则创建新baby skill existing_skill = db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id ).order_by(func.random()).first() if existing_skill: skill_name = existing_skill.skill_name # 更新SkillTrajectoryLink关联 links = db.query(SkillTrajectoryLink).filter( SkillTrajectoryLink.trajectory_id == traj.id, SkillTrajectoryLink.skill_name == "__pending__" ).all() for link in links: link.skill_name = skill_name existing_skill.use_count += 1 logger.info(f"轨迹 {traj.id} 归属于现有技能 {skill_name}") else: # 创建新baby skill skill_name = f"new_skill_{traj.id[:8]}" new_skill = EvolutionSkill( user_id=self.user_id, skill_name=skill_name, description=f"自动发掘的技能(源于轨迹{traj.id[:8]})", tags=[], category="未分类", state="baby", use_count=1, version=1, ) db.add(new_skill) db.flush() # 更新关联 links = db.query(SkillTrajectoryLink).filter( SkillTrajectoryLink.trajectory_id == traj.id, SkillTrajectoryLink.skill_name == "__pending__" ).all() for link in links: link.skill_name = skill_name logger.info(f"轨迹 {traj.id} 创建新baby技能 {skill_name}") db.commit() # 检查某skill的candidates（关联轨迹数）≥3，自动 baby→staging candidate_count = db.query(SkillTrajectoryLink).filter( SkillTrajectoryLink.user_id == self.user_id, SkillTrajectoryLink.skill_name == skill_name ).count() if candidate_count >= 3: skill_obj = db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id, EvolutionSkill.skill_name == skill_name, EvolutionSkill.state == "baby" ).first() if skill_obj: skill_obj.state = "staging" skill_obj.updated_at = datetime.now() # 记录事件 event = EvolutionEvent( user_id=self.user_id, skill_name=skill_name, event_type="promoted", description=f"技能 {skill_name} 自动晋升 baby->staging，关联轨迹数 {candidate_count}", ) db.add(event) db.commit() logger.info(f"技能 {skill_name} 自动晋升 baby->staging，关联轨迹数 {candidate_count}") traj.status = "done" traj.processed_at = datetime.now() db.commit() # 更新pipeline_processed计数 config = db.query(UserEvolutionConfig).filter( UserEvolutionConfig.user_id == self.user_id ).first() if config: config.pipeline_processed = (config.pipeline_processed or 0) + 1 db.commit() logger.info(f"轨迹 {traj.id} 处理完成，归属于技能 {skill_name}") class EvolutionWatcherManager: """全局watcher线程管理器（单例）""" _instance = None _lock = threading.Lock() def __new__(cls): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def __init__(self): if not hasattr(self, '_initialized'): self._schedulers: dict[str, PipelineScheduler] = {} self._max_schedulers = 100 self._initialized = True def start_user_evolution(self, user_id: str, db_session_factory) -> bool: """为用户启动进化引擎""" if user_id in self._schedulers: logger.info(f"用户 {user_id} 的进化引擎已在运行") return False if len(self._schedulers) >= self._max_schedulers: logger.warning(f"已达到最大scheduler数量限制 {self._max_schedulers}") return False # 创建并启动 PipelineScheduler scheduler = PipelineScheduler(user_id, db_session_factory) scheduler.start() self._schedulers[user_id] = scheduler # 更新 user_evolution_config 表 db = db_session_factory() try: config = db.query(UserEvolutionConfig).filter( UserEvolutionConfig.user_id == user_id ).first() if config: config.watcher_status = "running" config.pipeline_status = "running" config.watcher_error = None else: config = UserEvolutionConfig( user_id=user_id, watcher_status="running", pipeline_status="running", ) db.add(config) db.commit() logger.info(f"用户 {user_id} 进化引擎已启动") except Exception as e: logger.error(f"更新用户进化配置失败: {e}") return False finally: db.close() return True def stop_user_evolution(self, user_id: str) -> bool: """停止用户进化引擎""" scheduler = self._schedulers.pop(user_id, None) if not scheduler: logger.info(f"用户 {user_id} 的进化引擎未在运行") return False scheduler.stop() # 更新 user_evolution_config 表 db = self.db_session_factory() if hasattr(self, 'db_session_factory') else None if db: try: config = db.query(UserEvolutionConfig).filter( UserEvolutionConfig.user_id == user_id ).first() if config: config.watcher_status = "stopped" config.pipeline_status = "idle" db.commit() logger.info(f"用户 {user_id} 进化引擎已停止") except Exception as e: logger.error(f"更新用户进化配置失败: {e}") return False finally: db.close() return True def get_user_status(self, user_id: str, db_session_factory) -> dict: """获取用户进化引擎状态""" scheduler = self._schedulers.get(user_id) is_running = scheduler.is_running() if scheduler else False db = db_session_factory() try: config = db.query(UserEvolutionConfig).filter( UserEvolutionConfig.user_id == user_id ).first() watcher_status = config.watcher_status if config else "unknown" pipeline_status = config.pipeline_status if config else "unknown" pipeline_processed = config.pipeline_processed if config else 0 watcher_error = config.watcher_error if config else None # 统计该用户的技能数、轨迹数 skill_count = db.query(EvolutionSkill).filter( EvolutionSkill.user_id == user_id ).count() trajectory_count = db.query(EvolutionTrajectory).filter( EvolutionTrajectory.user_id == user_id ).count() return { "user_id": user_id, "scheduler_running": is_running, "watcher_status": watcher_status, "pipeline_status": pipeline_status, "pipeline_processed": pipeline_processed, "watcher_error": watcher_error, "skill_count": skill_count, "trajectory_count": trajectory_count, } except Exception as e: logger.error(f"获取用户进化状态失败: {e}") return {"error": str(e)} finally: db.close() def restart_user_evolution(self, user_id: str, db_session_factory) -> bool: """重启用户进化引擎""" self.stop_user_evolution(user_id) return self.start_user_evolution(user_id, db_session_factory) def shutdown_all(self): """优雅关闭所有scheduler""" logger.info("正在关闭所有进化引擎...") for user_id in list(self._schedulers.keys()): self.stop_user_evolution(user_id) logger.info("所有进化引擎已关闭") class EvolutionService: """AI自进化服务""" def __init__(self, db: Session, user_id: str): self.db = db self.user_id = user_id # ============================================================ # 技能管理 # ============================================================ def get_skills(self, state: Optional[str] = None, keyword: Optional[str] = None, sort_by: str = "updated_at", order: str = "desc") -> List[dict]: """获取用户技能列表""" query = self.db.query(EvolutionSkill).filter(EvolutionSkill.user_id == self.user_id) if state: query = query.filter(EvolutionSkill.state == state) if keyword: query = query.filter(EvolutionSkill.skill_name.contains(keyword)) if order == "desc": query = query.order_by(desc(getattr(EvolutionSkill, sort_by, EvolutionSkill.updated_at))) else: query = query.order_by(getattr(EvolutionSkill, sort_by, EvolutionSkill.updated_at)) skills = query.all() return [self._skill_to_dict(s) for s in skills] def get_skill_detail(self, skill_name: str) -> Optional[dict]: """获取技能详情（含SKILL.md内容）""" skill = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id, EvolutionSkill.skill_name == skill_name ).first() if not skill: return None result = self._skill_to_dict(skill) # 读取SKILL.md文件内容 if skill.skill_md_path and os.path.exists(skill.skill_md_path): with open(skill.skill_md_path, 'r') as f: result['skill_md_content'] = f.read() else: result['skill_md_content'] = None return result def update_skill(self, skill_name: str, data: dict) -> Optional[dict]: """编辑技能""" skill = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id, EvolutionSkill.skill_name == skill_name ).first() if not skill: return None for key, value in data.items(): if hasattr(skill, key) and key not in ('id', 'user_id', 'created_at'): setattr(skill, key, value) skill.version += 1 skill.updated_at = datetime.now() # 记录编辑事件 self._add_event(skill_name, "edited", f"技能 {skill_name} 已更新", version_before=skill.version - 1, version_after=skill.version) self.db.commit() self.db.refresh(skill) return self._skill_to_dict(skill) def delete_skill(self, skill_name: str) -> bool: """删除技能""" skill = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id, EvolutionSkill.skill_name == skill_name ).first() if not skill: return False self._add_event(skill_name, "deleted", f"技能 {skill_name} 已删除") self.db.delete(skill) self.db.commit() return True def freeze_skill(self, skill_name: str) -> Optional[dict]: """冻结技能""" skill = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id, EvolutionSkill.skill_name == skill_name ).first() if not skill: return None skill.state = "frozen" skill.updated_at = datetime.now() self._add_event(skill_name, "frozen", f"技能 {skill_name} 已冻结") self.db.commit() self.db.refresh(skill) return self._skill_to_dict(skill) def unfreeze_skill(self, skill_name: str) -> Optional[dict]: """解冻技能""" skill = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id, EvolutionSkill.skill_name == skill_name ).first() if not skill: return None skill.state = "main" skill.updated_at = datetime.now() self._add_event(skill_name, "unfrozen", f"技能 {skill_name} 已解冻") self.db.commit() self.db.refresh(skill) return self._skill_to_dict(skill) def rollback_skill(self, skill_name: str, target_version: int) -> Optional[dict]: """技能回滚：将技能恢复到指定的历史版本""" skill = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id, EvolutionSkill.skill_name == skill_name ).first() if not skill: logger.warning(f"rollback_skill: 技能 {skill_name} 不存在，用户 {self.user_id}") return None # 查询目标版本的SKILL.md内容 target_sv = self.db.query(SkillVersion).filter( SkillVersion.user_id == self.user_id, SkillVersion.skill_name == skill_name, SkillVersion.version == target_version ).first() if not target_sv: logger.warning(f"rollback_skill: 版本 {target_version} 不存在，技能 {skill_name}") return None version_before = skill.version # 如果skill_md_path存在，以新内容覆盖写入该文件 if skill.skill_md_path: try: os.makedirs(os.path.dirname(skill.skill_md_path), exist_ok=True) with open(skill.skill_md_path, 'w', encoding='utf-8') as f: f.write(target_sv.skill_md_content or "") logger.info(f"rollback_skill: 已覆盖SKILL.md文件 {skill.skill_md_path}") except Exception as e: logger.error(f"rollback_skill: 写入SKILL.md文件失败: {e}") return {"error": f"写入SKILL.md文件失败: {str(e)}"} # 保存回滚前的当前版本作为快照（新SkillVersion记录） current_content = "" if skill.skill_md_path and os.path.exists(skill.skill_md_path): with open(skill.skill_md_path, 'r', encoding='utf-8') as f: current_content = f.read() else: # 尝试从数据库获取当前版本内容 current_sv = self.db.query(SkillVersion).filter( SkillVersion.user_id == self.user_id, SkillVersion.skill_name == skill_name, SkillVersion.version == version_before ).first() if current_sv: current_content = current_sv.skill_md_content or "" snapshot_version = SkillVersion( user_id=self.user_id, skill_name=skill_name, version=version_before, skill_md_content=current_content, git_commit_sha=None, description=f"回滚前快照(版本{version_before})" ) self.db.add(snapshot_version) # 更新evolution_skills skill.version += 1 skill.git_commit_sha = None skill.updated_at = datetime.now() # 记录回滚事件 self._add_event( skill_name, "rolled_back", f"版本 V{version_before} 回滚至 V{target_version}", version_before=version_before, version_after=skill.version ) self.db.commit() self.db.refresh(skill) logger.info(f"rollback_skill: 技能 {skill_name} 从版本 {version_before} 回滚至 {target_version}") return self._skill_to_dict(skill) def get_skill_history(self, skill_name: str) -> List[dict]: """获取技能事件历史""" events = self.db.query(EvolutionEvent).filter( EvolutionEvent.user_id == self.user_id, EvolutionEvent.skill_name == skill_name ).order_by(desc(EvolutionEvent.created_at)).all() return [self._event_to_dict(e) for e in events] def get_skill_trajectories(self, skill_name: str) -> List[dict]: """获取技能来源轨迹（通过 skill_trajectory_links 表关联）""" from ..models import SkillTrajectoryLink # 通过关联表查询该技能关联的轨迹ID links = self.db.query(SkillTrajectoryLink).filter( SkillTrajectoryLink.user_id == self.user_id, SkillTrajectoryLink.skill_name == skill_name ).order_by(desc(SkillTrajectoryLink.created_at)).limit(20).all() if not links: return [] traj_ids = [link.trajectory_id for link in links if link.trajectory_id] trajectories = self.db.query(EvolutionTrajectory).filter( EvolutionTrajectory.id.in_(traj_ids) ).order_by(desc(EvolutionTrajectory.created_at)).all() result = [] for t in trajectories: d = self._trajectory_to_dict(t) # 附加权重分 for link in links: if link.trajectory_id == t.id: d['weightscore'] = link.weightscore d['atom_id'] = link.atom_id break result.append(d) return result # ============================================================ # 进化事件 # ============================================================ def get_events(self, event_type: Optional[str] = None, skill_name: Optional[str] = None, limit: int = 50) -> List[dict]: """获取进化事件列表""" query = self.db.query(EvolutionEvent).filter(EvolutionEvent.user_id == self.user_id) if event_type: query = query.filter(EvolutionEvent.event_type == event_type) if skill_name: query = query.filter(EvolutionEvent.skill_name == skill_name) events = query.order_by(desc(EvolutionEvent.created_at)).limit(limit).all() return [self._event_to_dict(e) for e in events] # ============================================================ # 驾驶舱数据 # ============================================================ def get_skill_tree(self) -> Dict: """获取技能图谱数据""" skills = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id ).all() categories = {} for s in skills: cat = s.category or "未分类" if cat not in categories: categories[cat] = [] categories[cat].append(self._skill_to_dict(s)) # 统计 total = len(skills) main_count = sum(1 for s in skills if s.state == "main") baby_count = sum(1 for s in skills if s.state == "baby") staging_count = sum(1 for s in skills if s.state == "staging") frozen_count = sum(1 for s in skills if s.state == "frozen") return { "categories": [{"name": k, "skills": v} for k, v in categories.items()], "total_count": total, "main_count": main_count, "baby_count": baby_count, "staging_count": staging_count, "frozen_count": frozen_count, } def get_timeline(self, event_type: Optional[str] = None, limit: int = 50) -> List[dict]: """获取进化时间线""" return self.get_events(event_type=event_type, limit=limit) def get_statistics(self) -> Dict: """获取使用统计数据""" skills = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id ).all() # Top排名 sorted_skills = sorted(skills, key=lambda s: s.use_count, reverse=True) ranking = [] for s in sorted_skills[:10]: ranking.append({ "skill_name": s.skill_name, "category": s.category or "未分类", "use_count": s.use_count, "state": s.state, "last_used_at": s.last_used_at.isoformat() if s.last_used_at else None, }) # 类别分布 category_dist = {} for s in skills: cat = s.category or "未分类" category_dist[cat] = category_dist.get(cat, 0) + s.use_count # 总调用次数 total_use = sum(s.use_count for s in skills) # 最近7天的趋势（按日期分组） seven_days_ago = datetime.now() - timedelta(days=7) recent_trajectories = self.db.query(EvolutionTrajectory).filter( EvolutionTrajectory.user_id == self.user_id, EvolutionTrajectory.created_at >= seven_days_ago, EvolutionTrajectory.status == "done" ).all() # 按天汇总 daily_counts = {} for t in recent_trajectories: day = t.created_at.strftime("%m-%d") daily_counts[day] = daily_counts.get(day, 0) + 1 # 低使用提醒（超过30天未使用） thirty_days_ago = datetime.now() - timedelta(days=30) zero_use_skills = [ self._skill_to_dict(s) for s in skills if s.use_count < 5 and (s.last_used_at is None or s.last_used_at < thirty_days_ago) ] return { "ranking": ranking, "total_use_count": total_use, "category_distribution": [{"category": k, "count": v} for k, v in category_dist.items()], "daily_trend": [{"date": k, "count": v} for k, v in sorted(daily_counts.items())], "low_use_skills": zero_use_skills, } def get_profile(self) -> Dict: """获取AI能力画像""" skills = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id ).all() total = len(skills) main_count = sum(1 for s in skills if s.state == "main") staging_count = sum(1 for s in skills if s.state == "staging") baby_count = sum(1 for s in skills if s.state == "baby") total_use = sum(s.use_count for s in skills) # 计算能力维度 dim_scores = {d["key"]: {"label": d["label"], "value": 0, "maxValue": 10} for d in ABILITY_DIMENSIONS} for s in skills: dim_key = CATEGORY_DIMENSION_MAP.get(s.category, "qa") if dim_key in dim_scores: # 成熟技能贡献更大 multiplier = {"main": 1.5, "staging": 1.0, "baby": 0.5, "frozen": 0.3}.get(s.state, 1.0) dim_scores[dim_key]["value"] += min(s.use_count * 0.2 * multiplier, 3) dimensions = list(dim_scores.values()) for d in dimensions: d["value"] = round(min(d["value"], 10), 1) # 进化等级计算 level_score = (total * 2 + main_count * 5 + staging_count * 3 + baby_count * 1 + total_use * 0.01) level = min(max(int(level_score / 10) + 1, 1), 10) level_titles = { 1: "AI初生", 2: "学习幼苗", 3: "技能萌芽", 4: "成长小树", 5: "能力之林", 6: "智慧果园", 7: "知识大师", 8: "AI专家", 9: "进化先驱", 10: "超级龙虾" } # 能力标签 all_tags = [] for s in skills: if s.tags: all_tags.extend(s.tags) # 去重并按技能数量加权排序 tag_counter = Counter(all_tags) top_tags = [tag for tag, _ in tag_counter.most_common(12)] if not top_tags: top_tags = ["AI进化中", "持续学习", "数据分析", "文字处理", "自动化", "智能助手"] # 成长建议 suggestions = [] if total == 0: suggestions = [ "多与AI进行各类对话，让它从实际交互中学习你的需求和风格", "尝试在不同领域使用AI：写作、编程、数据分析、创意生成等" ] else: dim_v = {d["label"]: d["value"] for d in dimensions} weak_dims = [d for d in dimensions if d["value"] < 3] if weak_dims: suggestions.append(f"建议多进行{weak_dims[0]['label']}类对话，培养该领域技能") if baby_count > 0: suggestions.append(f"你有{baby_count}个萌芽技能，保持使用帮助它们成长") if not any(s.state == "staging" for s in skills): suggestions.append("继续积累高质量对话，新技能将自动进入灰度验证") suggestions.append("持续在各类场景使用AI，系统会自动发掘你的专属技能") return { "level": level, "level_title": level_titles.get(level, "进化者"), "total_skills": total, "active_skills": main_count, "staging_skills": staging_count, "baby_skills": baby_count, "total_use_count": total_use, "dimensions": dimensions, "tags": top_tags, "growth_suggestions": suggestions[:4], } # ============================================================ # 轨迹管理 # ============================================================ def submit_trajectory(self, conversation_id: str, messages_data: List[dict]) -> dict: """提交对话轨迹（对话结束后调用）""" user_dir = UPLOAD_DIR.parent / "users" / self.user_id / "trajectories" user_dir.mkdir(parents=True, exist_ok=True) import uuid traj_id = str(uuid.uuid4())[:8] traj_dir = user_dir / f"traj_{traj_id}" traj_dir.mkdir(exist_ok=True) # 生成标准化轨迹文件 traj_md = self._format_trajectory_md(messages_data) traj_file = traj_dir / f"traj_{traj_id}.md" with open(traj_file, 'w', encoding='utf-8') as f: f.write(traj_md) # 记录到数据库 trajectory = EvolutionTrajectory( user_id=self.user_id, conversation_id=conversation_id, file_path=str(traj_file), status="discovered", atom_count=0, token_used=sum(m.get("tokens", 0) for m in messages_data), ) self.db.add(trajectory) self.db.commit() self.db.refresh(trajectory) logger.info(f"用户 {self.user_id} 提交轨迹 {traj_id}, 对话 {conversation_id}") return self._trajectory_to_dict(trajectory) def _format_trajectory_md(self, messages: List[dict]) -> str: """将消息列表格式化为xSkill兼容的轨迹Markdown""" lines = ["# Conversation Trajectory\n"] for msg in messages: role = msg.get("role", "unknown") content = msg.get("content", "") if role == "user": lines.append(f"## User\n\n{content}\n") elif role == "assistant": lines.append(f"## Assistant\n\n{content}\n") return "\n".join(lines) # ============================================================ # P0核心：流水线状态、发布到市场、进化报告、团队共享 # ============================================================ def get_pipeline_status(self) -> dict: """获取流水线运行状态""" # scheduler运行状态（通过全局单例） watcher = EvolutionWatcherManager() scheduler = watcher._schedulers.get(self.user_id) scheduler_running = scheduler.is_running() if scheduler else False # 查询 user_evolution_config config = self.db.query(UserEvolutionConfig).filter( UserEvolutionConfig.user_id == self.user_id ).first() watcher_status = config.watcher_status if config else "unknown" pipeline_status = config.pipeline_status if config else "unknown" # discovered_count（待处理轨迹数） discovered_count = self.db.query(EvolutionTrajectory).filter( EvolutionTrajectory.user_id == self.user_id, EvolutionTrajectory.status == "discovered" ).count() # done_count（已完成轨迹数） done_count = self.db.query(EvolutionTrajectory).filter( EvolutionTrajectory.user_id == self.user_id, EvolutionTrajectory.status == "done" ).count() # total_skills total_skills = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id ).count() # total_trajectories total_trajectories = self.db.query(EvolutionTrajectory).filter( EvolutionTrajectory.user_id == self.user_id ).count() logger.info(f"get_pipeline_status: 用户 {self.user_id}, scheduler_running={scheduler_running}") return { "scheduler_running": scheduler_running, "watcher_status": watcher_status, "pipeline_status": pipeline_status, "discovered_count": discovered_count, "done_count": done_count, "total_skills": total_skills, "total_trajectories": total_trajectories, } def publish_to_market(self, skill_name: str, price: float = 0.00, market_description: str = "") -> dict: """将技能发布到应用市场""" skill = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id, EvolutionSkill.skill_name == skill_name ).first() if not skill: logger.warning(f"publish_to_market: 技能 {skill_name} 不存在") return {"error": "技能不存在"} # 验证技能 state=main，否则返回错误 if skill.state != "main": logger.warning(f"publish_to_market: 技能 {skill_name} 状态为 {skill.state}，不是main") return {"error": f"只有main状态的技能才能发布到市场，当前状态: {skill.state}"} # 查询是否已经发布过 existing_record = self.db.query(SkillPublishRecord).filter( SkillPublishRecord.user_id == self.user_id, SkillPublishRecord.skill_name == skill_name, SkillPublishRecord.status == "published" ).first() if existing_record: logger.warning(f"publish_to_market: 技能 {skill_name} 已发布过") return {"error": "该技能已发布到市场，不能重复发布"} # 在 market_apps 表中创建记录 market_app = MarketApp( name=f"{skill_name}(进化)", type="skill", source="user", publisher_id=self.user_id, description=market_description or skill.description or "", price=price, skill_content="来自AI自进化技能的SKILL.md", status="published", ) self.db.add(market_app) self.db.flush() # 在 skill_publish_records 表中创建记录 publish_record = SkillPublishRecord( skill_id=skill.id, skill_name=skill_name, user_id=self.user_id, market_app_id=market_app.id, price=price, market_description=market_description, status="published", ) self.db.add(publish_record) # 记录事件 self._add_event(skill_name, "published", f"技能 {skill_name} 已发布到市场，价格 {price}") self.db.commit() logger.info(f"publish_to_market: 技能 {skill_name} 已发布到市场，market_app_id={market_app.id}") return { "skill_name": skill_name, "market_app_id": market_app.id, "price": price, "status": "published", "publish_record_id": publish_record.id, } def generate_report(self, report_type: str = "weekly") -> dict: """生成进化报告（weekly/monthly）""" now = datetime.now() if report_type == "weekly": period_start = (now - timedelta(days=7)).date() period_end = now.date() elif report_type == "monthly": period_start = (now - timedelta(days=30)).date() period_end = now.date() else: return {"error": f"不支持的报告类型: {report_type}，仅支持 weekly/monthly"} # 新技能数（skills表中created_at在时间范围内的） new_skills_count = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id, EvolutionSkill.created_at >= period_start, EvolutionSkill.created_at <= period_end ).count() # 技能升级数（events中state_changed事件，event_type含 promoted） upgrades_count = self.db.query(EvolutionEvent).filter( EvolutionEvent.user_id == self.user_id, EvolutionEvent.created_at >= period_start, EvolutionEvent.created_at <= period_end, EvolutionEvent.event_type.in_(["promoted", "edited", "milestone"]) ).count() # 最活跃技能TOP5 top_skills = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id ).order_by(desc(EvolutionSkill.use_count)).limit(5).all() top_skills_data = [self._skill_to_dict(s) for s in top_skills] # 总调用次数 total_calls_raw = self.db.query(func.sum(EvolutionSkill.use_count)).filter( EvolutionSkill.user_id == self.user_id ).scalar() or 0 total_calls = int(total_calls_raw) if total_calls_raw else 0 # 等级变化（查询报告期内的promoted事件数作为等级变化的近似） level_changes = self.db.query(EvolutionEvent).filter( EvolutionEvent.user_id == self.user_id, EvolutionEvent.created_at >= period_start, EvolutionEvent.created_at <= period_end, EvolutionEvent.event_type == "promoted" ).count() # 维度分析（基于当前skills计算） skills = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id ).all() dim_scores = {d["key"]: {"label": d["label"], "value": 0, "maxValue": 10} for d in ABILITY_DIMENSIONS} for s in skills: dim_key = CATEGORY_DIMENSION_MAP.get(s.category, "qa") if dim_key in dim_scores: multiplier = {"main": 1.5, "staging": 1.0, "baby": 0.5, "frozen": 0.3}.get(s.state, 1.0) dim_scores[dim_key]["value"] += min(s.use_count * 0.2 * multiplier, 3) dimensions = [{"label": v["label"], "value": round(min(v["value"], 10), 1)} for v in dim_scores.values()] # 构建报告数据 report_data = { "report_type": report_type, "period_start": period_start.isoformat(), "period_end": period_end.isoformat(), "generated_at": now.isoformat(), "summary": { "new_skills_count": new_skills_count, "upgrades_count": upgrades_count, "total_skills": len(skills), "total_calls": total_calls, "level_changes": level_changes, }, "top_skills": top_skills_data, "dimensions": dimensions, } # 存入 evolution_reports 表 evolution_report = EvolutionReport( user_id=self.user_id, report_type=report_type, period_start=period_start, period_end=period_end, report_data=report_data, ) self.db.add(evolution_report) # 发送通知给用户 notification = Notification( user_id=self.user_id, type="evolution_report", title=f"{'周报' if report_type == 'weekly' else '月报'}已生成", content=f"你的AI进化{'周报' if report_type == 'weekly' else '月报'}已生成。" f"新增技能 {new_skills_count} 个，技能升级 {upgrades_count} 次，总调用次数 {total_calls}。", ) self.db.add(notification) self.db.commit() self.db.refresh(evolution_report) logger.info(f"generate_report: 用户 {self.user_id} 生成 {report_type} 报告, report_id={evolution_report.id}") return report_data def get_reports(self, report_type: Optional[str] = None) -> List[dict]: """获取用户所有报告列表""" query = self.db.query(EvolutionReport).filter( EvolutionReport.user_id == self.user_id ) if report_type: query = query.filter(EvolutionReport.report_type == report_type) reports = query.order_by(desc(EvolutionReport.created_at)).all() result = [] for r in reports: result.append({ "id": r.id, "report_type": r.report_type, "period_start": r.period_start.isoformat() if r.period_start else None, "period_end": r.period_end.isoformat() if r.period_end else None, "created_at": r.created_at.isoformat() if r.created_at else None, }) return result def get_report_detail(self, report_id: str) -> Optional[dict]: """获取报告详情""" report = self.db.query(EvolutionReport).filter( EvolutionReport.user_id == self.user_id, EvolutionReport.id == report_id ).first() if not report: return None return { "id": report.id, "user_id": report.user_id, "report_type": report.report_type, "period_start": report.period_start.isoformat() if report.period_start else None, "period_end": report.period_end.isoformat() if report.period_end else None, "report_data": report.report_data or {}, "created_at": report.created_at.isoformat() if report.created_at else None, } def share_to_team(self, skill_name: str, team_id: str) -> dict: """将技能分享到团队""" # 验证技能存在且属于当前用户 skill = self.db.query(EvolutionSkill).filter( EvolutionSkill.user_id == self.user_id, EvolutionSkill.skill_name == skill_name ).first() if not skill: logger.warning(f"share_to_team: 技能 {skill_name} 不存在或不属于用户 {self.user_id}") return {"error": "技能不存在或不属于当前用户"} # 验证用户属于该team（检查team存在且用户是owner，或者有team_member关系需要检查） team = self.db.query(Team).filter( Team.id == team_id ).first() if not team: logger.warning(f"share_to_team: team {team_id} 不存在") return {"error": "团队不存在"} # 检查用户是team的owner或属于该team（根据模型结构，Team有owner_id字段） is_owner = team.owner_id == self.user_id if not is_owner: logger.warning(f"share_to_team: 用户 {self.user_id} 不属于 team {team_id}") return {"error": "用户不属于该团队"} # 检查是否已分享 existing = self.db.query(TeamSharedSkill).filter( TeamSharedSkill.team_id == team_id, TeamSharedSkill.skill_id == skill.id ).first() if existing: logger.warning(f"share_to_team: 技能 {skill_name} 已分享至 team {team_id}") return {"error": "该技能已分享至此团队"} # 在 team_shared_skills 表中创建记录 shared_skill = TeamSharedSkill( team_id=team_id, skill_id=skill.id, skill_name=skill_name, shared_by=self.user_id, shared_by_name=None, is_public=True, ) self.db.add(shared_skill) # 记录事件 self._add_event(skill_name, "shared", f"技能 {skill_name} 已分享至团队 {team.name}") self.db.commit() self.db.refresh(shared_skill) logger.info(f"share_to_team: 技能 {skill_name} 已分享至 team {team_id}, shared_id={shared_skill.id}") return { "shared_id": shared_skill.id, "skill_name": skill_name, "team_id": team_id, "team_name": team.name, "shared_by": self.user_id, "is_public": True, } # ============================================================ # 辅助方法 # ============================================================ def _skill_to_dict(self, skill: EvolutionSkill) -> dict: return { "id": skill.id, "skill_name": skill.skill_name, "description": skill.description or "", "tags": skill.tags or [], "category": skill.category or "未分类", "state": skill.state, "use_count": skill.use_count, "version": skill.version, "last_used_at": skill.last_used_at.isoformat() if skill.last_used_at else None, "created_at": skill.created_at.isoformat() if skill.created_at else None, "updated_at": skill.updated_at.isoformat() if skill.updated_at else None, } def _event_to_dict(self, event: EvolutionEvent) -> dict: return { "id": event.id, "skill_name": event.skill_name, "event_type": event.event_type, "description": event.description or "", "version_before": event.version_before, "version_after": event.version_after, "created_at": event.created_at.isoformat() if event.created_at else None, } def _trajectory_to_dict(self, traj: EvolutionTrajectory) -> dict: return { "id": traj.id, "conversation_id": traj.conversation_id, "file_path": traj.file_path, "status": traj.status, "atom_count": traj.atom_count, "token_used": traj.token_used, "created_at": traj.created_at.isoformat() if traj.created_at else None, "processed_at": traj.processed_at.isoformat() if traj.processed_at else None, } def _add_event(self, skill_name: str, event_type: str, description: str, version_before: Optional[int] = None, version_after: Optional[int] = None): event = EvolutionEvent( user_id=self.user_id, skill_name=skill_name, event_type=event_type, description=description, version_before=version_before, version_after=version_after, ) self.db.add(event) def get_evolution_service(db: Session, user_id: str) -> EvolutionService: return EvolutionService(db, user_id)