Wie man die Token-Nutzung mit Milvus optimiert (Schritt für Schritt)

Wie man die Token-Nutzung mit Milvus optimiert (Schritt für Schritt)

Eine effiziente Handhabung der Token-Nutzung mit Milvus kann unnötige Computerkosten reduzieren und Ihre Embeddings – und somit Ihre Vektorsuche – viel schneller und intelligenter machen. Während viele Leute die Suche nach “milvus optimize token usage” als Black Box behandeln, werde ich Ihnen genau zeigen, wie Sie die Token-Überlastung in Ihren RAG-Pipelines, der Vektorsuche und den nachgelagerten Abfragen reduzieren können, ohne dabei an Genauigkeit zu verlieren.

Voraussetzungen

• Python 3.11+
• Milvus Server 2.2.9+ (neueste stabile Version vom 21. März 2026)
• pymilvus>=2.2.9
• Grundkenntnisse über Embeddings und Vektorsuchen
• Zugang zu GPU- oder CPU-basiertem Vektorkodieren (wie OpenAI embeddings, Huggingface Modelle oder ähnliche)
• Vertrautheit mit den Token-Grenzen Ihrer LLMs (z.B. GPT-4’s 8k Token) und wie diese Kosten und Latenz beeinflussen

Was Sie Tatsächlich Erstellen

Wir entwickeln eine Vektorsuche-Pipeline, die die überflüssigen Teile aus Ihren Texteingaben entfernt, sodass Milvus nur das speichert, was Ihren Abfragekontext wirklich bereichert, während die Qualität der Embeddings ausgewogen bleibt. Falls Sie jemals alle Ihre Quelldokumente direkt in Milvus gepumpt und beobachtet haben, wie die Kosten und Token-Zählungen explodieren, ist dies die Lösung.

Schritt für Schritt

Schritt 1: Messen Sie Ihre Token, bevor Sie sie verwenden

from transformers import GPT2TokenizerFast

tokenizer = GPT2TokenizerFast.from_pretrained("gpt2")

def count_tokens(text: str) -> int:
 return len(tokenizer.encode(text))

sample_text = "Dieser Absatz wird tokenisiert und gezählt, um ein Verschwenden von Token zu verhindern."
print(f"Token-Zählung: {count_tokens(sample_text)}")

Warum: Wenn Sie massive Texte blindlings im Original zur Embedding-Phase senden, verbrennen Sie Token, für die Sie nicht bezahlen oder die Sie nicht speichern müssen. Der GPT-2-Tokenizer ist eine kostengünstige und einfache Proxy, die grob mit der Token-Zählung im OpenAI-Stil übereinstimmt. Diese anfängliche Zählstufe verhindert, dass zu lange Stücke in Milvus gelangen.

Fehler, die Sie machen werden: Die Verwendung eines Tokenizers, der nicht zu Ihrem LLM passt, führt zu Unter- oder Überzählungen. Zum Beispiel unterscheiden sich Huggingface-Tokenizer für T5 erheblich von GPT-3/4-Tokenizern. Überprüfen Sie immer, welcher Tokenizer mit der Nutzung Ihres Modells übereinstimmt.

Schritt 2: Text Intelligently Chunken – Gehen Sie Semantisch Über Statisch

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) # Beispiel für einen langen Text
split_chunks = chunk_text(long_text, max_tokens=500)
print(f"Erstellt {len(split_chunks)} Stücke.")

Warum: Das Aufteilen nach Token-Zählungen anstelle von festen Zeichenlängen verhindert, dass Sie versehentlich die Token-Grenzen bei der Embedding- oder Abfragezeit überschreiten. Ich habe gesehen, wie Pipelines abgestürzt sind oder sich verschlechtert haben, weil die Token-Zählungen unerwartet anstiegen, als Leerzeichen oder UTF-8-Zeichen auftraten. Semantisches Chunking (wie Satzgrenzen oder Absatzumbrüche) funktioniert oft besser, aber einfaches Maxing bei Token funktioniert zuverlässig.

Fehler, die Sie machen werden: Naive Zeichenstücke führen zu schlechten Abgleichungen bei Abfragen – Kontextfragmente repräsentieren keine kohärente Bedeutung. Übergroße Stücke verursachen Fehler bei der Embedding-API oder bringen Sie schnell aus den kostenlosen Tarifen heraus.

Schritt 3: Duplizieren vor dem Hochladen an 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"Auf {len(unique_chunks)} eindeutige Stücke dedupliziert.")

Warum: Redundanz ist der Feind. Ich kann das nicht genug betonen – viele echte Datensätze haben seltsame Wiederholungen, sei es durch beschädigte PDFs oder Protokolle. Die Duplikation vermeidet eine Verschwendung von Embedding- und Milvus-Speichertoken, spart Computerkosten und verhindert später Suchgeräusche.

