سير العمل المدفوع بالمهام

المهام ليست مجرد مهام عادية — إنها العمود الفقري لسير عمل LLM الدائم. بإنشاء مهام للعمل متعدد الخطوات، تضمن الاستمرارية عبر الجلسات وتوفر مسارات تدقيق لما تم.

لماذا مدفوع بالمهام؟

بدون مهام:

• يبدأ LLM كل جلسة غير متأكد ماذا يفعل
• العمل متعدد الخطوات يُنسى في منتصف التنفيذ
• لا يوجد سجل بما تم

مع المهام:

• يستأنف LLM المهام قيد التقدم فورًا
• العمل متعدد الخطوات يبقى عبر الجلسات
• مسار تدقيق مدمج لكل العمل

النمط

1. At session start: check in_progress tasks
2. If tasks exist: resume them
3. If no tasks: create new tasks for current work
4. Update task status as you progress
5. Mark done when complete

التنفيذ

الخطوة 1: أنشئ مهمة للعمل متعدد الخطوات

def start_workflow(title, steps):
    """Create a task for multi-step work."""
    task_id = create_task(
        title=title,
        description=f"Steps:\n" + "\n".join(f"  {i+1}. {s}" for i, s in enumerate(steps)),
        priority="high"
    )
    return task_id

# مثال
task_id = start_workflow("Deploy Synapse v1.6.0", [
    "Bump version in package.json",
    "Update CHANGELOG.md",
    "Commit and push",
    "Wait for CI green",
    "Verify deployment"
])

الخطوة 2: تتبع التقدم في وصف المهمة

def update_progress(task_id, current_step, total_steps, status_note):
    """Update task with current progress."""
    description = f"Progress: {current_step}/{total_steps}\nStatus: {status_note}"
    update_task(task_id, status="in_progress", description=description)

# مثال
update_progress(task_id, 2, 5, "CHANGELOG updated, committing now")

الخطوة 3: الاستئناف عبر الجلسات

def resume_work():
    """At session start, find and resume in-progress tasks."""
    tasks = list_tasks(status="in_progress")
    
    for task in tasks:
        print(f"Resuming: {task['title']}")
        print(f"Last status: {task['description']}")
        
        # حلل التقدم من الوصف
        progress = parse_progress(task['description'])
        next_step = progress['current_step'] + 1
        
        # واصل من الخطوة التالية
        continue_from_step(task['id'], next_step)

الخطوة 4: أكمل وأرشف

def complete_task(task_id, summary):
    """Mark task done with completion summary."""
    update_task(task_id, 
        status="done",
        description=f"COMPLETED. Summary: {summary}"
    )
    # خزّن أيضًا كذاكرة للمرجع طويل الأمد
    remember(
        category="project",
        key=f"completed_{task_id}",
        content=f"Task: {task_id}\nSummary: {summary}",
        tags=["completed", "task"],
        priority="normal"
    )

مثال كامل: سير عمل النشر

class DeployWorkflow:
    def __init__(self, version):
        self.version = version
        self.task_id = None
        self.steps = [
            ("Bump version", self.bump_version),
            ("Update changelog", self.update_changelog),
            ("Commit and push", self.commit_push),
            ("Wait for CI", self.wait_for_ci),
            ("Verify deployment", self.verify_deployment),
        ]
    
    def run(self):
        # تحقق إذا كان قيد التقدم بالفعل
        existing = self.find_existing()
        if existing:
            self.task_id = existing['id']
            start_step = self.parse_progress(existing['description'])
        else:
            self.task_id = create_task(
                title=f"Deploy Synapse v{self.version}",
                description=self.build_description(0),
                priority="high"
            )
            start_step = 0
        
        # نفّذ الخطوات المتبقية
        for i in range(start_step, len(self.steps)):
            step_name, step_fn = self.steps[i]
            self.update_progress(i, f"Running: {step_name}")
            try:
                step_fn()
            except Exception as e:
                self.update_progress(i, f"FAILED at {step_name}: {e}")
                raise
        
        self.complete()
    
    def update_progress(self, step_idx, status):
        update_task(self.task_id,
            status="in_progress",
            description=f"Step {step_idx+1}/{len(self.steps)}: {status}"
        )
    
    def complete(self):
        complete_task(self.task_id, f"Deployed v{self.version} successfully")

التسلسل الهرمي للمهام

للعمل المعقد، استخدم علاقات المهام الأب-الابن:

# مهمة أب
parent_id = create_task("v1.6.0 Release", priority="high")

# مهام فرعية (مربوطة عبر الوسوم)
create_task("Bump version", 
    description=f"Parent: {parent_id}",
    tags=["v1.6.0", f"parent-{parent_id}"],
    priority="high")

create_task("Update docs",
    description=f"Parent: {parent_id}",
    tags=["v1.6.0", f"parent-{parent_id}"],
    priority="normal")

ابحث عن المهام الفرعية:

curl -H "Authorization: Bearer $KEY" \
     ".../memory/search?q=parent-{parent_id}&tag=v1.6.0"

سير العمل للحالة

pending → in_progress → done
                ↘ cancelled

Pending

مهمة أُنشئت لكن لم تبدأ. استخدم للعمل المخطط.

In Progress

يجري العمل عليها حاليًا. حدّث الوصف بالتقدم.

Done

أُكملت بنجاح. يجب أن يتضمن الوصف الملخص.

Cancelled

متروكة. يجب أن يتضمن الوصف السبب.

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

أنماط شائعة

النمط: سير عمل إصلاح خطأ

def fix_bug(bug_id, description):
    task_id = create_task(
        title=f"Fix bug {bug_id}",
        description=description,
        priority="high"
    )
    
    # تحقق
    update_progress(task_id, "Investigating root cause")
    root_cause = investigate()
    
    # أصلح
    update_progress(task_id, f"Applying fix: {root_cause}")
    apply_fix(root_cause)
    
    # اختبر
    update_progress(task_id, "Testing fix")
    run_tests()
    
    # انشر
    update_progress(task_id, "Deploying fix")
    deploy()
    
    complete_task(task_id, f"Fixed: {root_cause}")

النمط: سير عمل البحث

def research_topic(topic):
    task_id = create_task(
        title=f"Research: {topic}",
        priority="normal"
    )
    
    update_progress(task_id, "Gathering sources")
    sources = gather_sources(topic)
    
    update_progress(task_id, "Analyzing")
    analysis = analyze(sources)
    
    update_progress(task_id, "Storing findings")
    remember("fact", f"research_{topic}", analysis,
             tags=["research", topic], priority="normal")
    
    complete_task(task_id, f"Research complete: {len(sources)} sources")

الخطوات التالية

• نمط بدء الجلسة
• نمط استطلاع الدردشة
• التعافي من الأخطاء