Медленное мышление для быстрых машин: архитектура рефлексирующего ИИ-агента нового поколения

Мы привыкли, что ИИ-агенты — это про скорость. Быстрее написать код, быстрее ответить, быстрее сгенерировать. Но скорость без размышления — это не интеллект ^[1], а рефлекс ^[2]. Настоящий прорыв происходит, когда агент начинает думать, прежде чем делать. Проверять себя после каждого шага. Сомневаться. Спрашивать разрешения. И только потом действовать.

Современные языковые модели отлично рассуждают в теории. Но стоит дать им инструменты и доступ к реальному миру, как проявляются системные уязвимости: галлюцинации, превращённые в действия; каскадные ошибки ^[3], где одна неточность тянет за собой цепочку неверных решений; слепое следование цели с игнорированием побочных эффектов; отсутствие самокритики. Решение — не в том, чтобы сделать модель умнее. Решение — в архитектуре, которая принудительно встраивает размышление в каждое звено цикла «восприятие → решение → действие → проверка».

Эта статья — про архитектуру, которая превращает бездумный конвейер генерации в мыслящего коллегу. И она применима не только к работе с кодом, а к любому домену, где цена ошибки высока: юриспруденция, медицина, финансы, управление инфраструктурой, маркетинг, поддержка клиентов, образование.

🐍 Реализация рефлексирующего ИИ-агента на Python

📁 Структура проекта

reflective_agent/
├── core/
│   ├── __init__.py
│   ├── agent.py           # Основной цикл агента
│   ├── context.py         # Динамический контекст
│   ├── state.py           # Управление состоянием
│   └── types.py           # Типы данных
├── tools/
│   ├── __init__.py
│   ├── base.py            # Базовый класс инструментов
│   ├── filesystem.py      # Файловая система
│   ├── terminal.py        # Терминальные команды
│   └── mcp_adapter.py     # MCP протокол
├── safety/
│   ├── __init__.py
│   ├── jail.py            # Scope Jail
│   ├── detectors.py       # Детекция циклов
│   └── snapshots.py       # Снапшоты и откат
├── ui/
│   ├── __init__.py
│   └── console.py         # Консольный интерфейс
└── main.py                # Точка входа

📦 Основные типы данных

# core/types.py
from dataclasses import dataclass, field
from typing import Any, Optional, List, Dict, Literal
from datetime import datetime
from enum import Enum

class AgentMode(Enum):
    CHAT = "chat"
    AGENT = "agent"
    TERMINAL = "terminal"

class ToolRisk(Enum):
    SAFE_READ = "safe_read"      # Автоматически
    MODIFY = "modify"            # Черновик
    DESTRUCTIVE = "destructive"  # Требует подтверждения

@dataclass
class Message:
    role: Literal["user", "assistant", "system", "tool"]
    content: Optional[str]
    timestamp: float = field(default_factory=datetime.now().timestamp)
    tool_calls: Optional[List[Dict]] = None
    tool_call_id: Optional[str] = None
    name: Optional[str] = None
    requires_action: Optional[str] = None

@dataclass
class ToolCall:
    id: str
    name: str
    arguments: Dict[str, Any]
    risk: ToolRisk

@dataclass
class Draft:
    file_path: str
    original_content: str
    new_content: str
    status: Literal["pending", "accepted", "rejected"] = "pending"

@dataclass
class Snapshot:
    id: str
    timestamp: float
    files: Dict[str, str]
    messages: List[Message]
    drafts: List[Draft]

🧠 Ядро агента

# core/agent.py
import asyncio
import json
import hashlib
from typing import List, Optional, Dict, Any, Callable
from datetime import datetime
from openai import AsyncOpenAI

from .types import (
    AgentMode, Message, ToolCall, Draft, Snapshot, ToolRisk
)
from .context import DynamicContext
from .state import AgentState
from safety.jail import ScopeJail
from safety.detectors import LoopDetector, IterationLimiter
from safety.snapshots import SnapshotManager
from tools.base import ToolRegistry


