Görev Odaklı İş Akışı

Görevler yalnızca yapılacaklar değil — kalıcı LLM iş akışlarının omurgasıdır. Çok adımlı iş için görevler oluşturarak, oturumlar arasında süreklilik sağlar ve nelerin yapıldığına dair denetim izleri sağlarsınız.

Neden Görev Odaklı?

Görevler olmadan:

• LLM her oturuma ne yapacağından emin olmadan başlar
• Çok adımlı iş yürütme sırasında unutulur
• Nelerin yapıldığına dair kayıt yoktur

Görevlerle:

• LLM devam eden görevleri hemen sürdürür
• Çok adımlı iş oturumlar arasında hayatta kalır
• Tüm çalışmanın yerleşik denetim izi

Desen

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

Uygulama

Adım 1: Çok adımlı iş için bir görev oluştur

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

# Example
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"
])

Adım 2: İlerlemeyi görev açıklamasında izle

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)

# Example
update_progress(task_id, 2, 5, "CHANGELOG updated, committing now")

Adım 3: Oturumlar arasında sürdür

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']}")
        
        # Parse progress from description
        progress = parse_progress(task['description'])
        next_step = progress['current_step'] + 1
        
        # Continue from next step
        continue_from_step(task['id'], next_step)

Adım 4: Tamamla ve arşivle

def complete_task(task_id, summary):
    """Mark task done with completion summary."""
    update_task(task_id, 
        status="done",
        description=f"COMPLETED. Summary: {summary}"
    )
    # Also store as memory for long-term reference
    remember(
        category="project",
        key=f"completed_{task_id}",
        content=f"Task: {task_id}\nSummary: {summary}",
        tags=["completed", "task"],
        priority="normal"
    )

Tam Örnek: Dağıtım İş Akışı

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):
        # Check if already in progress
        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
        
        # Execute remaining steps
        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")

Görev Hiyerarşisi

Karmaşık iş için ebeveyn-çocuk görev ilişkileri kullanın:

# Parent task
parent_id = create_task("v1.6.0 Release", priority="high")

# Sub-tasks (linked via tags)
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")

Alt görevleri ara:

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

Durum İş Akışı

pending → in_progress → done
                ↘ cancelled

Pending

Görev oluşturuldu ama başlanmadı. Planlı iş için kullanın.

In Progress

Şu anda üzerinde çalışılıyor. İlerlemeyi açıklamayla güncelleyin.

Done

Başarıyla tamamlandı. Açıklama özet içermelidir.

Cancelled

Terk edildi. Açıklama nedeni içermelidir.

En İyi Uygulamalar

Yaygın Desenler

Desen: Hata Düzeltme İş Akışı

def fix_bug(bug_id, description):
    task_id = create_task(
        title=f"Fix bug {bug_id}",
        description=description,
        priority="high"
    )
    
    # Investigate
    update_progress(task_id, "Investigating root cause")
    root_cause = investigate()
    
    # Fix
    update_progress(task_id, f"Applying fix: {root_cause}")
    apply_fix(root_cause)
    
    # Test
    update_progress(task_id, "Testing fix")
    run_tests()
    
    # Deploy
    update_progress(task_id, "Deploying fix")
    deploy()
    
    complete_task(task_id, f"Fixed: {root_cause}")

Desen: Araştırma İş Akışı

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")

Sonraki Adımlar

• Oturum Başlangıç Deseni
• Sohbet Polling Deseni
• Hata Kurtarma