Tác vụ không chỉ là todo — chúng là xương sống của quy trình LLM cố định. Bằng\ncách tạo tác vụ cho công việc nhiều bước, bạn đảm bảo tính liên tục qua các phiên\nvà cung cấp dấu vết kiểm toán cho những gì đã làm.
\nKhông có tác vụ:
\nVới tác vụ:
\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")Cho công việc phức tạp, sử dụng quan hệ tác vụ cha-con:
\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")Tìm tác vụ con:
\ncurl -H "Authorization: Bearer $KEY" \\\n ".../memory/search?q=parent-{parent_id}&tag=v1.6.0"pending → in_progress → done\n ↘ cancelledTác vụ đã tạo nhưng chưa bắt đầu. Sử dụng cho công việc đã lên kế hoạch.
\nĐang được thực hiện. Cập nhật mô tả với tiến độ.
\nHoàn thành thành công. Mô tả nên bao gồm tóm tắt.
\nĐã bỏ. Mô tả nên bao gồm lý do.
\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")