class ReflectiveAgent:
    """Рефлексирующий ИИ-агент с семиступенчатым циклом"""
    
    def __init__(
        self,
        model: str = "gpt-4",
        api_key: str = None,
        max_iterations: int = 15,
        require_confirmation: bool = True
    ):
        self.client = AsyncOpenAI(api_key=api_key)
        self.model = model
        self.state = AgentState()
        self.context = DynamicContext()
        self.tools = ToolRegistry()
        self.jail = ScopeJail()
        self.loop_detector = LoopDetector(max_repeats=3)
        self.iteration_limiter = IterationLimiter(max_iterations)
        self.snapshot_manager = SnapshotManager()
        self.require_confirmation = require_confirmation
        
        # Флаги
        self.is_reflecting = False
        self._abort_controller = False
        
    async def run(self, user_input: str) -> None:
        """Главный метод запуска агента"""
        
        # Создаем снапшот перед началом
        await self.snapshot_manager.create_snapshot(self.state)
        
        # Добавляем сообщение пользователя
        user_message = Message(
            role="user",
            content=user_input,
            timestamp=datetime.now().timestamp()
        )
        self.state.add_message(user_message)
        
        # Запускаем рекурсивный цикл
        await self._run_cycle(depth=1)
    
    async def _run_cycle(self, depth: int) -> None:
        """Рекурсивный цикл выполнения"""
        
        # Проверка прерывания
        if self._abort_controller:
            return
        
        # Проверка лимита итераций
        if not self.iteration_limiter.can_continue(depth):
            await self._handle_iteration_limit()
            return
        
        try:
            # 1. ВОСПРИЯТИЕ - Сбор динамического контекста
            context = await self.context.build_context(self.state)
            
            # 2. ПЛАНИРОВАНИЕ - Вызов LLM
            response = await self._call_llm(context)
            
            # Сохраняем ответ ассистента
            assistant_msg = Message(
                role="assistant",
                content=response.get("content"),
                tool_calls=response.get("tool_calls")
            )
            self.state.add_message(assistant_msg)
            
            # 3. ОБРАБОТКА TOOL CALLS
            if response.get("tool_calls"):
                await self._process_tool_calls(response["tool_calls"])
                
                # Рекурсивный вызов для следующего шага
                await self._run_cycle(depth + 1)
            else:
                # Нет вызовов инструментов -> завершение
                await self._complete_task()
                
        except Exception as e:
            await self._handle_error(e, depth)
    
    async def _call_llm(self, context: str) -> Dict[str, Any]:
        """Вызов LLM с системным промптом и инструментами"""
        
        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            *[self._serialize_message(m) for m in self.state.messages]
        ]
        
        # Добавляем контекст как системное сообщение
        messages.insert(1, {"role": "system", "content": context})
        
        tools = self.tools.get_openai_schema()
        
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            tools=tools if tools else None,
            tool_choice="auto" if tools else None,
            temperature=0.3
        )
        
        message = response.choices[0].message
        
        result = {"content": message.content or ""}
        
        if message.tool_calls:
            result["tool_calls"] = [
                {
                    "id": tc.id,
                    "name": tc.function.name,
                    "arguments": json.loads(tc.function.arguments)
                }
                for tc in message.tool_calls
            ]
        
        return result
    
    async def _process_tool_calls(self, tool_calls: List[Dict]) -> None:
        """Обработка вызовов инструментов"""
        
        for tool_call in tool_calls:
            if self._abort_controller:
                break
            
            # 4. ШЛЮЗ - Проверка безопасности
            risk = self.tools.get_risk(tool_call["name"])
            
            if risk == ToolRisk.DESTRUCTIVE and self.require_confirmation:
                # Пауза, ждем подтверждения пользователя
                await self._request_confirmation(tool_call)
                return  # Выход из цикла, ждем событие
            
            # 5. ДЕТЕКЦИЯ ЗАЦИКЛИВАНИЙ
            signature = self._get_tool_signature(tool_call)
            if self.loop_detector.check(signature):
                await self._handle_loop_detected(tool_call)
                return
            
            # 6. Scope Jail - проверка границ
            if not await self.jail.check_permissions(tool_call):
                await self._handle_access_denied(tool_call)
                continue
            
            # 7. ВЫПОЛНЕНИЕ ИНСТРУМЕНТА
            result = await self.tools.execute(
                tool_call["name"],
                tool_call["arguments"],
                is_draft=(risk == ToolRisk.MODIFY)
            )
            
            # Сохраняем результат
            tool_msg = Message(
                role="tool",
                content=result,
                tool_call_id=tool_call["id"],
                name=tool_call["name"]
            )
            self.state.add_message(tool_msg)
    
    async def _complete_task(self) -> None:
        """Завершение задачи с фазой рефлексии"""
        
        if not self.is_reflecting:
            # ФАЗА РЕФЛЕКСИИ
            self.is_reflecting = True
            
            # Запрашиваем самопроверку
            reflection_msg = Message(
                role="system",
                content="🔄 **SELF-REFLECTION PHASE**nn"
                        "Пожалуйста, проверьте свою работу:n"
                        "1. Все ли шаги выполнены?n"
                        "2. Нет ли ошибок или пропусков?n"
                        "3. Если есть проблемы - исправьте ихn"
                        "4. Если всё верно - вызовите task_completed(verified=true)",
                requires_action="reflect"
            )
            self.state.add_message(reflection_msg)
            
            # Продолжаем цикл для рефлексии
            await self._run_cycle(depth=self.state.current_depth + 1)
        else:
            # ФИНАЛЬНОЕ ЗАВЕРШЕНИЕ
            if self.state.drafts:
                # Показываем окно ревью
                await self._show_review_modal()
            
            print("n✅ Задача завершена!")
            print(f"📊 Статистика: {self.state.get_stats()}")
            self.is_reflecting = False
    
    def _build_system_prompt(self) -> str:
        """Построение системного промпта"""
        return """
ТЫ - Рефлексирующий ИИ-агент. Твоя задача - помогать пользователю, 
активно используя доступные инструменты.

ПРАВИЛА:
1. Перед действием объясни, что ты собираешься сделать
2. Используй create_plan для сложных задач
3. Проверяй свою работу перед завершением
4. Не гадай - используй инструменты для получения информации

ДОСТУПНЫЕ ПРИНЦИПЫ:
- Черновики: изменения сохраняются в черновик до подтверждения
- Рефлексия: всегда проверяй свою работу
- Безопасность: не выходи за пределы разрешенных границ
"""
    
    async def _request_confirmation(self, tool_call: Dict) -> None:
        """Запрос подтверждения у пользователя"""
        
        self.state.pending_confirmation = tool_call
        
        print(f"n⚠️  ТРЕБУЕТСЯ ПОДТВЕРЖДЕНИЕ")
        print(f"Инструмент: {tool_call['name']}")
        print(f"Аргументы: {json.dumps(tool_call['arguments'], indent=2, ensure_ascii=False)}")
        
        # В реальном приложении здесь был бы UI
        # Для консоли - ждем ввода
        response = input("nПодтвердить? (y/n): ").lower()
        
        if response == 'y':
            await self.confirm_action()
        else:
            await self.reject_action()
    
    async def confirm_action(self) -> None:
        """Подтверждение действия"""
        
        if not self.state.pending_confirmation:
            return
        
        tool_call = self.state.pending_confirmation
        self.state.pending_confirmation = None
        
        # Выполняем подтвержденное действие
        result = await self.tools.execute(
            tool_call["name"],
            tool_call["arguments"],
            is_draft=False  # Принудительное применение
        )
        
        tool_msg = Message(
            role="tool",
            content=result,
            tool_call_id=tool_call["id"],
            name=tool_call["name"]
        )
        self.state.add_message(tool_msg)
        
        # Продолжаем цикл
        await self._run_cycle(self.state.current_depth + 1)
    
    async def reject_action(self) -> None:
        """Отклонение действия"""
        
        if not self.state.pending_confirmation:
            return
        
        tool_call = self.state.pending_confirmation
        self.state.pending_confirmation = None
        
        tool_msg = Message(
            role="tool",
            content="Действие отклонено пользователем",
            tool_call_id=tool_call["id"],
            name=tool_call["name"]
        )
        self.state.add_message(tool_msg)
        
        await self._run_cycle(self.state.current_depth + 1)
    
    def stop(self) -> None:
        """Остановка агента"""
        self._abort_controller = True
    
    async def _handle_error(self, error: Exception, depth: int) -> None:
        """Обработка ошибок с авто-повтором"""
        
        if self.state.retry_count < 2:
            self.state.retry_count += 1
            delay = 5  # секунд
            
            print(f"⚠️ Ошибка: {error}. Повтор через {delay}с...")
            await asyncio.sleep(delay)
            
            # Повтор с той же глубиной
            await self._run_cycle(depth)
        else:
            print(f"❌ Критическая ошибка: {error}")
            self.state.add_message(Message(
                role="system",
                content=f"Ошибка: {error}"
            ))
    
    def _get_tool_signature(self, tool_call: Dict) -> str:
        """Генерация сигнатуры для детекции циклов"""
        content = f"{tool_call['name']}:{json.dumps(tool_call['arguments'], sort_keys=True)}"
        return hashlib.md5(content.encode()).hexdigest()
    
    async def _show_review_modal(self) -> None:
        """Показ окна ревью изменений"""
        
        print("n📝 ИЗМЕНЕНИЯ В ЧЕРНОВИКАХ:")
        for i, draft in enumerate(self.state.drafts):
            print(f"n{i+1}. {draft.file_path}")
            print(f"   Статус: {draft.status}")
            print(f"   Размер: {len(draft.new_content)} символов")
        
        response = input("nПрименить все изменения? (y/n): ").lower()
        
        if response == 'y':
            await self._commit_drafts()
        else:
            await self._rollback()
    
    async def _commit_drafts(self) -> None:
        """Применение черновиков"""
        for draft in self.state.drafts:
            if draft.status == "pending":
                await self.tools.apply_draft(draft)
                draft.status = "accepted"
        
        print("✅ Изменения применены")
    
    async def _rollback(self) -> None:
        """Откат изменений"""
        await self.snapshot_manager.restore_snapshot(self.state)
        print("↩️ Изменения отменены")
    
    async def _handle_iteration_limit(self) -> None:
        """Обработка превышения лимита итераций"""
        print(f"n⚠️ Достигнут лимит итераций ({self.iteration_limiter.max_iterations})")
        
        response = input("Продолжить? (y/n): ").lower()
        
        if response == 'y':
            self.iteration_limiter.reset()
            await self._run_cycle(depth=1)
        else:
            self.stop()
    
    async def _handle_loop_detected(self, tool_call: Dict) -> None:
        """Обработка обнаруженного цикла"""
        print(f"n🔄 ОБНАРУЖЕН ЦИКЛ: {tool_call['name']}")
        print("Агент повторяет одно и то же действие")
        self.stop()
    
    async def _handle_access_denied(self, tool_call: Dict) -> None:
        """Обработка нарушения доступа"""
        print(f"n🚫 ДОСТУП ЗАПРЕЩЕН: {tool_call['name']}")
        print(f"Попытка выхода за пределы разрешенной зоны")
        
        self.state.add_message(Message(
            role="system",
            content=f"Ошибка доступа: {tool_call['name']} - выход за границы"
        ))
    
    def _serialize_message(self, msg: Message) -> Dict:
        """Сериализация сообщения для OpenAI API"""
        result = {"role": msg.role}
        
        if msg.content:
            result["content"] = msg.content
        if msg.tool_calls:
            result["tool_calls"] = msg.tool_calls
        if msg.tool_call_id:
            result["tool_call_id"] = msg.tool_call_id
        if msg.name:
            result["name"] = msg.name
        
        return result

