Contexte et objectifs

L'article décrit la transformation d'un bouclage LLM basique en un système de production capable de planifier, d'agir, de récupérer et de prouver la validité de chaque étape. Le point de départ est un « basic harness » où un seul appel LLM génère une réponse linéaire. L’objectif est de rendre ce flux fiable à l’échelle d’une campagne aérienne : plusieurs missions parallèles, budgets de calcul, journalisation exhaustive et validation post‑exécution.

Architecture du harnais

Le cœur du nouveau harnais repose sur une orchestration très fine, découpée en trois rôles : Planner, Worker et Critic. Chaque rôle est implémenté comme une primitive testable et interconnectée via un graphe acyclique dirigé (DAG) qui décrit les dépendances entre les tâches. Le DAG permet l’exécution parallèle : dans l’exemple de comparaison de villes, neuf appels d’outils (trois villes × trois attributs) sont planifiés simultanément, alors que le rapport final attend la terminaison de tous les nœuds précédents.

class LLMProvider:
    """Shared interface. Subclass to plug in a different backend."""
    def complete(self, system: str, user: str, role: str = "default") -> str:
        raise NotImplementedError
    async def acomplete(self, system: str, user: str, role: str = "default") -> str:
        # Wrap sync call in a thread; works for any SDK.
        return await asyncio.to_thread(self.complete, system, user, role)

Cette abstraction évite le verrouillage fournisseur : le même orchestrateur peut appeler Anthropic, OpenAI ou un mock déterministe, facilitant les tests unitaires et la reproductibilité.

Primitives techniques

Les outils typés sont définis à l’aide de modèles Pydantic. Chaque outil expose un schéma JSON conforme aux API de tool‑use d’Anthropic et d’OpenAI, ce qui garantit une validation stricte avant l’exécution. Le code suivant montre la structure :

@dataclass
class TypedTool:
    name: str
    description: str
    args_model: type[BaseModel]
    fn: Callable[..., Any]
    cost_hint: float = 0.0
    def schema(self) -> dict:
        return {"name": self.name, "description": self.description,
                "input_schema": self.args_model.model_json_schema()}
    def run(self, raw_args: dict) -> Any:
        args, err = self.validate_args(raw_args)
        if err is not None:
            raise ValueError(err)
        return self.fn(**args.model_dump())

Le champ cost_hint alimente le système de budgeting multidimensionnel. Les outils get_population et get_timezone ont un cost_hint de 0,1, alors que summarize_city coûte 1,0, imposant une pression budgétaire réaliste. En cas de dépassement, le harnais applique une dégradation graduelle, par exemple en réduisant la granularité des résumés.

La mémoire à plusieurs niveaux combine un cache en‑mémoire pour les requêtes fréquentes et un récupérateur limité par un budget de tokens. Cette hiérarchie empêche le remplissage du contexte LLM avec des données inutiles, un problème fréquent dans les boucles séquentielles.

Une hiérarchie de vérification intercepte les sorties à chaque étape : le Critic valide les réponses des Workers, le Planner re‑planifie si une dépendance échoue, et le système journalise chaque décision via un tracer dédié.

Évaluation et limites

L’évaluation repose sur un jeu de tests automatisés qui mesurent la conformité du rapport final aux villes demandées, ainsi que sur des benchmarks de récupération qui comparent le temps d’accès aux différents niveaux de mémoire. Le prototype montre que, pour trois villes, le temps total diminue de 30 % grâce à l’exécution parallèle du DAG, tout en maintenant le coût tokenique sous le budget fixé.

Les limites restent liées à la dépendance à un modèle LLM fiable : si le modèle génère des arguments hors schéma, la validation Pydantic les rejette, mais le coût de la re‑planification n’est pas encore quantifié. De plus, le système de budgets repose sur des cost_hint statiques ; une adaptation dynamique aux variations de prix du modèle n’est pas implémentée.