النشر الإنتاجي والسلامة
المراقبة والقابلية للملاحظة
5 دقيقة للقراءة
تحتاج وكلاء استخدام الحاسوب في الإنتاج إلى مراقبة شاملة لضمان الموثوقية، وتصحيح المشكلات، والحفاظ على الأمان.
ما يجب مراقبته
| الفئة | المقاييس |
|---|---|
| الأداء | وقت الاستجابة، تكرارات الحلقة، معدل الإكمال |
| التكاليف | الرموز المستخدمة، لقطات الشاشة المعالجة، استدعاءات API |
| الموثوقية | معدل الخطأ، عدد إعادة المحاولات، تكرار المهلة |
| الأمان | الإجراءات المحظورة، الأنماط المشبوهة، فشل المصادقة |
إطار التسجيل
import logging
from datetime import datetime
logger = logging.getLogger("computer_use_agent")
class AgentLogger:
def __init__(self, session_id: str):
self.session_id = session_id
self.start_time = datetime.now()
def log_action(self, action: dict, result: dict):
logger.info({
"session": self.session_id,
"timestamp": datetime.now().isoformat(),
"action_type": action.get("type"),
"coordinates": action.get("coordinate"),
"success": result.get("success"),
"duration_ms": result.get("duration_ms")
})
def log_screenshot(self, size_bytes: int):
logger.info({
"session": self.session_id,
"event": "screenshot",
"size_bytes": size_bytes
})
تسجيل الجلسة
سجل الجلسات الكاملة للتصحيح:
class SessionRecorder:
def __init__(self):
self.screenshots = []
self.actions = []
def record_frame(self, screenshot_b64: str, action: dict):
self.screenshots.append(screenshot_b64)
self.actions.append({
"timestamp": time.time(),
"action": action
})
def save_recording(self, path: str):
# حفظ كفيديو أو تسلسل إطارات
with open(path, 'w') as f:
json.dump({
"actions": self.actions,
"frame_count": len(self.screenshots)
}, f)
تتبع التكاليف
class CostTracker:
COSTS = {
"input_token": 0.003 / 1000, # لكل رمز
"output_token": 0.015 / 1000, # لكل رمز
"image_token": 0.003 / 1000, # لكل رمز صورة
}
def __init__(self):
self.total_cost = 0
self.call_count = 0
def add_usage(self, response):
usage = response.usage
cost = (
usage.input_tokens * self.COSTS["input_token"] +
usage.output_tokens * self.COSTS["output_token"]
)
self.total_cost += cost
self.call_count += 1
return {
"call_cost": cost,
"total_cost": self.total_cost,
"call_count": self.call_count
}
فحوصات الصحة
async def health_check():
checks = {
"api_connection": await test_api_connection(),
"display_available": await test_display(),
"disk_space": check_disk_space(),
"memory_available": check_memory()
}
healthy = all(checks.values())
return {"healthy": healthy, "checks": checks}
التنبيهات
def check_and_alert(metrics):
alerts = []
if metrics["error_rate"] > 0.1:
alerts.append("معدل خطأ مرتفع: >10%")
if metrics["avg_response_time"] > 30:
alerts.append("استجابات بطيئة: متوسط >30 ثانية")
if metrics["cost_per_task"] > 1.0:
alerts.append("تكاليف مرتفعة: >$1 لكل مهمة")
if alerts:
send_alert(alerts)
مقاييس لوحة التحكم
المقاييس الأساسية للوحة التحكم:
| المقياس | الهدف | عتبة التنبيه |
|---|---|---|
| معدل النجاح | >95% | <90% |
| متوسط وقت الإكمال | <60 ثانية | >120 ثانية |
| التكلفة لكل مهمة | <$0.50 | >$1.00 |
| معدل الخطأ | <5% | >10% |
أدوات التصحيح
# إعادة تشغيل الجلسات الفاشلة
def replay_session(session_id: str):
session = load_session(session_id)
for i, (screenshot, action) in enumerate(session):
print(f"الخطوة {i}: {action}")
display_screenshot(screenshot)
input("اضغط Enter للمتابعة...")
نصيحة: خزّن تسجيلات الجلسات لمدة 7-30 يوماً لتصحيح المشكلات التي يبلغ عنها المستخدمون.
تهانينا! لقد أكملت الدورة. حان الوقت لبناء وكلاء استخدام الحاسوب الخاصين بك. :::