Wie man die Nutzung von Tokens mit Milvus optimiert (Schritt für Schritt)

📖 12 min read•2,312 words•Updated Mar 29, 2026

Wie man die Nutzung von Tokens mit Milvus optimiert (Schritt für Schritt)

Die effektive Verwaltung der Token-Nutzung mit Milvus kann unnötige Rechenkosten senken und Ihre Embeddings—und damit Ihre Vektorsuche—viel schneller und intelligenter machen. Obwohl viele „milvus optimieren die Nutzung von tokens“ als Black Box betrachten, werde ich Ihnen genau zeigen, wie Sie den Tokenfluss in Ihren RAG-Pipelines, Ihrer Vektorsuche und Ihren nachgelagerten Abfragen reduzieren können, ohne die Genauigkeit zu opfern.

Voraussetzungen

Python 3.11+
Milvus Server 2.2.9+ (letzte stabile Version ab dem 21. März 2026)
pymilvus>=2.2.9
Grundkenntnisse der Konzepte von Embeddings und Vektorsuche
Zugang zu einer GPU- oder CPU-basierten Vektorisierung (wie OpenAI-Embeddings, Huggingface-Modelle oder ähnliche)
Vertrautheit mit den Token-Limits Ihrer LLMs (z. B. die 8k Tokens von GPT-4) und wie sie die Kosten und Latenz beeinflussen

Was Sie Wirklich Erstellen

Wir entwerfen eine Vektorsuch-Pipeline, die das Überflüssige aus Ihren Texteingaben entfernt, sodass Milvus nur das behält, was den Kontext Ihrer Abfrage tatsächlich bereichert, während die Qualität der Embeddings erhalten bleibt. Wenn Sie bereits alle Ihre Quelldokumente direkt in Milvus gesendet haben und die Explosion der Kosten und Token-Zählungen beobachtet haben, ist dies die Lösung.

Schritt für Schritt

Schritt 1: Messen Sie Ihre Tokens, bevor Sie validieren

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 zu vermeiden, dass Tokens verschwendet werden."
print(f"Token-Zählung: {count_tokens(sample_text)}")

Warum: Wenn Sie blind große Texte unverändert in der Embedding-Phase senden, verbrennen Sie Tokens, für die Sie nicht bezahlen oder die Sie nicht speichern müssen. Der GPT-2-Tokenizer ist ein kostengünstiger und einfacher Proxy, der ungefähr mit den Token-Zählungen im OpenAI-Stil übereinstimmt. Dieser erste Zähl-Schritt verhindert, dass zu lange Stücke in Milvus eindringen.

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

Schritt 2: Teilen Sie den Text intelligent auf – Wählen Sie Semantik statt Statik

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

Warum: Das Teilen nach Token-Zählungen anstelle von festen Zeichenlängen verhindert, dass Sie versehentlich die Token-Limits in der Embedding- oder Abfrage-Phase überschreiten. Ich habe gesehen, dass Pipelines abstürzen oder sich verschlechtern, weil die Token-Zählungen unerwartet explodiert sind, wenn Leerzeichen oder UTF-8-Zeichen auftraten. Semantisches Chunking (wie Satzgrenzen oder Absatzumbrüche) funktioniert oft besser, aber eine einfache Maximierung der Tokens funktioniert zuverlässig.

Fehler, die Sie antreffen werden: Naive Zeichenaufteilungen führen zu schlechten Abfrageübereinstimmungen—die Kontextfragmente repräsentieren keinen kohärenten Sinn. Zu große Stücke verursachen Fehler in der Embedding-API oder bringen Sie schnell über die kostenlosen Kontingente hinaus.

Schritt 3: Duplizieren Sie vor dem Pushen zu 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"Dedupliziert auf {len(unique_chunks)} einzigartige Stücke.")

Warum: Redundanz ist der Feind. Ich kann nicht genug betonen—viele reale Datensätze weisen seltsame Wiederholungen auf, sei es durch beschädigte PDFs oder Protokolle. Die Duplikation vermeidet das Verschwenden von Embedding- und Speicher-Tokens in Milvus, was Ihnen Rechenkosten spart und späteren Suchrauschen verhindert.

