Concevoir un harness d'agents in-house

Pourquoi parler de harness in-house

Notre position est claire : LangGraph couvre 80 % des besoins agentiques en production. Si vous êtes au début d'un projet et que LangGraph répond à votre cahier des charges, n'écrivez pas votre propre harness — c'est du gaspillage.

Mais 20 % des projets, chez nos clients, sortent du cadre des frameworks génériques. Cet article documente quand et comment concevoir un harness sur-mesure, en s'inspirant de ce que les frameworks font bien et en allant au-delà sur les axes qui le justifient.

Quand un framework générique ne suffit plus

Quatre situations justifient, à nos yeux, l'investissement d'un harness in-house. Toutes les autres devraient se résoudre avec LangGraph + une discipline d'ingénierie correcte.

① Audit forensique exigé

Certains secteurs (banque sous ACPR, défense, OIV, santé HDS) demandent une traçabilité au-delà de ce qu'offrent les frameworks : signature cryptographique des transitions, immutabilité forte, time-travel queries avec garanties d'intégrité, archivage légal sur 7-10 ans.

LangGraph trace tout via son checkpointer, mais ne signe pas, n'horodate pas avec un tiers de confiance, n'archive pas. Si votre RSSI ou votre DPO l'exige, un harness avec event-sourcing signé devient nécessaire.

② Latence sub-second

Pour les usages temps-réel (voix conversationnelle, trading, copilote d'assistance), chaque milliseconde compte. Python avec asyncio + LangGraph ajoute 50 à 200 ms d'overhead par transition de nœud — parfois inacceptable.

Un harness optimisé (Rust, Go, ou Python avec instrumentation minimale) peut diviser cet overhead par 5 à 10. Au prix d'un effort d'ingénierie significatif.

③ Air-gap strict avec audit dépendances

Environnements isolés ANSSI ou OTAN : aucun package PyPI tiers sans audit individuel. LangGraph dépend transitivement de centaines de packages. Tracer la conformité de chacun est un cauchemar.

Un harness in-house minimal (200-500 lignes Python) avec uniquement les dépendances stdlib + httpx + pydantic est plus simple à faire valider.

④ Intégration profonde au SI existant

Quand votre SI est construit autour de Kafka, NATS, Temporal, Camunda, ou d'un workflow engine maison, LangGraph devient un corps étranger. Mieux vaut un harness qui s'aligne sur l'architecture existante.

Les cinq patterns de design

Notre harness de référence, que nous adaptons et livrons chez nos clients quand requis, repose sur cinq patterns.

① State machine explicite

Chaque transition est une fonction nommée, testable indépendamment. L'état est un dataclass(frozen=True) — immuable. Chaque nœud retourne un nouvel état via replace(), jamais ne mute l'état d'entrée.

from dataclasses import dataclass, replace

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

def planner(state: AgentState) -> AgentState:
    plan = llm.invoke(state.messages[-1]).content
    return replace(state, plan=plan, iteration=state.iteration + 1)

L'immutabilité élimine toute une classe de bugs (mutations cachées, races, deep copies oubliées). Le coût en mémoire est négligeable comparé aux gains en debuggability.

② Event sourcing

Chaque transition produit un événement persisté avant son exécution. Le replay devient déterministe — en présence d'outils déterministes, ou via snapshots LLM (cache du prompt + réponse).

class EventStore(Protocol):
    def append(self, event_type: str, node: str, state: AgentState) -> str: ...
    def replay(self, run_id: str) -> Iterable[AgentState]: ...

# Implémentation Postgres
class PostgresEventStore:
    def append(self, event_type, node, state):
        event_id = str(uuid4())
        self.conn.execute(
            "INSERT INTO events (id, ts, run_id, event_type, node, state_json) VALUES (...)",
            (event_id, datetime.utcnow(), self.run_id, event_type, node, asdict(state))
        )
        return event_id

Avantages : audit forensique trivial, debug par replay, A/B testing sur trajectoires capturées, dataset d'évaluation gratuit (chaque run en prod génère des trajectoires).

③ Supervision arborescente

Calque sur OTP (Erlang). Chaque agent enfant a un parent qui décide de la stratégie de reprise en cas d'erreur : restart (relance le nœud), escalate (remonte au parent), ignore (continue malgré l'échec).

class Supervisor:
    def handle(self, state: AgentState, failed_node: str, exc: Exception) -> AgentState:
        strategy = self.strategy_for(failed_node, exc)

        if strategy == "restart":
            # Retry avec backoff exponentiel
            return state  # le router renverra vers failed_node

        elif strategy == "escalate":
            # Marque l'erreur dans l'état et passe à un nœud de handling
            return replace(state, decisions=(*state.decisions, f"failed:{failed_node}"))

        elif strategy == "ignore":
            # Continue comme si rien n'avait échoué (dangereux mais parfois utile)
            return state

④ 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.

def budget_check(state: AgentState) -> AgentState:
    if state.iteration >= MAX_STEPS:
        raise BudgetExceeded("max_steps")
    if state.budget_eur > MAX_BUDGET:
        raise BudgetExceeded("max_cost")
    return state

Pattern complémentaire : circuit breaker sur les outils externes. Si un appel API échoue 5 fois en 60 secondes, on coupe l'accès à cet outil pour 5 minutes. L'agent fonctionne en dégradé plutôt que de boucler sur l'échec.

⑤ Observabilité native

Chaque transition émet : OpenTelemetry GenAI (le standard 2026), trace MLflow, métriques Prometheus. Pas d'instrumentation ajoutée après coup — c'est dans le harness lui-même.

def step(self, state: AgentState) -> AgentState:
    next_node = self.router(state)
    if next_node == "END":
        return state

    with mlflow.start_span(name=next_node) as span:
        self.event_store.append("transition_start", next_node, state)
        prom_counter.labels(node=next_node).inc()

        try:
            t0 = time.time()
            new_state = self.transitions[next_node](state)
            prom_latency.labels(node=next_node).observe(time.time() - t0)
            self.event_store.append("transition_ok", next_node, new_state)
        except Exception as e:
            self.event_store.append("transition_fail", next_node, e)
            new_state = self.supervisor.handle(state, next_node, e)

        return new_state

Pièges fréquents quand on écrit son harness

Piège 1 — Réinventer LangGraph en moins bien

Si vous écrivez votre propre StateGraph, vos propres conditional_edges, votre propre checkpointer Postgres — vous êtes en train de réécrire LangGraph. Reprenez LangGraph. Le seul vrai harness in-house apporte quelque chose que LangGraph ne fait structurellement pas (signature crypto, latence Rust, air-gap absolu).

Piège 2 — Pas de typage strict

Un harness sans mypy --strict ou équivalent finit par dériver. L'état circule entre 20 fonctions, personne ne sait plus quels champs sont définis quand. Typage strict, dès la première ligne.

Piège 3 — Sérialisation cassée

L'état doit être sérialisable : pour le checkpoint, pour le replay, pour l'audit. Évitez de mettre dans l'état des objets non-sérialisables (instances LangChain, sockets, connexions DB). Tout ce qui passe par l'état doit être pur Python typé.

Piège 4 — Tests d'agents oubliés

Un harness in-house est du code maison : il doit avoir une suite de tests unitaires (chaque transition testée en isolation) et d'intégration (trajectoires complètes avec LLM mockés). Sans ça, vous serez plus fragile que LangGraph dans 6 mois.

Notre recommandation

Si vous hésitez entre LangGraph et un harness in-house : commencez par LangGraph. Mesurez. Si à 6 mois vous identifiez une contrainte structurelle (signature crypto, latence sub-second, air-gap), extrayez le harness à partir de votre code LangGraph existant. Vous gagnez sur les deux tableaux : time-to-market initial, et liberté à terme.

Chez DEEP-5, c'est exactement le service que nous délivrons : nous démarrons sur LangGraph, et nous extrayons un harness sur-mesure si et seulement si vos contraintes le justifient. Pas par défaut. Pas par dogme.