Written with AI assistance from the source video. The video is the authority.
Log in to leave a comment
No posts yet
Darle a un agente acceso directo a Bash mediante subprocess o exec() es equivalente a entregarle las llaves de tu casa. Un solo ataque de inyección de prompts podría comprometer todo el servidor host, o el agente podría sufrir alucinaciones y ejecutar rm -rf /. Por otro lado, si divides y registras todas las herramientas en esquemas OpenAPI detallados, añadir solo un par de herramientas hará que la ventana de contexto de tu LLM explote.
La respuesta definitiva es adjuntar contenedores Docker efímeros que se inicien y destruyan en 0,1 segundos por sesión. Al aislar de manera rigurosa el entorno de seguridad y procesar grandes volúmenes de datos mediante canalizaciones internas en el contenedor, es posible reducir el consumo de tokens de API en más de un 70%.
Incluso si el agente entra en un bucle infinito o ejecuta una bomba fork, el servidor host debe mantenerse intacto. Si no limitas de forma estricta la CPU y la memoria a nivel de Cgroups en el kernel de Linux, un solo hilo fuera de control del agente puede convertirse en una bomba de costes en la nube.
A continuación, se muestra la estructura de un sandbox aislado construido con el SDK de Docker para Python:
`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)
`
Ejecutar docker run cada vez que se envía un comando introduce una latencia de arranque en frío terrible de 4,7 segundos, lo cual es inviable en entornos de producción. En su lugar, debes mantener el sandbox activo en segundo plano como demonio (detach=True, stdin_open=True, tty=True) e inyectar los comandos mediante exec_run, logrando así tiempos de respuesta inferiores a 100 ms.
Hay tres puntos clave de configuración:
mem_limit="512m", cpu_quota=50000 y pids_limit=50.network_disabled=True y read_only=True, montando de forma limitada únicamente los espacios de trabajo /workspace y /tmp.user="1000:1000" e implementar manejadores de señales POSIX para garantizar que el contenedor se destruya de forma limpia cuando el proceso finalice.El enfoque tradicional de definir e inyectar esquemas de API individuales en JSON consume entre 550 y 1.400 tokens por herramienta. Con solo 20 herramientas, habrás gastado 20.000 tokens antes de poder hacer una sola pregunta.
Según un informe técnico del motor de búsqueda You.com, al adoptar la ejecución de scripts en Bash (CodeAct) en lugar de la inyección simple de esquemas JSON, el consumo de tokens se redujo en un 61% y la velocidad de procesamiento mejoró en un 40%.
| Criterio de evaluación | Inyección de esquemas JSON | Model Context Protocol (MCP) | Canalización CLI en Bash |
|---|---|---|---|
| Tokens de definición de herramientas | ~550–1.400 tokens por herramienta | ~550–1.400 tokens por herramienta | 1 interfaz meta (~100 tokens) |
| Contaminación de contexto por datos intermedios | Muy alta (envía el payload completo) | Alta (envía el payload completo) | Ninguna (se procesa y devuelve dentro del sandbox) |
| Interacciones de ida y vuelta con el LLM | N iteraciones secuenciales | N iteraciones secuenciales | 1 iteración (lote de scripts multipaso) |
| Latencia en la finalización de tareas | Referencia | Sobrecarga por serialización en el servidor | Reducción promedio del 48,5% |
| Tasa de ahorro de tokens | 0% (referencia) | 0% | Reducción del 61% al 98,7% |
Los datos de análisis interno del equipo de Anthropic muestran resultados similares. Al cambiar el análisis de archivos de gran tamaño por un enfoque basado en la ejecución de código, el uso de tokens disminuyó hasta un 98,7%. Además, al reducirse a una sola interacción de ida y vuelta entre el LLM y la herramienta, la latencia se redujo casi a la mitad.
Es fundamental establecer restricciones claras para la canalización de texto en CLI dentro del prompt del sistema suministrado al LLM:
`text
You operate inside a sandboxed Linux Bash environment.
To process data files or API responses, follow these constraints:
`
Introducir datos sin procesar en el contexto para analizar un CSV de 100.000 filas es tirar el dinero. Induce al modelo a analizar primero la estructura con head -n 5, realizar las agregaciones dentro del sandbox con awk o python, y devolver solo una línea con el resultado final.
Si le confías el entorno Bash a un agente, eventualmente terminará devolviendo un código de salida Exit Code 127 (Command Not Found) o parámetros con flags incorrectos.
Según un informe del equipo de análisis del framework de evaluación de agentes PASTE, cuando una ejecución de comando falla, activar un bucle de autocorrección en lugar de finalizar la sesión de inmediato reduce la tasa de fallos de ejecución a menos del 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
`
El flujo de trabajo del módulo de recuperación de errores es sencillo:
apt-get o pip.stderr generado se reenvía al LLM para que reescriba el código por sí mismo.chmod +x) dentro de /usr/local/bin/ en el contenedor. En ejecuciones posteriores, bastará con llamar directamente a este binario sin gastar tokens adicionales.Si un script dentro del sandbox entra en un bucle infinito, toda la sesión quedará bloqueada. Para evitarlo, es necesario definir un esquema de timeouts jerárquicos: 30 segundos para comandos generales, 60 segundos para instalación de paquetes y 300 segundos para la sesión completa. Si se supera el límite de tiempo, un hilo de monitoreo debe enviar una señal SIGKILL al PID correspondiente para eliminar de forma limpia únicamente el proceso problemático.
Otro desafío son las condiciones de carrera (race conditions) que ocurren cuando múltiples instancias de agentes acceden al mismo archivo en un volumen compartido. Intentar escribir directamente con open(path, 'w') puede vaciar el archivo a 0 bytes antes de obtener el bloqueo. Por ello, es imprescindible utilizar un sistema de bloqueo basado en la llamada al sistema fcntl.flock a nivel del kernel POSIX.
`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)
`
La clave reside en abrir el archivo mediante os.open con los flags os.O_RDWR | os.O_CREAT. Esto evita que el archivo se truncase antes de adquirir el bloqueo. A continuación, se intenta un bloqueo asíncrono con fcntl.LOCK_EX | fcntl.LOCK_NB y, si se alcanza el tiempo de espera, se lanza una excepción para evitar que los hilos en espera se queden bloqueados indefinidamente.