""" Neural Sync - Role Dispatcher 根据任务的技术指纹自动路由到对应角色的工作台。 """ from enum import Enum from dataclasses import dataclass, field from typing import Dict, List, Optional, Any, Tuple from datetime import datetime import re import logging from pydantic import BaseModel, Field logger = logging.getLogger(__name__) class Role(str, Enum): """系统角色定义""" BOSS = "boss" # 老板 - 战略决策 AI_ENGINEER = "ai_engineer" # AI 工程师 - Prompt/模型/RAG BACKEND = "backend" # 后端开发 - API/数据库/爬虫 FRONTEND = "frontend" # 前端开发 - UI/组件/交互 @dataclass class RoutingRule: """路由规则定义""" role: Role name: str description: str triggers: List[str] = field(default_factory=list) # 触发关键词 keywords: List[str] = field(default_factory=list) # 技术关键词 patterns: List[str] = field(default_factory=list) # 正则模式 threshold: Optional[Dict[str, Any]] = None # 阈值条件 priority: int = 0 # 优先级，数值越大越优先 class RoleRouter: """ 角色路由引擎 根据信号的技术指纹自动路由到对应角色： - Boss: 高成本、高风险、架构变更决策 - AI Engineer: LLM、Prompt、RAG、数据处理 - Backend: API、数据库、爬虫、性能优化 - Frontend: UI、可视化、交互、响应式 """ # 角色路由规则定义 ROLE_RULES: List[RoutingRule] = [ RoutingRule( role=Role.BOSS, name="战略决策", description="需要老板决策的高影响任务", triggers=["cost_estimate", "architecture_change", "high_risk", "security"], keywords=["预算", "架构", "安全", "风险", "审批", "决策"], threshold={"dev_days": 5, "risk_level": ["CRITICAL", "HIGH"]}, priority=100, ), RoutingRule( role=Role.AI_ENGINEER, name="AI 工程", description="AI 模型和数据处理相关任务", triggers=["prompt", "model", "embedding", "rag", "llm"], keywords=[ "ollama", "openai", "claude", "prompt", "embedding", "rag", "llm", "向量", "模型", "训练", "数据清洗", "token", "context", ], patterns=[r"(prompt|embedding|rag|llm|vector|model)", r"(ai|machine.?learning|ml)"], priority=50, ), RoutingRule( role=Role.BACKEND, name="后端开发", description="API、数据库、爬虫等后端任务", triggers=["api", "database", "crawler", "performance", "docker"], keywords=[ "fastapi", "sqlalchemy", "postgres", "sql", "async", "docker", "kubernetes", "redis", "celery", "爬虫", "api", "接口", "数据库", "后端", "性能", "缓存", ], patterns=[ r"(api|endpoint|route)", r"(sql|database|db)", r"(docker|k8s|kubernetes)", r"(async|await|celery)", ], priority=30, ), RoutingRule( role=Role.FRONTEND, name="前端开发", description="UI、可视化、交互等前端任务", triggers=["ui", "visualization", "interaction", "responsive", "chart"], keywords=[ "vue", "react", "typescript", "component", "chart", "echarts", "d3", "可视化", "ui", "ux", "样式", "动画", "响应式", "组件", "交互", "页面", "布局", ], patterns=[ r"(vue|react|tsx?|jsx?)", r"(css|scss|tailwind)", r"(component|props|emit)", r"(chart|graph|visualization)", ], priority=30, ), ] # 技术栈关键词映射 TECH_KEYWORDS = { "frontend": [ "vue", "react", "typescript", "javascript", "css", "scss", "tailwind", "component", "props", "emit", "template", "style", ], "backend": [ "fastapi", "flask", "django", "sqlalchemy", "postgres", "mysql", "redis", "celery", "async", "await", "docker", "kubernetes", "k8s", "api", "endpoint", "route", ], "ai": [ "llm", "ollama", "openai", "claude", "prompt", "embedding", "rag", "vector", "model", "training", "inference", "token", "context", ], } def __init__(self): # 编译正则模式 self._compiled_patterns: Dict[Role, List[re.Pattern]] = {} for rule in self.ROLE_RULES: patterns = [re.compile(p, re.IGNORECASE) for p in rule.patterns] self._compiled_patterns[rule.role] = patterns def route( self, signal_content: str, analysis: Optional[Dict[str, Any]] = None, source: Optional[str] = None, ) -> Tuple[Role, float, Dict[str, Any]]: """ 根据信号内容和分析结果决定路由角色 Args: signal_content: 信号内容 analysis: AI 分析结果（可选） source: 信号来源（可选） Returns: (角色, 置信度, 路由详情) """ analysis = analysis or {} details = { "matched_rules": [], "scores": {}, "reasons": [], } # 1. 首先检查是否需要升级到老板决策 if self._should_escalate_to_boss(analysis): details["reasons"].append("触发老板决策阈值") details["matched_rules"].append("threshold_check") return Role.BOSS, 1.0, details # 2. 计算每个角色的匹配分数 scores: Dict[Role, float] = {rule.role: 0.0 for rule in self.ROLE_RULES} # 基于关键词匹配 content_lower = signal_content.lower() for rule in self.ROLE_RULES: keyword_score = sum( 1 for kw in rule.keywords if kw.lower() in content_lower ) if keyword_score > 0: scores[rule.role] += keyword_score * 0.3 details["scores"][f"{rule.role}_keywords"] = keyword_score # 基于正则模式匹配 for rule in self.ROLE_RULES: if rule.role in self._compiled_patterns: pattern_score = sum( 1 for pattern in self._compiled_patterns[rule.role] if pattern.search(signal_content) ) if pattern_score > 0: scores[rule.role] += pattern_score * 0.5 details["scores"][f"{rule.role}_patterns"] = pattern_score # 基于触发词匹配 for rule in self.ROLE_RULES: trigger_score = sum( 1 for trigger in rule.triggers if trigger.lower() in content_lower ) if trigger_score > 0: scores[rule.role] += trigger_score * 0.7 details["scores"][f"{rule.role}_triggers"] = trigger_score # 基于优先级加权 for rule in self.ROLE_RULES: if scores[rule.role] > 0: scores[rule.role] *= (1 + rule.priority * 0.01) # 3. 找出最高分的角色 max_role = Role.AI_ENGINEER # 默认 AI 工程师 max_score = 0.0 for role, score in scores.items(): if score > max_score: max_score = score max_role = role elif score == max_score and score > 0: # 分数相同时，优先级高的胜出 for rule in self.ROLE_RULES: if rule.role == role: role_priority = rule.priority if rule.role == max_role: max_role_priority = rule.priority if role_priority > max_role_priority: max_role = role # 归一化置信度 confidence = min(max_score / 3.0, 1.0) if max_score > 0 else 0.5 # 记录详情 details["matched_rules"] = [ rule.name for rule in self.ROLE_RULES if scores[rule.role] > 0 ] details["final_scores"] = {role.value: score for role, score in scores.items()} details["reasons"].append(f"基于关键词和模式匹配") return max_role, confidence, details def _should_escalate_to_boss(self, analysis: Dict[str, Any]) -> bool: """检查是否需要升级到老板决策""" # 检查开发成本 dev_days = analysis.get("resource_cost", {}).get("dev_days", 0) if dev_days >= 5: return True # 检查风险等级 risk_level = analysis.get("risk_level", "") if risk_level in ["CRITICAL", "HIGH"]: return True # 检查是否涉及架构变更 is_architecture = analysis.get("is_architecture_change", False) if is_architecture: return True return False def get_role_info(self, role: Role) -> Dict[str, Any]: """获取角色信息""" for rule in self.ROLE_RULES: if rule.role == role: return { "role": role.value, "name": rule.name, "description": rule.description, "keywords": rule.keywords, "priority": rule.priority, } return {} def get_all_roles(self) -> List[Dict[str, Any]]: """获取所有角色信息""" return [ { "role": rule.role.value, "name": rule.name, "description": rule.description, "priority": rule.priority, } for rule in sorted(self.ROLE_RULES, key=lambda r: -r.priority) ] class RoutingResult(BaseModel): """路由结果模型""" role: str confidence: float details: Dict[str, Any] signal_id: Optional[str] = None timestamp: float = Field(default_factory=lambda: datetime.utcnow().timestamp()) # 全局路由器实例 _global_router: Optional[RoleRouter] = None def get_router() -> RoleRouter: """获取全局路由器实例""" global _global_router if _global_router is None: _global_router = RoleRouter() return _global_router # 便捷函数 def route_signal( signal_content: str, analysis: Optional[Dict[str, Any]] = None, signal_id: Optional[str] = None, ) -> RoutingResult: """ 路由信号到对应角色 Args: signal_content: 信号内容 analysis: AI 分析结果 signal_id: 信号 ID Returns: 路由结果 """ router = get_router() role, confidence, details = router.route(signal_content, analysis) return RoutingResult( role=role.value, confidence=confidence, details=details, signal_id=signal_id, )