pipeline-composer

Agent Pipeline Composer

Quand utiliser ce skill

Ce skill est adapté lorsqu'une tâche complexe doit être décomposée en étapes séquentielles ou parallèles, chaque étape étant assurée par un sous-agent spécialisé dont le résultat alimente directement l'étape suivante.

Cas d'usage typiques :

• Pipelines ETL : Extract (scraping/API) → Transform (parsing/enrichissement) → Load (DB/fichier)
• Traitement de contenu : Recherche → Analyse → Rédaction → Révision → Publication
• Qualification de leads : Enrichissement → Scoring → Segmentation → Routage CRM
• Traitement de documents : OCR → NLP → Extraction → Validation → Stockage

Ne pas utiliser si : les étapes n'ont pas de dépendances de données entre elles (utiliser un pool d'agents parallèles indépendants à la place).

---

Étapes de conception

1. Choisir la topologie

Topologie	Quand l'utiliser	Structure
linear	Étapes strictement séquentielles, chaque output = input du suivant	A → B → C
DAG	Dépendances multiples, parallélisme possible	A → (B ‖ C) → D
conditional branch	Routing selon le résultat d'une étape	A → if X then B else C
map-reduce	Même traitement sur N items, puis agrégation	A → [B₁‖B₂‖B₃] → C
loop	Itérer jusqu'à un critère de satisfaction (qualité, score)	A → B → if OK then fin else A

2. Définir les schémas d'interface

Chaque stage doit avoir un contrat explicite. C'est la source n°1 de bugs quand il est flou.

from pydantic import BaseModel

class ResearchOutput(BaseModel):
    raw_text: str
    sources: list[str]
    confidence: float  # 0.0 – 1.0

class AnalysisInput(BaseModel):
    raw_text: str       # mappé depuis ResearchOutput.raw_text
    sources: list[str]  # mappé depuis ResearchOutput.sources

class AnalysisOutput(BaseModel):
    summary: str
    key_points: list[str]
    confidence: float

Valider à chaque frontière entre stages (model.model_validate(output_dict)) — erreur de parsing = bug de mapping attrapé immédiatement, pas deux stages plus loin.