ai-workflow-orchestrator

AI Workflow Orchestrator

Critères de décision : quel framework choisir ?

Complexité	Pattern	Framework recommandé
1–3 étapes linéaires	Séquentiel simple	Python natif ou LCEL
Fan-out / fan-in sans état	Parallèle sans mémoire	LCEL RunnableParallel
Boucles, conditions, état persistant	Agent / cycle	LangGraph
RAG + retrieval hybride	Search pipeline	Haystack
Stack Azure / .NET	Intégration Microsoft	Semantic Kernel
Contrôle total, zéro dépendance	Custom	Script async pur

Règle d'or : n'ajouter un framework que quand la complexité le justifie. 50 lignes de Python natif > architecture LangGraph mal comprise.

---

Workflow en étapes

1. Décomposer en DAG avant de coder

Identifier pour chaque étape : input attendu, output produit, dépendances. Dessiner un DAG — économise des heures de débogage.

Pipeline d'analyse de document :
[Chargement] → [Extraction] ─┬─ [Résumé]      ─┐
                               ├─ [Entités NER]  ├─ [Rapport final]
                               └─ [Sentiment]   ─┘

Questions à poser : quelles étapes sont indépendantes (candidats au parallélisme) ? Quelles étapes nécessitent un état partagé ? Y a-t-il des points de décision conditionnels ?

---

2. Implémenter les patterns de chaînes

Séquentiel (LCEL | operator)

from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
pipeline = extract_prompt | llm | parse_output | report_prompt | llm
result = pipeline.invoke({"document": text})

Parallèle (fan-out / fan-in)

from langchain_core.runnables import RunnableParallel

parallel = RunnableParallel({
    "summary":   summarize_prompt | llm,
    "entities":  extract_prompt   | llm,
    "sentiment": sentiment_prompt | llm,
})
full = parallel | merge_results | report_prompt | llm
result = full.invoke({"document": text})

Map-Reduce (N documents)

from langchain_core.runnables import RunnableLambda
import asyncio

async def map_reduce(docs: list[str]) -> str:
    summaries = await asyncio.gather(*[
        summarize_chain.ainvoke({"text": d}) for d in docs
    ])
    return await aggregate_chain.ainvoke({"summaries": summaries})