작업 기반 워크플로우

작업은 단순한 할 일이 아닙니다 — 영구 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

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

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)

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

4단계: 완료 및 보관

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

전체 예시: 배포 워크플로우

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

작업 계층

복잡한 작업의 경우, 부모-자식 작업 관계 사용:

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

하위 작업 검색:

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

패턴: 연구 워크플로우

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

다음 단계

• 세션 시작 패턴
• 채팅 폴링 패턴
• 오류 복구