النشر الإنتاجي والسلامة

المراقبة والقابلية للملاحظة

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 يوماً لتصحيح المشكلات التي يبلغ عنها المستخدمون.

تهانينا! لقد أكملت الدورة. حان الوقت لبناء وكلاء استخدام الحاسوب الخاصين بك. :::

اختبار

الوحدة 5: النشر الإنتاجي والسلامة

خذ الاختبار