🛠️ Базовые инструменты

# tools/base.py
from abc import ABC, abstractmethod
from typing import Dict, Any, List, Optional
from core.types import ToolRisk


class BaseTool(ABC):
    """Базовый класс для всех инструментов"""
    
    def __init__(self, name: str, description: str, risk: ToolRisk):
        self.name = name
        self.description = description
        self.risk = risk
        self.parameters = self.get_parameters()
    
    @abstractmethod
    def get_parameters(self) -> Dict[str, Any]:
        """Схема параметров для OpenAI API"""
        pass
    
    @abstractmethod
    async def execute(self, **kwargs) -> str:
        """Выполнение инструмента"""
        pass


class FileSystemTool(BaseTool):
    """Инструменты файловой системы"""
    
    def __init__(self, workspace_root: str):
        self.workspace_root = workspace_root
        super().__init__(
            name="read_file",
            description="Read file content",
            risk=ToolRisk.SAFE_READ
        )
    
    def get_parameters(self) -> Dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "File path"}
            },
            "required": ["path"]
        }
    
    async def execute(self, path: str, **kwargs) -> str:
        # Проверка безопасности пути
        full_path = self._safe_path(path)
        
        try:
            with open(full_path, 'r', encoding='utf-8') as f:
                return f.read()
        except Exception as e:
            return f"Error reading file: {e}"
    
    def _safe_path(self, path: str) -> str:
        """Проверка безопасного пути"""
        import os
        full = os.path.join(self.workspace_root, path.lstrip('/'))
        if not full.startswith(self.workspace_root):
            raise PermissionError("Access denied: path outside workspace")
        return full


