الدرس 7 من 20

الوكلاء طويلو المدى

تتبع التقدم

4 دقيقة للقراءة

الوكلاء طويلو المدى يحتاجون معرفة أين هم، ما اكتمل، وما التالي. تتبع التقدم الفعال ليس فقط عن الحالة—إنه عن الحالة القابلة للاسترداد.

ثلاثة مستويات من التتبع

1. تقدم مستوى المهمة

تتبع إكمال المهام الفرعية عالية المستوى:

class TaskTracker:
    def __init__(self):
        self.tasks = []

    def add_task(self, task_id: str, description: str, dependencies: list = None):
        self.tasks.append({
            "id": task_id,
            "description": description,
            "dependencies": dependencies or [],
            "status": "pending",  # pending, in_progress, completed, failed
            "started_at": None,
            "completed_at": None,
            "result": None
        })

    def start_task(self, task_id: str):
        task = self.get_task(task_id)
        task["status"] = "in_progress"
        task["started_at"] = datetime.now().isoformat()

    def complete_task(self, task_id: str, result: dict):
        task = self.get_task(task_id)
        task["status"] = "completed"
        task["completed_at"] = datetime.now().isoformat()
        task["result"] = result

    def get_next_task(self) -> dict | None:
        """احصل على المهمة التالية مع تبعيات مُرضية."""
        for task in self.tasks:
            if task["status"] != "pending":
                continue
            deps_satisfied = all(
                self.get_task(dep)["status"] == "completed"
                for dep in task["dependencies"]
            )
            if deps_satisfied:
                return task
        return None

2. تقدم مستوى الميزة

لمهام التطوير، تتبع الميزات ليس فقط المهام:

# features.json - تقدم قابل للقراءة من البشر
{
  "feature": "مصادقة المستخدم",
  "subtasks": [
    {"name": "إنشاء نموذج User", "status": "completed", "files": ["models/user.py"]},
    {"name": "إضافة نقطة نهاية تسجيل الدخول", "status": "completed", "files": ["api/auth.py"]},
    {"name": "إضافة وسيط JWT", "status": "in_progress", "files": ["middleware/auth.py"]},
    {"name": "كتابة الاختبارات", "status": "pending", "files": []}
  ],
  "overall_progress": "60%",
  "blockers": [],
  "last_updated": "2025-12-16T10:30:00Z"
}

3. تقدم مستوى Git

استخدم التحكم بالإصدار كنظام نقاط حفظ:

import subprocess

class GitCheckpointer:
    def __init__(self, repo_path: str):
        self.repo_path = repo_path

    def checkpoint(self, message: str, task_id: str):
        """إنشاء commit نقطة حفظ."""
        subprocess.run(["git", "add", "."], cwd=self.repo_path)
        subprocess.run([
            "git", "commit", "-m",
            f"[AGENT] {message}\n\nTask: {task_id}\nTimestamp: {datetime.now().isoformat()}"
        ], cwd=self.repo_path)

    def rollback_to_checkpoint(self, task_id: str):
        """التراجع لآخر نقطة حفظ لمهمة."""
        # البحث عن commit بواسطة task_id في الرسالة
        result = subprocess.run(
            ["git", "log", "--oneline", "--grep", f"Task: {task_id}"],
            capture_output=True, text=True, cwd=self.repo_path
        )
        if result.stdout:
            commit_hash = result.stdout.split()[0]
            subprocess.run(["git", "reset", "--hard", commit_hash], cwd=self.repo_path)
            return True
        return False

    def get_changes_since(self, task_id: str) -> list:
        """احصل على جميع الملفات المتغيرة منذ بدء المهمة."""
        result = subprocess.run(
            ["git", "diff", "--name-only", f"HEAD~10"],
            capture_output=True, text=True, cwd=self.repo_path
        )
        return result.stdout.strip().split("\n")

تقارير التقدم

اجعل التقدم مرئياً للبشر:

def generate_progress_report(tracker: TaskTracker) -> str:
    completed = len([t for t in tracker.tasks if t["status"] == "completed"])
    total = len(tracker.tasks)
    current = next((t for t in tracker.tasks if t["status"] == "in_progress"), None)

    report = f"""
## تقرير التقدم
**الإجمالي**: {completed}/{total} مهام ({completed/total*100:.0f}%)

### مكتمل
{chr(10).join(f"✓ {t['description']}" for t in tracker.tasks if t["status"] == "completed")}

### قيد التنفيذ
{"→ " + current["description"] if current else "لا شيء"}

### متبقي
{chr(10).join(f"○ {t['description']}" for t in tracker.tasks if t["status"] == "pending")}

**آخر تحديث**: {datetime.now().strftime("%Y-%m-%d %H:%M")}
"""
    return report

أفضل الممارسات

الممارسة لماذا
نقطة حفظ بعد كل مهمة فرعية تقليل العمل المفقود عند الفشل
استخدم git commits كنقاط حفظ تراجع سهل، تاريخ مدمج
أبقِ ملفات التقدم قابلة للقراءة من البشر البشر قد يحتاجون للتدخل
تتبع الوقت المستغرق لكل مهمة يساعد في تقدير العمل المتبقي
سجّل القرارات، ليس فقط الإجراءات فهم لماذا أُجريت التغييرات

ملاحظة نيردية: ملف progress.json يمكن للإنسان قراءته وفهمه يستحق أكثر من تسلسل الكائنات الذكي. عندما تسوء الأمور الساعة 3 صباحاً، ستشكر نفسك.

التالي: ماذا يحدث عندما تفشل الأمور—الاسترداد والاستئناف. :::

اختبار

الوحدة 2: الوكلاء طويلو المدى

خذ الاختبار