Fehler, die Sie antreffen werden: Die Duplikation zu überspringen, füllt Milvus mit doppelten Vektoren, verlangsamt die Suche und bläht den Speicher auf. Ihr tokenbasierter Budget wird explodieren.

Schritt 4: Kodieren Sie die Stücke effizient in Vektoren

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

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

Fehler, die Sie antreffen werden: Der Versuch, OpenAI-Embeddings für Tausende von langen Stücken ohne Vorverarbeitung zu verwenden, verbrennt Tokens und erreicht schnell die API-Rate-Limits. Außerdem verringert das Embedding ohne Batch-Verarbeitung den Durchsatz.

Schritt 5: Speichern Sie mit Metadaten, 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, "Dokumentenstücke 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 der Token-Zählungen neben den Embeddings ermöglicht es Ihnen, die Stücke kostengünstig zu filtern oder zu sortieren, ohne später erneut tokenisieren zu müssen. Die Token-Metadaten reduzieren die Abfragen, die versuchen, zu viel Kontext zu komprimieren—und geben Ihnen die Kontrolle über die Größe der Nutzlast von Milvus zum Zeitpunkt der Suche.

Schritt 6: Abfragen unter Berücksichtigung der 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"Die Abfrage überschreitet das 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 Nutzung von Tokens mit Milvus"
result_docs = search_similar(query)
for hit in result_docs:
 print(hit.entity.get("chunk_text"))

Warum: Das Kontextfenster Ihres LLM ist wertvoll. Wenn Sie vergessen, Ihre Abfrage-Tokens plus die Tokens relevanter Stücke zu überprüfen, überschreiten Sie Ihre Grenzen—was zu Fehlern oder abgeschnittenen Eingaben führt. Das Filtern von Milvus basierend auf den gespeicherten Token-Metadaten hilft Ihnen, dynamisch unter Budget zu bleiben.

Fehler, die Sie antreffen werden: Das Übergeben von zu großen kombinierten Token-Sets an Ihren Generator führt zu Completion-Fehlern oder seltsamen Kontextwechseln. Einmal hatte ich einen Systemabsturz, nachdem ich die Token-Limits ignoriert hatte. Nicht lustig.

Die Fallstricke

Die Token-Zählung der Embedding-API wird Ihnen schaden: Die OpenAI-Embeddings zählen Tokens, die Sie nicht immer erwarten, wie implizite Eingabetokens oder Trennzeichen. Führen Sie immer Trockenversuche mit Stückzählungen durch, bevor Sie in großen Mengen einbetten.
Die Speicherkosten von Milvus steigen schnell: Das Milvus-Repository unter der Apache-2.0-Lizenz milvus-io/milvus hat 43.421 Sterne – ja, es ist beliebt – aber die Vektordimension und die Anzahl der Vektoren, die Sie speichern, führen zu einem schnellen Verbrauch von RAM/Speicher. Überdimensionierte Vektoren ohne Token-Pruning erhöhen die Kosten.
Die Tokenizer stimmen nicht überein: Wenn der Tokenizer für die Erstellung von Stücken und der Tokenizer Ihres LLM nicht übereinstimmen, werden Sie entweder die Tokens überschätzen oder unterschätzen. Verwenden Sie den genauen Tokenizer, den Ihr LLM benötigt.
Indexierungszeit & Speicher: Die Verwendung großer nlist-Werte in IVF_FLAT-Indizes verbessert den Rückruf, erhöht jedoch die Latenz und den RAM-Verbrauch. Finden Sie Ihr Gleichgewicht. Ich beginne normalerweise mit nlist=128.
Stückkohärenz vs. Größe: Größere Stücke enthalten mehr Kontext, kosten jedoch mehr Tokens. Kleinere Stücke verursachen Fragmentierung und verringern die Genauigkeit. Experimentieren Sie.

Vollständiger Code

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

# Schritt 1: Tokenizer und Modelle initialisieren
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

# Mit Milvus verbinden
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, "Dokumentstü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 Abfrage ({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

