Written with AI assistance from the source video. The video is the authority.
Log in to leave a comment
No posts yet
Donner un accès direct à Bash à un agent via subprocess ou exec(), c'est comme lui confier les clés de votre maison. Une seule injection de prompt peut compromettre l'ensemble du serveur hôte, ou l'agent peut halluciner et exécuter un rm -rf /. Mais si vous découpez tous les outils en schémas OpenAPI pour les enregistrer ? Ajouter seulement quelques outils fera exploser la fenêtre de contexte du LLM.
En fin de compte, la solution consiste à attacher des conteneurs Docker éphémères qui apparaissent et disparaissent en 0,1 seconde par session. En isolant clairement les périmètres de sécurité et en traitant les gros volumes de données au sein du pipeline interne du conteneur, vous pouvez réduire l'utilisation des jetons d'API de plus de 70 %.
Même si l'agent entre dans une boucle infinie ou déclenche une bombe fork, le serveur hôte doit rester intact. Sans une restriction stricte du CPU et de la mémoire au niveau des Cgroups du noyau Linux, un seul thread d'agent en roue libre entraînera une explosion des coûts de cloud.
Voici une structure de bac à sable isolé construite avec le SDK Docker pour 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)
`
Lancer un nouveau docker run à chaque exécution de commande entraîne une latence de démarrage à froid terrible de 4,7 secondes. C'est inutilisable en production. Au lieu de cela, vous devez maintenir le bac à sable actif en mode démon d'arrière-plan (detach=True, stdin_open=True, tty=True) et injecter uniquement les commandes via exec_run pour réduire le temps de réponse à moins de 100 ms.
Voici les trois points de configuration essentiels :
mem_limit="512m", cpu_quota=50000 et pids_limit=50.network_disabled=True et read_only=True, tout en ne montant de manière restreinte que les espaces de travail /workspace et /tmp.user="1000:1000" et configurer des gestionnaires de signaux POSIX pour garantir que le conteneur soit proprement détruit lors de l'arrêt du processus.La méthode traditionnelle consistant à définir et injecter des schémas d'API individuels en JSON consomme entre 550 et 1 400 jetons par outil. Avec seulement 20 outils, vous gâchez 20 000 jetons avant même de poser une question.
Selon un rapport technique du moteur de recherche You.com, l'adoption de l'exécution de scripts Bash (CodeAct) à la place de l'injection de schémas JSON simples a réduit l'utilisation des jetons de 61 % et accéléré la vitesse de traitement de 40 %.
| Élément d'évaluation | Injection de schéma JSON | Model Context Protocol (MCP) | Pipeline CLI Bash |
|---|---|---|---|
| Jetons de définition d'outils | ~550–1 400 jetons par outil | ~550–1 400 jetons par outil | 1 méta-interface (~100 jetons) |
| Pollution du contexte par données intermédiaires | Très élevée (transmission de tout le payload) | Élevée (transmission de tout le payload) | Aucune (renvoyée après nettoyage interne dans le bac à sable) |
| Allers-retours LLM | N allers-retours séquentiels | N allers-retours séquentiels | 1 seul (regroupement de scripts multi-étapes) |
| Latence d'accomplissement des tâches | Référence | Surcoût de sérilisation serveur | Réduction moyenne de 48,5 % |
| Taux d'économie de jetons | 0 % (référence) | 0 % | Économie de 61 % à 98,7 % |
Les données d'analyse interne de l'équipe Anthropic montrent des résultats similaires. Le passage au mode d'exécution de code pour l'analyse de gros fichiers a réduit l'utilisation des jetons jusqu'à 98,7 %. Le nombre d'allers-retours entre le LLM et les outils étant réduit à un seul, la latence est également divisée par près de deux.
Le prompt système fourni au LLM doit imposer clairement les contraintes du pipeline texte CLI :
`text
You operate inside a sandboxed Linux Bash environment.
To process data files or API responses, follow these constraints:
`
Lors de l'analyse d'un fichier CSV de 100 000 lignes, injecter les données brutes dans le contexte revient à jeter de l'argent par la fenêtre. Inspectez la structure avec head -n 5, effectuez l'agrégation avec awk ou python à l'intérieur du bac à sable, puis incitez le modèle à ne renvoyer qu'une seule ligne de résultat final.
Confier Bash à un agent entraînera inévitablement des erreurs Exit Code 127 (Command Not Found) ou des flags d'options incorrects.
Selon le rapport de l'équipe d'analyse du framework d'évaluation d'agents PASTE, le taux d'échec d'exécution est tombé à moins de 5 % lorsqu'une boucle de correction automatique a été intégrée au lieu de fermer la session immédiatement en cas d'échec.
`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
`
Le flux de fonctionnement du module de récupération d'erreur est simple :
apt-get ou pip.stderr produit au LLM pour qu'il génère lui-même un code corrigé./usr/local/bin/ à l'intérieur du conteneur en leur donnant les permissions d'exécution (chmod +x). La prochaine fois, il suffira d'appeler directement ce binaire sans consommer de jetons supplémentaires.Si un script dans le bac à sable entre dans une boucle infinie, toute la session se retrouve verrouillée. Il faut établir des timeouts hiérarchiques : 30 secondes pour les commandes ordinaires, 60 secondes pour l'installation de paquets et 300 secondes pour l'ensemble de la session. Si le timeout est dépassé, un thread de surveillance doit envoyer un SIGKILL au PID concerné pour nettoyer proprement le processus posant problème.
Les conditions de concurrence (race conditions) qui surviennent lorsque plusieurs instances d'agents accèdent au même fichier sur un volume partagé constituent un autre problème. Exécuter un simple open(path, 'w') peut vider le fichier à 0 octet avant même d'avoir obtenu le verrou. Un verrouillage basé sur l'appel système fcntl.flock au niveau du noyau POSIX est indispensable.
`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)
`
L'élément clé est d'ouvrir os.open avec les flags os.O_RDWR | os.O_CREAT. Cela empêche la troncature du fichier avant l'obtention du verrou. Ensuite, une tentative de verrouillage synchrone/asynchrone est effectuée via fcntl.LOCK_EX | fcntl.LOCK_NB, et une exception est levée si le timeout est atteint, évitant ainsi qu'un thread en attente ne reste bloqué indéfiniment.