class PatchFileTool(BaseTool):
    """Инструмент патчинга файлов (создает черновик)"""
    
    def __init__(self, draft_manager):
        self.draft_manager = draft_manager
        super().__init__(
            name="patch_file",
            description="Patch file content (creates draft)",
            risk=ToolRisk.MODIFY
        )
    
    def get_parameters(self) -> Dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "search_str": {"type": "string"},
                "replace_str": {"type": "string"}
            },
            "required": ["path", "search_str", "replace_str"]
        }
    
    async def execute(self, path: str, search_str: str, replace_str: str, **kwargs) -> str:
        # Создаем черновик вместо прямого изменения
        draft = await self.draft_manager.create_draft(
            path=path,
            search_str=search_str,
            replace_str=replace_str
        )
        
        return f"Draft created: {path}nPreview: {draft.new_content[:200]}..."


class ToolRegistry:
    """Реестр всех доступных инструментов"""
    
    def __init__(self):
        self._tools: Dict[str, BaseTool] = {}
    
    def register(self, tool: BaseTool) -> None:
        self._tools[tool.name] = tool
    
    async def execute(self, name: str, arguments: Dict[str, Any], is_draft: bool = False) -> str:
        if name not in self._tools:
            return f"Error: Tool '{name}' not found"
        
        tool = self._tools[name]
        
        # Для модифицирующих операций используем черновик
        if tool.risk == ToolRisk.MODIFY and is_draft:
            return await tool.execute(**arguments, _is_draft=True)
        
        return await tool.execute(**arguments)
    
    def get_openai_schema(self) -> List[Dict]:
        """Получение схемы для OpenAI API"""
        return [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.parameters
                }
            }
            for tool in self._tools.values()
        ]
    
    def get_risk(self, tool_name: str) -> ToolRisk:
        """Получение уровня риска инструмента"""
        if tool_name not in self._tools:
            return ToolRisk.SAFE_READ
        return self._tools[tool_name].risk

🔒 Система безопасности

# safety/jail.py
from typing import Dict, Any, List
import re


