الاسترداد والاستئناف

الأشياء ستفشل. الشبكات تنقطع، واجهات API تنتهي مهلتها، الأخطاء تعطل وكيلك. السؤال ليس إذا ولكن كم بسلاسة تسترد.

أنماط الفشل

نوع الفشل	السبب	استراتيجية الاسترداد
انتهاء مهلة الشبكة	API بطيء، انقطاع الاتصال	إعادة المحاولة مع تراجع
فيض السياق	سياق متراكم كثير	جلسة جديدة، تحميل نقطة الحفظ
خطأ الأداة	خدمة خارجية فشلت	تخطي أو إعادة محاولة المهمة
ارتباك الوكيل	مخرج سيء، هلوسة	تراجع، إعادة طلب
تعطل العملية	نفاد الذاكرة، استثناء	استئناف من آخر نقطة حفظ

نمط إعادة التشغيل السلس

class ResilientAgent:
    def __init__(self, state_file: str, max_retries: int = 3):
        self.state_file = state_file
        self.max_retries = max_retries
        self.state = self.load_state()

    def run_with_recovery(self):
        """الحلقة الرئيسية مع استرداد تلقائي."""
        while not self.is_complete():
            task = self.get_current_task()

            for attempt in range(self.max_retries):
                try:
                    result = self.execute_task(task)
                    self.checkpoint(task, result)
                    break
                except RecoverableError as e:
                    self.log(f"المحاولة {attempt + 1} فشلت: {e}")
                    if attempt < self.max_retries - 1:
                        time.sleep(2 ** attempt)  # تراجع أسي
                    else:
                        self.mark_task_failed(task, str(e))
                        self.handle_failed_task(task)
                except FatalError as e:
                    self.log(f"خطأ قاتل: {e}")
                    self.save_state()  # حفظ التقدم
                    raise  # لا يمكن الاسترداد، الخروج

    def handle_failed_task(self, task: dict):
        """قرر ماذا تفعل مع المهام الفاشلة."""
        if task.get("critical"):
            # لا يمكن المتابعة بدون هذه المهمة
            raise FatalError(f"مهمة حرجة فشلت: {task['id']}")
        else:
            # تخطي والمتابعة
            self.log(f"تخطي مهمة غير حرجة: {task['id']}")
            self.advance_to_next_task()

إعداد البيئة عند الاستئناف

عند الاستئناف، البيئة قد تكون تغيرت:

class EnvironmentManager:
    def __init__(self, required_state: dict):
        self.required = required_state

    def verify_environment(self) -> list[str]:
        """تحقق إذا البيئة تطابق الحالة المتوقعة."""
        issues = []

        # تحقق من دليل العمل
        if not os.path.exists(self.required.get("work_dir", ".")):
            issues.append("دليل العمل مفقود")

        # تحقق من وجود الملفات المطلوبة
        for file in self.required.get("required_files", []):
            if not os.path.exists(file):
                issues.append(f"ملف مطلوب مفقود: {file}")

        # تحقق من حالة git
        if self.required.get("expected_branch"):
            current = self.get_current_branch()
            if current != self.required["expected_branch"]:
                issues.append(f"فرع خاطئ: {current} مقابل {self.required['expected_branch']}")

        return issues

    def restore_environment(self):
        """محاولة استعادة البيئة المتوقعة."""
        # الانتقال للفرع الصحيح
        if self.required.get("expected_branch"):
            subprocess.run(["git", "checkout", self.required["expected_branch"]])

        # استعادة من commit نقطة الحفظ إذا متوفر
        if self.required.get("checkpoint_commit"):
            subprocess.run(["git", "reset", "--hard", self.required["checkpoint_commit"]])

        # إعادة تثبيت التبعيات إذا لزم
        if os.path.exists("requirements.txt"):
            subprocess.run(["pip", "install", "-r", "requirements.txt"])

التحقق من نقطة الحفظ

لا تثق بنقاط الحفظ أعمى—تحقق منها:

def validate_checkpoint(state: dict) -> tuple[bool, list[str]]:
    """تحقق من سلامة نقطة الحفظ قبل الاستئناف."""
    errors = []

    # تحقق من الحقول المطلوبة
    required = ["task", "subtasks", "current_index", "created_at"]
    for field in required:
        if field not in state:
            errors.append(f"حقل مفقود: {field}")

    # تحقق من حدود الفهرس
    if state.get("current_index", 0) > len(state.get("subtasks", [])):
        errors.append("الفهرس الحالي خارج الحدود")

    # تحقق من صحة الطوابع الزمنية
    try:
        datetime.fromisoformat(state["created_at"])
    except (KeyError, ValueError):
        errors.append("تنسيق طابع زمني غير صالح")

    # تحقق من أن المهام المكتملة لها نتائج
    for completed in state.get("completed", []):
        if "result" not in completed:
            errors.append(f"مهمة مكتملة تفتقد النتيجة: {completed.get('task')}")

    return len(errors) == 0, errors

تدفق الاستئناف

def resume_agent(state_file: str):
    """تدفق استئناف كامل مع التحقق."""

    # 1. تحميل نقطة الحفظ
    if not os.path.exists(state_file):
        raise FileNotFoundError("لم يُعثر على نقطة حفظ")

    state = json.load(open(state_file))

    # 2. التحقق من نقطة الحفظ
    valid, errors = validate_checkpoint(state)
    if not valid:
        print(f"فشل التحقق من نقطة الحفظ: {errors}")
        if input("محاولة الإصلاح؟ (y/n): ") == "y":
            state = repair_checkpoint(state, errors)
        else:
            raise ValueError("نقطة حفظ غير صالحة")

    # 3. التحقق من البيئة
    env_manager = EnvironmentManager(state.get("environment", {}))
    issues = env_manager.verify_environment()
    if issues:
        print(f"مشاكل البيئة: {issues}")
        env_manager.restore_environment()

    # 4. إنشاء الوكيل والاستئناف
    agent = ResilientAgent(state_file)
    agent.run_with_recovery()

ملاحظة نيردية: اختبر منطق الاسترداد بقتل وكيلك عشوائياً في منتصف المهمة. إذا لم تستطع الاستئناف بنظافة، نقاط الحفظ معطلة.

الوحدة التالية: بروتوكول سياق النموذج وبناء مهارات الوكيل. :::