task ไม่ใช่แค่ todo — แต่เป็นกระดูกสันหลังของ workflow LLM ที่ถาวร โดยการสร้าง task สำหรับงานหลายขั้นตอน คุณรับประกัน continuity ข้าม session และให้ audit trail ของสิ่งที่ทำ
\nไม่มี task:
\nมี task:
\n1. ที่จุดเริ่มต้น session: ตรวจ task in_progress\n2. หากมี task: ทำต่อ\n3. หากไม่มี task: สร้าง task ใหม่สำหรับงานปัจจุบัน\n4. อัปเดตสถานะ task ขณะคืบหน้า\n5. ทำเครื่องหมายเสร็จเมื่อเสร็จสิ้นdef 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")สำหรับงานซับซ้อน ใช้ความสัมพันธ์ parent-child task:
\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")ค้นหา sub-task:
\ncurl -H "Authorization: Bearer $KEY" \\\n ".../memory/search?q=parent-{parent_id}&tag=v1.6.0"pending → in_progress → done\n ↘ cancelledtask สร้างแล้วแต่ยังไม่เริ่ม ใช้สำหรับงานที่วางแผนไว้
\nกำลังดำเนินการ อัปเดต description ด้วยความคืบหน้า
\nเสร็จสิ้นสำเร็จ description ควรรวมสรุป
\nถูกยกเลิก description ควรรวมเหตุผล
\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")