class ScopeJail:
    """Проверка границ доступа"""
    
    def __init__(self, allowed_paths: List[str] = None, blocked_patterns: List[str] = None):
        self.allowed_paths = allowed_paths or ["."]
        self.blocked_patterns = blocked_patterns or [
            r"../",           # Обход директорий
            r"/etc/passwd",     # Системные файлы
            r"rms+-rf",        # Опасные команды
            r"sudos+",         # Повышение привилегий
            r"|s*shs",       # Инъекции
        ]
    
    async def check_permissions(self, tool_call: Dict[str, Any]) -> bool:
        """Проверка разрешений для вызова инструмента"""
        
        tool_name = tool_call.get("name")
        args = tool_call.get("arguments", {})
        
        # Проверка путей
        for path_key in ["path", "oldPath", "newPath", "source", "destination"]:
            if path_key in args:
                if not self._check_path(args[path_key]):
                    return False
        
        # Проверка команд
        if tool_name == "exec_command":
            command = args.get("command", "")
            if not self._check_command(command):
                return False
        
        return True
    
    def _check_path(self, path: str) -> bool:
        """Проверка безопасности пути"""
        
        # Проверка на паттерны обхода
        for pattern in self.blocked_patterns:
            if re.search(pattern, path):
                return False
        
        return True
    
    def _check_command(self, command: str) -> bool:
        """Проверка безопасности команды"""
        
        dangerous = ["rm -rf", "dd if=", "mkfs", ":(){ :|:& };:"]
        for cmd in dangerous:
            if cmd in command.lower():
                return False
        
        return True


class LoopDetector:
    """Детектор зацикливаний"""
    
    def __init__(self, max_repeats: int = 3, window_size: int = 10):
        self.max_repeats = max_repeats
        self.window_size = window_size
        self._signatures: List[str] = []
    
    def check(self, signature: str) -> bool:
        """Проверка на повторение"""
        
        self._signatures.append(signature)
        
        # Оставляем только последние N
        if len(self._signatures) > self.window_size:
            self._signatures.pop(0)
        
        # Подсчет повторений
        repeats = sum(1 for s in self._signatures if s == signature)
        
        return repeats >= self.max_repeats
    
    def reset(self) -> None:
        """Сброс детектора"""
        self._signatures.clear()


class IterationLimiter:
    """Лимитер итераций"""
    
    def __init__(self, max_iterations: int = 15):
        self.max_iterations = max_iterations
    
    def can_continue(self, current_depth: int) -> bool:
        return current_depth <= self.max_iterations
    
    def reset(self) -> None:
        pass


class SnapshotManager:
    """Управление снапшотами и откатами"""
    
    def __init__(self):
        self._snapshots: List[Snapshot] = []
    
    async def create_snapshot(self, state: 'AgentState') -> Snapshot:
        """Создание снапшота текущего состояния"""
        
        snapshot = Snapshot(
            id=self._generate_id(),
            timestamp=datetime.now().timestamp(),
            files=state.files.copy(),
            messages=state.messages.copy(),
            drafts=state.drafts.copy()
        )
        
        self._snapshots.append(snapshot)
        return snapshot
    
    async def restore_snapshot(self, state: 'AgentState', snapshot_id: str = None) -> bool:
        """Восстановление состояния из снапшота"""
        
        if not self._snapshots:
            return False
        
        if snapshot_id:
            snapshot = next((s for s in self._snapshots if s.id == snapshot_id), None)
        else:
            snapshot = self._snapshots[-1]  # Последний
        
        if not snapshot:
            return False
        
        # Восстановление состояния
        state.files = snapshot.files.copy()
        state.messages = snapshot.messages.copy()
        state.drafts = snapshot.drafts.copy()
        
        return True
    
    def _generate_id(self) -> str:
        """Генерация ID снапшота"""
        import uuid
        return str(uuid.uuid4())[:8]

🎮 Консольный UI

# ui/console.py
import asyncio
from core.agent import ReflectiveAgent


