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؟ إضافة بضع أدوات فقط ستؤدي إلى انفجار نافذة السياق (Context Window) لنموذج اللغات الكبيرة (LLM).
في النهاية، الحل يكمن في إرفاق حاويات Docker قصيرة العمر (Ephemeral) تظهر وتختفي في غضون 0.1 ثانية لكل جلسة. من خلال تقسيم المناطق الأمنية بوضوح ومعالجة البيانات الضخمة داخل أنابيب المعالجة (Pipelines) للداخلية للحاوية، يمكنك تقليل استهلاك رموز API (Tokens) بنسبة تتجاوز 70%.
حتى لو دخل الوكيل في حلقة مفرغة أو أطلق قنبلة شعبية (Fork Bomb)، يجب أن يظل الخادم المستضيف سليمًا. ما لم تضغط على موارد المعالج والذاكرة بالقوة على مستوى Cgroups في نواة Linux، فإن مسار تنفيذ (Thread) واحد للوكيل قد يخرج عن السيطرة، مما يؤدي إلى انفجار في تكاليف السحابة.
فيما يلي بنية البيئة المعزولة باستخدام 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 جديد في كل مرة ينفذ فيها أمر، فسينتج عن ذلك تأخير بدء تشغيل بارد (Cold Start) كئيب يستغرق 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"، وتعيين معالجات إشارات POSIX لضمان تدمير الحاوية بشكل نظيف عند إنهاء العملية.الطريقة التقليدية لتحديد وتحديد كل مخطط API كـ JSON تستهلك ما بين 550 إلى 1,400 رمز لكل أداة. وبإضافة 20 أداة فقط، يضيع 20,000 رمز قبل أن تطرح سؤالاً واحدًا.
وفقًا للتقرير التقني لمحرّك البحث You.com، فإن اعتماد طريقة تنفيذ البرامج النصية القائمة على Bash (المعروفة بـ CodeAct) بدلاً من الحقن البسيط لمخطط JSON أدى إلى تقليل استهلاك الرموز بنسبة 61% وزيادة سرعة المعالجة بنسبة 40%.
| عنصر التقييم | حقن مخطط JSON | بروتوكول سياق النموذج (MCP) | أنبوب Bash CLI |
|---|---|---|---|
| رموز تعريف الأدوات | ~550–1,400 رمز لكل أداة | ~550–1,400 رمز لكل أداة | واجهة وصفية واحدة (~100 رمز) |
| تلوث سياق البيانات المتوسطة | مرتفع جدًا (نقل الحمولة بالكامل) | مرتفع (نقل الحمولة بالكامل) | معدوم (تصفية البيانات داخل البيئة المعزولة وإعادتها) |
| عدد رحلات الذهاب والإياب لـ LLM | N من رحلات الذهاب والإياب المتسلسلة | N من رحلات الذهاب والإياب المتسلسلة | رحلة واحدة (تجميع البرامج النصية متعددة الخطوات) |
| تأخير إكمال المهمة | نقطة المرجعية | حدوث عبء إضافي لترتيب البيانات على الخادم | انخفاض بنسبة 48.5% في المتوسط |
| نسبة توفير الرموز | 0% (المرجع) | 0% | توفير بنسبة 61%–98.7% |
تظهر بيانات التحليل الداخلي لفريق Anthropic نتائج مماثلة. فعند التحول إلى طريقة تنفيذ الكود لتحليل الملفات الضخمة، انخفض استهلاك الرموز بنسبة تصل إلى 98.7%. ومع تقليل عدد رحلات الذهاب والإياب بين LLM والأدوات إلى رحلة واحدة، تم تقليل زمن التأخير إلى النصف تقريبًا.
يجب وضع قيود واضحة على أنابيب نصوص CLI في موجه النظام (System Prompt) الممرر إلى LLM:
`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 ليدعه يصلح الكود بنفسه./usr/local/bin/ مع منحها صلاحية التنفيذ (chmod +x). وفي المرات القادمة، يمكنك استدعاء هذا الملف التنفيذي مباشرةً دون الحاجة لاستهلاك المزيد من الرموز.إذا دخل سكريبت داخل البيئة المعزولة في حلقة مفرغة، فستتوقف الجلسة بأكملها. لذلك يجب وضع مهلة زمنية متدرجة (Hierarchical Timeout)؛ مثلاً: 30 ثانية للأوامر العادية، 60 ثانية لتثبيت الحزم، و300 ثانية للجلسة بأكملها. وفي حال تجاوز المهلة، يجب أن يرسل مسار المراقبة إشارة SIGKILL إلى معرّف العملية (PID) المعني لتدمير العملية المسببة للمشكلة ونزع فتيلا بشكل نظيف.
مشكلة أخرى تظهر عند محاولة عدة نماذج لوكلاء الوصول إلى نفس الملف في وحدة التخزين المشتركة، مما يسبب حالة سباق (Race Condition). فإذا قمت بتنفيذ open(path, 'w') ببساطة، فقد يتم تفريغ الملف إلى 0 بايت قبل الحصول على القفل. هنا يصبح القفل القائم على استدعاء النظام fcntl.flock على مستوى نواة 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)
`
السر يكمن في فتح os.open باستخدام العَلمين os.O_RDWR | os.O_CREAT. هذا يمنع تفريغ الملف وتدمير محتواه قبل الحصول على القفل. بعد ذلك، يتم محاولة القفل غير المتزامن باستخدام fcntl.LOCK_EX | fcntl.LOCK_NB، وإذا تم الوصول إلى مهلة الانتظار، يتم إلقاء استثناء لمنع المسارات المنتظرة من التوقف إلى الأبد.