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

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

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

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

🐍 Реализация рефлексирующего ИИ-агента на 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.pyfrom dataclasses import dataclass, fieldfrom typing import Any, Optional, List, Dict, Literalfrom datetime import datetimefrom enum import Enumclass AgentMode(Enum):    CHAT = "chat"    AGENT = "agent"    TERMINAL = "terminal"class ToolRisk(Enum):    SAFE_READ = "safe_read"      # Автоматически    MODIFY = "modify"            # Черновик    DESTRUCTIVE = "destructive"  # Требует подтверждения@dataclassclass 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@dataclassclass ToolCall:    id: str    name: str    arguments: Dict[str, Any]    risk: ToolRisk@dataclassclass Draft:    file_path: str    original_content: str    new_content: str    status: Literal["pending", "accepted", "rejected"] = "pending"@dataclassclass Snapshot:    id: str    timestamp: float    files: Dict[str, str]    messages: List[Message]    drafts: List[Draft]

🧠 Ядро агента

# core/agent.pyimport asyncioimport jsonimport hashlibfrom typing import List, Optional, Dict, Any, Callablefrom datetime import datetimefrom openai import AsyncOpenAIfrom .types import (    AgentMode, Message, ToolCall, Draft, Snapshot, ToolRisk)from .context import DynamicContextfrom .state import AgentStatefrom safety.jail import ScopeJailfrom safety.detectors import LoopDetector, IterationLimiterfrom safety.snapshots import SnapshotManagerfrom tools.base import ToolRegistryclass 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**\n\n"                        "Пожалуйста, проверьте свою работу:\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.pyfrom abc import ABC, abstractmethodfrom typing import Dict, Any, List, Optionalfrom core.types import ToolRiskclass 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:        """Выполнение инструмента"""        passclass 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 fullclass 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.pyfrom typing import Dict, Any, Listimport reclass 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"rm\s+-rf",        # Опасные команды            r"sudo\s+",         # Повышение привилегий            r"\|\s*sh\s",       # Инъекции        ]        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 Trueclass 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:        passclass 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.pyimport asynciofrom core.agent import ReflectiveAgentclass 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("\n\n👋 До свидания!")                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.pyimport asynciofrom ui.console import ConsoleUIfrom core.agent import ReflectiveAgentfrom tools.filesystem import FileSystemTool, PatchFileToolfrom tools.terminal import TerminalToolfrom tools.mcp_adapter import MCPAdapterasync 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.txtopenai>=1.0.0asynciopydantic>=2.0.0python-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 критические ошибки. Применить правки?»   Пользователь видит дифф по каждому пункту, может отредактировать или отклонить          ↓ПОЛЬЗОВАТЕЛЬ: Подтверждает          ↓✅ КОММИТ: Черновики атомарно применены к документу. Снапшот сохранён.

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

---

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

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

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

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

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

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

---

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

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

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

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

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

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

---

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

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

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

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

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

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

---

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