class ConsoleUI:
    """Консольный интерфейс для взаимодействия с агентом"""
    
    def __init__(self):
        self.agent = None
    
    async def run(self):
        """Запуск консольного интерфейса"""
        
        print("=" * 60)
        print("🤖 РЕФЛЕКСИРУЮЩИЙ ИИ-АГЕНТ")
        print("=" * 60)
        print("nКоманды:")
        print("  /stop - остановить агента")
        print("  /status - показать статус")
        print("  /snapshot - создать снапшот")
        print("  /rollback - откат к последнему снапшоту")
        print("  /help - показать помощь")
        print("n" + "=" * 60)
        
        # Инициализация агента
        api_key = input("n🔑 OpenAI API Key: ").strip()
        
        self.agent = ReflectiveAgent(
            model="gpt-4",
            api_key=api_key,
            max_iterations=15,
            require_confirmation=True
        )
        
        print("n✅ Агент готов к работе!")
        print("Введите ваш запрос:n")
        
        while True:
            try:
                user_input = input("👤 Вы: ").strip()
                
                if not user_input:
                    continue
                
                if user_input.startswith("/"):
                    await self._handle_command(user_input)
                    continue
                
                print("n🤖 Агент думает...n")
                
                # Запуск агента
                await self.agent.run(user_input)
                
                print("n" + "-" * 40 + "n")
                
            except KeyboardInterrupt:
                print("nn👋 До свидания!")
                break
            except Exception as e:
                print(f"n❌ Ошибка: {e}n")
    
    async def _handle_command(self, command: str):
        """Обработка команд"""
        
        if command == "/stop":
            self.agent.stop()
            print("⏹️ Агент остановлен")
        
        elif command == "/status":
            print(f"n📊 Статус:")
            print(f"  Сообщений: {len(self.agent.state.messages)}")
            print(f"  Черновиков: {len(self.agent.state.drafts)}")
            print(f"  Итераций: {self.agent.state.current_depth}")
        
        elif command == "/snapshot":
            snapshot = await self.agent.snapshot_manager.create_snapshot(self.agent.state)
            print(f"📸 Снапшот создан: {snapshot.id}")
        
        elif command == "/rollback":
            await self.agent.snapshot_manager.restore_snapshot(self.agent.state)
            print("↩️ Выполнен откат к последнему снапшоту")
        
        elif command == "/help":
            print("nДоступные команды:")
            print("  /stop - остановить агента")
            print("  /status - показать статус")
            print("  /snapshot - создать снапшот")
            print("  /rollback - откат к последнему снапшоту")
            print("  /help - показать помощь")
        
        else:
            print(f"❌ Неизвестная команда: {command}")


async def main():
    ui = ConsoleUI()
    await ui.run()


if __name__ == "__main__":
    asyncio.run(main())

🚀 Точка входа

# main.py
import asyncio
from ui.console import ConsoleUI
from core.agent import ReflectiveAgent
from tools.filesystem import FileSystemTool, PatchFileTool
from tools.terminal import TerminalTool
from tools.mcp_adapter import MCPAdapter


async def main():
    """Точка входа в приложение"""
    
    # Запуск консольного интерфейса
    ui = ConsoleUI()
    await ui.run()


async def demo():
    """Демонстрация работы агента без UI"""
    
    agent = ReflectiveAgent(
        model="gpt-4",
        api_key="your-api-key",  # Замените на ваш ключ
        max_iterations=10
    )
    
    # Регистрация инструментов
    from tools.filesystem import FileSystemTool, PatchFileTool
    from tools.terminal import TerminalTool
    
    agent.tools.register(FileSystemTool(workspace_root="./workspace"))
    agent.tools.register(PatchFileTool(agent.state.draft_manager))
    agent.tools.register(TerminalTool())
    
    # Запросы пользователя
    queries = [
        "Прочитай файл README.md",
        "Создай новый Python файл с функцией hello_world",
        "Какой сейчас Python version?",
    ]
    
    for query in queries:
        print(f"n{'='*60}")
        print(f"👤 Пользователь: {query}")
        print(f"{'='*60}n")
        
        await agent.run(query)
        
        print("n" + "="*60 + "n")


if __name__ == "__main__":
    # Для демонстрации
    # asyncio.run(demo())
    
    # Для интерактивного режима
    asyncio.run(main())

📦 Установка зависимостей

# requirements.txt
openai>=1.0.0
asyncio
pydantic>=2.0.0
python-dotenv>=1.0.0

pip install -r requirements.txt

🎯 Ключевые особенности реализации

1. Рекурсивный цикл – асинхронная рекурсия вместо while

2. Черновики – все изменения проходят через DraftManager

3. Рефлексия – принудительная самопроверка перед завершением

4. Безопасность – многоуровневые проверки (Jail, детектор циклов)

5. Снапшоты – возможность отката любого изменения

6. Human-in-the-loop – подтверждение опасных действий

---

🧩 Почему «быстрый» агент — это проблема

Большинство агентов живут по схеме think → do. Это быстро, но хрупко. Достаточно одной галлюцинации, неоднозначного промпта или непредвиденного состояния среды — и агент молча отправляет не тому клиенту письмо, ломает интеграцию, генерирует некорректный отчёт или уходит в бесконечный цикл запросов, сжигая бюджет.

Универсальная архитектура заменяет эту схему на семиступенчатый цикл:

Восприятие → Планирование → Черновик → Рефлексия → Шлюз → Коммит → Откат

Это не замедляет работу. Это делает её предсказуемой, аудируемой и безопасной для продакшена.

---

🏗️ Семь столпов универсальной архитектуры

1. Динамическая сборка контекста

Агент не действует вслепую. Перед каждым шагом он собирает актуальное состояние среды: какие данные доступны, какие инструменты подключены, какие политики и ограничения действуют, что уже было сделано в текущей сессии. Это как пилот, проверяющий приборы перед манёвром.

Вместо того чтобы запихивать в промпт мегабайты данных, архитектура собирает динамический контекст:

