任务不仅仅是待办 — 它们是持久化 LLM 工作流的骨架。为多步工作创建任务，可确保跨会话连续性，并提供完成情况的审计轨迹。
\n无任务：
\n有任务：
\n1. 会话开始：检查 in_progress 任务\n2. 若有任务：恢复它们\n3. 若无任务：为当前工作创建新任务\n4. 随进度更新任务状态\n5. 完成时标记 donedef start_workflow(title, steps):\n """为多步工作创建任务。"""\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# 示例\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 """更新任务的当前进度。"""\n description = f"Progress: {current_step}/{total_steps}\\nStatus: {status_note}"\n update_task(task_id, status="in_progress", description=description)\n\n# 示例\nupdate_progress(task_id, 2, 5, "CHANGELOG updated, committing now")def resume_work():\n """会话开始时，找出并恢复进行中的任务。"""\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 # 从描述中解析进度\n progress = parse_progress(task['description'])\n next_step = progress['current_step'] + 1\n \n # 从下一步继续\n continue_from_step(task['id'], next_step)def complete_task(task_id, summary):\n """标记任务为完成，附上完成总结。"""\n update_task(task_id, \n status="done",\n description=f"COMPLETED. Summary: {summary}"\n )\n # 同时存为记忆供长期参考\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 # 检查是否已在进行中\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 # 执行剩余步骤\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# 父任务\nparent_id = create_task("v1.6.0 Release", priority="high")\n\n# 子任务（通过标签关联）\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 # 调查\n update_progress(task_id, "Investigating root cause")\n root_cause = investigate()\n \n # 修复\n update_progress(task_id, f"Applying fix: {root_cause}")\n apply_fix(root_cause)\n \n # 测试\n update_progress(task_id, "Testing fix")\n run_tests()\n \n # 部署\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")