Workflow basado en tareas
Use las tareas de Synapse para impulsar flujos de trabajo LLM multi-paso que sobreviven entre sesiones.
Workflow basado en tareas
Las tareas no son solo tareas pendientes — son la columna vertebral de los flujos de trabajo LLM persistentes. Al crear tareas para el trabajo multi-paso, garantiza la continuidad entre sesiones y proporciona pistas de auditoría de lo que se ha hecho.
¿Por qué basado en tareas?
Sin tareas:
- El LLM empieza cada sesión sin saber qué hacer
- El trabajo multi-paso se olvida a mitad de ejecución
- No hay registro de lo que se ha hecho
Con tareas:
- El LLM retoma tareas en progreso de inmediato
- El trabajo multi-paso sobrevive entre sesiones
- Pista de auditoría integrada de todo el trabajo
El patrón
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 completeImplementación
Paso 1: Crear una tarea para trabajo multi-paso
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"
])Paso 2: Hacer seguimiento del progreso en la descripción de la tarea
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")Paso 3: Retomar entre sesiones
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)Paso 4: Completar y archivar
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"
)Ejemplo completo: workflow de despliegue
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")Jerarquía de tareas
Para trabajo complejo, use relaciones de tarea padre-hijo:
# 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")Buscar subtareas:
curl -H "Authorization: Bearer $KEY" \
".../memory/search?q=parent-{parent_id}&tag=v1.6.0"Flujo de estados
pending → in_progress → done
↘ cancelledPending
Tarea creada pero no iniciada. Úselo para trabajo planificado.
In Progress
En ejecución actualmente. Actualice la descripción con el progreso.
Done
Completada con éxito. La descripción debería incluir un resumen.
Cancelled
Abandonada. La descripción debería incluir el motivo.
Mejores prácticas
Patrones comunes
Patrón: workflow de corrección de 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}")Patrón: workflow de investigación
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")