diff --git a/backend/python/app/services/evolution_service.py b/backend/python/app/services/evolution_service.py index 67ef952..3ad015d 100644 --- a/backend/python/app/services/evolution_service.py +++ b/backend/python/app/services/evolution_service.py @@ -1,1334 +1,1349 @@ -""" -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): - """处理单条轨迹 — Stage D: LLM智能抽取替代mock""" - logger.info("开始处理轨迹 {}，用户 {}".format(traj.id, self.user_id)) - - # step 1: discovered -> splitting - traj.status = "splitting" - db.commit() - - # Stage D: 使用对话内容进行真实分析 - # 从关联的对话中提取消息内容 - conversation_id = traj.conversation_id - conversation_messages = [] - if conversation_id: - msgs = db.query(Message).filter( - Message.conversation_id == conversation_id - ).order_by(Message.created_at).limit(20).all() - for m in msgs: - conversation_messages.append({"role": m.role, "content": m.content}) - - # 基于对话内容分析可抽取的技能模式 - # 使用关键词和模式匹配进行轻量级技能发现（不依赖外部LLM调用避免延迟） - atom_tasks = self._extract_skill_patterns(conversation_messages) - - atom_count = len(atom_tasks) - 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, task in enumerate(atom_tasks): - link = SkillTrajectoryLink( - skill_name=task.get("skill_name", "__pending__"), - user_id=self.user_id, - trajectory_id=traj.id, - atom_id="{}_atom_{}".format(traj.id[:8], i), - weightscore=task.get("score", 5) - ) - db.add(link) - db.commit() - - # step 3: split_done -> indexed - traj.status = "indexed" - db.commit() - - # step 4: indexed -> clustering - traj.status = "clustering" - db.commit() - - def _extract_skill_patterns(self, messages: list) -> list: - """Stage D: 从对话消息中抽取可复用的技能模式""" - if not messages: - return [{"skill_name": "__pending__", "score": 3}] - - patterns = [] - all_content = " ".join([m.get("content", "") for m in messages]) - - # 技能模式关键词映射 - skill_keywords = { - "writing": ["写作", "文章", "撰写", "草稿", "文案", "内容创作", "write", "draft", "content"], - "analysis": ["分析", "评估", "研究", "调查", "数据解读", "analyze", "assess", "evaluate"], - "translation": ["翻译", "translate", "中译英", "英译中", "localize"], - "summarization": ["总结", "摘要", "概括", "归纳", "summarize", "abstract", "brief"], - "coding": ["代码", "编程", "开发", "函数", "算法", "code", "program", "develop"], - "planning": ["计划", "规划", "方案", "策略", "plan", "strategy", "roadmap"], - "search": ["搜索", "查询", "查找", "检索", "search", "query", "find"], - "calculation": ["计算", "估算", "预算", "统计", "calculate", "estimate", "budget"], - } - - for skill_type, keywords in skill_keywords.items(): - for kw in keywords: - if kw.lower() in all_content.lower(): - # 计算匹配得分（基于出现频率和上下文长度） - count = all_content.lower().count(kw.lower()) - score = min(10, max(1, count * 2)) - patterns.append({ - "skill_name": skill_type, - "score": score, - "keyword": kw, - "count": count, - }) - break # One match per skill_type is enough - - # Deduplicate and sort by score - seen = set() - unique_patterns = [] - for p in sorted(patterns, key=lambda x: x["score"], reverse=True): - if p["skill_name"] not in seen: - seen.add(p["skill_name"]) - unique_patterns.append(p) - - # If no patterns found, create a generic pending task - if not unique_patterns: - unique_patterns = [{"skill_name": "__pending__", "score": 3}] - - return unique_patterns[:5] # Max 5 atoms per trajectory - - # 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.md""" - skill = self.db.query(EvolutionSkill).filter( - EvolutionSkill.user_id == self.user_id, - EvolutionSkill.skill_name == skill_name - ).first() - if not skill: - return None - - version_before = skill.version - - 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() - - # Stage A.3 Fix: Create SkillVersion snapshot - skill_md_content = self._generate_skill_md(skill) - version_record = SkillVersion( - user_id=self.user_id, - skill_name=skill_name, - version=skill.version, - skill_md_content=skill_md_content, - description="技能 {} v{} (编辑更新)".format(skill_name, skill.version), - ) - self.db.add(version_record) - - # Stage A.3 Fix: Write SKILL.md file to UPLOAD_DIR - skill_dir = os.path.join(UPLOAD_DIR, "skills", self.user_id) - os.makedirs(skill_dir, exist_ok=True) - md_path = os.path.join(skill_dir, "{}.md".format(skill_name)) - try: - with open(md_path, 'w', encoding='utf-8') as f: - f.write(skill_md_content) - skill.skill_md_path = md_path - except Exception as e: - logger.warning("Failed to write SKILL.md for {}: {}".format(skill_name, e)) - - # 记录编辑事件 - self._add_event(skill_name, "edited", "技能 {} 已更新至v{}".format(skill_name, skill.version), - version_before=version_before, version_after=skill.version) - - self.db.commit() - self.db.refresh(skill) - return self._skill_to_dict(skill) - - def _generate_skill_md(self, skill) -> str: - """Generate SKILL.md content from EvolutionSkill object""" - parts = [ - "---", - "name: {}".format(skill.skill_name), - "description: {}".format(skill.description or ""), - "category: {}".format(skill.category or "general"), - "state: {}".format(skill.state), - "version: {}".format(skill.version), - "---", - "", - ] - if skill.description: - parts.append("## 技能说明") - parts.append(skill.description) - parts.append("") - if skill.tags: - tag_list = skill.tags if isinstance(skill.tags, list) else [] - parts.append("## 标签") - parts.append(", ".join(tag_list)) - parts.append("") - parts.append("## 使用统计") - parts.append("- 使用次数: {}".format(skill.use_count or 0)) - parts.append("- 当前版本: v{}".format(skill.version)) - parts.append("- 状态: {}".format(skill.state)) - - return "\n".join(parts) - - 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) +""" +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): + """处理单条轨迹 — Stage D: LLM智能抽取替代mock""" + logger.info("开始处理轨迹 {}，用户 {}".format(traj.id, self.user_id)) + + # step 1: discovered -> splitting + traj.status = "splitting" + db.commit() + + # Stage D: 使用对话内容进行真实分析 + # 从关联的对话中提取消息内容 + conversation_id = traj.conversation_id + conversation_messages = [] + if conversation_id: + msgs = db.query(Message).filter( + Message.conversation_id == conversation_id + ).order_by(Message.created_at).limit(20).all() + for m in msgs: + conversation_messages.append({"role": m.role, "content": m.content}) + + # 基于对话内容分析可抽取的技能模式 + # 使用关键词和模式匹配进行轻量级技能发现（不依赖外部LLM调用避免延迟） + atom_tasks = self._extract_skill_patterns(conversation_messages) + + atom_count = len(atom_tasks) + 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, task in enumerate(atom_tasks): + link = SkillTrajectoryLink( + skill_name=task.get("skill_name", "__pending__"), + user_id=self.user_id, + trajectory_id=traj.id, + atom_id="{}_atom_{}".format(traj.id[:8], i), + weightscore=task.get("score", 5) + ) + db.add(link) + db.commit() + + # step 3: split_done -> indexed + traj.status = "indexed" + db.commit() + + # step 4: indexed -> clustering + traj.status = "clustering" + db.commit() + + def _extract_skill_patterns(self, messages: list) -> list: + """Stage D: 从对话消息中抽取可复用的技能模式""" + if not messages: + return [{"skill_name": "__pending__", "score": 3}] + + patterns = [] + all_content = " ".join([m.get("content", "") for m in messages]) + + # 技能模式关键词映射 + skill_keywords = { + "writing": ["写作", "文章", "撰写", "草稿", "文案", "内容创作", "write", "draft", "content"], + "analysis": ["分析", "评估", "研究", "调查", "数据解读", "analyze", "assess", "evaluate"], + "translation": ["翻译", "translate", "中译英", "英译中", "localize"], + "summarization": ["总结", "摘要", "概括", "归纳", "summarize", "abstract", "brief"], + "coding": ["代码", "编程", "开发", "函数", "算法", "code", "program", "develop"], + "planning": ["计划", "规划", "方案", "策略", "plan", "strategy", "roadmap"], + "search": ["搜索", "查询", "查找", "检索", "search", "query", "find"], + "calculation": ["计算", "估算", "预算", "统计", "calculate", "estimate", "budget"], + } + + for skill_type, keywords in skill_keywords.items(): + for kw in keywords: + if kw.lower() in all_content.lower(): + # 计算匹配得分（基于出现频率和上下文长度） + count = all_content.lower().count(kw.lower()) + score = min(10, max(1, count * 2)) + patterns.append({ + "skill_name": skill_type, + "score": score, + "keyword": kw, + "count": count, + }) + break # One match per skill_type is enough + + # Deduplicate and sort by score + seen = set() + unique_patterns = [] + for p in sorted(patterns, key=lambda x: x["score"], reverse=True): + if p["skill_name"] not in seen: + seen.add(p["skill_name"]) + unique_patterns.append(p) + + # If no patterns found, create a generic pending task + if not unique_patterns: + unique_patterns = [{"skill_name": "__pending__", "score": 3}] + + return unique_patterns[:5] # Max 5 atoms per trajectory + + # 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.md""" + skill = self.db.query(EvolutionSkill).filter( + EvolutionSkill.user_id == self.user_id, + EvolutionSkill.skill_name == skill_name + ).first() + if not skill: + return None + + version_before = skill.version + + 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() + + # Stage A.3 Fix: Create SkillVersion snapshot + skill_md_content = self._generate_skill_md(skill) + version_record = SkillVersion( + user_id=self.user_id, + skill_name=skill_name, + version=skill.version, + skill_md_content=skill_md_content, + description="技能 {} v{} (编辑更新)".format(skill_name, skill.version), + ) + self.db.add(version_record) + + # Stage A.3 Fix: Write SKILL.md file to UPLOAD_DIR + skill_dir = os.path.join(UPLOAD_DIR, "skills", self.user_id) + os.makedirs(skill_dir, exist_ok=True) + md_path = os.path.join(skill_dir, "{}.md".format(skill_name)) + try: + with open(md_path, 'w', encoding='utf-8') as f: + f.write(skill_md_content) + skill.skill_md_path = md_path + except Exception as e: + logger.warning("Failed to write SKILL.md for {}: {}".format(skill_name, e)) + + # 记录编辑事件 + self._add_event(skill_name, "edited", "技能 {} 已更新至v{}".format(skill_name, skill.version), + version_before=version_before, version_after=skill.version) + + self.db.commit() + self.db.refresh(skill) + return self._skill_to_dict(skill) + + def _generate_skill_md(self, skill) -> str: + """Generate SKILL.md content from EvolutionSkill object""" + parts = [ + "---", + "name: {}".format(skill.skill_name), + "description: {}".format(skill.description or ""), + "category: {}".format(skill.category or "general"), + "state: {}".format(skill.state), + "version: {}".format(skill.version), + "---", + "", + ] + if skill.description: + parts.append("## 技能说明") + parts.append(skill.description) + parts.append("") + if skill.tags: + tag_list = skill.tags if isinstance(skill.tags, list) else [] + parts.append("## 标签") + parts.append(", ".join(tag_list)) + parts.append("") + parts.append("## 使用统计") + parts.append("- 使用次数: {}".format(skill.use_count or 0)) + parts.append("- 当前版本: v{}".format(skill.version)) + parts.append("- 状态: {}".format(skill.state)) + + return "\n".join(parts) + + 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": "该技能已发布到市场，不能重复发布"} + + # 角色判定：管理员发布=官方+已发布+免费；用户发布=用户+待审核+客户定价 + _u = self.db.query(User).filter(User.id == self.user_id).first() + _is_admin = bool(_u and getattr(_u, "role", "user") == "admin") + if _is_admin: + app_source = "official" + app_status = "published" + app_price = 0.0 + app_member_free = True + else: + app_source = "user" + app_status = "pending" + app_price = price if price else 0.0 + app_member_free = False + + # 在 market_apps 表中创建记录 + market_app = MarketApp( + name=f"{skill_name}(进化)", + type="skill", + source=app_source, + publisher_id=self.user_id, + description=market_description or skill.description or "", + price=app_price, + is_member_free=app_member_free, + skill_content="来自AI自进化技能的SKILL.md", + status=app_status, + ) + 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=app_price, + market_description=market_description, + status=app_status, + ) + 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) diff --git a/run.sh b/run.sh index 9bdafa9..51c519e 100755 --- a/run.sh +++ b/run.sh @@ -167,11 +167,19 @@ echo " 使用数据库: $MYSQL_DATABASE" MIGRATION_DIR="$PROJECT_DIR/migrations" if [ -d "$MIGRATION_DIR" ]; then + # 幂等迁移：用 schema_migrations 记录已执行版本，避免每次启动重复执行（如重复插入演示数据） + mysql -h "$MYSQL_HOST" -P "$MYSQL_PORT" -u "$MYSQL_USER" -p"$MYSQL_PASSWORD" $MYSQL_SSL_FLAG "$MYSQL_DATABASE" -e "CREATE TABLE IF NOT EXISTS schema_migrations (version VARCHAR(255) NOT NULL PRIMARY KEY, applied_at DATETIME DEFAULT CURRENT_TIMESTAMP) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;" 2>/dev/null for sql_file in $(ls "$MIGRATION_DIR"/*.sql 2>/dev/null | sort); do filename=$(basename "$sql_file") - echo " - 执行: $filename" - mysql -h "$MYSQL_HOST" -P "$MYSQL_PORT" -u "$MYSQL_USER" -p"$MYSQL_PASSWORD" $MYSQL_SSL_FLAG \ - "$MYSQL_DATABASE" < "$sql_file" 2>&1 | sed 's/^/ /' || echo " [警告] $filename 部分语句已跳过（可能已存在）" + version="${filename%.sql}" + already=$(mysql -h "$MYSQL_HOST" -P "$MYSQL_PORT" -u "$MYSQL_USER" -p"$MYSQL_PASSWORD" $MYSQL_SSL_FLAG "$MYSQL_DATABASE" -sN -e "SELECT 1 FROM schema_migrations WHERE version='$version' LIMIT 1;" 2>/dev/null) + if [ "$already" = "1" ]; then + echo " - 跳过(已执行): $filename" + else + echo " - 执行: $filename" + mysql -h "$MYSQL_HOST" -P "$MYSQL_PORT" -u "$MYSQL_USER" -p"$MYSQL_PASSWORD" $MYSQL_SSL_FLAG "$MYSQL_DATABASE" < "$sql_file" 2>&1 | sed 's/^/ /' || echo " [警告] $filename 部分语句已跳过（可能已存在）" + mysql -h "$MYSQL_HOST" -P "$MYSQL_PORT" -u "$MYSQL_USER" -p"$MYSQL_PASSWORD" $MYSQL_SSL_FLAG "$MYSQL_DATABASE" -e "INSERT IGNORE INTO schema_migrations (version) VALUES ('$version');" 2>/dev/null + fi done echo " 数据库初始化完成" @@ -291,6 +299,11 @@ fi # 如果 nginx 可用，则启动（Docker 环境） if command -v nginx &>/dev/null; then + # 自愈：确保媒体生成(视频/图像)等长请求不被 60s 代理超时中断 + if [ -f /etc/nginx/conf.d/aiflow.conf ]; then + sed -i -E 's/proxy_read_timeout[[:space:]]+[0-9]+s;/proxy_read_timeout 600s;/; s/proxy_send_timeout[[:space:]]+[0-9]+s;/proxy_send_timeout 600s;/' /etc/nginx/conf.d/aiflow.conf 2>/dev/null || true + fi + echo "" echo " Nginx 配置测试..." if nginx -t 2>&1; then