{"title":"Workflow piloté par les tâches","slug":"task-driven-workflow","category":"llm-cookbook","summary":"Utilisez les tâches Synapse pour piloter des workflows LLM multi-étapes qui survivent entre les sessions.","audience":["llm"],"tags":["cookbook","tasks","workflow","persistence"],"difficulty":"intermediate","updated":"2026-06-27","word_count":354,"read_minutes":2,"lang":"fr","translated":true,"requested_lang":"fr","content_markdown":"\n# Workflow piloté par les tâches\n\nLes tâches ne sont pas juste des todos — elles sont l'épine dorsale des workflows LLM\npersistants. En créant des tâches pour le travail multi-étapes, vous garantissez la\ncontinuité entre les sessions et fournissez des pistes d'audit de ce qui a été fait.\n\n## Pourquoi piloté par les tâches ?\n\nSans tâches :\n- Le LLM commence chaque session incertain de ce qu'il faut faire\n- Le travail multi-étapes est oublié en cours d'exécution\n- Aucun enregistrement de ce qui a été fait\n\nAvec des tâches :\n- Le LLM reprend les tâches en cours immédiatement\n- Le travail multi-étapes survit entre les sessions\n- Piste d'audit intégrée de tout le travail\n\n## Le schéma\n\n```\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 complete\n```\n\n## Implémentation\n\n### Étape 1 : créer une tâche pour un travail multi-étapes\n\n```python\ndef 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# Exemple\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])\n```\n\n### Étape 2 : suivre la progression dans la description de tâche\n\n```python\ndef 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# Exemple\nupdate_progress(task_id, 2, 5, \"CHANGELOG updated, committing now\")\n```\n\n### Étape 3 : reprendre entre les sessions\n\n```python\ndef 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 # Analyser la progression depuis la description\n progress = parse_progress(task['description'])\n next_step = progress['current_step'] + 1\n \n # Continuer depuis l'étape suivante\n continue_from_step(task['id'], next_step)\n```\n\n### Étape 4 : compléter et archiver\n\n```python\ndef 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 # Aussi stocker en mémoire pour référence longue durée\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 )\n```\n\n## Exemple complet : workflow de déploiement\n\n```python\nclass 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 # Vérifier si déjà en cours\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 # Exécuter les étapes restantes\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```\n\n## Hiérarchie de tâches\n\nPour le travail complexe, utilisez des relations parent-enfant :\n\n```python\n# Tâche parent\nparent_id = create_task(\"v1.6.0 Release\", priority=\"high\")\n\n# Sous-tâches (liées 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\")\n```\n\nRecherchez les sous-tâches :\n\n```bash\ncurl -H \"Authorization: Bearer $KEY\" \\\n \".../memory/search?q=parent-{parent_id}&tag=v1.6.0\"\n```\n\n## Workflow de statut\n\n```\npending → in_progress → done\n ↘ cancelled\n```\n\n### Pending\n\nTâche créée mais non commencée. À utiliser pour le travail planifié.\n\n### In Progress\n\nEn cours de traitement. **Mettez à jour la description avec la progression.**\n\n### Done\n\nTerminée avec succès. La description devrait inclure un résumé.\n\n### Cancelled\n\nAbandonnée. La description devrait inclure la raison.\n\n## Bonnes pratiques\n\n> [!TIP]\n> - **Créez des tâches pour le travail multi-étapes** — le travail mono-étape n'a pas besoin de tâche\n> - **Mettez à jour la description avec la progression** — permet la reprise\n> - **Utilisez une priorité élevée pour le travail actif** — apparaît dans le rappel\n> - **Complétez les tâches quand elles sont terminées** — ne les laissez pas in_progress\n> - **Stockez les résumés de complétion en mémoire** — référence longue durée\n\n## Schémas courants\n\n### Schéma : workflow de correction de bug\n\n```python\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 # Investiguer\n update_progress(task_id, \"Investigating root cause\")\n root_cause = investigate()\n \n # Corriger\n update_progress(task_id, f\"Applying fix: {root_cause}\")\n apply_fix(root_cause)\n \n # Tester\n update_progress(task_id, \"Testing fix\")\n run_tests()\n \n # Déployer\n update_progress(task_id, \"Deploying fix\")\n deploy()\n \n complete_task(task_id, f\"Fixed: {root_cause}\")\n```\n\n### Schéma : workflow de recherche\n\n```python\ndef 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\")\n```\n\n## Prochaines étapes\n\n- [Schéma de début de session](/docs/llm-cookbook/session-start-pattern)\n- [Schéma de polling de chat](/docs/llm-cookbook/chat-polling-pattern)\n- [Récupération d'erreurs](/docs/llm-cookbook/error-recovery)\n","content_html":"

Workflow piloté par les tâches

\n

Les tâches ne sont pas juste des todos — elles sont l'épine dorsale des workflows LLM\npersistants. En créant des tâches pour le travail multi-étapes, vous garantissez la\ncontinuité entre les sessions et fournissez des pistes d'audit de ce qui a été fait.

\n

Pourquoi piloté par les tâches ?

\n

Sans tâches :

\n\n

Avec des tâches :

\n\n

Le schéma

\n
1. 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 complete

Implémentation

\n

Étape 1 : créer une tâche pour un travail multi-étapes

\n
def 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# Exemple\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])

Étape 2 : suivre la progression dans la description de tâche

\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# Exemple\nupdate_progress(task_id, 2, 5, "CHANGELOG updated, committing now")

Étape 3 : reprendre entre les sessions

\n
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        # Analyser la progression depuis la description\n        progress = parse_progress(task['description'])\n        next_step = progress['current_step'] + 1\n        \n        # Continuer depuis l'étape suivante\n        continue_from_step(task['id'], next_step)

Étape 4 : compléter et archiver

\n
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    # Aussi stocker en mémoire pour référence longue durée\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    )

Exemple complet : workflow de déploiement

\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        # Vérifier si déjà en cours\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        # Exécuter les étapes restantes\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")

Hiérarchie de tâches

\n

Pour le travail complexe, utilisez des relations parent-enfant :

\n
# Tâche parent\nparent_id = create_task("v1.6.0 Release", priority="high")\n\n# Sous-tâches (liées 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")

Recherchez les sous-tâches :

\n
curl -H "Authorization: Bearer $KEY" \\\n     ".../memory/search?q=parent-{parent_id}&tag=v1.6.0"

Workflow de statut

\n
pending → in_progress → done\n                ↘ cancelled

Pending

\n

Tâche créée mais non commencée. À utiliser pour le travail planifié.

\n

In Progress

\n

En cours de traitement. Mettez à jour la description avec la progression.

\n

Done

\n

Terminée avec succès. La description devrait inclure un résumé.

\n

Cancelled

\n

Abandonnée. La description devrait inclure la raison.

\n

Bonnes pratiques

\n

Schémas courants

\n

Schéma : workflow de correction de bug

\n
def 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    # Investiguer\n    update_progress(task_id, "Investigating root cause")\n    root_cause = investigate()\n    \n    # Corriger\n    update_progress(task_id, f"Applying fix: {root_cause}")\n    apply_fix(root_cause)\n    \n    # Tester\n    update_progress(task_id, "Testing fix")\n    run_tests()\n    \n    # Déployer\n    update_progress(task_id, "Deploying fix")\n    deploy()\n    \n    complete_task(task_id, f"Fixed: {root_cause}")

Schéma : workflow de recherche

\n
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")

Prochaines étapes

\n\n","urls":{"html":"/docs/llm-cookbook/task-driven-workflow","text":"/docs/llm-cookbook/task-driven-workflow?format=text","json":"/docs/llm-cookbook/task-driven-workflow?format=json","llm":"/docs/llm-cookbook/task-driven-workflow?format=llm"},"translations_available":["en","zh","hi","es","fr","ar","pt","ru","ja","de","it","ko","nl","pl","tr","sv","vi","th","id","uk"]}