|
Professor Seleznov
|
Мы привыкли, что ИИ-агенты — это про скорость. Быстрее написать код, быстрее ответить, быстрее сгенерировать. Но скорость без размышления — это не интеллект, а рефлекс. Настоящий прорыв происходит, когда агент начинает думать, прежде чем делать. Проверять себя после каждого шага. Сомневаться. Спрашивать разрешения. И только потом действовать.
Современные языковые модели отлично рассуждают в теории. Но стоит дать им инструменты и доступ к реальному миру, как проявляются системные уязвимости: галлюцинации, превращённые в действия; каскадные ошибки, где одна неточность тянет за собой цепочку неверных решений; слепое следование цели с игнорированием побочных эффектов; отсутствие самокритики. Решение — не в том, чтобы сделать модель умнее. Решение — в архитектуре, которая принудительно встраивает размышление в каждое звено цикла «восприятие → решение → действие → проверка».
Эта статья — про архитектуру, которая превращает бездумный конвейер генерации в мыслящего коллегу. И она применима не только к работе с кодом, а к любому домену, где цена ошибки высока: юриспруденция, медицина, финансы, управление инфраструктурой, маркетинг, поддержка клиентов, образование.
🐍 Реализация рефлексирующего ИИ-агента на 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**\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.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"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 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("\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.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
🎯 Ключевые особенности реализации
- Рекурсивный цикл - асинхронная рекурсия вместо while
- Черновики - все изменения проходят через DraftManager
- Рефлексия - принудительная самопроверка перед завершением
- Безопасность - многоуровневые проверки (Jail, детектор циклов)
- Снапшоты - возможность отката любого изменения
- Human-in-the-loop - подтверждение опасных действий
-
🧩 Почему «быстрый» агент — это проблема
Большинство агентов живут по схеме think → do. Это быстро, но хрупко. Достаточно одной галлюцинации, неоднозначного промпта или непредвиденного состояния среды — и агент молча отправляет не тому клиенту письмо, ломает интеграцию, генерирует некорректный отчёт или уходит в бесконечный цикл запросов, сжигая бюджет.
Универсальная архитектура заменяет эту схему на семиступенчатый цикл:
Восприятие → Планирование → Черновик → Рефлексия → Шлюз → Коммит → Откат
Это не замедляет работу. Это делает её предсказуемой, аудируемой и безопасной для продакшена.
-
🏗️ Семь столпов универсальной архитектуры
1. Динамическая сборка контекста
Агент не действует вслепую. Перед каждым шагом он собирает актуальное состояние среды: какие данные доступны, какие инструменты подключены, какие политики и ограничения действуют, что уже было сделано в текущей сессии. Это как пилот, проверяющий приборы перед манёвром.
Вместо того чтобы запихивать в промпт мегабайты данных, архитектура собирает динамический контекст:
| Компонент |
Что содержит |
Аналогия |
| Дерево доступных объектов |
Файлы, таблицы, документы, API-эндпоинты — в виде структуры, а не содержимого |
Карта местности |
| Загруженные сущности |
Полное содержимое только того, с чем агент работает прямо сейчас |
Открытые вкладки |
| Границы области (Scope) |
Чёткий периметр, за который нельзя выходить |
Красная лента на месте происшествия |
| Подключённые расширения |
Список доступных инструментов с сигнатурами |
Пояс с инструментами |
| История взаимодействия |
Полная цепочка действий и результатов в рамках сессии |
Оперативная память |
Это даёт агенту не «всезнание», а ситуационную осведомлённость — ровно то, что нужно для принятия решений в конкретный момент. Контекст формируется заново на каждой итерации, чтобы избежать работы с устаревшими допущениями.
-
2. Рекурсивный цикл с правом на паузу
Вместо жёсткого while(true) используется асинхронная рекурсия. Почему это важно? Потому что ИИ-агент — не конвейер, а диалог.
Агент может остановиться, запросить уточнение, дождаться ответа человека, внешнего вебхука или результата долгой операции, и продолжить ровно с того же места. История диалога, состояние инструментов и бюджет вычислений сохраняются. Пауза не ломает процесс — она встроена в его ДНК. Человек остаётся в контуре, а не выпадает из него.
-
3. Виртуальная песочница и черновики
Одно из самых мощных архитектурных решений — концепция черновиков. Все модифицирующие операции не применяются к реальным данным сразу. Вместо этого:
- Создаётся виртуальный слой изменений в памяти.
- Все последующие чтения внутри сессии видят уже изменённые данные (чтобы агент работал с актуальным состоянием).
- Реальные объекты остаются нетронутыми.
- Пользователь видит окно ревью: дифф всех изменений, возможность принять, отклонить или отредактировать.
- Только после явного «Сохранить» черновики атомарно применяются.
Это принцип git staging, перенесённый на любые действия. Будь то правка документа, обновление записи в CRM, отправка сообщения, запуск ETL-пайплайна или изменение конфигурации — система фиксирует «что было» и «что станет», но не трогает продакшен-среду. Черновики превращают агента из «исполнителя» в «советника», оставляя финальное слово за человеком.
-
4. Обязательная фаза рефлексии
Это ключевой архитектурный блок, который отличает «думающего» агента от «генерирующего». Когда агент считает задачу выполненной, система не принимает это на веру.
Включается фаза принудительной рефлексии — внутренний аудит:
- Соответствует ли результат исходной цели?
- Нет ли логических противоречий или пропущенных шагов?
- Соблюдены ли бизнес-правила, форматы, ограничения?
- Можно ли запустить сухую проверку или тест?
Агент пересматривает свои действия, перечитывает изменённые объекты, запрашивает недостающий контекст. Если находит проблемы — исправляет их через те же инструменты. Только после явного подтверждения (verified: true) задача считается завершённой.
Это аналог чек-листа пилота перед взлётом: лучше потратить лишний шаг на проверку, чем получить критическую ошибку в продакшене. В юридическом домене это означало бы перепроверку ссылок на статьи закона. В медицинском — перекрёстный анализ назначений с противопоказаниями. В финансовом — сверку расчётов с контрольными суммами. Рефлексия превращает агента из «генератора текста» в ответственного исполнителя.
-
5. Человеческий шлюз для критических действий
Все действия делятся на три категории риска:
- 🟢 Безопасные чтения — выполняются автоматически
- 🟡 Изменения — создают черновики, не затрагивая оригинал
- 🔴 Деструктивные операции — требуют явного подтверждения
Удаление данных, финансовые транзакции, публикация контента, изменение ролей, массовые рассылки, перезапуск сервисов — всё, что необратимо или влияет на внешнюю среду, требует паузы. Интерфейс показывает карточку с контекстом, рисками, альтернативами и превью изменений. Агент замирает, ждёт решения и продолжает работу только после «зелёного света».
Это не чат с кнопками, а полноценный паттерн Human-in-the-Loop: человек остаётся архитектором последствий, а скорость агента не превращается в безрассудство.
-
6. Многоуровневые предохранители
Защита не может быть фильтром на выходе. Она должна быть частью цикла выполнения. Каждое действие агента проходит через независимые предохранители, встроенные в исполнительный слой:
| Механизм |
Что делает |
Аналогия |
| Scope Jail |
Блокирует выход за пределы разрешённых ресурсов, ролей, политик |
Забор вокруг стройплощадки |
| Детектор зацикливаний |
3 одинаковых действия подряд с одной сигнатурой → остановка с диагностикой |
Автоматический выключатель |
| Лимиты итераций и стоимости |
Жёсткие потолки на шаги и токены с предложением «продолжить?» |
Таймер и счётчик бюджета |
| Снапшоты и откат |
Точка восстановления перед каждой сессией. Ошибка? Откат в один клик |
Кнопка «Отменить всё» |
| Abort-контроллер |
Мгновенная остановка по сигналу пользователя или системы |
Аварийный тормоз |
Эти механизмы работают не «поверх» агента, а встроены в уровень вызова инструментов. Модель не может их обойти, потому что проверка происходит на уровне исполнения, а не генерации текста. Агент не может «договориться» с системой обойти защиту.
-
7. Универсальный протокол инструментов
Архитектура не привязана к конкретному домену. Через стандартизированный интерфейс (аналог Model Context Protocol) подключаются любые сервисы: базы данных, CRM, аналитические платформы, браузеры, почта, IoT-устройства, внешние API.
Это как USB-C для ИИ-агентов:
| Домен |
Что подключается через MCP |
| Разработка |
Файловая система, терминал, Git, линтеры |
| Право |
Базы законов, конструкторы договоров, проверка контрагентов |
| Медицина |
Медицинские справочники, системы расшифровки анализов |
| Финансы |
Платёжные шлюзы, CRM, системы бюджетирования |
| Маркетинг |
Платформы рассылок, аналитика, CMS |
| Образование |
Базы знаний, системы проверки заданий |
Агент видит инструменты как единый набор «рук», а система безопасности проверяет каждый вызов независимо от источника. Сегодня агент работает с таблицами, завтра — с чат-ботами, послезавтра — с логистическими системами. Ядро не меняется.
-
🌍 Как это работает в разных сферах?
| Домен |
Пример цикла агента |
| Аналитика данных |
Формирует запрос → запускает в сухом режиме → проверяет структуру и объём выборки → показывает превью → после одобрения экспортирует отчёт и фиксирует метаданные |
| Маркетинг и коммуникации |
Генерирует черновик рассылки → проверяет тон, персонализацию, соответствие регламенту → ждёт одобрения менеджера → отправляет → логирует результат |
| Исследования и синтез |
Собирает источники → проверяет пересечения и достоверность → формирует черновик обзора → запускает самопроверку на логические разрывы → предлагает финальную версию с цитатами |
| Управление процессами |
Читает задачу из трекера → распределяет подзадачи → проверяет доступность исполнителей → создаёт черновик плана → после утверждения создаёт тикеты и ставит напоминания |
| Поддержка клиентов |
Анализирует тикет → ищет похожие кейсы в базе → формирует ответ-черновик → проверяет на соответствие SLA и тону бренда → предлагает оператору → отправляет после клика |
| Юриспруденция |
Анализирует документ → выявляет риски по секциям → сверяет выводы с судебной практикой → формирует черновик заключения → юрист проверяет и утверждает |
Во всех сценариях сохраняется один паттерн: агент думает, примеряет, проверяет, спрашивает и только потом действует.
-
🔄 Анатомия одного цикла: что происходит под капотом
Рассмотрим универсальный пример — анализ документа с рисками:
ПОЛЬЗОВАТЕЛЬ: "Проанализируй договор аренды на предмет рисков"
↓
🔵 ВОСПРИЯТИЕ: Агент подгружает файл, видит его структуру и метаданные
↓
🟡 ПЛАНИРОВАНИЕ: Определяет, что это договор аренды. Планирует анализ по секциям:
предмет, сроки, ответственность, расторжение. Загружает релевантные статьи закона.
↓
┌─→🔴 ДЕЙСТВИЕ: Читает секцию «Ответственность», формирует оценку неустойки
│ ↓
│ 🟣 РЕФЛЕКСИЯ: Находит пункт с завышенной неустойкой.
│ Сверяет с актуальной судебной практикой. Фиксирует риск в черновик.
│ ↓
└───Продолжает цикл для остальных секций
↓
🟣 ФИНАЛЬНАЯ РЕФЛЕКСИЯ: Перепроверяет все выводы, ищет упущенные риски,
проверяет, не противоречат ли выводы по разным секциям друг другу
↓
🔴 ШЛЮЗ: Окно ревью — «Найдено 12 рисков, 3 критические ошибки. Применить правки?»
Пользователь видит дифф по каждому пункту, может отредактировать или отклонить
↓
ПОЛЬЗОВАТЕЛЬ: Подтверждает
↓
✅ КОММИТ: Черновики атомарно применены к документу. Снапшот сохранён.
Здесь нет спешки. Каждый шаг осознан, проверен, подтверждён. Агент работает не как «быстрый генератор», а как внимательный аналитик.
-
💡 Почему это меняет правила игры?
Доверие строится на архитектуре, а не на обещаниях. Черновики, снапшоты и шлюзы делают ошибки обратимыми. Стоимость сбоя стремится к нулю. Разработчик или бизнес-пользователь больше не боится запустить агента на рабочем проекте, потому что до явного подтверждения ни одно действие не покинет песочницу.
ИИ становится коллегой, а не чёрным ящиком. Каждый шаг прозрачен, логируем и объясним. Человек видит не только результат, но и процесс. Вы не гадаете, что сделал ИИ — вы видите каждый шаг и держите руку на пульсе.
Безопасность встроена, а не добавлена. Предохранители работают на уровне исполнения, а не промпта. Модель не может «договориться» с системой обойти защиту, потому что проверка происходит после генерации, на уровне вызова инструментов.
Масштабируемость без страха. Один и тот же цикл разворачивается в коде, данных, текстах, процессах и интеграциях. Меняются инструменты — не меняется логика. Универсальный протокол инструментов позволяет агенту работать с чем угодно, не теряя в безопасности.
Экономика под контролем. Лимиты итераций, детектор циклов и паузы на подтверждение предотвращают runaway-сценарии и неконтролируемые расходы токенов. Бюджет расходуется осмысленно, а не сжигается в бесконечной «мыслительной жвачке».
-
🔮 Что дальше: эволюция рефлексирующих агентов
Текущая архитектура закладывает фундамент. Следующие шаги эволюции:
- Коллективная рефлексия: несколько агентов проверяют работу друг друга — как код-ревью, только для любых решений.
- Предиктивная рефлексия: агент предсказывает, что может пойти не так, до выполнения действия, а не после.
- Обучение на ошибках: паттерны неудач сохраняются в «память агента» и учитываются в будущих задачах.
- Этическая рефлексия: проверка решений на соответствие не только целям, но и этическим нормам и регуляторным требованиям.
-
🔚 Заключение: медленное мышление — это и есть прогресс
Мы привыкли измерять прогресс ИИ скоростью. Но настоящий прорыв происходит, когда мы учим машины не торопиться. Когда между стимулом и реакцией появляется пауза — на размышление, проверку, сомнение.
Гонка за «полной автономностью» часто игнорирует главный принцип инженерии: надёжность важнее скорости. Агент, который работает быстро, но ломает процессы, бесполезен. Агент, который работает чуть медленнее, но предсказуемо, проверяет себя и уважает границы, становится частью рабочего потока.
Универсальная архитектура рефлексирующего агента — это не про то, чтобы сделать ИИ «безопаснее» как маркетинговый слоган. Это про другой класс агентов: тех, кто работает не вместо человека, а вместе с ним. Кто не скрывает свои действия, а делает их прозрачными. Кто не боится перепроверить себя и признать ошибку. Кто берёт на себя рутину, синтез и исполнение, но оставляет за человеком право вето, контекст и финальное слово.
Именно такие агенты выйдут из песочниц в реальный мир. В больницы, суды, банки, фабрики. Не потому что они быстрее. А потому что им можно доверять.
Самый умный агент — не тот, который делает всё сам. А тот, который умеет останавливаться, проверять себя и спрашивать: «Я всё правильно понял?»-Если вам интересны технические детали реализации (управление токенами, fuzzy-поиск, интеграция MCP, логика рекурсивного цикла) или хотите увидеть схему развёртывания для конкретного домена — дайте знать, разберём любой компонент под микроскопом.-Источник
|
