Written with AI assistance from the source video. The video is the authority.
Log in to leave a comment
No posts yet
एजेंट को subprocess या exec() के माध्यम से सीधे Bash की पहुँच देना, अपने घर की चाबियाँ उसे सौंप देने जैसा है। केवल एक प्रॉम्प्ट इंजेक्शन हमले (prompt injection attack) से आपका पूरा होस्ट सर्वर हैक हो सकता है, या एजेंट मतिभ्रम (hallucination) में आकर rm -rf / चला सकता है। तो क्या सभी टूल्स को OpenAPI स्कीमा के रूप में बारीक टुकड़ों में बाँटकर पंजीकृत कर देना सही उपाय है? केवल कुछ टूल्स जोड़ने पर ही LLM का कॉन्टेक्स्ट विंडो (context window) ओवरफ्लो हो जाएगा।
अंततः इसका समाधान एक अल्पकालिक (Ephemeral) Docker कंटेनर जोड़ना है, जो प्रत्येक सेशन के लिए 0.1 सेकंड में चालू होता है और गायब हो जाता है। सुरक्षा सीमाओं को स्पष्ट रूप से विभाजित करके और बड़े डेटा को कंटेनर के आंतरिक पाइपलाइन के माध्यम से प्रोसेस करके, आप API टोकन के उपयोग को 70% से अधिक कम कर सकते हैं।
यदि एजेंट अनंत लूप (infinite loop) चलाता है या फोर्क बम (fork bomb) फोड़ता है, तब भी आपका होस्ट सर्वर पूरी तरह सुरक्षित रहना चाहिए। यदि आप Linux कर्नेल Cgroups स्तर पर CPU और मेमोरी को जबरन सीमित नहीं करते हैं, तो एक अकेला एजेंट थ्रेड नियंत्रण से बाहर होकर क्लाउड लागत का भारी बोझ पैदा कर सकता है।
यहाँ Python Docker SDK का उपयोग करके बनाया गया एक पृथक (isolated) सैंडबॉक्स ढाँचा दिया गया है:
`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 सेकंड का भयानक कोल्ड स्टार्ट (cold start) विलंब उत्पन्न होता है। इसे प्रोडक्शन सेवा में उपयोग नहीं किया जा सकता। इसके बजाय, बैकग्राउंड डेमन मोड (detach=True, stdin_open=True, tty=True) में सैंडबॉक्स को चालू रखें और exec_run का उपयोग करके केवल कमांड इंजेक्ट करें, जिससे रिस्पॉन्स टाइम घटकर 100ms से कम हो जाता है।
यहाँ तीन मुख्य कॉन्फ़िगरेशन बिंदु दिए गए हैं:
mem_limit="512m", cpu_quota=50000, pids_limit=50 का उपयोग करके हार्डवेयर संसाधनों को छोटे थ्रेड स्तर तक सीमित करें।network_disabled=True और read_only=True द्वारा आंतरिक नेटवर्क पहुँच को रोकें और केवल कार्य स्थान /workspace और /tmp को सीमित रूप से माउंट करें।user="1000:1000" के साथ non-root विशेषाधिकार लागू करें और POSIX सिग्नल हैंडलर का उपयोग करें ताकि प्रक्रिया समाप्त होने पर कंटेनर को सफाई से नष्ट किया जा सके।प्रत्येक API स्कीमा को JSON के रूप में परिभाषित करके इंजेक्ट करने का पारंपरिक तरीका प्रति टूल 550 से 1,400 टोकन खर्च करता है। यदि आप केवल 20 टूल्स जोड़ते हैं, तो एक भी प्रश्न पूछने से पहले ही 20,000 टोकन समाप्त हो जाएंगे।
सर्च इंजन You.com की तकनीकी रिपोर्ट के अनुसार, सरल JSON स्कीमा इंजेक्शन के बजाय Bash-आधारित स्क्रिप्ट निष्पादन विधि (CodeAct) को अपनाने से टोकन का उपयोग 61% घट गया और प्रोसेसिंग गति 40% बढ़ गई।
| मूल्यांकन आइटम | JSON स्कीमा इंजेक्शन | Model Context Protocol (MCP) | Bash CLI पाइपलाइन |
|---|---|---|---|
| टूल परिभाषा टोकन | प्रति टूल ~550–1,400 टोकन | प्रति टूल ~550–1,400 टोकन | 1 मेटा इंटरफ़ेस (~100 टोकन) |
| इंटरमीडिएट डेटा कॉन्टेक्स्ट प्रदूषण | अत्यधिक उच्च (पूरा पेलोड भेजा जाता है) | उच्च (पूरा पेलोड भेजा जाता है) | कोई नहीं (सैंडबॉक्स के भीतर फ़िल्टर करके वापस किया जाता है) |
| LLM राउंड ट्रिप की संख्या | N बार क्रमिक राउंड ट्रिप | N बार क्रमिक राउंड ट्रिप | 1 बार (बहु-स्तरीय स्क्रिप्ट बंडल) |
| कार्य पूरा करने में विलंबता (Latency) | बेसलाइन | सर्वर सीरियलाइज़ेशन ओवरहेड | औसतन 48.5% की कमी |
| टोकन बचत दर | 0% (बेसलाइन) | 0% | 61%–98.7% की बचत |
Anthropic टीम का आंतरिक विश्लेषण डेटा भी इसी तरह के परिणाम दिखाता है। बड़े फ़ाइल विश्लेषण कार्यों के लिए कोड निष्पादन विधि पर स्विच करने पर टोकन का उपयोग 98.7% तक कम हो गया। LLM और टूल्स के बीच राउंड-ट्रिप की संख्या घटकर केवल 1 रह जाने से विलंबता (latency) भी लगभग आधी हो जाती है।
LLM को दिए जाने वाले सिस्टम प्रॉम्प्ट में CLI टेक्स्ट पाइपलाइन की सीमाओं को स्पष्ट रूप से परिभाषित किया जाना चाहिए:
`text
You operate inside a sandboxed Linux Bash environment.
To process data files or API responses, follow these constraints:
`
100,000 पंक्तियों वाली CSV फ़ाइल का विश्लेषण करते समय, मूल डेटा को सीधे कॉन्टेक्स्ट में डालना पैसे की बर्बादी है। 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 को भेजें ताकि वह स्वयं संशोधित कोड तैयार कर सके।/usr/local/bin/ में निष्पादन अनुमति (chmod +x) देकर सहेजें। अगली बार केवल इसी बाइनरी को सीधे कॉल किया जा सकता है, जिससे टोकन दोबारा खर्च करने की आवश्यकता नहीं होगी।यदि सैंडबॉक्स के अंदर की स्क्रिप्ट एक अनंत लूप में फंस जाती है, तो पूरा सेशन लॉक हो जाएगा। आपको एक पदानुक्रमित समय-सीमा (hierarchical timeout) निर्धारित करनी चाहिए। उदाहरण के लिए, सामान्य कमांड के लिए 30 सेकंड, पैकेज इंस्टॉलेशन के लिए 60 सेकंड और पूरे सेशन के लिए 300 सेकंड की सीमा तय करें। यदि समय-सीमा पार हो जाती है, तो मॉनिटरिंग थ्रेड को संबंधित प्रोसेस PID पर SIGKILL भेजना चाहिए ताकि केवल समस्याग्रस्त प्रक्रिया को सफाई से समाप्त किया जा सके।
एक और समस्या रेस कंडीशन (race condition) है, जो तब उत्पन्न होती है जब कई एजेंट इंस्टेंस एक ही साझा वॉल्यूम फ़ाइल तक पहुँचने का प्रयास करते हैं। लॉक प्राप्त करने से पहले केवल open(path, 'w') चलाने से फ़ाइल 0 बाइट तक रीसेट हो सकती है। इसलिए 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 फ़्लैग के साथ खोला जाए। यह लॉक हासिल करने से पहले फ़ाइल को छोटा (truncate) होने से बचाता है। इसके बाद, fcntl.LOCK_EX | fcntl.LOCK_NB के माध्यम से एक एसिंक्रोनस लॉक का प्रयास करें, और समय सीमा पूरी होने पर एक एक्सेप्शन (exception) थ्रो करें ताकि प्रतीक्षारत थ्रेड्स को अनिश्चित काल के लिए ब्लॉक होने से रोका जा सके।