Comment optimiser l’utilisation des tokens avec Milvus (Étape par étape)
Gérer efficacement l’utilisation des tokens 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. Alors que beaucoup considèrent “milvus optimise l’utilisation des tokens” comme une boîte noire, je vais vous montrer exactement comment vous pouvez réduire le gaspillage de tokens dans vos pipelines RAG, la recherche vectorielle et les 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 embeddings et des concepts de recherche vectorielle
- Accès à un encodage vectoriel basé sur GPU ou CPU (comme les embeddings OpenAI, les modèles Huggingface ou similaires)
- Connaissance des limites de tokens de vos LLM (par exemple, les 8k tokens de GPT-4) et de leur impact sur le coût et la latence
Ce que vous êtes en train de construire
Nous construisons un pipeline de recherche vectorielle qui élimine les éléments superflus de vos entrées textuelles afin que Milvus stocke uniquement ce qui enrichit réellement le contexte de votre requête, tout en équilibrant la qualité de l’embedding. Si vous avez déjà envoyé tous vos documents sources directement dans Milvus et regardé les coûts et le nombre de tokens exploser, voici la solution.
Étape par étape
Étape 1 : Mesurez vos tokens avant de vous engager
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 le gaspillage de tokens."
print(f"Nombre de tokens : {count_tokens(sample_text)}")Pourquoi : Si vous envoyez aveuglément d’énormes 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 première étape de comptage vous évite de laisser des morceaux trop longs entrer dans Milvus.
Erreurs que vous rencontrerez : Utiliser un tokenizer qui ne correspond pas à votre LLM entraîne un sous-comptage ou un surcomptage. Par exemple, les tokenizers Huggingface pour T5 diffèrent considérablement des tokenizers GPT-3/4. Vérifiez toujours quel tokenizer correspond à 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(["word"] * 2000) # Exemple de long texte
split_chunks = chunk_text(long_text, max_tokens=500)
print(f"Créé {len(split_chunks)} morceaux.")Pourquoi : Diviser par le nombre de tokens au lieu de longueurs de caractères fixes empêche d’éventuels dépassements de limites de tokens lors de l’embedding ou de la requête. J’ai vu des pipelines se bloquer ou se dégrader parce que le nombre de tokens a soudainement augmenté lorsque des espaces ou des caractères UTF-8 sont apparus. La découpe sémantique (comme les limites de phrases ou les sauts de paragraphes) fonctionne souvent mieux, mais un découpage simple par maximum de tokens est fiable.
Erreurs que vous rencontrerez : Des découpages naïfs par caractères créent de mauvaises correspondances de requêtes — les fragments de contexte ne représentent pas un sens cohérent. Des morceaux trop grands causent des erreurs d’API d’embedding ou vous font sortir rapidement des niveaux 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 peux pas insister assez—de nombreux jeux de données réels contiennent 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, économisant ainsi des coûts de calcul et prévenant le bruit de recherche par la suite.
Erreurs que vous rencontrerez : Négliger 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 générés : {embeddings.shape}")Pourquoi : Choisissez des modèles d’embedding plus petits et plus rapides, sauf si vous avez spécifiquement besoin des plus gros 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 du vecteur (384), vitesse et budget de tokens. Les embeddings à haute dimension ne sont pas toujours meilleurs ; ils peuvent gonfler votre index Milvus et ralentir la recherche.
Erreurs que vous rencontrerez : Tenter d’utiliser l’embedding OpenAI pour des milliers de morceaux longs sans prétraitement consommera des tokens et atteindra rapidement les limites de taux API. De plus, l’embedding sans traitement par lots 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 document avec métadonnées de token")
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 de manière économique sans re-tokeniser plus tard. Les métadonnées de 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 lors de la recherche.
Étape 6 : Requêter en tenant compte des 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 dépassez 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 le budget.
Erreurs que vous rencontrerez : Passer des ensembles de tokens combinés trop importants à votre générateur entraîne des échecs de complétion ou des sauts de contexte bizarres. J'ai déjà eu un système qui a planté après avoir ignoré les limites de tokens. Pas amusant.
Les pièges
- Le comptage des tokens de l'API d'embedding vous nuit : Les embeddings OpenAI comptent des tokens que vous n'attendez pas toujours, comme des tokens de prompt implicites ou des séparateurs. Faites toujours des essais à sec avec des comptes par morceau avant un embedding en masse.
- Les coûts de stockage Milvus grimpent rapidement : Le dépôt avec licence Apache-2.0 de Milvus milvus-io/milvus a 43 421 étoiles—oui, il est populaire—mais la dimension du vecteur et le nombre de vecteurs que vous stockez provoquent une utilisation rapide de la RAM/du stockage. Des vecteurs surdimensionnés sans élagage de tokens font gonfler les coûts.
- Les tokenizers ne s'accordent pas : Si votre tokenizer de création de morceaux et le tokenizer du LLM ne correspondent pas, vous allez soit surévaluer, soit sous-évaluer les tokens. Utilisez le tokenizer exact dont votre LLM a besoin.
- Temps de création d'index et mémoire : Utiliser des valeurs nlist élevées dans les index IVF_FLAT améliore le rappel mais ajoute de la latence et consomme de la RAM. Trouvez votre juste milieu. Je commence généralement avec nlist=128.
- Coherence des morceaux vs taille : Des morceaux plus grands contiennent plus de contexte mais coûtent plus de tokens. Des morceaux plus petits entraînent une fragmentation et diminuent 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
# Connexion à 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, "Segments de document avec métadonnées de token")
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 tokens dans 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 tokens d'exemple"
found = search_similar(query)
for hit in found:
print(hit.entity.get("chunk_text"))
Qu'est-ce qui suit
Maintenant que vous avez maîtrisé l'augmentation des tokens envoyés à Milvus, l'étape logique suivante est d'implémenter un trimming dynamique des invites de requête—c'est-à-dire que votre application doit surveiller la longueur combinée des tokens (requête plus contexte récupéré) et supprimer ou paraphraser automatiquement les segments de faible valeur avant d'appeler votre LLM. Cela vous fera économiser de l'argent et préviendra les erreurs de limite de tokens à l'exécution en production.
FAQ
Q : Comment puis-je confirmer que mes nombres de tokens correspondent au comptage interne du LLM ?
A : La meilleure option 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 soient affectées ?
A : Milvus est optimisé pour des millions de vecteurs, mais pratiquement, le type d'index, la dimension des vecteurs et le matériel dictent les performances. Par exemple, IVF_FLAT avec nlist=128 est gérable à quelques millions de vecteurs sur de bons serveurs, mais la latence et la RAM peuvent augmenter sans un traitement par lots et un élagage.
Q : Puis-je automatiser l'élagage des tokens au moment de l'insertion ?
A : Absolument, mais prenez garde. Vous pouvez supprimer ou résumer les segments dépassant les limites de tokens 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.
Aperçu des statistiques de Milvus
| Métrique | Valeur | Commentaire |
|---|---|---|
| Étoiles GitHub | 43,421 | Indique une forte adoption et un soutien communautaire |
| Forks | 3,909 | Démontre une contribution active et des cas d'utilisation personnalisés |
| Problèmes ouverts | 1,098 | Énonce le développement en cours et le suivi des bugs |
| 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 tokens basique pour réduire le dérive et les coûts. Utilisez l'indexation intégrée de Milvus et surveillez votre utilisation des tokens avec des compteurs simples.
2. Le Data Scientist : Essayez des approches de découpage sémantique—essayez la détection de frontières de phrases ou l'encodage de paragraphes—pour améliorer la fidélité de l'intégration. Incorporez des métadonnées sur le nombre de tokens pour l'élagage lors de la requête. Vous pourriez également envisager un ajustement fin de l'intégration personnalisé basé sur la complexité des tokens des segments.
3. L'ingénieur d'entreprise : Créez un pipeline adaptatif qui incorpore une surveillance en temps réel du budget de tokens, déduplication des segments, dimensionnalité vectorielle dynamique et réglage de l'index sur Milvus. Intégrez vos pipelines LLM de manière étroite pour éviter les dépassements de scénarios et optimiser les dépenses de calcul.
Données au 21 mars 2026. Sources : https://github.com/milvus-io/milvus, Limites de tokens Milvus LangChain, Optimisation du LLM de Milvus
Articles connexes
- Conteneuriser des agents avec Docker Compose
- Récupération après sinistre de déploiement d'agent IA
- Mise à l'échelle des agents IA horizontalement
🕒 Published: