L'Atelier — Réseau francophone IA (produit, tech, business)

Pourquoi construire un orchestrateur custom ?

Les frameworks (LangGraph, CrewAI, AutoGen) couvrent 80% des cas. Mais quand vous avez besoin de contrôle fin sur le routing, la gestion d'erreurs, ou l'intégration avec des systèmes existants, un orchestrateur custom devient nécessaire.

Signaux qu'un framework ne suffit plus

Besoin de routing dynamique basé sur des règles métier complexes
Intégration profonde avec une infrastructure existante
Contraintes de latence ou de coût non supportées par le framework
Logique de fallback et retry spécifique au domaine
Audit trail et compliance nécessitant un contrôle total

Architecture de base

from abc import ABC, abstractmethod
from typing import Any
from dataclasses import dataclass, field

@dataclass
class AgentState:
    messages: list = field(default_factory=list)
    context: dict = field(default_factory=dict)
    current_step: str = "start"
    iteration: int = 0
    max_iterations: int = 10

class BaseAgent(ABC):
    @abstractmethod
    async def execute(self, state: AgentState) -> AgentState:
        pass

class Orchestrator:
    def __init__(self):
        self.agents: dict[str, BaseAgent] = {}
        self.router = Router()

    def register(self, name: str, agent: BaseAgent):
        self.agents[name] = agent

    async def run(self, initial_state: AgentState) -> AgentState:
        state = initial_state
        while state.current_step != "end":
            if state.iteration >= state.max_iterations:
                raise MaxIterationsExceeded()

            agent_name = self.router.route(state)
            agent = self.agents[agent_name]
            state = await agent.execute(state)
            state.iteration += 1
        return state

Router intelligent

class Router:
    def __init__(self):
        self.rules: list[RoutingRule] = []

    def route(self, state: AgentState) -> str:
        for rule in self.rules:
            if rule.matches(state):
                return rule.target_agent
        return "default_agent"

    def add_llm_routing(self, llm):
        """Utiliser un LLM pour le routing quand les règles ne matchent pas."""
        self.fallback_llm = llm

Gestion d'erreurs et retry

class ResilientOrchestrator(Orchestrator):
    async def run_with_retry(self, state: AgentState) -> AgentState:
        for attempt in range(3):
            try:
                return await self.run(state)
            except RateLimitError:
                await asyncio.sleep(2 ** attempt)
            except ToolExecutionError as e:
                state.context["last_error"] = str(e)
                state.current_step = "error_recovery"
                return await self.run(state)

Observabilité intégrée

import structlog

class ObservableOrchestrator(Orchestrator):
    def __init__(self):
        super().__init__()
        self.logger = structlog.get_logger()

    async def run(self, state: AgentState) -> AgentState:
        with self.trace_span("orchestrator.run"):
            self.logger.info("step_start",
                step=state.current_step,
                iteration=state.iteration)
            # ...

Quand revenir à un framework

Si votre orchestrateur custom dépasse 500 lignes, réévaluez. Souvent, LangGraph avec des nœuds custom offre le meilleur compromis entre contrôle et maintenabilité.

Orchestrateur custom : quand les frameworks ne suffisent pas