# Beispielverwendung
if __name__ == "__main__":
 raw_text = ("Dies ist ein Beispiel für einen Absatz, den wir gerne aufteilen, deduplizieren, einbetten und speichern würden. " * 100)
 chunks = chunk_text(raw_text)
 insert_chunks(chunks)

 query = "Nutzung von Tokens in einem Beispielabsatz"
 found = search_similar(query)
 for hit in found:
 print(hit.entity.get("chunk_text"))

Und jetzt

Jetzt, da Sie die Token-Proliferation in Milvus gemeistert haben, besteht der nächste logische Schritt darin, ein dynamisches Chunking von Abfrageaufforderungen zu implementieren, was bedeutet, dass Ihre Anwendung die kombinierte Länge der Tokens (Abfrage plus abgerufener Kontext) überwachen und automatisch weniger wertvolle Stücke vor dem Aufruf Ihres LLM entfernen oder umformulieren muss. Das spart Ihnen Geld und vermeidet Token-Limitfehler bei der Ausführung in der Produktion.

FAQ

F: Wie kann ich bestätigen, dass meine Token-Zählungen mit denen des LLM übereinstimmen?

A: Der beste Ansatz ist, den Tokenizer zu verwenden, der von Ihrem LLM bereitgestellt wird. Für OpenAI-Modelle ist tiktoken der kanonische Tokenizer. Der GPT-2-Tokenizer ist ein vernünftiger, aber nicht genauer Proxy. 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 verwalten, bevor die Leistung leidet?

A: Milvus ist für Millionen von Vektoren optimiert, aber in der Praxis bestimmen Ihr Index-Typ, die Vektordimension und die Hardware die Leistung. Zum Beispiel ist IVF_FLAT mit nlist=128 auf einigen Millionen Vektoren auf geeigneten Servern handhabbar, aber die Latenz und der RAM können ohne Batch-Verarbeitung und Pruning steigen.

F: Kann ich das Pruning von Tokens beim Einfügen automatisieren?

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

Zusammenfassung der Milvus-Statistiken

Metrik	Wert	Kommentar
GitHub-Sterne	43.421	Zeigt eine hohe Akzeptanz und Unterstützung durch die Community an
Forks	3.909	Zeigt aktive Beiträge und benutzerdefinierte Anwendungsfälle
Offene Probleme	1.098	Zeigt laufende Entwicklungen und Bug-Tracking an
Lizenz	Apache-2.0	Permissive Lizenz, die eine Nutzung im Unternehmensumfeld begünstigt
Letzte Aktualisierung	21. März 2026	Das Projekt wird aktiv gewartet

Empfehlungen für verschiedene Entwicklertypen

1. Der Startup-Hacker: Wenn Sie MVPs schnell erstellen, konzentrieren Sie sich auf einsatzbereite Integrationsmodelle wie 'all-MiniLM-L6-v2' und eine einfache Token-Zerlegung, um Drift und Kosten zu reduzieren. Nutzen Sie die integrierte Indizierung von Milvus und überwachen Sie Ihre Token-Nutzung mit einfachen Zählern.

2. Der Data Scientist: Experimentieren Sie mit semantischen Chunking-Ansätzen: Versuchen Sie die Satzgrenzenerkennung oder die Kodierung von Absätzen, um die Integrationsgenauigkeit zu verbessern. Integrieren Sie Token-Zählmetadaten für das Pruning zur Abfragezeit. Sie sollten auch eine Feinabstimmung der Integration basierend auf der Token-Komplexität der Stücke in Betracht ziehen.

3. Der Unternehmensingenieur: Bauen Sie eine adaptive Pipeline, die eine Echtzeitüberwachung des Token-Budgets, die De-Duplizierung von Stücken, die dynamische Vektordimensionalität und die Indexanpassung in Milvus umfasst. Integrieren Sie Ihre LLM-Pipelines eng, um Überlauf-Szenarien zu vermeiden und die Rechenkosten zu optimieren.

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

Wie man die Nutzung von Tokens mit Milvus optimiert (Schritt für Schritt)