작업은 단순한 할 일이 아닙니다 — 영구 LLM 워크플로우의 근간입니다. 다단계\n작업에 대해 작업을 생성함으로써 세션 간 연속성을 보장하고 수행된 작업에 대한\n감사 추적을 제공합니다.
\n작업이 없으면:
\n작업이 있으면:
\n1. At session start: check in_progress tasks\n2. If tasks exist: resume them\n3. If no tasks: create new tasks for current work\n4. Update task status as you progress\n5. Mark done when completedef start_workflow(title, steps):\n """Create a task for multi-step work."""\n task_id = create_task(\n title=title,\n description=f"Steps:\\n" + "\\n".join(f" {i+1}. {s}" for i, s in enumerate(steps)),\n priority="high"\n )\n return task_id\n\n# Example\ntask_id = start_workflow("Deploy Synapse v1.6.0", [\n "Bump version in package.json",\n "Update CHANGELOG.md",\n "Commit and push",\n "Wait for CI green",\n "Verify deployment"\n])def update_progress(task_id, current_step, total_steps, status_note):\n """Update task with current progress."""\n description = f"Progress: {current_step}/{total_steps}\\nStatus: {status_note}"\n update_task(task_id, status="in_progress", description=description)\n\n# Example\nupdate_progress(task_id, 2, 5, "CHANGELOG updated, committing now")def resume_work():\n """At session start, find and resume in-progress tasks."""\n tasks = list_tasks(status="in_progress")\n \n for task in tasks:\n print(f"Resuming: {task['title']}")\n print(f"Last status: {task['description']}")\n \n # Parse progress from description\n progress = parse_progress(task['description'])\n next_step = progress['current_step'] + 1\n \n # Continue from next step\n continue_from_step(task['id'], next_step)def complete_task(task_id, summary):\n """Mark task done with completion summary."""\n update_task(task_id, \n status="done",\n description=f"COMPLETED. Summary: {summary}"\n )\n # Also store as memory for long-term reference\n remember(\n category="project",\n key=f"completed_{task_id}",\n content=f"Task: {task_id}\\nSummary: {summary}",\n tags=["completed", "task"],\n priority="normal"\n )class DeployWorkflow:\n def __init__(self, version):\n self.version = version\n self.task_id = None\n self.steps = [\n ("Bump version", self.bump_version),\n ("Update changelog", self.update_changelog),\n ("Commit and push", self.commit_push),\n ("Wait for CI", self.wait_for_ci),\n ("Verify deployment", self.verify_deployment),\n ]\n \n def run(self):\n # Check if already in progress\n existing = self.find_existing()\n if existing:\n self.task_id = existing['id']\n start_step = self.parse_progress(existing['description'])\n else:\n self.task_id = create_task(\n title=f"Deploy Synapse v{self.version}",\n description=self.build_description(0),\n priority="high"\n )\n start_step = 0\n \n # Execute remaining steps\n for i in range(start_step, len(self.steps)):\n step_name, step_fn = self.steps[i]\n self.update_progress(i, f"Running: {step_name}")\n try:\n step_fn()\n except Exception as e:\n self.update_progress(i, f"FAILED at {step_name}: {e}")\n raise\n \n self.complete()\n \n def update_progress(self, step_idx, status):\n update_task(self.task_id,\n status="in_progress",\n description=f"Step {step_idx+1}/{len(self.steps)}: {status}"\n )\n \n def complete(self):\n complete_task(self.task_id, f"Deployed v{self.version} successfully")복잡한 작업의 경우, 부모-자식 작업 관계 사용:
\n# Parent task\nparent_id = create_task("v1.6.0 Release", priority="high")\n\n# Sub-tasks (linked via tags)\ncreate_task("Bump version", \n description=f"Parent: {parent_id}",\n tags=["v1.6.0", f"parent-{parent_id}"],\n priority="high")\n\ncreate_task("Update docs",\n description=f"Parent: {parent_id}",\n tags=["v1.6.0", f"parent-{parent_id}"],\n priority="normal")하위 작업 검색:
\ncurl -H "Authorization: Bearer $KEY" \\\n ".../memory/search?q=parent-{parent_id}&tag=v1.6.0"pending → in_progress → done\n ↘ cancelled생성되었지만 시작되지 않은 작업. 계획된 작업에 사용.
\n현재 작업 중. 설명에 진행 상황 업데이트.
\n성공적으로 완료. 설명에 요약 포함.
\n중단. 설명에 이유 포함.
\ndef fix_bug(bug_id, description):\n task_id = create_task(\n title=f"Fix bug {bug_id}",\n description=description,\n priority="high"\n )\n \n # Investigate\n update_progress(task_id, "Investigating root cause")\n root_cause = investigate()\n \n # Fix\n update_progress(task_id, f"Applying fix: {root_cause}")\n apply_fix(root_cause)\n \n # Test\n update_progress(task_id, "Testing fix")\n run_tests()\n \n # Deploy\n update_progress(task_id, "Deploying fix")\n deploy()\n \n complete_task(task_id, f"Fixed: {root_cause}")def research_topic(topic):\n task_id = create_task(\n title=f"Research: {topic}",\n priority="normal"\n )\n \n update_progress(task_id, "Gathering sources")\n sources = gather_sources(topic)\n \n update_progress(task_id, "Analyzing")\n analysis = analyze(sources)\n \n update_progress(task_id, "Storing findings")\n remember("fact", f"research_{topic}", analysis,\n tags=["research", topic], priority="normal")\n \n complete_task(task_id, f"Research complete: {len(sources)} sources")