Это даёт агенту не «всезнание», а ситуационную осведомлённость — ровно то, что нужно для принятия решений в конкретный момент. Контекст формируется заново на каждой итерации, чтобы избежать работы с устаревшими допущениями.

---

2. Рекурсивный цикл с правом на паузу

Вместо жёсткого while(true) используется асинхронная рекурсия. Почему это важно? Потому что ИИ-агент — не конвейер, а диалог.

Агент может остановиться, запросить уточнение, дождаться ответа человека, внешнего вебхука или результата долгой операции, и продолжить ровно с того же места. История диалога, состояние инструментов и бюджет вычислений сохраняются. Пауза не ломает процесс — она встроена в его ДНК. Человек остаётся в контуре, а не выпадает из него.

---

3. Виртуальная песочница и черновики

Одно из самых мощных архитектурных решений — концепция черновиков. Все модифицирующие операции не применяются к реальным данным сразу. Вместо этого:

1. Создаётся виртуальный слой изменений в памяти.

2. Все последующие чтения внутри сессии видят уже изменённые данные (чтобы агент работал с актуальным состоянием).

3. Реальные объекты остаются нетронутыми.

4. Пользователь видит окно ревью: дифф всех изменений, возможность принять, отклонить или отредактировать.

5. Только после явного «Сохранить» черновики атомарно применяются.

Это принцип git staging, перенесённый на любые действия. Будь то правка документа, обновление записи в CRM, отправка сообщения, запуск ETL-пайплайна или изменение конфигурации — система фиксирует «что было» и «что станет», но не трогает продакшен-среду. Черновики превращают агента из «исполнителя» в «советника», оставляя финальное слово за человеком.

---

4. Обязательная фаза рефлексии

Это ключевой архитектурный блок, который отличает «думающего» агента от «генерирующего». Когда агент считает задачу выполненной, система не принимает это на веру.

Включается фаза принудительной рефлексии — внутренний аудит:

• Соответствует ли результат исходной цели?

• Нет ли логических противоречий или пропущенных шагов?

• Соблюдены ли бизнес-правила, форматы, ограничения?

• Можно ли запустить сухую проверку или тест?

Агент пересматривает свои действия, перечитывает изменённые объекты, запрашивает недостающий контекст. Если находит проблемы — исправляет их через те же инструменты. Только после явного подтверждения (verified: true) задача считается завершённой.

Это аналог чек-листа пилота перед взлётом: лучше потратить лишний шаг на проверку, чем получить критическую ошибку в продакшене. В юридическом домене это означало бы перепроверку ссылок на статьи закона. В медицинском — перекрёстный анализ назначений с противопоказаниями. В финансовом — сверку расчётов с контрольными суммами. Рефлексия превращает агента из «генератора текста» в ответственного исполнителя.

---

5. Человеческий шлюз для критических действий

Все действия делятся на три категории риска:

• 🟢 Безопасные чтения — выполняются автоматически

• 🟡 Изменения — создают черновики, не затрагивая оригинал

• 🔴 Деструктивные операции — требуют явного подтверждения

Удаление данных, финансовые транзакции, публикация контента, изменение ролей, массовые рассылки, перезапуск сервисов — всё, что необратимо или влияет на внешнюю среду, требует паузы. Интерфейс показывает карточку с контекстом, рисками, альтернативами и превью изменений. Агент замирает, ждёт решения и продолжает работу только после «зелёного света».

Это не чат с кнопками, а полноценный паттерн Human-in-the-Loop: человек остаётся архитектором последствий, а скорость агента не превращается в безрассудство.

---

6. Многоуровневые предохранители

Защита не может быть фильтром на выходе. Она должна быть частью цикла выполнения. Каждое действие агента проходит через независимые предохранители, встроенные в исполнительный слой:

Эти механизмы работают не «поверх» агента, а встроены в уровень вызова инструментов. Модель не может их обойти, потому что проверка происходит на уровне исполнения, а не генерации текста. Агент не может «договориться» с системой обойти защиту.

---

7. Универсальный протокол инструментов

Архитектура не привязана к конкретному домену. Через стандартизированный интерфейс (аналог Model Context Protocol) подключаются любые сервисы: базы данных, CRM, аналитические платформы, браузеры, почта, IoT-устройства, внешние API.

Это как USB-C для ИИ-агентов:

Агент видит инструменты как единый набор «рук», а система безопасности проверяет каждый вызов независимо от источника. Сегодня агент работает с таблицами, завтра — с чат-ботами, послезавтра — с логистическими системами. Ядро не меняется.

---

🌍 Как это работает в разных сферах?

Во всех сценариях сохраняется один паттерн: агент думает, примеряет, проверяет, спрашивает и только потом действует.

---

🔄 Анатомия одного цикла: что происходит под капотом

Рассмотрим универсальный пример — анализ документа с рисками:

ПОЛЬЗОВАТЕЛЬ: "Проанализируй договор аренды на предмет рисков"
          ↓
🔵 ВОСПРИЯТИЕ: Агент подгружает файл, видит его структуру и метаданные
          ↓
