Comment Optimiser l’Utilisation des Tokens avec Milvus (Étape par Étape)
Gérer l’utilisation des tokens de manière efficace avec Milvus peut réduire les coûts de calcul inutiles et rendre vos embeddings—et donc votre recherche vectorielle—beaucoup plus rapides et intelligents. Bien que beaucoup considèrent “milvus optimiser l’utilisation des tokens” comme une boîte noire, je vais vous montrer exactement comment vous pouvez réduire le flux de tokens dans vos pipelines RAG, votre recherche vectorielle et vos requêtes en aval sans sacrifier la précision.
Prérequis
- Python 3.11+
- Milvus Server 2.2.9+ (dernière version stable à partir du 21 mars 2026)
pymilvus>=2.2.9- Connaissance de base des concepts d’embeddings et de recherche vectorielle
- Accès à un encodage vectoriel basé sur GPU ou CPU (comme les embeddings OpenAI, les modèles Huggingface, ou similaires)
- Familiarité avec les limites de tokens de vos LLM (par exemple, les 8k tokens de GPT-4) et comment elles influencent le coût et la latence
Ce que Vous Construisez Réellement
Nous concevons un pipeline de recherche vectorielle qui élimine le superflu de vos entrées textuelles afin que Milvus ne conserve que ce qui enrichit réellement le contexte de votre requête, tout en maintenant la qualité des embeddings. Si vous avez déjà envoyé tous vos documents sources directement dans Milvus et observé l’explosion des coûts et des comptes de tokens, c’est la solution.
Étape par Étape
Étape 1 : Mesurez Vos Tokens Avant de Valider
from transformers import GPT2TokenizerFast
tokenizer = GPT2TokenizerFast.from_pretrained("gpt2")
def count_tokens(text: str) -> int:
return len(tokenizer.encode(text))
sample_text = "Ce paragraphe sera tokenisé et compté pour éviter de gaspiller des tokens."
print(f"Compte de tokens : {count_tokens(sample_text)}")Pourquoi : Si vous envoyez aveuglément de gros textes tels quels à l’étape d’embedding, vous brûlez des tokens que vous n’avez pas besoin de payer ou de stocker. Le tokenizer GPT-2 est un proxy bon marché et facile qui correspond à peu près aux comptes de tokens de style OpenAI. Cette étape de comptage initial empêche des morceaux trop longs de s’infiltrer dans Milvus.
Erreurs que vous rencontrerez : Utiliser un tokenizer qui ne correspond pas à votre LLM entraîne un sous-comptage ou un sur-comptage. Par exemple, les tokenizers Huggingface pour T5 diffèrent considérablement de ceux de GPT-3/4. Vérifiez toujours quel tokenizer est aligné avec l’utilisation de votre modèle.
Étape 2 : Découpez le Texte de Manière Intelligente – Optez pour le Sémantique plutôt que le Statique
def chunk_text(text: str, max_tokens: int = 500):
words = text.split()
chunks = []
current_chunk = []
current_tokens = 0
for word in words:
word_tokens = count_tokens(word)
if current_tokens + word_tokens > max_tokens:
chunks.append(" ".join(current_chunk))
current_chunk = [word]
current_tokens = word_tokens
else:
current_chunk.append(word)
current_tokens += word_tokens
if current_chunk:
chunks.append(" ".join(current_chunk))
return chunks
long_text = " ".join(["mot"] * 2000) # Exemple de long texte
split_chunks = chunk_text(long_text, max_tokens=500)
print(f"Créé {len(split_chunks)} morceaux.")Pourquoi : Diviser par des comptes de tokens au lieu de longueurs de caractères fixes évite de dépasser accidentellement les limites de tokens à l’étape d’embedding ou de requête. J’ai vu des pipelines planter ou se dégrader parce que les comptes de tokens ont explosé de manière inattendue lorsque des espaces ou des caractères UTF-8 sont apparus. Le découpage sémantique (comme les limites de phrases ou les sauts de paragraphe) fonctionne souvent mieux, mais une simple maximisation des tokens fonctionne de manière fiable.
Erreurs que vous rencontrerez : Des découpes naïves de caractères créent de mauvaises correspondances de requête—les fragments de contexte ne représentent pas un sens cohérent. Des morceaux trop volumineux provoquent des erreurs d’API d’embedding ou vous poussent rapidement hors des paliers gratuits.
Étape 3 : Dédupliquez Avant de Pousser vers Milvus
from hashlib import sha256
def deduplicate_chunks(chunks):
seen = set()
unique_chunks = []
for chunk in chunks:
fingerprint = sha256(chunk.encode("utf-8")).hexdigest()
if fingerprint not in seen:
unique_chunks.append(chunk)
seen.add(fingerprint)
return unique_chunks
unique_chunks = deduplicate_chunks(split_chunks)
print(f"Dédupliqué à {len(unique_chunks)} morceaux uniques.")Pourquoi : La redondance est l’ennemi. Je ne saurais trop insister là-dessus—de nombreux ensembles de données réels présentent des répétitions étranges, qu’il s’agisse de PDF corrompus ou de journaux. La déduplication évite le gaspillage de tokens d’embedding et de stockage dans Milvus, ce qui vous fait économiser de l’argent de calcul et empêche le bruit de recherche plus tard.
Erreurs que vous rencontrerez : Sauter la déduplication remplit Milvus de vecteurs en double, ralentit la recherche et gonfle le stockage. Votre budget basé sur les tokens explosera.
Étape 4 : Encodez les Morceaux en Vecteurs de Manière Efficace
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('all-MiniLM-L6-v2')
def encode_chunks(chunks):
embeddings = model.encode(chunks, convert_to_tensor=True)
return embeddings
embeddings = encode_chunks(unique_chunks)
print(f"Forme des embeddings produits : {embeddings.shape}")Pourquoi : Choisissez des modèles d’embedding plus petits et plus rapides à moins que vous ayez spécifiquement besoin des plus grands transformers pour une précision sémantique. Pour la plupart des applications, des modèles comme “all-MiniLM-L6-v2” représentent le meilleur compromis entre dimension vectorielle (384), vitesse et budget de tokens. Les embeddings de haute dimension ne sont pas toujours meilleurs ; ils peuvent gonfler votre index Milvus et ralentir la recherche.
Erreurs que vous rencontrerez : Tenter d’utiliser les embeddings OpenAI pour des milliers de longs morceaux sans prétraitement brûlera des tokens et atteindra rapidement les limites de taux de l’API. De plus, l’embedding sans lotir réduit le débit.
Étape 5 : Stockez avec des Métadonnées pour Filtrer le Contexte
from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection
connections.connect("default", host="localhost", port="19530")
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=384),
FieldSchema(name="token_count", dtype=DataType.INT64),
FieldSchema(name="chunk_text", dtype=DataType.VARCHAR, max_length=1024)
]
schema = CollectionSchema(fields, "Morceaux de documents avec métadonnées de tokens")
collection = Collection("doc_chunks", schema)
token_counts = [count_tokens(chunk) for chunk in unique_chunks]
entities = [
token_counts,
embeddings.tolist(),
unique_chunks
]
collection.insert(entities)
collection.create_index("embedding", {"index_type": "IVF_FLAT", "params": {"nlist": 128}, "metric_type": "L2"})
collection.load()Pourquoi : Stocker les comptes de tokens aux côtés des embeddings vous permet de filtrer ou de classer les morceaux à peu de frais sans avoir besoin de retokeniser plus tard. Les métadonnées des tokens réduisent les requêtes qui essaient de comprimer trop de contexte—et vous donnent le contrôle sur la taille de la charge utile de Milvus au moment de la recherche.
Étape 6 : Interrogez en Gardant à l’Esprit les Budgets de Tokens
def search_similar(query: str, top_k=5, max_query_tokens=1000):
query_token_count = count_tokens(query)
if query_token_count > max_query_tokens:
raise ValueError(f"La requête dépasse le budget de tokens : {query_token_count} > {max_query_tokens}")
query_embedding = model.encode([query])[0].tolist()
results = collection.search(
[query_embedding],
"embedding",
param={"metric_type": "L2", "params": {"nprobe": 10}},
limit=top_k,
output_fields=["token_count", "chunk_text"]
)
filtered_results = [res for res in results[0] if res.entity.get("token_count", 0) + query_token_count < max_query_tokens]
return filtered_results
query = "Utilisation efficace des tokens avec Milvus"
result_docs = search_similar(query)
for hit in result_docs:
print(hit.entity.get("chunk_text"))Pourquoi : La fenêtre de contexte de votre LLM est précieuse. Si vous oubliez de vérifier vos tokens de requête plus les tokens de morceaux pertinents, vous franchissez vos limites—ce qui entraîne des erreurs ou des invites tronquées. Le filtrage de Milvus basé sur les métadonnées de tokens stockées vous aide à rester dynamiquement sous budget.
Erreurs que vous rencontrerez : Passer des ensembles de tokens combinés trop volumineux à votre générateur conduit à des échecs de complétions ou à des sauts de contexte étranges. Une fois, j'ai eu un crash système après avoir ignoré les limites de tokens. Pas amusant.
Les Pièges
- Le Comptage des Tokens de l'API d'Embedding Vous Nuira : Les embeddings OpenAI comptent des tokens que vous n'attendez pas toujours, comme les tokens d'invite implicites ou les séparateurs. Effectuez toujours des essais à sec avec des comptes par morceau avant d'embedding en masse.
- Les Coûts de Stockage de Milvus Grimpent Rapidement : Le dépôt Milvus sous licence Apache-2.0 milvus-io/milvus a 43 421 étoiles—oui, il est populaire—mais la dimension vectorielle et le nombre de vecteurs que vous stockez entraînent une utilisation rapide de la RAM/stockage. Des vecteurs surdimensionnés sans élagage de tokens augmentent les coûts.
- Les Tokenizers ne S'accordent Pas : Si le tokenizer de création de morceaux et le tokenizer de votre LLM ne correspondent pas, vous serez soit en surestimation, soit en sous-estimation des tokens. Utilisez le tokenizer exact dont votre LLM a besoin.
- Temps de Création d'Index & Mémoire : Utiliser de grandes valeurs nlist dans les index IVF_FLAT améliore le rappel mais ajoute de la latence et une consommation de RAM. Trouvez votre juste milieu. Je commence généralement à nlist=128.
- Coherence des Morceaux vs Taille : Des morceaux plus gros contiennent plus de contexte mais coûtent plus de tokens. Des morceaux plus petits causent de la fragmentation et réduisent la précision. Expérimentez.
Code Complet
from transformers import GPT2TokenizerFast
from sentence_transformers import SentenceTransformer
from hashlib import sha256
from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection
# Étape 1 : Initialiser le tokenizer et les modèles
tokenizer = GPT2TokenizerFast.from_pretrained("gpt2")
model = SentenceTransformer('all-MiniLM-L6-v2')
def count_tokens(text: str) -> int:
return len(tokenizer.encode(text))
def chunk_text(text: str, max_tokens: int = 500):
words = text.split()
chunks = []
current_chunk = []
current_tokens = 0
for word in words:
word_tokens = count_tokens(word)
if current_tokens + word_tokens > max_tokens:
chunks.append(" ".join(current_chunk))
current_chunk = [word]
current_tokens = word_tokens
else:
current_chunk.append(word)
current_tokens += word_tokens
if current_chunk:
chunks.append(" ".join(current_chunk))
return chunks
def deduplicate_chunks(chunks):
seen = set()
unique_chunks = []
for chunk in chunks:
fingerprint = sha256(chunk.encode("utf-8")).hexdigest()
if fingerprint not in seen:
unique_chunks.append(chunk)
seen.add(fingerprint)
return unique_chunks
def encode_chunks(chunks):
embeddings = model.encode(chunks, convert_to_tensor=True)
return embeddings
# Se connecter à Milvus
connections.connect("default", host="localhost", port="19530")
# Définir le schéma
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=384),
FieldSchema(name="token_count", dtype=DataType.INT64),
FieldSchema(name="chunk_text", dtype=DataType.VARCHAR, max_length=1024)
]
schema = CollectionSchema(fields, "Morceaux de document avec métadonnées de jetons")
collection = Collection("doc_chunks", schema)
def insert_chunks(chunks):
unique_chunks = deduplicate_chunks(chunks)
embeddings = encode_chunks(unique_chunks)
token_counts = [count_tokens(chunk) for chunk in unique_chunks]
entities = [
token_counts,
embeddings.tolist(),
unique_chunks
]
collection.insert(entities)
collection.create_index("embedding", {"index_type": "IVF_FLAT", "params": {"nlist": 128}, "metric_type": "L2"})
collection.load()
def search_similar(query: str, top_k=5, max_query_tokens=1000):
query_token_count = count_tokens(query)
if query_token_count > max_query_tokens:
raise ValueError(f"Le nombre de jetons de la requête ({query_token_count}) dépasse la limite ({max_query_tokens})")
query_embedding = model.encode([query])[0].tolist()
results = collection.search(
[query_embedding],
"embedding",
param={"metric_type": "L2", "params": {"nprobe": 10}},
limit=top_k,
output_fields=["token_count", "chunk_text"]
)
filtered_results = [res for res in results[0] if res.entity.get("token_count", 0) + query_token_count < max_query_tokens]
return filtered_results
# Exemple d'utilisation
if __name__ == "__main__":
raw_text = ("Ceci est un exemple de paragraphe que nous aimerions découper, dédupliquer, intégrer et stocker. " * 100)
chunks = chunk_text(raw_text)
insert_chunks(chunks)
query = "utilisation des jetons dans un paragraphe exemple"
found = search_similar(query)
for hit in found:
print(hit.entity.get("chunk_text"))
Et maintenant
Maintenant que vous avez maîtrisé la prolifération des jetons dans Milvus, la prochaine étape logique consiste à mettre en œuvre un découpage dynamique des invites de requête, ce qui signifie que votre application doit surveiller la longueur combinée des jetons (requête plus contexte récupéré) et automatiquement supprimer ou reformuler les morceaux de faible valeur avant d'appeler votre LLM. Cela vous fera économiser de l'argent et évitera les erreurs de limite de jetons à l'exécution en production.
FAQ
Q : Comment puis-je confirmer que mes comptes de jetons correspondent à ceux de l'LLM ?
A : Le meilleur pari est d'utiliser le tokenizer fourni par votre LLM. Pour les modèles OpenAI, tiktoken est le tokenizer canonique. Le tokenizer GPT-2 est un proxy raisonnable mais pas exact. Exécutez toujours des cas de test avec le comptage de votre modèle pour vérifier.
Q : Quel est le nombre maximum de vecteurs que Milvus peut gérer avant que les performances ne souffrent ?
A : Milvus est optimisé pour des millions de vecteurs, mais en pratique, votre type d'index, la dimension du vecteur et le matériel dictent les performances. Par exemple, IVF_FLAT avec nlist=128 est gérable à quelques millions de vecteurs sur des serveurs corrects, mais la latence et la RAM peuvent augmenter sans traitement par lots et élagage.
Q : Puis-je automatiser l'élagage des jetons lors de l'insertion ?
A : Absolument, mais faites attention. Vous pouvez supprimer ou résumer les morceaux dépassant les limites de jetons avant l'intégration, mais un élagage excessif réduit la richesse sémantique, nuisant à la qualité de recherche en aval. Utilisez des seuils adaptatifs ajustés sur votre ensemble de données.
Résumé des statistiques Milvus
| Métrique | Valeur | Commentaire |
|---|---|---|
| Étoiles GitHub | 43,421 | Indique une forte adoption et un soutien communautaire |
| Forks | 3,909 | Montre une contribution active et des cas d'utilisation personnalisés |
| Problèmes ouverts | 1,098 | Signale le développement en cours et le suivi des bogues |
| Licence | Apache-2.0 | Licence permissive favorable à une utilisation en entreprise |
| Dernière mise à jour | 21 mars 2026 | Le projet est activement maintenu |
Recommandations pour différents types de développeurs
1. Le hacker de startup : Si vous construisez des MVP rapidement, concentrez-vous sur des modèles d'intégration prêts à l'emploi comme 'all-MiniLM-L6-v2' et un découpage de jetons simple pour réduire la dérive et les coûts. Utilisez l'indexation intégrée de Milvus et surveillez votre utilisation des jetons avec des compteurs simples.
2. Le data scientist : Expérimentez avec des approches de découpage sémantique : essayez la détection de limites de phrases ou l'encodage de paragraphes pour améliorer la fidélité des intégrations. Incorporez des métadonnées de compte de jetons pour l'élagage au moment de la requête. Vous devriez également envisager un ajustement fin de l'intégration basé sur la complexité des jetons des morceaux.
3. L'ingénieur d'entreprise : Construisez un pipeline adaptatif qui inclut une surveillance en temps réel du budget de jetons, la dé-duplication des morceaux, la dimensionnalité des vecteurs dynamique et le réglage de l'index sur Milvus. Intégrez étroitement vos pipelines LLM pour éviter les scénarios de dépassement et optimiser les dépenses informatiques.
Données au 21 mars 2026. Sources : https://github.com/milvus-io/milvus, Limites des jetons Milvus LangChain, Optimisation LLM Milvus
Articles connexes
- Containerisation des agents avec Docker Compose
- Récupération après sinistre pour le déploiement des agents IA
- Mise à l'échelle des agents IA horizontalement
🕒 Published: