Uppgiftsdrivet arbetsflöde

Uppgifter är inte bara att-göra-listor — de är ryggraden i beständiga LLM-arbetsflöden. Genom att skapa uppgifter för arbete i flera steg säkerställer ni kontinuitet över sessioner och tillhandahåller granskningsspår för vad som har gjorts.

Varför uppgiftsdrivet?

Utan uppgifter:

• LLM startar varje session osäker på vad som ska göras
• Arbete i flera steg glöms mitt i utförande
• Ingen registrering av vad som har gjorts

Med uppgifter:

• LLM återupptar pågående uppgifter omedelbart
• Arbete i flera steg överlever över sessioner
• Inbyggt granskningsspår av allt arbete

Mönstret

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

Implementering

Steg 1: Skapa en uppgift för arbete i flera steg

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

Steg 2: Spåra förlopp i uppgiftsbeskrivning

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

Steg 3: Återuppta över sessioner

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)

Steg 4: Slutför och arkivera

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

Fullständigt exempel: Driftsättningsarbetsflöde

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

Uppgiftshierarki

För komplext arbete, använd förälder-barn-uppgiftsrelationer:

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

Sök efter deluppgifter:

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

Statusarbetsflöde

pending → in_progress → done
                ↘ cancelled

Pending

Uppgift skapad men inte startad. Används för planerat arbete.

In Progress

Pågår. Uppdatera beskrivning med förlopp.

Done

Slutförd. Beskrivning bör inkludera sammanfattning.

Cancelled

Övergiven. Beskrivning bör inkludera anledning.

Bästa praxis

Vanliga mönster

Mönster: Buggfix-arbetsflöde

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

Mönster: Forskningsarbetsflöde

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ästa steg

• Sessionsstart-mönster
• Chattpollningsmönster
• Felåterställning