Surveillance de l'utilisation des tokens et création d'un tableau de bord de contrôle des coûts pour Langchain-Chatchat

Dans le contexte des applications d'IA d'entreprise, un aspect crucial mais souvent négligé est la maîtrise des coûts associés aux appels aux modèles de langage (LLM). Pour les systèmes comme Langchain-Chatchat, qui exploitent des LLM pour des réponses sur des bases de connaissances locales, le suivi de l'utilisation des tokens devient essentiel lorsque des API payantes telles que Qwen ou Wenxin Yiyan sont utilisées.

Les tokens représentent les unités de base dans lesquelles le texte est divisé par les LLM. Par exemple, en français, le mot « intelligence » pourrait être un token, tandis qu'en anglais, « un » et « happiness » pourraient être des tokens séparés selon le modèle. Chaque modèle utilise son propre tokenizer, ce qui signifie que le même texte peut produire des nombres de tokens différents. Par conséquent, les API facturées au token nécessitent une surveillance précise pour éviter des coûts imprévus.

Statistique précise des tokens

Une estimation approximative basée sur les caractères est sujette à erreurs. Il faut utiliser le tokenizer officiel du modèle cible. Par exemple, avec le tokenizer Qwen :

from transformers import AutoTokenizer

# Charger le tokenizer correspondant au modèle utilisé
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen-7B-Chat", trust_remote_code=True)

def calculate_token_count(texte: str) -> int:
    return len(tokenizer.encode(texte))

Cette fonction doit être intégrée dans le pipeline de Langchain-Chatchat, par exemple dans un wrapper personnalisé, pour compter les tokens avant et après chaque requête. Pour d'autres modèles comme Baichuan ou Zhipu, il est préférable d'utiliser leurs SDK spécifiques.

Conception d'un tableau de bord de contrôle des coûts

Un tableau de bord visuel aide à transformer les données brutes en informations exploitables. La première étape consiste à enregistrer chaque appel LLM dans une base de données. Utilisons SQLite pour un prototype rapide :

import sqlite3
from datetime import datetime

connection = sqlite3.connect('metrics_cout.db')
cursor = connection.cursor()

cursor.execute('''
CREATE TABLE IF NOT EXISTS usage_tokens (
    identifiant INTEGER PRIMARY KEY AUTOINCREMENT,
    id_requete TEXT,
    horodatage DATETIME DEFAULT CURRENT_TIMESTAMP,
    modele TEXT,
    tokens_entree INTEGER,
    tokens_sortie INTEGER,
    cout_estime REAL,
    texte_requete TEXT,
    nombre_docs_source INTEGER
)
''')
connection.commit()

Ensuite, encapsuler l'enregistrement des métriques :

def consigner_utilisation_tokens(id_requete, modele, tokens_entree, tokens_sortie, requete, nb_docs=3):
    total = tokens_entree + tokens_sortie
    cout = total * 0.008 / 1000  # Prix par millier de tokens

    cursor.execute('''
        INSERT INTO usage_tokens 
        (id_requete, modele, tokens_entree, tokens_sortie, cout_estime, texte_requete, nombre_docs_source)
        VALUES (?, ?, ?, ?, ?, ?, ?)
    ''', (id_requete, modele, tokens_entree, tokens_sortie, cout, requete, nb_docs))

    connection.commit()

Cette fonction est appelée après chaque interaction, capturant ainsi toutes les données pertinentes pour l'analyse des coûts.

Visualisation avec un tableau de bord interactif

Pour créer une interface web, on peut utiliser Streamlit, qui permet de développer rapidement des tableaux de bord. Après extraction des données agrégées :

import streamlit as st
import pandas as pd

# Lire les données de la base
df = pd.read_sql_query("""
    SELECT 
        DATE(horodatage) as date,
        SUM(cout_estime) as cout_journalier,
        AVG(tokens_entree + tokens_sortie) as moyenne_tokens,
        COUNT(*) as nombre_appels
    FROM usage_tokens 
    GROUP BY DATE(horodatage)
""", connection)

st.title("Tableau de bord de contrôle des coûts Langchain-Chatchat")
st.line_chart(df, x='date', y='cout_journalier')
st.bar_chart(df, x='date', y='moyenne_tokens')

On peut ajouter des filtres par utilisateur ou type de requête pour une analyse plus fine. Ce tableau de bord peut être déployé indépendamment du service principal pour éviter les interférences de performance.

Considérations d'ingénierie pour la production

Pour une mise en œuvre robuste, plusieurs aspects méritent attention :

• Écriture asynchrone : Utiliser une file de messages comme Redis pour ne pas bloquer le pipeline d'inférence.
• Anonymisation des données : Appliuqer des expressions régulières pour masquer les informations sensibles dans le texte des requêtes avant l'enregistrement.
• Interface unifiée pour multi-modèles : Abstraire la logique de comptage des tokens pour supporter plusieurs modèles via une fonction générique.
• Surveillance des modèles locaux : Même sans coût API direct, surveiller les tokens aide à optimiser les performances et l'utilisation des ressources.

En mettant en place cette surveilalnce, on résout des problèmes concrets : identifier les coûts excessifs par département, détecter les abus d'utilisation, et comparer l'efficacité des modèles. Cela permet de justifier les investissements en IA avec des mesures quantifiables et d'améliorer continuellement le système.