Task-getriebener Workflow

Tasks sind nicht nur Todos — sie sind das Rückgrat persistenter LLM-Workflows. Indem du für mehrschrittige Arbeit Tasks anlegst, sicherst du Kontinuität über Sessions und stellst Audit-Trails für getane Arbeit bereit.

Warum Task-getrieben?

Ohne Tasks:

• LLM startet jede Session unsicher, was zu tun ist
• Mehrschrittige Arbeit wird mitten in der Ausführung vergessen
• Kein Record darüber, was getan wurde

Mit Tasks:

• LLM nimmt in-progress Tasks sofort wieder auf
• Mehrschrittige Arbeit überdauert Sessions
• Eingebauter Audit-Trail aller Arbeit

Das Pattern

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

Implementierung

Schritt 1: Einen Task für mehrschrittige Arbeit anlegen

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

Schritt 2: Fortschritt in Task-Beschreibung tracken

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

Schritt 3: Über Sessions hinweg fortsetzen

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)

Schritt 4: Abschließen und archivieren

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

Vollständiges Beispiel: Deploy-Workflow

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

Task-Hierarchie

Für komplexe Arbeit verwende Parent-Child-Task-Beziehungen:

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

Nach Sub-Tasks suchen:

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

Status-Workflow

pending → in_progress → done
                ↘ cancelled

Pending

Task angelegt, aber nicht begonnen. Für geplante Arbeit verwenden.

In Progress

Wird gerade bearbeitet. Beschreibung mit Fortschritt aktualisieren.

Done

Erfolgreich abgeschlossen. Beschreibung sollte Zusammenfassung enthalten.

Cancelled

Abgebrochen. Beschreibung sollte den Grund enthalten.

Best Practices

Häufige Patterns

Pattern: Bug-Fix-Workflow

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

Pattern: Recherche-Workflow

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

Nächste Schritte

• Session-Start-Pattern
• Chat-Polling-Pattern
• Error-Recovery