Quy trình điều khiển bởi tác vụ
Sử dụng tác vụ Synapse để điều khiển quy trình LLM nhiều bước tồn tại qua các phiên.
Quy trình điều khiển bởi tác vụ
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 cá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 và cung cấp dấu vết kiểm toán cho những gì đã làm.
Tại sao điều khiển bởi tác vụ?
Không có tác vụ:
- LLM bắt đầu mỗi phiên không chắc chắn phải làm gì
- Công việc nhiều bước bị quên giữa thực thi
- Không có bản ghi cái gì đã làm
Với tác vụ:
- LLM tiếp tục tác vụ đang tiến hành ngay lập tức
- Công việc nhiều bước tồn tại qua các phiên
- Dấu vết kiểm toán tích hợp cho tất cả công việc
Mẫu
1. At session start: check in_progress tasks
2. If tasks exist: resume them
3. If no tasks: create new tasks for current work
4. Update task status as you progress
5. Mark done when completeTriển khai
Bước 1: Tạo tác vụ cho công việc nhiều bước
def start_workflow(title, steps):
"""Create a task for multi-step work."""
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
# Example
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"
])Bước 2: Theo dõi tiến độ trong mô tả tác vụ
def update_progress(task_id, current_step, total_steps, status_note):
"""Update task with current progress."""
description = f"Progress: {current_step}/{total_steps}\nStatus: {status_note}"
update_task(task_id, status="in_progress", description=description)
# Example
update_progress(task_id, 2, 5, "CHANGELOG updated, committing now")Bước 3: Tiếp tục qua các phiên
def resume_work():
"""At session start, find and resume in-progress tasks."""
tasks = list_tasks(status="in_progress")
for task in tasks:
print(f"Resuming: {task['title']}")
print(f"Last status: {task['description']}")
# Parse progress from description
progress = parse_progress(task['description'])
next_step = progress['current_step'] + 1
# Continue from next step
continue_from_step(task['id'], next_step)Bước 4: Hoàn thành và lưu trữ
def complete_task(task_id, summary):
"""Mark task done with completion summary."""
update_task(task_id,
status="done",
description=f"COMPLETED. Summary: {summary}"
)
# Also store as memory for long-term reference
remember(
category="project",
key=f"completed_{task_id}",
content=f"Task: {task_id}\nSummary: {summary}",
tags=["completed", "task"],
priority="normal"
)Ví dụ đầy đủ: Quy trình triển khai
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):
# Check if already in progress
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
# Execute remaining steps
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")Hệ thống tác vụ
Cho công việc phức tạp, sử dụng quan hệ tác vụ cha-con:
# Parent task
parent_id = create_task("v1.6.0 Release", priority="high")
# Sub-tasks (linked via tags)
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")Tìm tác vụ con:
curl -H "Authorization: Bearer $KEY" \
".../memory/search?q=parent-{parent_id}&tag=v1.6.0"Quy trình trạng thái
pending → in_progress → done
↘ cancelledPending
Tác vụ đã tạo nhưng chưa bắt đầu. Sử dụng cho công việc đã lên kế hoạch.
In Progress
Đang được thực hiện. Cập nhật mô tả với tiến độ.
Done
Hoàn thành thành công. Mô tả nên bao gồm tóm tắt.
Cancelled
Đã bỏ. Mô tả nên bao gồm lý do.
Thực hành tốt nhất
Mẫu phổ biến
Mẫu: Quy trình sửa bug
def fix_bug(bug_id, description):
task_id = create_task(
title=f"Fix bug {bug_id}",
description=description,
priority="high"
)
# Investigate
update_progress(task_id, "Investigating root cause")
root_cause = investigate()
# Fix
update_progress(task_id, f"Applying fix: {root_cause}")
apply_fix(root_cause)
# Test
update_progress(task_id, "Testing fix")
run_tests()
# Deploy
update_progress(task_id, "Deploying fix")
deploy()
complete_task(task_id, f"Fixed: {root_cause}")Mẫu: Quy trình nghiên cứu
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")