Cómo Optimizar el Uso de Tokens con Milvus (Paso a Paso)
Manejar el uso de tokens de manera eficiente con Milvus puede reducir costos computacionales innecesarios y hacer que tus embeddings —y por lo tanto, tu búsqueda vectorial— sean mucho más rápidos e inteligentes. Mientras que muchas personas tratan “milvus optimize token usage” como una caja negra, voy a mostrarte exactamente cómo puedes reducir el exceso de tokens en tus tuberías RAG, búsqueda vectorial y consultas posteriores sin sacrificar precisión.
Requisitos Previos
- Python 3.11+
- Servidor Milvus 2.2.9+ (última versión estable a partir del 21 de marzo de 2026)
pymilvus>=2.2.9- Familiaridad básica con conceptos de embeddings y búsqueda vectorial
- Acceso a codificación de vectores basada en GPU o CPU (como embeddings de OpenAI, modelos de Huggingface o similares)
- Familiaridad con los límites de tokens de tus LLMs (por ejemplo, los 8k tokens de GPT-4) y cómo afectan al costo y la latencia
Lo que Realmente Estás Construyendo
Estamos creando una tubería de búsqueda vectorial que elimina lo innecesario de tus entradas de texto para que Milvus solo almacene lo que realmente enriquece el contexto de tu consulta, todo mientras se mantiene la calidad del embedding. Si alguna vez has introducido todos tus documentos fuente directamente en Milvus y has visto cómo los costos y el conteo de tokens explotan, esta es la solución.
Paso a Paso
Paso 1: Mide Tus Tokens Antes de Comprometerte
from transformers import GPT2TokenizerFast
tokenizer = GPT2TokenizerFast.from_pretrained("gpt2")
def count_tokens(text: str) -> int:
return len(tokenizer.encode(text))
sample_text = "Este párrafo será tokenizado y contado para evitar gastar tokens innecesarios."
print(f"Token count: {count_tokens(sample_text)}")Por qué: Si envías textos masivos tal cual a la etapa de embedding, estás quemando tokens por los que no necesitas pagar o almacenar. El tokenizador GPT-2 es un proxy barato y fácil que se mapea aproximadamente a los conteos de tokens al estilo OpenAI. Esta etapa de conteo inicial evita que fragmentos demasiado largos se cuelen en Milvus.
Errores que encontrarás: Usar un tokenizador que no coincide con tu LLM lleva a contajes erróneos. Por ejemplo, los tokenizadores de Huggingface para T5 difieren significativamente de los tokenizadores de GPT-3/4. Siempre verifica qué tokenizador alinea con el uso de tu modelo.
Paso 2: Divide el Texto de Manera Inteligente – Ve por lo Semántico en Lugar de lo Estático
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) # Ejemplo de texto largo
split_chunks = chunk_text(long_text, max_tokens=500)
print(f"Creado {len(split_chunks)} fragmentos.")Por qué: Dividir por conteos de tokens en lugar de longitudes fijas de caracteres evita sobrepasar accidentalmente los límites de tokens en el momento de hacer embedding o consultar. He visto tuberías fallar o degradarse porque los conteos de tokens aumentaron inesperadamente cuando aparecieron espacios o caracteres UTF-8. El fragmentado semántico (como los límites de oraciones o saltos de párrafo) a menudo funciona mejor, pero un simple maxeo de tokens funciona de manera confiable.
Errores que encontrarás: Cortes de caracteres ingenuos crean coincidencias de consulta pobres: los fragmentos de contexto no representan un significado coherente. Fragmentos de gran tamaño causan errores en la API de embedding o te sacan rápidamente de los niveles gratuitos.
Paso 3: Desduplicar Antes de Enviar a 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"Desduplicado a {len(unique_chunks)} fragmentos únicos.")Por qué: La redundancia es el enemigo. No puedo enfatizar esto lo suficiente: muchos conjuntos de datos reales tienen repeticiones extrañas, ya sean PDFs corrompidos o registros. La deduplicación evita el desperdicio de tokens de embedding y almacenamiento en Milvus, ahorrando costos de cómputo y previniendo ruido de búsqueda más adelante.
Errores que encontrarás: Saltar la deduplicación llena a Milvus de vectores duplicados, ralentiza la búsqueda e inflaciona el almacenamiento. Tu presupuesto basado en tokens se disparará.
Paso 4: Codificar Fragmentos en Vectores de Manera Eficiente
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"Producción de forma de embeddings: {embeddings.shape}")Por qué: Elige modelos de embedding más pequeños y rápidos a menos que necesites específicamente los transformers más grandes para precisión semántica. Para la mayoría de las aplicaciones, modelos como “all-MiniLM-L6-v2” logran el mejor compromiso entre dimensión del vector (384), velocidad y presupuesto de tokens. Los embeddings de alta dimensión no siempre son mejores; pueden inflar tu índice de Milvus y ralentizar la búsqueda.
Errores que encontrarás: Intentar embedding de OpenAI para miles de fragmentos largos sin preprocesamiento quemará tokens y alcanzará rápidamente los límites de tasa de la API. Además, hacer embedding sin agrupamiento reduce el rendimiento.
Paso 5: Almacenar con Metadatos para Filtrar el Contexto
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, "Fragmentos de documentos con metadatos 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()Por qué: Almacenar los conteos de tokens junto con los embeddings te permite filtrar o clasificar fragmentos de manera económica sin necesidad de re-tokenizar más tarde. Los metadatos de tokens disminuyen las consultas que intentan acumular demasiado contexto y te dan control sobre el tamaño de la carga útil de Milvus en el momento de la búsqueda.
Paso 6: Consultar Teniedo en Cuenta los Presupuestos 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 consulta excede el presupuesto 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 = "Uso eficiente de tokens con Milvus"
result_docs = search_similar(query)
for hit in result_docs:
print(hit.entity.get("chunk_text"))Por qué: La ventana de contexto de tu LLM es valiosa. Si olvidas verificar los tokens de tu consulta más los tokens de fragmentos relevantes, sobrepasas tus límites, lo que lleva a errores o mensajes truncados. El filtrado de Milvus basado en los metadatos de tokens almacenados te ayuda a mantenerte dinámicamente dentro del presupuesto.
Errores que encontrarás: Pasar conjuntos de tokens combinados demasiado grandes a tu generador lleva a completaciones fallidas o saltos extraños de contexto. Una vez, tuve un sistema que se bloqueó después de ignorar los límites de tokens. No es divertido.
Las Trampas
- Contar Tokens en la API de Embedding te Está Afectando: Los embeddings de OpenAI cuentan tokens que no siempre esperas, como tokens de indicación implícitos o separadores. Siempre haz pruebas previas con conteos por fragmento antes de realizar embedding en masa.
- Los Costos de Almacenamiento de Milvus Aumentan Rápidamente: El repositorio de Milvus con licencia Apache-2.0 milvus-io/milvus tiene 43,421 estrellas: sí, es popular, pero la dimensión del vector y el número de vectores que almacenas causan un rápido uso de RAM/almacenamiento. Los vectores de gran tamaño sin poda de tokens inflan los costos.
- Los Tokenizadores No Coinciden: Si tu tokenizador de creación de fragmentos y el tokenizador de LLM no coinciden, sobreestimarás o subestimarás los tokens. Usa el tokenizador exacto que tu LLM requiere.
- Tiempo y Memoria de Creación de Índices: Usar valores altos de nlist en índices IVF_FLAT mejora el recuerdo, pero añade latencia y consumo de RAM. Encuentra tu punto óptimo. Yo normalmente empiezo en nlist=128.
- Coherencia del Fragmento vs Tamaño: Los fragmentos más grandes retienen más contexto pero cuestan más tokens. Los fragmentos más pequeños causan fragmentación y reducen la precisión. Experimenta.
Código Completo
from transformers import GPT2TokenizerFast
from sentence_transformers import SentenceTransformer
from hashlib import sha256
from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection
# Paso 1: Inicializar el tokenizador y los modelos
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
# Conectar a Milvus
connections.connect("default", host="localhost", port="19530")
# Definir esquema
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, "Fragmentos de documento con metadatos de tokens")
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"El conteo de tokens de la consulta ({query_token_count}) supera el límite ({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
# Ejemplo de uso
if __name__ == "__main__":
raw_text = ("Este es un párrafo de ejemplo que nos gustaría dividir, desduplicar, incrustar y almacenar. " * 100)
chunks = chunk_text(raw_text)
insert_chunks(chunks)
query = "uso de tokens de párrafo de ejemplo"
found = search_similar(query)
for hit in found:
print(hit.entity.get("chunk_text"))
Qué sigue
Ahora que has controlado el crecimiento de los tokens que se envían a Milvus, el siguiente paso lógico es implementar un recorte dinámico de los prompts de consulta, es decir, tu aplicación debería monitorear la longitud combinada de los tokens (consulta más contexto recuperado) y eliminar o parafrasear fragmentos de bajo valor automáticamente antes de llamar a tu LLM. Esto te ahorrará dinero y evitará errores de límite de tokens en tiempo de ejecución en producción.
Preguntas Frecuentes
Q: ¿Cómo puedo confirmar que mis conteos de tokens coinciden con el conteo interno de la LLM?
A: La opción más segura es usar el tokenizador proporcionado por tu LLM. Para los modelos de OpenAI, tiktoken es el tokenizador canónico. El tokenizador GPT-2 es un proxy razonable, pero no exacto. Siempre ejecuta casos de prueba con el conteo de tu modelo para verificar.
Q: ¿Cuál es el número máximo de vectores que Milvus puede manejar antes de que el rendimiento se vea afectado?
A: Milvus está optimizado para millones de vectores, pero en la práctica, tu tipo de índice, dimensión del vector y hardware dictan el rendimiento. Por ejemplo, IVF_FLAT con nlist=128 es manejable en unos pocos millones de vectores en servidores decentes, pero la latencia y la RAM pueden dispararse sin procesamiento por lotes y poda.
Q: ¿Puedo automatizar la poda de tokens en el momento de la inserción?
A: Absolutamente, pero ten cuidado. Puedes eliminar o resumir fragmentos que superen los límites de token antes de la incrustación, pero una poda excesiva reduce la riqueza semántica, perjudicando la calidad de búsqueda posterior. Utiliza umbrales adaptativos ajustados a tu conjunto de datos.
Resumen de Estadísticas de Milvus
| Métrica | Valor | Comentario |
|---|---|---|
| Estrellas en GitHub | 43,421 | Indica alta adopción y soporte de la comunidad |
| Forks | 3,909 | Demuestra contribución activa y casos de uso personalizados |
| Problemas abiertos | 1,098 | Señala el desarrollo continuo y el seguimiento de errores |
| Licencia | Apache-2.0 | Licencia permisiva favorable para uso empresarial |
| Última actualización | 21 de marzo de 2026 | El proyecto se mantiene activamente |
Recomendaciones para Diferentes Tipos de Desarrolladores
1. El Hacker de Startups: Si estás construyendo MVPs rápidamente, enfócate en modelos de incrustación listos para usar como 'all-MiniLM-L6-v2' y en el procesamiento básico de tokens para mantener bajos el desvío y los costos. Usa la indexación incorporada de Milvus y controla tu uso de tokens con contadores sencillos.
2. El Científico de Datos: Experimenta con enfoques de agrupación semántica: prueba la detección de límites de oración o la codificación de párrafos para mejorar la fidelidad de la incrustación. Incorpora metadatos de conteo de tokens para la poda en el momento de la consulta. También podrías considerar el ajuste fino de incrustaciones personalizadas basado en la complejidad de los tokens de los fragmentos.
3. El Ingeniero Empresarial: Construye un pipeline adaptativo que incorpore monitoreo en tiempo real del presupuesto de tokens, desduplicación de fragmentos, dimensionalidad vectorial dinámica y ajuste de índices en Milvus. Integra estrechamente con tus pipelines de LLM para prevenir escenarios de sobrecarga y optimizar el gasto computacional.
Datos a partir del 21 de marzo de 2026. Fuentes: https://github.com/milvus-io/milvus, Límites de Tokens de Milvus LangChain, Optimización de LLM de Milvus
Artículos Relacionados
- Contenerizando Agentes con Docker Compose
- Recuperación de desastres en el despliegue de agentes de IA
- Escalando agentes de IA horizontalmente
🕒 Published: