Written with AI assistance from the source video. The video is the authority.
Log in to leave a comment
No posts yet
Предоставлять агенту прямой доступ к Bash через subprocess или exec() — это все равно что отдать ему ключи от квартиры. Одна атака типа Prompt Injection — и весь хост-сервер будет взломан, либо агент галлюцинирует и выполнит rm -rf /. Но что, если разбить все инструменты на мелкие части и зарегистрировать их через OpenAPI-схему? В этом случае добавление всего нескольких инструментов сразу перегрузит контекстное окно LLM.
В конечном счете решение заключается в подключении одноразовых (Ephemeral) контейнеров Docker, которые поднимаются и уничтожаются за 0,1 секунды на каждую сессию. Четкое разграничение периметра безопасности и обработка больших объемов данных внутри контейнерного пайплайна позволяют сократить расход токенов API более чем на 70%.
Даже если агент войдет в бесконечный цикл или устроит форк-бомбу, хост-сервер должен оставаться в порядке. Если не ограничить ресурсы CPU и памяти на уровне Cgroups ядра Linux, один зависший поток агента может привести к огромным счетам за облако.
Вот архитектура изолированной песочницы, построенная с использованием Python Docker SDK.
`python
import atexit
import os
import signal
import docker
from docker.errors import DockerException
class EphemeralBashSandbox:
def init(self, workspace_host_path: str, image: str = "python:3.11-slim"):
self.client = docker.from_env()
self.image = image
self.workspace_host_path = os.path.abspath(workspace_host_path)
self.container = None
self._start_sandbox()
atexit.register(self.cleanup)
signal.signal(signal.SIGINT, self._signal_handler)
signal.signal(signal.SIGTERM, self._signal_handler)
def _start_sandbox(self):
self.container = self.client.containers.run(
image=self.image,
command="/bin/bash",
detach=True,
stdin_open=True,
tty=True,
network_disabled=True,
read_only=True,
mem_limit="512m",
cpu_quota=50000,
pids_limit=50,
user="1000:1000",
volumes={
self.workspace_host_path: {
"bind": "/workspace",
"mode": "rw"
},
"/tmp": {
"bind": "/tmp",
"mode": "rw"
}
},
working_dir="/workspace",
environment={"HOME": "/tmp"}
)
def execute_command(self, cmd: str, timeout: int = 30) -> tuple[int, str, str]:
if not self.container:
raise RuntimeError("Sandbox container is not active.")
exec_res = self.container.exec_run(
cmd=["/bin/bash", "-c", cmd],
workdir="/workspace",
demux=True
)
exit_code = exec_res.exit_code
stdout = exec_res.output[0].decode('utf-8', errors='replace') if exec_res.output and exec_res.output[0] else ""
stderr = exec_res.output[1].decode('utf-8', errors='replace') if exec_res.output and exec_res.output[1] else ""
return exit_code, stdout, stderr
def cleanup(self):
if self.container:
try:
self.container.stop(timeout=2)
self.container.remove(force=True)
except DockerException:
pass
finally:
self.container = None
def _signal_handler(self, signum, frame):
self.cleanup()
os._exit(0)
`
Если выполнять docker run заново при каждом вызове команды, возникнет ужасная задержка холодного старта в 4,7 секунды. Это абсолютно неприемлемо для реального сервиса. Вместо этого нужно держать песочницу запущенной в фоновом режиме демона (detach=True, stdin_open=True, tty=True) и передавать команды через exec_run — тогда время отклика снизится до менее чем 100 мс.
Выделим три ключевых момента настройки:
mem_limit="512m", cpu_quota=50000, pids_limit=50 жестко ограничивают аппаратные ресурсы на уровне небольшого процесса.network_disabled=True и read_only=True блокируют доступ к внутренней сети и монтируют с ограниченными правами только рабочие директории /workspace и /tmp.user="1000:1000" принудительно задает права non-root, а обработчики сигналов POSIX гарантируют корректное удаление контейнера при завершении процесса.Традиционный подход с передачей JSON-схем для каждого отдельного API съедает от 550 до 1 400 токенов на один инструмент. Если у вас 20 инструментов, 20 000 токенов улетят еще до того, как пользователь задаст первый вопрос.
Согласно техническому отчету поисковой системы You.com, переход от простой передачи JSON-схем к исполнению скриптов на базе Bash (CodeAct) сократил использование токенов на 61%, а скорость обработки выросла на 40%.
| Критерий оценки | Внедрение JSON-схемы | Model Context Protocol (MCP) | CLI-пайплайн Bash |
|---|---|---|---|
| Токены на определение инструмента | ~550–1,400 токенов на инструмент | ~550–1,400 токенов на инструмент | 1 мета-интерфейс (~100 токенов) |
| Загрязнение контекста промежуточными данными | Очень высокое (передача всего payload) | Высокое (передача всего payload) | Отсутствует (возврат после очистки внутри песочницы) |
| Количество циклов LLM (Round-trips) | N последовательных циклов | N последовательных циклов | 1 цикл (пакет многошаговых скриптов) |
| Задержка выполнения задачи | Базовый уровень | Издержки сериализации на сервере | Снижение в среднем на 48.5% |
| Экономия токенов | 0% (базовый уровень) | 0% | Экономия 61%–98.7% |
Внутренние аналитические данные команды Anthropic показывают аналогичные результаты. При переходе на исполнение кода для задач анализа больших файлов использование токенов сократилось вплоть до 98,7%. А благодаря тому, что количество итераций между LLM и инструментами сократилось до одного цикла, задержка уменьшилась почти вдвое.
В системном промпте для LLM необходимо четко прописать ограничения для текстового CLI-пайплайна:
`text
You operate inside a sandboxed Linux Bash environment.
To process data files or API responses, follow these constraints:
`
При анализе CSV-файла на 100 000 строк загрузка сырых данных в контекст — это просто выброс денег на ветер. Заставьте модель сначала изучить структуру с помощью head -n 5, агрегировать данные внутри песочницы с помощью awk или python, и вернуть только одну итоговую строку.
Если доверить Bash агенту, он гарантированно будет выдавать Exit Code 127 (Command Not Found) или неверные флаги опций.
Согласно отчету команды фреймворка оценки агентов PASTE, если при неудачном выполнении команды не завершать сессию сразу, а запускать цикл автоматической коррекции, доля ошибок выполнения снижается до уровня менее 5%.
`python
import re
from typing import Callable, Optional
class SelfHealingBashRunner:
def init(self, sandbox: EphemeralBashSandbox, llm_repair_fn: Callable[[str, str], str]):
self.sandbox = sandbox
self.llm_repair_fn = llm_repair_fn
self.max_retries = 3
def run_with_healing(self, initial_cmd: str) -> tuple[bool, str]:
current_cmd = initial_cmd
for attempt in range(self.max_retries):
exit_code, stdout, stderr = self.sandbox.execute_command(current_cmd)
if exit_code == 0:
return True, stdout
if exit_code == 127 or "command not found" in stderr.lower():
missing_binary = self._extract_missing_command(stderr)
if missing_binary:
install_success = self._try_install_package(missing_binary)
if install_success:
continue
repair_prompt = (
f"The executed Bash command failed.\n"
f"Failed Command: {current_cmd}\n"
f"Exit Code: {exit_code}\n"
f"Stderr Output: {stderr}\n"
f"Stdout Output: {stdout}\n"
f"Analyze the error. Return ONLY a corrected single-line Bash command to fulfill the objective."
)
current_cmd = self.llm_repair_fn(stderr, repair_prompt).strip()
return False, f"Failed after {self.max_retries} attempts. Last Stderr: {stderr}"
def _extract_missing_command(self, stderr: str) -> Optional[str]:
match = re.search(r"([a-zA-Z0-9_-]+):\s*command not found", stderr) or re.search(r"command not found:\s*([a-zA-Z0-9_-]+)", stderr)
return match.group(1) if match else None
def _try_install_package(self, binary_name: str) -> bool:
install_cmd = f"apt-get update && apt-get install -y {binary_name} || pip install {binary_name}"
exit_code, _, _ = self.sandbox.execute_command(install_cmd)
return exit_code == 0
`
Логика работы модуля восстановления крайне проста:
apt-get или pip.stderr отправляется обратно в LLM, чтобы модель сама сгенерировала исправленный код.chmod +x), и они сохраняются в /usr/local/bin/ внутри контейнера. В следующий раз можно сразу вызывать готовый бинарник, не расходуя токены повторно.Если скрипт внутри песочницы попадет в бесконечный цикл, заблокируется вся сессия. Для этого необходимы иерархические таймауты: например, 30 секунд для обычных команд, 60 секунд для установки пакетов и 300 секунд для всей сессии. При превышении таймаута мониторинговый поток отправляет SIGKILL по PID этого процесса, аккуратно завершая только проблемный процесс.
Еще одна проблема — состояния гонки (race conditions), возникающие при одновременном обращении нескольких экземпляров агентов к файлу в общем монтируемом томе. Простой вызов open(path, 'w') обнулит файл еще до получения блокировки. В этом случае необходима системная блокировка на уровне ядра POSIX с помощью системного вызова fcntl.flock.
`python
import fcntl
import os
import time
from contextlib import contextmanager
class SafeFileLockTimeout(Exception):
pass
@contextmanager
def safe_file_lock(lock_file_path: str, timeout: float = 10.0, poll_interval: float = 0.05):
lock_dir = os.path.dirname(os.path.abspath(lock_file_path))
if lock_dir:
os.makedirs(lock_dir, exist_ok=True)
fd = os.open(lock_file_path, os.O_RDWR | os.O_CREAT, 0o666)
start_time = time.time()
try:
while True:
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
break
except (OSError, IOError):
if time.time() - start_time >= timeout:
raise SafeFileLockTimeout(
f"Timed out after {timeout} seconds waiting for lock on: {lock_file_path}"
)
time.sleep(poll_interval)
yield fd
finally:
try:
fcntl.flock(fd, fcntl.LOCK_UN)
except (OSError, IOError):
pass
os.close(fd)
`
Ключевым моментом является открытие файла через os.open с флагами os.O_RDWR | os.O_CREAT. Это предотвращает обрезание содержимого файла до момента получения блокировки. Далее выполняется попытка асинхронной блокировки с помощью fcntl.LOCK_EX | fcntl.LOCK_NB, а при достижении таймаута выбрасывается исключение, исключая бесконечное зависание ожидающего потока.