Architecture agentique & LangGraph — notre cœur de métier.

Concevoir un harness d'agents in-house.

Les frameworks (LangGraph, CrewAI, AutoGen) couvrent 80 % des besoins. Mais pour les usages critiques — audit AI Act, replay forensique, contrôle de flux financier, contraintes de latence sub-second, environnements air-gap — il faut souvent concevoir un harness in-house qui s'inspire des frameworks mais reste votre code.

Quand un framework générique ne suffit plus

Les 5 patterns de design qu'on applique

1. State machine explicite — chaque transition est une fonction nommée et testable, l'état est un dataclass immuable copié à chaque pas (pas de mutations cachées).
2. Event sourcing — chaque transition produit un événement persisté avant exécution. Le replay devient déterministe (en présence d'outils déterministes ou avec snapshots LLM).
3. Supervision arborescente — calque sur OTP/Erlang. Chaque agent enfant a un parent qui décide de la stratégie de reprise (restart, escalate, ignore).
4. Circuit breakers & budgets — limites strictes : max_tokens, max_steps, max_cost_eur, max_latency_ms. Un agent qui dépasse est tué, pas redémarré indéfiniment.
5. Observabilité native — chaque transition émet OpenTelemetry GenAI + MLflow Trace + métriques Prometheus. Pas d'instrumentation ajoutée après coup.

from dataclasses import dataclass, replace
from typing import Callable, Protocol
import mlflow

@dataclass(frozen=True)
class AgentState:
    messages: tuple
    plan: str | None
    budget_eur: float
    budget_steps: int
    decisions: tuple

class Transition(Protocol):
    name: str
    def __call__(self, state: AgentState) -> AgentState: ...

class Harness:
    def __init__(self, transitions, router, event_store, supervisor):
        self.transitions = {t.name: t for t in transitions}
        self.router = router
        self.event_store = event_store
        self.supervisor = supervisor

    def step(self, state: AgentState) -> AgentState:
        next_name = self.router(state)
        if next_name == "END" or state.budget_steps <= 0:
            return state

        with mlflow.start_span(name=next_name) as span:
            self.event_store.append("transition_start", next_name, state)
            try:
                new_state = self.transitions[next_name](state)
                self.event_store.append("transition_ok", next_name, new_state)
            except Exception as e:
                self.event_store.append("transition_fail", next_name, e)
                return self.supervisor.handle(state, next_name, e)
            return replace(new_state, budget_steps=new_state.budget_steps - 1)

    def run(self, initial: AgentState) -> AgentState:
        state = initial
        while self.router(state) != "END" and state.budget_steps > 0:
            state = self.step(state)
        return state

Mémoire longue : Letta, Mem0, MLflow Trace.

Un agent sans mémoire est un poisson rouge. Trois couches de mémoire à concevoir séparément : la mémoire de travail (état du graphe), la mémoire épisodique (historiques de conversations passées), et la mémoire sémantique (faits stables, préférences utilisateur).