Como Otimizar o Uso dos Tokens com Milvus (Passo a Passo)
Gerenciar o uso dos tokens de forma eficiente com Milvus pode reduzir custos computacionais desnecessários e tornar suas embeddings—e, portanto, sua pesquisa vetorial—muito mais rápidas e inteligentes. Enquanto muitos consideram “milvus otimizar o uso dos tokens” como uma caixa preta, vou mostrar exatamente como reduzir a sobrecarga de tokens em seus pipelines RAG, na pesquisa vetorial e nas consultas downstream sem sacrificar a precisão.
Pré-requisitos
- Python 3.11+
- Milvus Server 2.2.9+ (última versão estável em 21 de março de 2026)
pymilvus>=2.2.9- Conhecimento básico de embeddings e conceitos de pesquisa vetorial
- Acesso à codificação vetorial em GPU ou CPU (como OpenAI embeddings, modelos Huggingface ou similares)
- Familiaridade com os limites de tokens dos seus LLM (por exemplo, os 8k tokens do GPT-4) e como isso influencia custos e latência
O Que Você Está Realmente Construindo
Estamos criando um pipeline de pesquisa vetorial que elimina o supérfluo de seus inputs textuais, de modo que Milvus armazene apenas o que realmente enriquece o contexto da sua consulta, equilibrando ao mesmo tempo a qualidade das embeddings. Se você já enviou todos os seus documentos fonte diretamente para Milvus e viu os custos e a contagem de tokens explodirem, essa é a solução.
Passo a Passo
Passo 1: Meça Seus Tokens Antes de Prosseguir
from transformers import GPT2TokenizerFast
tokenizer = GPT2TokenizerFast.from_pretrained("gpt2")
def count_tokens(text: str) -> int:
return len(tokenizer.encode(text))
sample_text = "Este parágrafo será tokenizado e contado para evitar desperdiçar tokens."
print(f"Contagem de tokens: {count_tokens(sample_text)}")Por quê: Se você enviar cegamente textos massivos assim como estão para a fase de embedding, está queimando tokens que não precisa pagar ou armazenar. O tokenizer GPT-2 é um proxy econômico e simples que mapeia grosseiramente as contagens de tokens no estilo OpenAI. Esta fase inicial de contagem impede que você envie partes excessivamente longas para Milvus.
Erros que você pode encontrar: Usar um tokenizer que não corresponde ao seu LLM leva a contagens incorretas, tanto em excesso quanto em falta. Por exemplo, os tokenizers da Huggingface para T5 diferem significativamente dos tokenizers do GPT-3/4. Sempre verifique qual tokenizer se alinha ao uso do seu modelo.
Passo 2: Divida o Texto de Forma Inteligente – Escolha Semântico em vez de 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(["palavra"] * 2000) # Exemplo de texto longo
split_chunks = chunk_text(long_text, max_tokens=500)
print(f"Criados {len(split_chunks)} chunks.")Por quê: Dividir por contagens de tokens em vez de comprimentos de caracteres fixos impede que você ultrapasse acidentalmente os limites de tokens no momento da embedding ou da consulta. Já vi pipelines travarem ou degradassem porque as contagens de tokens aumentaram inesperadamente quando surgiram espaços ou caracteres UTF-8. A divisão semântica (como os limites de frases ou quebras de parágrafo) geralmente funciona melhor, mas um máximo de tokens mais simples funciona de forma confiável.
Erros que você pode encontrar: Simples divisões por caracteres criam correspondências ruins nas consultas — os fragmentos de contexto não representam um significado coerente. Chunks excessivamente grandes causam erros na API de embedding ou rapidamente o tiram dos níveis gratuitos.
Passo 3: Deduplicate Antes de Enviar para Milvus
“`html
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"Deduplicato a {len(unique_chunks)} chunk unici.")Por que: A redundância é o inimigo. Não posso enfatizar isso o suficiente—muitos conjuntos de dados reais têm uma repetição estranha, sejam PDFs corrompidos ou arquivos de log. A deduplicação evita desperdiçar embedding e tokens de armazenamento no Milvus, economizando dinheiro em cálculos e prevenindo ruídos de busca posteriormente.
Erros que você pode encontrar: Saltar a deduplicação enche Milvus com vetores duplicados, desacelera a busca e incha o armazenamento. Seu orçamento baseado em tokens explodirá.
Passo 4: Codifique os Chunk em Vetores de Forma 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"Forma das embeddings produzidas: {embeddings.shape}")Por que: Escolha modelos de embedding menores e mais rápidos, a menos que você precise especificamente de modelos de transformadores maiores para precisão semântica. Para a maioria das aplicações, modelos como “all-MiniLM-L6-v2” oferecem o melhor compromisso entre o tamanho dos vetores (384), velocidade e orçamento de tokens. Embeddings de alta dimensão nem sempre são melhores; podem inchá-lo seu índice Milvus e desacelerar a busca.
Erros que você pode encontrar: Tentar usar o embedding do OpenAI para milhares de chunks longos sem pré-processamento consumirá tokens e rapidamente atingirá os limites de frequência da API. Além disso, embeddings sem agrupamento reduzem a taxa de transferência.
Passo 5: Armazene Com Metadados para Filtrar o 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 documento com metadados sobre os 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 que: Armazenar as contagens de tokens junto com as embeddings permite que você filtre ou ordene os chunks de baixo custo sem precisar retokenizar posteriormente. Os metadados sobre os tokens reduzem as consultas que tentam comprimir contexto demais—e lhe dão controle sobre o tamanho do payload do Milvus no momento da busca.
Passo 6: Execute Consultas Levando em Conta os Orçamentos 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"A consulta excede o orçamento 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 dos tokens com Milvus"
result_docs = search_similar(query)
for hit in result_docs:
print(hit.entity.get("chunk_text"))Por que: A janela de contexto do seu LLM é valiosa. Se você esquecer de verificar sua contagem de tokens da consulta mais os tokens dos chunks pertinentes, você ultrapassa seus limites—levando a erros ou prompts truncados. O filtragem de Milvus baseada nos metadados dos tokens armazenados ajuda você a se manter dinamicamente dentro do orçamento.
``````html
Erros que você pode encontrar: Passar conjuntos de tokens combinados muito grandes para o seu gerador leva a completamentos falhados ou saltos de contexto estranhos. Uma vez tive uma falha de sistema após ignorar os limites de tokens. Não foi divertido.
Os Perigos
- A Contagem de Tokens na API de Embedding Está Te Prejudicando: As embeddings da OpenAI contam tokens que você nem sempre espera, como tokens de prompt implícitos ou separadores. Sempre faça testes preliminares com as contagens por chunk antes do embedding em massa.
- Os Custos de Armazenamento do Milvus Crescem Rapidamente: O repositório do Milvus, licenciado sob Apache-2.0 milvus-io/milvus, tem 43.421 estrelas—sim, é popular—mas o tamanho do vetor e o número de vetores que você armazena causam um rápido uso de RAM/armazenamento. Vetores excessivamente grandes sem poda de tokens aumentam os custos.
- Os Tokenizers Não Concordam: Se o tokenizer para a criação dos chunks e o tokenizer LLM não correspondem, você superestimará ou subestimará os tokens. Use o tokenizer exato necessário para o seu LLM.
- Tempo de Criação do Índice e Memória: Usar valores nlist altos nos índices IVF_FLAT melhora a recuperação, mas adiciona latência e uso de RAM. Encontre seu ponto ideal. Normalmente começo com nlist=128.
- Consistência do Chunk vs Tamanho: Chunks maiores contêm mais contexto, mas custam mais tokens. Chunks menores causam fragmentação e reduzem a precisão. Experimente.
Código Completo
```
from transformers import GPT2TokenizerFast
from sentence_transformers import SentenceTransformer
from hashlib import sha256
from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection
# Etapa 1: Inicializar o tokenizer e 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 ao Milvus
connections.connect("default", host="localhost", port="19530")
# Definir o 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, "Folhas de documentos com metadados sobre os 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"O número de tokens da consulta ({query_token_count}) excede o 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
# Exemplo de uso
if __name__ == "__main__":
raw_text = ("Este é um exemplo de parágrafo que gostaríamos de fragmentar, deduplicar, incorporar e armazenar. " * 100)
chunks = chunk_text(raw_text)
insert_chunks(chunks)
query = "exemplo de uso dos tokens no parágrafo"
found = search_similar(query)
for hit in found:
print(hit.entity.get("chunk_text"))
O que vem a seguir
Agora que você domou a expansão de tokens que alimenta o Milvus, o próximo passo lógico é implementar um algoritmo de recorte dinâmico da consulta—o que significa que seu aplicativo deve monitorar o comprimento combinado dos tokens (consulta mais contexto recuperado) e remover ou parafrasear automaticamente os fragmentos de baixo valor antes de chamar seu LLM. Isso economizará dinheiro e previne erros de limite de tokens em produção.
FAQ
P: Como posso confirmar que minhas contagens de tokens correspondem às internas do LLM?
A: A aposta mais segura é usar o tokenizer fornecido pelo seu LLM. Para os modelos OpenAI, tiktoken é o tokenizer canônico. O tokenizer GPT-2 é um proxy razoável, mas não exato. Sempre execute casos de teste com a contagem do seu modelo para verificar.
P: Qual é o número máximo de vetores que o Milvus pode gerenciar antes que o desempenho seja afetado?
A: O Milvus é otimizado para milhões de vetores, mas na prática, o tipo de índice, o tamanho do vetor e o hardware determinam o desempenho. Por exemplo, IVF_FLAT com nlist=128 é gerenciável com alguns milhões de vetores em servidores decentes, mas a latência e a RAM podem aumentar sem batching e poda.
P: Posso automatizar a poda dos tokens no momento da inserção?
A: Absolutamente, mas tenha cuidado. Você pode eliminar ou resumir fragmentos que excedem os limites de tokens antes da incorporação, mas uma poda excessiva reduz a riqueza semântica, prejudicando a qualidade da pesquisa subsequente. Use limiares adaptativos calibrados com seu conjunto de dados.
Visão geral das estatísticas do Milvus
| Métricas | Valor | Comentário |
|---|---|---|
| Estrelas no GitHub | 43.421 | Indica alta adoção e suporte da comunidade |
| Forks | 3.909 | Demonstra uma contribuição ativa e casos de uso personalizados |
| Problemas abertos | 1.098 | Indica desenvolvimento e rastreamento de bugs em andamento |
| Licença | Apache-2.0 | Licença permissiva favorável ao uso empresarial |
| Última atualização | 21 de março de 2026 | O projeto é ativamente mantido |
Recomendações para diferentes tipos de desenvolvedores
1. O Hacker de Startup: Se você está criando MVPs rápidos, concentre-se em modelos de incorporação prontos para uso, como 'all-MiniLM-L6-v2', e em uma simples fragmentação dos tokens para manter o drift e os custos baixos. Use a indexação integrada do Milvus e fique de olho no uso dos tokens com contadores simples.
2. O Cientista de Dados: Experimente abordagens de fragmentação semântica—tente a detecção de limites de frases ou a codificação de parágrafos—para melhorar a fidelidade da incorporação. Incorpore os metadados sobre a contagem de tokens para a poda no momento da consulta. Você também pode considerar o fine-tuning personalizado da incorporação com base na complexidade dos tokens dos fragmentos.
3. O Engenheiro Empresarial: Construa um pipeline adaptativo que incorpora o monitoramento em tempo real do orçamento dos tokens, a deduplicação dos fragmentos, a dimensionalidade dos vetores dinâmica e a regulação dos índices no Milvus. Integre com seus pipelines LLM para prevenir cenários de sobrecarga e otimizar os custos de computação.
Dados atualizados em 21 de março de 2026. Fontes: https://github.com/milvus-io/milvus, Limites de tokens do Milvus LangChain, Otimização dos LLMs do Milvus
Artigos Correlacionados
- Containerizando agentes com Docker Compose
- Recuperação de desastres para o deployment de agentes de IA
- Escalando agentes de IA horizontalmente
🕒 Published: