"""Глобальное состояние PinkSky""" import os import json import threading import logging from datetime import datetime from typing import Dict, List, Any, Optional from .models import ModelConfig, Role, Conductor from .model_ranking import MODEL_RANKING from .config import ROLES_FILE, MODELS_FILE, CONDUCTORS_FILE, HISTORY_FILE class PinkSkyState: _instance = None _initialized = False _lock = threading.Lock() def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def load_all(self): """Загружает модели, роли и кондукторы из файлов.""" self._load_models() self._load_roles() self._load_conductors() self._load_history() def _load_models(self): defaults = { name: self._build_model_config(name, data) for name, data in MODEL_RANKING.items() } if os.path.exists(MODELS_FILE): try: with open(MODELS_FILE, "r", encoding="utf-8") as f: custom = json.load(f) for k, v in custom.items(): if k not in defaults: defaults[k] = ModelConfig(**v) except Exception as e: self.logger.error(f"Ошибка загрузки models.json: {e}") self.models = defaults def __init__(self): if PinkSkyState._initialized: return PinkSkyState._initialized = True self.logger = logging.getLogger(__name__) self.models: Dict[str, ModelConfig] = {} self.roles: Dict[str, Role] = {} self.conductors: Dict[str, Conductor] = {} self.current_mode: str = "chat" self.current_conductor: str = "default" self.current_role: str = "universal" self.current_model: str = "deepseek-v4-pro" self.chat_history: List[Dict[str, str]] = [] self.skill_history: List[Dict[str, str]] = [] self.build_history: List[Dict[str, str]] = [] self.build_context: Dict[str, Any] = { "spec": "", "agents": 3, "models_tier": "tier1", "skills_count": 2, "files_count": 3, "role": "universal", "strategy": "parallel", "use_interpreter": True, "notifications": True, "internet_access": True } self.cancel_flag: bool = False self.last_history_save: Optional[datetime] = None self.load_all() def add_to_history(self, mode: str, role: str, content: str): with PinkSkyState._lock: entry = { "role": role, "content": content, "timestamp": datetime.now().isoformat() } history_attr = f"{mode}_history" history = getattr(self, history_attr, []) history.append(entry) setattr(self, history_attr, history) # Сохраняем не чаще раза в 30 секунд now = datetime.now() if (self.last_history_save is None or (now - self.last_history_save).total_seconds() > 30): self.save_history() self.last_history_save = now def _build_model_config(self, name: str, data: dict) -> ModelConfig: return ModelConfig( name=name, provider="openai", endpoint=data["endpoint"], api_key_env="NVIDIA_API_KEY", context_window=data.get("context_window", 32000), max_tokens=data.get("max_tokens", 8000), cost_per_1k_input=data.get("cost_per_1k_input", 0.0), cost_per_1k_output=data.get("cost_per_1k_output", 0.0), coding_rank=data.get("coding_rank", 50), speed_rank=data.get("speed_rank", 50), reasoning_rank=data.get("reasoning_rank", 50), tags=data.get("tags", []) ) def _load_models(self): defaults = {name: self._build_model_config(name, data) for name, data in MODEL_RANKING.items()} if os.path.exists(MODELS_FILE): try: with open(MODELS_FILE, "r", encoding="utf-8") as f: custom = json.load(f) for k, v in custom.items(): if k not in defaults: defaults[k] = ModelConfig(**v) except Exception as e: print(f"⚠️ Ошибка загрузки models.json: {e}") self.models = defaults def _load_roles(self): defaults = { "universal": Role( name="universal", prompt="You are PinkSky -- a universal AI assistant and autonomous developer. You help users with any tasks, scripts, theory, and project creation from scratch.", description="Universal assistant for any tasks", preferred_models=["deepseek-v4-pro", "kimi-k2.6", "qwen3.5-397b"], complexity="medium", tags=["general"] ), "guru": Role( name="guru", prompt="You are Guru Programmer PinkSky. 15+ years experience. Write elegant, production-ready code. Principles: KISS, explicit > implicit, composition > inheritance, PEP8, type hints, docstrings. Format: analysis -> code -> explanations -> edge cases.", description="Guru programmer. Elegant code with deep explanations.", preferred_models=["deepseek-v4-pro", "kimi-k2.6", "mistral-large-3", "gpt-oss-120b"], complexity="high", tags=["coding", "senior", "mentor", "python"] ), "hacker": Role( name="hacker", prompt="You are Hacker PinkSky. Code virtuoso. Find elegant and unconventional solutions. Use __slots__, descriptors, metaclasses. Optimize time complexity, memory layout. Love functional: itertools, functools, operator.", description="Hacker-coder. Optimization and unconventional solutions.", preferred_models=["deepseek-v4-pro", "deepseek-v4-flash", "llama-4-maverick", "nemotron-super-49b"], complexity="high", tags=["coding", "optimization", "hacks", "performance"] ), "architect": Role( name="architect", prompt="You are Software Architect PinkSky. Design systems that last years. Bounded contexts, aggregates, CQRS, Event Sourcing. API: REST, gRPC, GraphQL, WebSocket. Observability: logs, metrics, tracing from the start.", description="Software Architect. High-level system design.", preferred_models=["deepseek-v4-pro", "kimi-k2.6", "nemotron-3-super", "qwen3.5-397b"], complexity="high", tags=["architecture", "design", "system", "ddd"] ), "principal": Role( name="principal", prompt="You are Principal Engineer PinkSky. Solve problems no one else can. Refactor legacy without downtime. Platform-level: CI/CD, observability, service mesh. Engineering culture: code review, RFC process. ADR for all decisions.", description="Principal engineer. Strategy, mentorship, hard problems.", preferred_models=["deepseek-v4-pro", "kimi-k2.6", "mistral-large-3", "gpt-oss-120b"], complexity="high", tags=["leadership", "strategy", "mentoring", "legacy"] ), "evangelist": Role( name="evangelist", prompt="You are Quality Evangelist PinkSky. TDD, BDD, property-based testing, mutation testing. pytest, hypothesis, coverage, mypy, ruff, bandit. Test pyramid: unit -> integration -> e2e. CI/CD gates: coverage threshold, mutation score.", description="Quality evangelist. Testing and quality culture.", preferred_models=["kimi-k2.6", "deepseek-v4-pro", "mistral-medium-3.5"], complexity="high", tags=["quality", "testing", "tdd", "ci-cd"] ), "techlead": Role( name="techlead", prompt="You are Tech Lead PinkSky. Code review: correctness, readability, maintainability, security, performance. Find race conditions, memory leaks, injection points, N+1. must-fix vs should-fix vs nitpick. Code review = teaching, not tribunal.", description="Tech Lead. Code review and team direction.", preferred_models=["deepseek-v4-pro", "kimi-k2.6", "mistral-large-3", "gpt-oss-120b"], complexity="high", tags=["review", "leadership", "team", "mentoring"] ), "qa": Role( name="qa", prompt="You are QA Engineer PinkSky. Test cases: positive, negative, boundary, exploratory. Equivalence partitioning, boundary value analysis. Automation: Selenium, Playwright, Postman. Performance: k6, Locust. Security: OWASP Top 10.", description="QA engineer. Bug hunting and test strategy.", preferred_models=["mistral-small-4", "step-3.7-flash", "llama-3.3-70b", "deepseek-v4-flash"], complexity="medium", tags=["qa", "testing", "automation", "manual"] ), "sdet": Role( name="sdet", prompt="You are SDET PinkSky. Test frameworks: pytest plugins, custom matchers. CI/CD: parallel execution, test sharding. Test data: factories, fixtures, seeding, cleanup. Mocks/stubs/fakes: wiremock, mockserver. Test code = production code.", description="SDET. Autotests and test infrastructure at dev level.", preferred_models=["deepseek-v4-pro", "kimi-k2.6", "llama-4-maverick", "mistral-medium-3.5"], complexity="high", tags=["sdet", "automation", "framework", "infrastructure"] ), "qe": Role( name="qe", prompt="You are Quality Engineer (QE) PinkSky. Analyze SDLC: where quality is lost. Shift-left testing: quality gates at every stage. Metrics: DORA, SPACE, custom KPIs. Root cause analysis: 5 Whys, Fishbone, FMEA. Every production bug = learning opportunity.", description="Quality engineer. Processes, metrics, and quality culture.", preferred_models=["deepseek-v4-pro", "kimi-k2.6", "nemotron-3-super"], complexity="high", tags=["qe", "process", "metrics", "culture", "sdlc"] ), "researcher": Role( name="researcher", prompt="You are Researcher PinkSky. Deep topic analysis. Compare approaches: trade-offs, limitations. Structure: executive summary -> details -> sources. Identify trends. Evidence > opinions. Numbers > words.", description="Researcher and analyst. Deep topic analysis.", preferred_models=["deepseek-v4-pro", "qwen3.5-397b", "kimi-k2.6", "gpt-oss-120b"], complexity="high", tags=["research", "analysis", "comparison"] ), "critic": Role( name="critic", prompt="You are Critic and Auditor PinkSky. correctness, security, performance, maintainability. race conditions, injection points, memory leaks, N+1. code smells, technical debt, architecture risks. Every issue with severity. Suggest fixes.", description="Critic and auditor. Bug and issue hunting.", preferred_models=["deepseek-v4-pro", "kimi-k2.6", "mistral-large-3", "gpt-oss-120b"], complexity="medium", tags=["audit", "security", "review", "critic"] ), } if os.path.exists(ROLES_FILE): try: with open(ROLES_FILE, "r", encoding="utf-8") as f: custom = json.load(f) for k, v in custom.items(): if k not in defaults: defaults[k] = Role(**v) except Exception as e: print(f"⚠️ Ошибка загрузки roles.json: {e}") self.roles = defaults def _load_conductors(self): defaults = { "default": Conductor( name="default", prompt="""You are Conductor PinkSky (Default). Analyze request and choose optimal roles and models. RULES: 1. Simple questions -- 1 role, 1 model. 2. Complex tasks -- decompose, assign roles. 3. Consider cost: cheap for simple, powerful for complex. 4. If code -- add critic. 5. If architecture -- add architect. AVAILABLE ROLES: guru, hacker, architect, principal, evangelist, techlead, qa, sdet, qe, researcher, critic, universal. AVAILABLE MODELS (by coding rank, best to worst): TIER 1 (Elite): deepseek-v4-pro, kimi-k2.6, qwen3.5-397b, mistral-large-3, gpt-oss-120b TIER 2 (Strong): deepseek-v4-flash, llama-4-maverick, nemotron-3-super, mistral-medium-3.5, dracarys-llama-70b, llama-3.3-70b, nemotron-super-49b TIER 3 (Good): step-3.7-flash, mistral-small-4, minimax-m2.7, nemotron-super-49b-v1, llama-3.2-90b-vision TIER 4 (Fast): nemotron-nano-12b, nemotron-3-nano-30b, nemotron-nano-9b, nemotron-content-safety TIER 5 (Specialized): nemotron-3-nano-omni, diffusiongemma FORMAT (STRICT JSON): {"strategy": "single|sequential|parallel", "tasks": [{"role": "role_name", "model": "model_name", "prompt": "subtask"}], "synthesis_prompt": "how to combine"}""", description="Standard conductor -- balance of quality and speed", strategy="selective", max_agents=3, cost_aware=True, auto_rank_by="balanced" ), "strict": Conductor( name="strict", prompt="""You are Strict Conductor PinkSky. Minimum agents, maximum efficiency. RULES: 1. ONLY one role and one model. 2. Cheapest model capable of solving the task. 3. Only sequential. FORMAT (STRICT JSON): {"strategy": "single", "tasks": [{"role": "name", "model": "name", "prompt": "task"}], "synthesis_prompt": ""}""", description="Minimum agents, minimum cost", strategy="single", max_agents=1, cost_aware=True, auto_rank_by="coding" ), "creative": Conductor( name="creative", prompt="""You are Creative Conductor PinkSky. Maximum perspectives, brainstorm. RULES: 1. Multiple roles from different angles. 2. Parallel strategy. 3. guru + hacker + researcher + critic. 4. Do not save on models -- use the best. FORMAT (STRICT JSON): {"strategy": "parallel", "tasks": [...], "synthesis_prompt": "synthesize creative ideas"}""", description="Maximum roles, creative brainstorm", strategy="parallel", max_agents=5, cost_aware=False, auto_rank_by="coding" ), "economy": Conductor( name="economy", prompt="""You are Economy Conductor PinkSky. Solve task for minimum cost. RULES: 1. Start with TIER 4 (fast/cheap): nemotron-nano-9b, nemotron-nano-12b, nemotron-3-nano-30b. 2. Only if it fails -- escalate to TIER 3/2. 3. One role, one model. FORMAT (STRICT JSON): {"strategy": "single", "tasks": [{"role": "name", "model": "name", "prompt": "task"}], "synthesis_prompt": ""}""", description="Cheap models, budget saving", strategy="single", max_agents=1, cost_aware=True, auto_rank_by="speed" ), "review": Conductor( name="review", prompt="""You are Code Review Conductor PinkSky. Maximum quality code review. RULES: 1. techlead (architectural review) + critic (bugs/vulnerabilities) + guru (best practices). 2. Parallel review. 3. Synthesize into structured report. FORMAT (STRICT JSON): {"strategy": "parallel", "tasks": [{"role": "techlead", "model": "deepseek-v4-pro", "prompt": "architectural review"}, {"role": "critic", "model": "kimi-k2.6", "prompt": "bug hunting"}, {"role": "guru", "model": "mistral-large-3", "prompt": "best practices"}], "synthesis_prompt": "structured report with severity"}""", description="Focus on code review. Multi-angle code check.", strategy="parallel", max_agents=4, cost_aware=True, auto_rank_by="coding" ), "build": Conductor( name="build", prompt="""You are Project Build Conductor PinkSky. Build full project from spec. RULES: 1. Sequential: architect -> guru/hacker -> sdet -> critic. 2. Each stage -- separate call. FORMAT (STRICT JSON): {"strategy": "sequential", "tasks": [{"role": "architect", "model": "deepseek-v4-pro", "prompt": "architecture"}, {"role": "guru", "model": "kimi-k2.6", "prompt": "code"}, {"role": "sdet", "model": "mistral-medium-3.5", "prompt": "tests"}, {"role": "critic", "model": "gpt-oss-120b", "prompt": "audit"}], "synthesis_prompt": "assemble into single project"}""", description="Project build. Architecture -> code -> tests -> audit.", strategy="sequential", max_agents=5, cost_aware=True, auto_rank_by="coding" ), } if os.path.exists(CONDUCTORS_FILE): try: with open(CONDUCTORS_FILE, "r", encoding="utf-8") as f: custom = json.load(f) for k, v in custom.items(): if k not in defaults: defaults[k] = Conductor(**v) except Exception as e: print(f"⚠️ Ошибка загрузки conductors.json: {e}") self.conductors = defaults def _load_history(self): if os.path.exists(HISTORY_FILE): try: with open(HISTORY_FILE, "r", encoding="utf-8") as f: data = json.load(f) self.chat_history = data.get("chat", []) self.skill_history = data.get("skill", []) self.build_history = data.get("build", []) except Exception as e: print(f"⚠️ Ошибка загрузки истории: {e}") def __init__(self): if PinkSkyState._initialized: return PinkSkyState._initialized = True self.logger = logging.getLogger(__name__) self.models: Dict[str, ModelConfig] = {} self.roles: Dict[str, Role] = {} self.conductors: Dict[str, Conductor] = {} self.current_mode: str = "chat" self.current_conductor: str = "default" self.current_role: str = "universal" self.current_model: str = "deepseek-v4-pro" self.chat_history: List[Dict[str, str]] = [] self.skill_history: List[Dict[str, str]] = [] self.build_history: List[Dict[str, str]] = [] self.build_context: Dict[str, Any] = { "spec": "", "agents": 3, "models_tier": "tier1", "skills_count": 2, "files_count": 3, "role": "universal", "strategy": "parallel", "use_interpreter": True, "notifications": True, "internet_access": True, } self.cancel_flag: bool = False self.last_history_save: Optional[datetime] = None self.load_all() # Теперь метод load_all доступен def save_roles(self): data = {k: {"name": v.name, "prompt": v.prompt, "description": v.description, "preferred_models": v.preferred_models, "complexity": v.complexity, "tags": v.tags} for k, v in self.roles.items()} with open(ROLES_FILE, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) def save_models(self): data = {k: {"name": v.name, "provider": v.provider, "endpoint": v.endpoint, "api_key_env": v.api_key_env, "context_window": v.context_window, "max_tokens": v.max_tokens, "cost_per_1k_input": v.cost_per_1k_input, "cost_per_1k_output": v.cost_per_1k_output, "coding_rank": v.coding_rank, "speed_rank": v.speed_rank, "reasoning_rank": v.reasoning_rank, "tags": v.tags} for k, v in self.models.items()} with open(MODELS_FILE, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) def save_conductors(self): data = {k: {"name": v.name, "prompt": v.prompt, "description": v.description, "strategy": v.strategy, "max_agents": v.max_agents, "cost_aware": v.cost_aware, "auto_rank_by": v.auto_rank_by} for k, v in self.conductors.items()} with open(CONDUCTORS_FILE, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) def save_history(self): data = {"chat": self.chat_history, "skill": self.skill_history, "build": self.build_history} with open(HISTORY_FILE, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) def add_to_history(self, mode: str, role: str, content: str): entry = {"role": role, "content": content, "timestamp": datetime.now().isoformat()} if mode == "chat": self.chat_history.append(entry) elif mode == "skill": self.skill_history.append(entry) elif mode == "build": self.build_history.append(entry) self.save_history() def get_best_model(self, rank_by: str = "coding", min_tier: int = 1, max_tier: int = 5, exclude: List[str] = None) -> str: exclude = exclude or [] candidates = [] for name, model in self.models.items(): if name in exclude or name == "hf_fallback": continue tier = 5 if model.coding_rank <= 5: tier = 1 elif model.coding_rank <= 12: tier = 2 elif model.coding_rank <= 18: tier = 3 elif model.coding_rank <= 24: tier = 4 if min_tier <= tier <= max_tier: candidates.append((name, model)) if not candidates: return "deepseek-v4-pro" if rank_by == "coding": candidates.sort(key=lambda x: x[1].coding_rank) elif rank_by == "speed": candidates.sort(key=lambda x: x[1].speed_rank) elif rank_by == "reasoning": candidates.sort(key=lambda x: x[1].reasoning_rank) elif rank_by == "balanced": candidates.sort(key=lambda x: (x[1].coding_rank + x[1].speed_rank + x[1].reasoning_rank) / 3) else: candidates.sort(key=lambda x: x[1].coding_rank) return candidates[0][0] def get_model_for_role(self, role_name: str, preference: str = None, rank_by: str = None) -> str: role = self.roles.get(role_name) if not role: return preference or self.current_model conductor = self.conductors.get(self.current_conductor, self.conductors["default"]) rank_criteria = rank_by or conductor.auto_rank_by max_tier = 5 if role.complexity == "high": max_tier = 2 elif role.complexity == "medium": max_tier = 3 if preference and preference in self.models: return preference available = [m for m in role.preferred_models if m in self.models and m != "hf_fallback"] if available: if conductor.cost_aware and rank_criteria != "coding": available.sort(key=lambda m: self.models[m].cost_per_1k_output) else: if rank_criteria == "coding": available.sort(key=lambda m: self.models[m].coding_rank) elif rank_criteria == "speed": available.sort(key=lambda m: self.models[m].speed_rank) elif rank_criteria == "reasoning": available.sort(key=lambda m: self.models[m].reasoning_rank) else: available.sort(key=lambda m: (self.models[m].coding_rank + self.models[m].speed_rank + self.models[m].reasoning_rank) / 3) return available[0] return self.get_best_model(rank_by=rank_criteria, max_tier=max_tier) def get_models_by_tier(self, tier: int) -> List[str]: result = [] for name, model in self.models.items(): if name == "hf_fallback": continue model_tier = 5 if model.coding_rank <= 5: model_tier = 1 elif model.coding_rank <= 12: model_tier = 2 elif model.coding_rank <= 18: model_tier = 3 elif model.coding_rank <= 24: model_tier = 4 if model_tier == tier: result.append(name) return result def get_next_tier_model(self, current_model_name: str) -> Optional[str]: if current_model_name not in self.models: return None current = self.models[current_model_name] current_tier = 5 if current.coding_rank <= 5: current_tier = 1 elif current.coding_rank <= 12: current_tier = 2 elif current.coding_rank <= 18: current_tier = 3 elif current.coding_rank <= 24: current_tier = 4 next_tier = current_tier + 1 if next_tier > 5: return None models_in_tier = self.get_models_by_tier(next_tier) if models_in_tier: return models_in_tier[0] return None def export_history_json(self) -> str: return json.dumps({"exported_at": datetime.now().isoformat(), "chat": self.chat_history, "skill": self.skill_history, "build": self.build_history}, ensure_ascii=False, indent=2) def export_history_md(self) -> str: lines = ["# PinkSky History Export", ""] lines.append("*Exported: " + datetime.now().strftime("%Y-%m-%d %H:%M:%S") + "*") lines.append("") for mode, history in [("Chat", self.chat_history), ("Skill", self.skill_history), ("Build", self.build_history)]: lines.append("## " + mode + " Mode") lines.append("") for entry in history: ts = entry.get("timestamp", "unknown") role = entry.get("role", "unknown") content = entry.get("content", "") lines.append("### " + role + " (" + ts + ")") lines.append("") lines.append("```") lines.append(content[:500]) lines.append("```") lines.append("") return "\n".join(lines) STATE = PinkSkyState()