NOPE LinkedIn

Catégories:
Blog

Valider un Agent LoRA : Vérification Fonctionnelle par Injection CAP v1

Valider un Agent LoRA : Vérification Fonctionnelle par Injection CAP v1 image

Rubrique: Blog Tag: LoRA Tag: Qwen2.5 Tag: Fine-tuning Tag: Debug Tag: Vérification Tag: CAP v1 Tag: function calling Tag: Tokenizer

Après entraînement, la question n’est pas “quelle est la loss ?”, c’est “l’agent appelle-t-il la bonne fonction quand on lui donne une directive réelle ?”.

C’est cette distinction qui a motivé la construction d’un système de vérification comportementale, distinct et indépendant du pipeline d’entraînement.


Le format CAP v1

Le coordinateur communique avec les agents via un format structuré appelé CAP v1 (Coordinator-Agent Packet). C’est le format de production — ce que reçoit l’agent dans un déploiement réel.

{
  "directive": "block_ip",
  "entities": {
    "ip": "10.0.0.1"
  },
  "args": {},
  "context": "Directive émise suite à détection CrowdSec"
}

Le script de vérification injecte des paquets CAP v1 directement dans le modèle, sans passer par le coordinateur ni par l’API des équipements. C’est un court-circuit délibéré : on teste l’adapter en isolation, dans les conditions exactes de production.


Architecture des scripts de vérification

Trois scripts couvrent les trois agents. Chacun suit la même structure :

verify_opnsense_qwen25.py  →  wrapper → verify_opnsense_v2.py  (102 cas de test)
verify_wireguard_qwen25.py →  wrapper → verify_wireguard.py    (11 cas de test)
verify_crowdsec_qwen25.py  →  wrapper → verify_crowdsec.py     (15 cas de test)

Les wrappers *_qwen25.py surchargent deux constantes avant l’exécution :

import scripts.verify_wireguard as _v

_v.DEFAULT_ADAPTER = PROJECT_ROOT / "loras" / "wireguard_qwen25_lora" / "adapter"
_v.BASE_MODEL      = "Qwen/Qwen2.5-3B-Instruct"

Cela permet de réutiliser les cas de test existants sans les dupliquer.


Structure d’un script de vérification

Les cas de test

Chaque cas est un triplet (prompt, expected_function, description) :

TEST_CASES = [
    (
        "Créez un tunnel Site-to-Site entre Paris (192.168.1.0/24) et Lyon (192.168.2.0/24).",
        "create_site_to_site_tunnel",
        "create_site_to_site_tunnel — tunnel inter-sites",
    ),
    (
        "Set up a point-to-point WireGuard tunnel between server 10.0.0.1 and client 10.0.0.2.",
        "create_point_to_point_tunnel",
        "create_point_to_point_tunnel — tunnel point-à-point",
    ),
    # ...
]

Les prompts sont délibérément bilingues FR/EN — le dataset d’entraînement l’est aussi. Chaque fonction est testée une fois, avec une formulation naturelle différente de celles du dataset. L’objectif est de vérifier la généralisation, pas la mémorisation.

La construction du prompt

def build_prompt(tokenizer, user_request: str) -> str:
    messages = [
        {"role": "user", "content": user_request},
    ]
    return tokenizer.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=True,
    )

apply_chat_template est utilisé ici pour la même raison que dans l’entraînement : garantir la cohérence de format entre les deux phases. C’est directement la leçon de l’article précédent.

L’inférence

def generate(model, tokenizer, prompt: str, max_new_tokens: int = 256) -> str:
    inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
    input_len = inputs["input_ids"].shape[1]

    im_end_id = tokenizer.convert_tokens_to_ids("<|im_end|>")
    eos_ids = []
    base_eos = tokenizer.eos_token_id
    if isinstance(base_eos, list):
        eos_ids.extend([x for x in base_eos if x is not None])
    elif base_eos is not None:
        eos_ids.append(base_eos)
    if isinstance(im_end_id, int) and im_end_id not in eos_ids:
        eos_ids.append(im_end_id)
    if not eos_ids:
        eos_ids = [151645]  # <|im_end|> Qwen2.5 fallback

    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=max_new_tokens,
            do_sample=False,     # Greedy — déterministe pour la vérification
            temperature=1.0,
            pad_token_id=eos_ids[0],
            eos_token_id=eos_ids,
        )
    return tokenizer.decode(outputs[0][input_len:], skip_special_tokens=True).strip()

Deux points importants :

do_sample=False — On veut un comportement déterministe. La vérification doit être reproductible ; une température > 0 introduirait de la variance dans les résultats.

La gestion des EOS tokens — C’est là qu’un bug s’est glissé lors de la migration.


Bug 1 : Le crash des EOS tokens Phi-3

Les scripts de vérification originaux avaient été écrits pour Phi-3 :

# Version Phi-3 — CRASH sur Qwen2.5
eos_token_id = [tokenizer.eos_token_id, tokenizer.convert_tokens_to_ids("<|end|>")]

Le token <|end|> est propre à Phi-3. Quand on passe au tokenizer Qwen2.5, convert_tokens_to_ids("<|end|>") retourne None — le token n’existe pas.

