Written with AI assistance from the source video. The video is the authority.
Log in to leave a comment
No posts yet
Dar acesso direto ao Bash para um agente via subprocess ou exec() é o equivalente a entregar as chaves da sua casa de bandeja. Com um único ataque de injeção de prompt, todo o servidor hospedeiro pode ser comprometido, ou o agente pode alucinar e executar um rm -rf /. Por outro lado, fragmentar e registrar todas as ferramentas como esquemas OpenAPI fará com que a janela de contexto do seu LLM estoure assim que adicionar apenas algumas delas.
A solução definitiva é acoplar um contêiner Docker efêmero (Ephemeral) que surge e desaparece em 0,1 segundo por sessão. Ao isolar a camada de segurança e processar grandes volumes de dados por pipelines internos no contêiner, é possível reduzir o consumo de tokens de API em mais de 70%.
Mesmo que o agente entre em um loop infinito ou execute uma bomba fork, o servidor hospedeiro deve permanecer intacto. Se você não limitar rigidamente o uso de CPU e memória no nível de Cgroups do kernel do Linux, uma única thread do agente pode descontrolar-se e gerar uma explosão de custos na nuvem.
Abaixo está a estrutura de um sandbox isolado usando o SDK do 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)
`
Executar docker run a cada comando gera um tempo de latência de cold start terrível de 4,7 segundos, inviabilizando o uso em produção. Em vez disso, mantenha o sandbox ativo em modo daemon no segundo plano (detach=True, stdin_open=True, tty=True) e injete apenas os comandos com exec_run, reduzindo o tempo de resposta para menos de 100 ms.
Existem três pontos-chave de configuração:
mem_limit="512m", cpu_quota=50000 e pids_limit=50 restringem os recursos de hardware ao nível de uma thread leve.network_disabled=True e read_only=True impedem a varredura da rede interna e montam de forma restrita apenas os diretórios de trabalho /workspace e /tmp.user="1000:1000" força privilégios de usuário não-root, enquanto manipuladores de sinais POSIX garantem que o contêiner seja destruído adequadamente caso o processo seja interrompido.O método tradicional de definir e injetar esquemas de API individuais em JSON consome de 550 a 1.400 tokens por ferramenta. Com apenas 20 ferramentas, você gasta 20.000 tokens antes mesmo de fazer uma única pergunta.
De acordo com um relatório técnico do mecanismo de busca You.com, adotar uma abordagem de execução de scripts via Bash (CodeAct) em vez da simples injeção de esquemas JSON reduziu o uso de tokens em 61% e aumentou a velocidade de processamento em 40%.
| Item de avaliação | Injeção de esquema JSON | Model Context Protocol (MCP) | Pipeline Bash CLI |
|---|---|---|---|
| Tokens de definição de ferramentas | ~550–1.400 tokens por ferramenta | ~550–1.400 tokens por ferramenta | 1 meta-interface (~100 tokens) |
| Poluição de contexto com dados intermediários | Muito alta (envia o payload completo) | Alta (envia o payload completo) | Nenhuma (retorna após filtragem interna no sandbox) |
| Viagens de ida e volta do LLM (Round-trips) | N vezes sequenciais | N vezes sequenciais | 1 vez (agrupamento em script multietapa) |
| Latência para conclusão da tarefa | Linha de base | Overhead de serialização no servidor | Redução média de 48,5% |
| Taxa de economia de tokens | 0% (referência) | 0% | 61%–98,7% de economia |
Dados de análise interna da equipe da Anthropic mostram resultados semelhantes. A transição para a execução de código em tarefas de análise de arquivos de grande volume reduziu o consumo de tokens em até 98,7%. Como o número de viagens de ida e volta entre o LLM e as ferramentas caiu para 1, a latência também foi reduzida quase pela metade.
No prompt de sistema fornecido ao LLM, é fundamental definir claramente as restrições do pipeline de texto via CLI:
`text
You operate inside a sandboxed Linux Bash environment.
To process data files or API responses, follow these constraints:
`
Inserir o arquivo bruto no contexto para analisar um CSV de 100 mil linhas é jogar dinheiro fora. Incentive o modelo a entender a estrutura com head -n 5, agregar os dados dentro do sandbox usando awk ou python e retornar apenas uma linha com o resultado final.
Se você deixar o Bash sob responsabilidade do agente, é 100% certo que ele retornará Exit Code 127 (Command Not Found) ou sinalizadores de opção incorretos.
De acordo com um relatório da equipe de análise do framework de avaliação de agentes PASTE, a taxa de falha na execução caiu para menos de 5% ao implementar um loop de autocorreção em vez de encerrar a sessão imediatamente quando um comando falhava.
`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
`
O fluxo de funcionamento do módulo de recuperação de erros é simples:
apt-get ou pip.stderr gerada de volta ao LLM para que ele próprio corrija o código./usr/local/bin/ com permissão de execução (chmod +x). Da próxima vez, basta chamar esse binário diretamente, sem a necessidade de gastar tokens novamente.Se um script dentro do sandbox entrar em um loop infinito, toda a sessão ficará bloqueada. É necessário estabelecer timeouts hierárquicos: 30 segundos para comandos comuns, 60 segundos para instalação de pacotes e 300 segundos para a sessão inteira. Se o limite de tempo for ultrapassado, uma thread de monitoramento deve enviar um SIGKILL para o PID do processo correspondente, eliminando limpos apenas os processos com problemas.
Outro problema são as condições de corrida (race conditions) que ocorrem quando várias instâncias de agentes acessam o mesmo arquivo em um volume compartilhado. Fazer um simples open(path, 'w') fará com que o arquivo seja truncado para 0 bytes antes mesmo de obter o bloqueio. O bloqueio baseado em chamadas de sistema fcntl.flock no nível do kernel POSIX é indispensável.
`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)
`
O ponto crítico aqui é abrir o os.open com as flags os.O_RDWR | os.O_CREAT. Isso evita que o arquivo seja truncado antes de adquirir o bloqueio. Em seguida, tenta-se o bloqueio assíncrono com fcntl.LOCK_EX | fcntl.LOCK_NB, lançando uma exceção se o limite de tempo for atingido para evitar que a thread de espera fique bloqueada indefinidamente.