🟡 ПЛАНИРОВАНИЕ: Определяет, что это договор аренды. Планирует анализ по секциям:
   предмет, сроки, ответственность, расторжение. Загружает релевантные статьи закона.
          ↓
   ┌─→🔴 ДЕЙСТВИЕ: Читает секцию «Ответственность», формирует оценку неустойки
   │       ↓
   │   🟣 РЕФЛЕКСИЯ: Находит пункт с завышенной неустойкой.
   │       Сверяет с актуальной судебной практикой. Фиксирует риск в черновик.
   │       ↓
   └───Продолжает цикл для остальных секций
          ↓
🟣 ФИНАЛЬНАЯ РЕФЛЕКСИЯ: Перепроверяет все выводы, ищет упущенные риски,
   проверяет, не противоречат ли выводы по разным секциям друг другу
          ↓
🔴 ШЛЮЗ: Окно ревью — «Найдено 12 рисков, 3 критические ошибки. Применить правки?»
   Пользователь видит дифф по каждому пункту, может отредактировать или отклонить
          ↓
ПОЛЬЗОВАТЕЛЬ: Подтверждает
          ↓
✅ КОММИТ: Черновики атомарно применены к документу. Снапшот сохранён.

Здесь нет спешки. Каждый шаг осознан, проверен, подтверждён. Агент работает не как «быстрый генератор», а как внимательный аналитик.

---

💡 Почему это меняет правила игры?

Доверие строится на архитектуре, а не на обещаниях. Черновики, снапшоты и шлюзы делают ошибки обратимыми. Стоимость сбоя стремится к нулю. Разработчик или бизнес-пользователь больше не боится запустить агента на рабочем проекте, потому что до явного подтверждения ни одно действие не покинет песочницу.

ИИ становится коллегой, а не чёрным ящиком. Каждый шаг прозрачен, логируем и объясним. Человек видит не только результат, но и процесс. Вы не гадаете, что сделал ИИ — вы видите каждый шаг и держите руку на пульсе.

Безопасность встроена, а не добавлена. Предохранители работают на уровне исполнения, а не промпта. Модель не может «договориться» с системой обойти защиту, потому что проверка происходит после генерации, на уровне вызова инструментов.

Масштабируемость без страха. Один и тот же цикл разворачивается в коде, данных, текстах, процессах и интеграциях. Меняются инструменты — не меняется логика ^[5]. Универсальный протокол инструментов позволяет агенту работать с чем угодно, не теряя в безопасности.

Экономика под контролем. Лимиты итераций, детектор циклов и паузы на подтверждение предотвращают runaway-сценарии и неконтролируемые расходы токенов. Бюджет расходуется осмысленно, а не сжигается в бесконечной «мыслительной жвачке».

---

🔮 Что дальше: эволюция рефлексирующих агентов

Текущая архитектура закладывает фундамент. Следующие шаги эволюции:

1. Коллективная рефлексия: несколько агентов проверяют работу друг друга — как код-ревью, только для любых решений.

2. Предиктивная рефлексия: агент предсказывает, что может пойти не так, до выполнения действия, а не после.

3. Обучение ^[6] на ошибках: паттерны неудач сохраняются в «память агента» и учитываются в будущих задачах.

4. Этическая рефлексия: проверка решений на соответствие не только целям, но и этическим нормам и регуляторным требованиям.

---

🔚 Заключение: медленное мышление — это и есть прогресс

Мы привыкли измерять прогресс ИИ скоростью. Но настоящий прорыв происходит, когда мы учим машины не торопиться. Когда между стимулом ^[7] и реакцией ^[8] появляется пауза — на размышление, проверку, сомнение.

Гонка за «полной автономностью» часто игнорирует главный принцип инженерии: надёжность важнее скорости. Агент, который работает быстро, но ломает процессы, бесполезен. Агент, который работает чуть медленнее, но предсказуемо, проверяет себя и уважает границы, становится частью рабочего потока.

Универсальная архитектура рефлексирующего агента — это не про то, чтобы сделать ИИ «безопаснее» как маркетинговый слоган. Это про другой класс агентов: тех, кто работает не вместо человека, а вместе с ним. Кто не скрывает свои действия, а делает их прозрачными. Кто не боится перепроверить себя и признать ошибку. Кто берёт на себя рутину, синтез и исполнение, но оставляет за человеком право вето, контекст и финальное слово.

Именно такие агенты выйдут из песочниц в реальный мир. В больницы, суды, банки, фабрики. Не потому что они быстрее. А потому что им можно доверять.

Самый умный агент — не тот, который делает всё сам. А тот, который умеет останавливаться, проверять себя и спрашивать: «Я всё правильно понял?»

---

Если вам интересны технические детали реализации (управление токенами, fuzzy-поиск, интеграция MCP, логика рекурсивного цикла) или хотите увидеть схему развёртывания для конкретного домена — дайте знать, разберём любой компонент под микроскопом.