任务驱动工作流

intermediate 🤖 LLM 友好 1 分钟 · 116 字更新于： 2026-06-27 中文

使用 Synapse 任务驱动可跨会话存活的多步 LLM 工作流。

阅读本文： 📄 纯文本 ⚡ JSON 🤖 LLM 格式

任务驱动工作流

任务不仅仅是待办 — 它们是持久化 LLM 工作流的骨架。为多步工作创建任务，可确保跨会话连续性，并提供完成情况的审计轨迹。

为什么用任务驱动？

无任务：

LLM 每次会话开始时不知道要做什么
多步工作执行到一半被遗忘
没有已完成工作的记录

有任务：

LLM 立即恢复进行中的任务
多步工作跨会话存活
内置完成工作的审计轨迹

模式

1. 会话开始：检查 in_progress 任务
2. 若有任务：恢复它们
3. 若无任务：为当前工作创建新任务
4. 随进度更新任务状态
5. 完成时标记 done

实现

第 1 步：为多步工作创建任务

def start_workflow(title, steps):
    """为多步工作创建任务。"""
    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

# 示例
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):
    """更新任务的当前进度。"""
    description = f"Progress: {current_step}/{total_steps}\nStatus: {status_note}"
    update_task(task_id, status="in_progress", description=description)

# 示例
update_progress(task_id, 2, 5, "CHANGELOG updated, committing now")

第 3 步：跨会话恢复

def resume_work():
    """会话开始时，找出并恢复进行中的任务。"""
    tasks = list_tasks(status="in_progress")
    
    for task in tasks:
        print(f"Resuming: {task['title']}")
        print(f"Last status: {task['description']}")
        
        # 从描述中解析进度
        progress = parse_progress(task['description'])
        next_step = progress['current_step'] + 1
        
        # 从下一步继续
        continue_from_step(task['id'], next_step)

第 4 步：完成并归档

def complete_task(task_id, summary):
    """标记任务为完成，附上完成总结。"""
    update_task(task_id, 
        status="done",
        description=f"COMPLETED. Summary: {summary}"
    )
    # 同时存为记忆供长期参考
    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):
        # 检查是否已在进行中
        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
        
        # 执行剩余步骤
        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_id = create_task("v1.6.0 Release", priority="high")

# 子任务（通过标签关联）
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

已放弃。描述应包含原因。

最佳实践

常见模式

模式：Bug 修复工作流

def fix_bug(bug_id, description):
    task_id = create_task(
        title=f"Fix bug {bug_id}",
        description=description,
        priority="high"
    )
    
    # 调查
    update_progress(task_id, "Investigating root cause")
    root_cause = investigate()
    
    # 修复
    update_progress(task_id, f"Applying fix: {root_cause}")
    apply_fix(root_cause)
    
    # 测试
    update_progress(task_id, "Testing fix")
    run_tests()
    
    # 部署
    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")