Fehler, die Sie machen werden: Wenn Sie die Duplikation überspringen, füllen Sie Milvus mit duplizierten Vektoren, verlangsamen die Suche und erhöhen den Speicherbedarf. Ihr auf Token basierendes Budget wird explodieren.

Schritt 4: Chunk effizient in Vektoren kodieren

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"Produzierte Embeddings Form: {embeddings.shape}")

Warum: Wählen Sie kleinere, schnellere Embedding-Modelle, es sei denn, Sie benötigen speziell die größten Transformer für semantische Präzision. Für die meisten Anwendungen bieten Modelle wie „all-MiniLM-L6-v2“ den besten Kompromiss zwischen Vektorgröße (384), Geschwindigkeit und Token-Budget. Hochdimensionale Embeddings sind nicht immer besser; sie können Ihren Milvus-Index aufblähen und die Suche verlangsamen.

Fehler, die Sie machen werden: Der Versuch, OpenAI-Embedding für Tausende von langen Stücken ohne Vorverarbeitung zu verwenden, wird schnell Token verbrauchen und API-Rate-Limits erreichen. Zudem reduziert das Embedding ohne Batching den Durchsatz.

Schritt 5: Mit Metadaten speichern, um den Kontext zu filtern

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, "Dokumentchunks mit Token-Metadaten")
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()

Warum: Das Speichern von Token-Zählungen zusammen mit Embeddings ermöglicht es Ihnen, Stücke kostengünstig zu filtern oder zu bewerten, ohne später erneut zu tokenisieren. Die Token-Metadaten reduzieren Abfragen, die versuchen, zu viel Kontext zu quetschen – und geben Ihnen Kontrolle über die Payload-Größe von Milvus zur Suchzeit.

Schritt 6: Abfragen mit Blick auf Token-Budgets

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"Abfrage überschreitet Token-Budget: {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 = "Effiziente Token-Nutzung mit Milvus"
result_docs = search_similar(query)
for hit in result_docs:
 print(hit.entity.get("chunk_text"))

Warum: Das Kontextfenster Ihres LLM ist kostbar. Wenn Sie vergessen, Ihre Abfrage-Token und die relevanten Chunk-Token zu überprüfen, überschreiten Sie Ihre Grenzen, was zu Fehlern oder abgeschnittenen Aufforderungen führt. Milvus’ Filterung basierend auf gespeicherten Token-Metadaten hilft Ihnen, dynamisch im Budget zu bleiben.

Fehler, die Sie machen werden: Das Übergeben von zu großen kombinierten Token-Sets an Ihren Generator führt zu fehlgeschlagenen Vervollständigungen oder seltsamen Kontext-Sprüngen. Ich hatte einmal einen Systemabsturz, nachdem ich die Token-Grenzen ignoriert hatte. Nicht lustig.

Die Fallstricke

1. Die Token-Zählung der Embedding-API schadet Ihnen: OpenAI-Embeddings zählen Tokens, die Sie nicht immer erwarten, wie implizite Aufforderungs-Token oder Trennzeichen. Führen Sie immer Testläufe mit Zählungen pro Stück durch, bevor Sie in großen Mengen einfügen.
2. Die Speicherkosten für Milvus steigen schnell: Das Apache-2.0-lizenzierte Repository von Milvus milvus-io/milvus hat 43.421 Sterne – ja, es ist beliebt – aber die Vektordimension und die Anzahl der Vektoren, die Sie speichern, verursachen eine schnelle Nutzung von RAM/Speicher. Übergroße Vektoren ohne Token-Pruning erhöhen die Kosten.
3. Tokenizer stimmen nicht überein: Wenn Ihr Tokenizer zur Chunk-Erstellung und der LLM-Tokenizer nicht übereinstimmen, werden Sie entweder die Tokens überschätzen oder unterschätzen. Verwenden Sie den exakten Tokenizer, den Ihr LLM benötigt.
4. Index-Erstellungszeit & Speicher: Die Verwendung hoher nlist-Werte bei IVF_FLAT-Indizes verbessert den Recall, erhöht jedoch die Latenz und den RAM-Verbrauch. Finden Sie Ihren Sweet Spot. Ich beginne normalerweise bei nlist=128.
5. Chunk-Kohärenz vs. Größe: Größere Stücke halten mehr Kontext, kosten jedoch mehr Tokens. Kleinere Stücke verursachen Fragmentierung und verringern die Precision. Experimentieren Sie.

Voller Code

from transformers import GPT2TokenizerFast
from sentence_transformers import SentenceTransformer
from hashlib import sha256
from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection

# Schritt 1: Initialisierung des Tokenizers und der Modelle
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

# Verbindung zu Milvus herstellen
connections.connect("default", host="localhost", port="19530")

# Schema definieren
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, "Dokumentenstücke mit Token-Metadaten")

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"Die Anzahl der Tokens in der Anfrage ({query_token_count}) überschreitet das Limit ({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