model.generate() reçoit alors [151645, None] et lève une TypeError :

TypeError: 'NoneType' object cannot be interpreted as an integer

Le fix est la construction robuste de eos_ids visible ci-dessus : on teste le type de chaque valeur avant de l’inclure, avec un fallback explicite si la liste reste vide.

Pour Qwen2.5, le token d’arrêt correct est <|im_end|> (id 151645).


Bug 2 : Le system prompt absent du dataset

Les scripts WireGuard et CrowdSec avaient un autre problème : ils injectaient un system_prompt en préfixe du message user.

# Version initiale
messages = [
    {"role": "system", "content": "Tu es un agent WireGuard..."},
    {"role": "user", "content": user_request},
]

Mais les datasets wireguard_combined_train.jsonl et crowdsec_train.jsonl n’ont pas de message system — uniquement des paires user/assistant. Le modèle n’a jamais vu de system prompt pendant l’entraînement.

Envoyer un system prompt en inférence introduit un contexte inconnu. Le fix : supprimer le system prompt et ne passer que le message user, comme dans le dataset.

# Après fix
messages = [{"role": "user", "content": user_request}]

OPNsense, lui, a un system prompt dans son dataset — le script OPNsense était donc correct dès le départ.


L’extraction de la fonction prédite

Le modèle génère du texte ; il faut en extraire le nom de fonction. L’extracteur utilise une cascade de stratégies, du plus structuré au moins structuré :

def extract_function_name(raw_output: str) -> str | None:
    # 1. Format tableau tool_calls : [{"function": {"name": "..."}}]
    array_match = re.search(r'\[.*?\]', raw_output, re.DOTALL)
    if array_match:
        try:
            arr = json.loads(array_match.group())
            for item in arr:
                if isinstance(item, dict) and "function" in item:
                    fn = item["function"].get("name")
                    if fn in WIREGUARD_FUNCTIONS:
                        return fn
        except json.JSONDecodeError:
            pass

    # 2. Blocs JSON simples {"name": "..."} ou {"function": {"name": "..."}}
    for candidate in re.findall(r'\{[^{}]+\}', raw_output, re.DOTALL):
        try:
            obj = json.loads(candidate)
            if "name" in obj and obj["name"] in WIREGUARD_FUNCTIONS:
                return obj["name"]
        except json.JSONDecodeError:
            pass

    # 3. Fallback regex — tri du plus long au plus court (évite les sous-chaînes)
    for fn in sorted(WIREGUARD_FUNCTIONS, key=len, reverse=True):
        if re.search(r'\b' + re.escape(fn) + r'\b', raw_output):
            return fn

    return None

Le tri par longueur décroissante au fallback est important : si une fonction s’appelle get_tunnel_status et une autre get_wireguard_status, un match sur status sans word boundary trouverait les deux. Le tri garantit que le nom le plus spécifique est testé en premier.


La sortie du script

=================================================================
              VÉRIFICATION LORA WIREGUARD
               wireguard_qwen25_lora
=================================================================

✅ [01/11] create_site_to_site_tunnel — tunnel inter-sites
✅ [02/11] create_point_to_point_tunnel — tunnel point-à-point
✅ [03/11] create_mesh_network — réseau mesh multi-nœuds
✅ [04/11] get_tunnel_status — état d'un tunnel
✅ [05/11] rotate_keys — rotation de clés
✅ [06/11] verify_routing — vérification du routage
✅ [07/11] add_wireguard_server — configurer un serveur
✅ [08/11] add_wireguard_client — ajouter un peer client
✅ [09/11] generate_wireguard_keypair — générer une paire de clés
✅ [10/11] generate_wireguard_psk — clé pré-partagée
✅ [11/11] get_wireguard_status — état global WireGuard

=================================================================
  Score : 11/11  (100%)
  ✅  ADAPTATEUR VALIDÉ
=================================================================

Résultats finaux des trois agents

Après correction des deux bugs et réentraînement OPNsense avec le bon format de prompt :

Agent Fonctions Score Durée vérification
OPNsense 102 102/102 — 100% ~18 min (chargement + 102 inférences)
WireGuard 11 11/11 — 100% ~2 min
CrowdSec 15 15/15 — 100% ~3 min

Ce que cette vérification garantit — et ce qu’elle ne garantit pas

Ce qu’elle garantit : le modèle sélectionne la bonne fonction face à des formulations naturelles non vues pendant l’entraînement. C’est une validation comportementale de la capacité de routing.

Ce qu’elle ne garantit pas : que les arguments extraits sont corrects, que l’API de l’équipement accepte la requête, que le résultat est cohérent dans un contexte multi-tours. Ce sont des niveaux de test supplémentaires, non encore implémentés.

La vérification est une condition nécessaire, pas suffisante. Mais passer de 0/102 à 102/102 après correction du format de prompt est un signal clair que le problème était bien identifié.


Prochain article : comment servir trois agents LoRA sur un seul GPU avec vLLM en mode multi-LoRA dynamique — budget VRAM, découverte automatique, hot-swap.