api/assistant_factory.py

import json
import logging
import asyncio
from pathlib import Path
from typing import Dict, Any, Optional

# Настройка локального логгера для сервера
logger = logging.getLogger("CyberAgentFactory")

class AssistantFactory:
    def __init__(self, logger_agent=None, rules_path: str = 'api/coding_rules.json'):
        """
        Фабрика для сборки контекста и системных инструкций ИИ-агента.
        Реализована поддержка неблокирующего чтения конфигураций.
        """
        self.logger_agent = logger_agent
        self.rules_path = Path(rules_path)
        
        # Кэш в оперативной памяти для исключения избыточного дискового I/O
        self._cached_rules: Optional[Dict[str, Any]] = None
        # Блокировка для предотвращения Race Condition при горячей перезагрузке кэша
        self._lock = asyncio.Lock()

    def _read_file_sync(self) -> Dict[str, Any]:
        """Синхронный низкоуровневый запуск чтения (для выполнения в отдельном потоке)"""
        with open(self.rules_path, 'r', encoding='utf-8') as file:
            return json.load(file)

    async def _load_rules(self) -> Dict[str, Any]:
        """
        Безопасный асинхронный метод загрузки и кэширования правил.
        Выносит дисковое чтение из основного event loop приложения FastAPI.
        """
        if self._cached_rules is not None:
            return self._cached_rules

        async with self._lock:
            # Двойная проверка (Double-checked locking паттерн)
            if self._cached_rules is not None:
                return self._cached_rules

            if not self.rules_path.exists():
                warn_msg = f"Файл гайдлайнов не найден по пути: {self.rules_path}. Используется дефолтный промпт."
                logger.warning(warn_msg)
                if self.logger_agent and hasattr(self.logger_agent, 'log'):
                    self.logger_agent.log('factory_warning', warn_msg)
                self._cached_rules = {}
                return self._cached_rules

            try:
                # Делегируем блокирующий ввод-вывод в пул потоков операционной системы
                self._cached_rules = await asyncio.to_thread(self._read_file_sync)
                
                log_msg = f"Конфигурация правил успешно инициализирована из {self.rules_path} и кэширована."
                logger.info(log_msg)
                if self.logger_agent and hasattr(self.logger_agent, 'log'):
                    self.logger_agent.log('factory_init', log_msg)
                    
                return self._cached_rules

            except json.JSONDecodeError as e:
                err_msg = f"Критическая ошибка синтаксиса JSON в {self.rules_path}: {str(e)}. Активирован защитный режим."
                logger.error(err_msg)
                if self.logger_agent and hasattr(self.logger_agent, 'log'):
                    self.logger_agent.log('factory_json_error', err_msg)
                self._cached_rules = {}
                return self._cached_rules

            except Exception as e:
                err_msg = f"Непредвиденный сбой ввода-вывода при чтении {self.rules_path}: {str(e)}"
                logger.error(err_msg)
                if self.logger_agent and hasattr(self.logger_agent, 'log'):
                    self.logger_agent.log('factory_io_error', err_msg)
                self._cached_rules = {}
                return self._cached_rules

    def _format_rules_to_markdown(self, rules: Dict[str, Any], depth: int = 2) -> str:
        """
        Рекурсивно преобразует JSON-словарь любой вложенности в валидный Markdown список.
        Максимизирует показатель Prompt Adherence для LLM-моделей.
        """
        md_lines = []
        prefix = "#" * min(depth, 6)
        
        if not isinstance(rules, dict):
            return str(rules)

        for key, val in rules.items():
            display_key = key.replace("_", " ").capitalize()
            
            if isinstance(val, dict):
                md_lines.append(f"\n{prefix} {display_key}")
                md_lines.append(self._format_rules_to_markdown(val, depth + 1))
            elif isinstance(val, list):
                md_lines.append(f"\n{prefix} {display_key}:")
                for item in val:
                    md_lines.append(f" - {item}")
            else:
                # Базовый тип данных (строка, число, булево)
                if depth == 2:
                    md_lines.append(f"**{display_key}**: {val}")
                else:
                    md_lines.append(f" * **{key}**: {val}")
                    
        return "\n".join(md_lines)

    async def create_assistant_prompt(self, user_prompt: str) -> str:
        """
        Формирует структурированный системный промпт, изолируя правила безопасности
        в оптимизированные под LLM-инференс Markdown теги.
        """
        rules = await self._load_rules()
        if not rules:
            return user_prompt

        # Трансляция древовидной JSON-структуры в читаемый ИИ-моделями Markdown-контекст
        formatted_rules = self._format_rules_to_markdown(rules)

        system_prompt = (
            "### SYSTEM SECURITY GUARDRAILS AND OPERATIONAL RULES ###\n"
            f"{formatted_rules}\n"
            "### END OF GUARDRAILS ###\n\n"
            "Текущая операционная задача и контекст окружения:\n"
            f"{user_prompt}"
        )
        return system_prompt

    async def hot_reload_rules(self) -> str:
        """
        Метод 'горячей перезагрузки'. Атомарно обновляет правила в памяти 
        на лету без блокирования асинхронного цикла обработки входящих вебхуков.
        """
        async with self._lock:
            old_cache = self._cached_rules
            self._cached_rules = None
            
            # Принудительный вызов загрузки через пул потоков
            new_rules = await self._load_rules()
            if new_rules and len(new_rules) > 0:
                return "🔄 Контур управления: Кэш сброшен. Новые правила успешно развернуты в оперативной памяти!"
            
            # В случае повреждения структуры файла возвращаем стабильный старый кэш
            self._cached_rules = old_cache
            return "⚠️ Отказ перезагрузки: Файл поврежден или пуст. Ядро продолжает работу на старой конфигурации из кэша."