# Beispielnutzung
if __name__ == "__main__":
 raw_text = ("Dies ist ein Beispielabsatz, den wir chunkieren, deduplizieren, einbetten und speichern möchten. " * 100)
 chunks = chunk_text(raw_text)
 insert_chunks(chunks)

 query = "Beispielabsatz Token-Nutzung"
 found = search_similar(query)
 for hit in found:
 print(hit.entity.get("chunk_text"))

Was kommt als Nächstes

Jetzt, wo Sie das Token-Wachstum, das in Milvus speist, gezähmt haben, besteht der logische nächste Schritt darin, die dynamische Anpassung von Anfrage-Prompts umzusetzen – das bedeutet, Ihr Anwendung sollte die kombinierte Tokenlänge (Anfrage plus abgerufener Kontext) überwachen und automatisch weniger wertvolle Chunks entfernen oder umformulieren, bevor Sie Ihr LLM aufrufen. Dies wird Ihnen Geld sparen und verhindern, dass während der Laufzeit Token-Limit-Fehler in der Produktion auftreten.

FAQ

F: Wie kann ich bestätigen, dass meine Token-Anzahlen mit der internen Zählung des LLM übereinstimmen?

A: Die sicherste Methode ist, den von Ihrem LLM bereitgestellten Tokenizer zu verwenden. Für OpenAI-Modelle ist tiktoken der maßgebliche Tokenizer. Der GPT-2 Tokenizer ist ein angemessener Proxy, aber nicht exakt. Führen Sie immer Testfälle mit der Zählung Ihres Modells durch, um dies zu überprüfen.

F: Wie viele Vektoren kann Milvus maximal verarbeiten, bevor die Leistung leidet?

A: Milvus ist für Millionen von Vektoren optimiert, aber praktisch bestimmen Ihr Index-Typ, die Vektor-Dimension und die Hardware die Leistung. Zum Beispiel ist IVF_FLAT mit nlist=128 bei ein paar Millionen Vektoren auf anständigen Servern handhabbar, aber Latenz und RAM können ohne Batchverarbeitung und Pruning steigen.

F: Kann ich das Token-Pruning bei der Einfügezeit automatisieren?

A: Absolut, aber seien Sie vorsichtig. Sie können Chunks, die die Token-Grenzen überschreiten, vor dem Einbetten entfernen oder zusammenfassen, aber übermäßiges Pruning verringert die semantische Tiefe, was die Qualität der nachgelagerten Suche beeinträchtigen kann. Verwenden Sie adaptive Schwellenwerte, die auf Ihrem Datensatz abgestimmt sind.

Milvus Statistiken Übersicht

Metrik	Wert	Kommentar
GitHub Sterne	43,421	Zeigt hohe Akzeptanz und Unterstützung der Community an
Forks	3,909	Demonstriert aktive Beiträge und benutzerdefinierte Anwendungsfälle
Offene Probleme	1,098	Hinweis auf laufende Entwicklung und Fehlerverfolgung
Lizenz	Apache-2.0	Erlaubende Lizenz, die für die Nutzung in Unternehmen günstig ist
Letzte Aktualisierung	21. März 2026	Projekt wird aktiv gewartet

Empfehlungen für verschiedene Entwicklertypen

1. Der Startup-Hacker: Wenn Sie schnell MVPs erstellen, konzentrieren Sie sich auf handelsübliche Einbettungsmodelle wie 'all-MiniLM-L6-v2' und grundlegende Token-Chunks, um Abweichungen und Kosten gering zu halten. Nutzen Sie die integrierte Indizierung von Milvus und behalten Sie Ihre Token-Nutzung mit einfachen Zählern im Auge.

2. Der Datenwissenschaftler: Experimentieren Sie mit semantischen Chunking-Ansätzen – versuchen Sie die Erkennung von Satzgrenzen oder die Kodierung von Absätzen –, um die Treue der Einbettung zu verbessern. Integrieren Sie Metadaten zur Tokenanzahl für das Pruning zur Abfragezeit. Sie könnten auch in Betracht ziehen, benutzerdefinierte Feineinstellungen der Einbettung basierend auf der Komplexität der Chunk-Tokens vorzunehmen.

3. Der Unternehmensingenieur: Bauen Sie eine adaptive Pipeline, die die Echtzeitüberwachung des Tokenbudgets, die Deduplication von Chunks, die dynamische Vektordimensionalität und die Indexanpassung in Milvus umfasst. Integrieren Sie dies eng mit Ihren LLM-Pipelines, um Überlaufszenarien zu verhindern und die Rechenkosten zu optimieren.

Daten vom 21. März 2026. Quellen: https://github.com/milvus-io/milvus, Milvus LangChain Token Limits, Milvus LLM Optimization

Verwandte Artikel

• Containerisierung von Agenten mit Docker Compose
• Katastrophenwiederherstellung bei der Bereitstellung von KI-Agenten
• Horizontale Skalierung von KI-Agenten

🕒 Published: March 28, 2026