Con iteradores

Milvus proporciona iteradores de búsqueda y consulta para iterar resultados con un gran volumen de entidades. Dado que Milvus limita TopK a 16384, los usuarios pueden utilizar iteradores para devolver grandes números o incluso entidades enteras en una colección en modo por lotes.

Visión general

Los iteradores son potentes herramientas que le ayudan a iterar a través de un gran volumen de datos o de todos los datos de una colección utilizando valores de clave primaria y expresiones booleanas. Esto puede mejorar significativamente la forma de recuperar datos. A diferencia del uso tradicional de parámetros de desplazamiento y límite, que pueden perder eficacia con el tiempo, los iteradores ofrecen una solución más escalable.

Ventajas del uso de iteradores

• Simplicidad: Elimina los complejos parámetros offset y limit.

• Eficacia: Proporciona una recuperación de datos escalable al obtener sólo los datos necesarios.

• Coherencia: Garantiza un tamaño coherente del conjunto de datos con filtros booleanos.

Preparativos

Los siguientes pasos reutilizan el código para conectarse a Milvus, configurar rápidamente una colección e insertar más de 10.000 entidades generadas aleatoriamente en la colección.

Paso 1: Crear una colección

from pymilvus import MilvusClient

# 1. Set up a Milvus client
client = MilvusClient(
    uri="http://localhost:19530"
)

# 2. Create a collection
client.create_collection(
    collection_name="quick_setup",
    dimension=5,
)

import io.milvus.client.MilvusServiceClient;
import io.milvus.param.ConnectParam;
import io.milvus.param.highlevel.collection.CreateSimpleCollectionParam;

String CLUSTER_ENDPOINT = "http://localhost:19530";

// 1. Connect to Milvus server
ConnectParam connectParam = ConnectParam.newBuilder()
        .withUri(CLUSTER_ENDPOINT)
        .build();

MilvusServiceClient client  = new MilvusServiceClient(connectParam);

// 2. Create a collection
CreateSimpleCollectionParam createCollectionParam = CreateSimpleCollectionParam.newBuilder()
        .withCollectionName("quick_setup")
        .withDimension(5)
        .build();

client.createCollection(createCollectionParam);

Paso 2: Insertar entidades generadas aleatoriamente

# 3. Insert randomly generated vectors 
colors = ["green", "blue", "yellow", "red", "black", "white", "purple", "pink", "orange", "brown", "grey"]
data = []

for i in range(10000):
    current_color = random.choice(colors)
    current_tag = random.randint(1000, 9999)
    data.append({
        "id": i,
        "vector": [ random.uniform(-1, 1) for _ in range(5) ],
        "color": current_color,
        "tag": current_tag,
        "color_tag": f"{current_color}_{str(current_tag)}"
    })

print(data[0])

# Output
#
# {
#     "id": 0,
#     "vector": [
#         -0.5705990742218152,
#         0.39844925120642083,
#         -0.8791287928610869,
#         0.024163154953680932,
#         0.6837669917169638
#     ],
#     "color": "purple",
#     "tag": 7774,
#     "color_tag": "purple_7774"
# }

res = client.insert(
    collection_name="quick_setup",
    data=data,
)

print(res)

# Output
#
# {
#     "insert_count": 10000,
#     "ids": [
#         0,
#         1,
#         2,
#         3,
#         4,
#         5,
#         6,
#         7,
#         8,
#         9,
#         "(9990 more items hidden)"
#     ]
# }

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;

import com.alibaba.fastjson.JSONObject;

import io.milvus.param.R;
import io.milvus.param.dml.InsertParam;
import io.milvus.response.MutationResultWrapper;
import io.milvus.grpc.MutationResult;


// 3. Insert randomly generated vectors into the collection
List<String> colors = Arrays.asList("green", "blue", "yellow", "red", "black", "white", "purple", "pink", "orange", "brown", "grey");
List<JSONObject> data = new ArrayList<>();

for (int i=0; i<10000; i++) {
    Random rand = new Random();
    String current_color = colors.get(rand.nextInt(colors.size()-1));
    JSONObject row = new JSONObject();
    row.put("id", Long.valueOf(i));
    row.put("vector", Arrays.asList(rand.nextFloat(), rand.nextFloat(), rand.nextFloat(), rand.nextFloat(), rand.nextFloat()));
    row.put("color_tag", current_color + "_" + String.valueOf(rand.nextInt(8999) + 1000));
    data.add(row);
}

InsertParam insertParam = InsertParam.newBuilder()
    .withCollectionName("quick_setup")
    .withRows(data)
    .build();

R<MutationResult> insertRes = client.insert(insertParam);

if (insertRes.getStatus() != R.Status.Success.getCode()) {
    System.err.println(insertRes.getMessage());
}

MutationResultWrapper wrapper = new MutationResultWrapper(insertRes.getData());
System.out.println(wrapper.getInsertCount());

Búsqueda con iterador

Los iteradores hacen que las búsquedas por similitud sean más escalables.

1. Inicialice el iterador de búsqueda para definir los parámetros de búsqueda y los campos de salida.

2. Utilice el método next() dentro de un bucle para paginar los resultados de la búsqueda.

   • Si el método devuelve un array vacío, el bucle finaliza y no hay más páginas disponibles.

   • Todos los resultados llevan los campos de salida especificados.

3. Llame manualmente al método close() para cerrar el iterador una vez que se hayan recuperado todos los datos.

from pymilvus import Collection

# 4. Search with iterator
connections.connect(host="127.0.0.1", port=19530)
collection = Collection("quick_setup")

query_vectors = [[0.3580376395471989, -0.6023495712049978, 0.18414012509913835, -0.26286205330961354, 0.9029438446296592]]
search_params = {
    "metric_type": "IP",
    "params": {"nprobe": 10}
}

iterator = collection.search_iterator(
    data=query_vectors,
    anns_field="vector",
    batch_size=10,
    param=search_params,
    output_fields=["color_tag"],
    limit=3
)

results = []

while True:
    result = iterator.next()
    if not result:
        iterator.close()
        break
        
    results.extend(result)
    
    for hit in result:
        results.append(hit.to_dict())

print(results)

# Output
#
# [
#     {
#         "id": 1756,
#         "distance": 2.0642056465148926,
#         "entity": {
#             "color_tag": "black_9109"
#         }
#     },
#     {
#         "id": 6488,
#         "distance": 1.9437453746795654,
#         "entity": {
#             "color_tag": "purple_8164"
#         }
#     },
#     {
#         "id": 3338,
#         "distance": 1.9107104539871216,
#         "entity": {
#             "color_tag": "brown_8121"
#         }
#     }
# ]

import io.milvus.param.dml.QueryIteratorParam;
import io.milvus.param.dml.SearchIteratorParam;
import io.milvus.response.QueryResultsWrapper;
import io.milvus.orm.iterator.SearchIterator;

// 4. Search with iterators
SearchIteratorParam iteratorParam = SearchIteratorParam.newBuilder()
    .withCollectionName("quick_setup")
    .withVectorFieldName("vector")
    // Use withFloatVectors() in clusters compatible with Milvus 2.4.x
    .withVectors(Arrays.asList(0.3580376395471989f, -0.6023495712049978f, 0.18414012509913835f, -0.26286205330961354f, 0.9029438446296592f))
    .withBatchSize(10L)
    .withParams("{\"metric_type\": \"COSINE\", \"params\": {\"level\": 1}}")
    .build();
        

R<SearchIterator> searchIteratorRes = client.searchIterator(iteratorParam);

if (searchIteratorRes.getStatus() != R.Status.Success.getCode()) {
    System.err.println(searchIteratorRes.getMessage());
}

SearchIterator searchIterator = searchIteratorRes.getData();
List<QueryResultsWrapper.RowRecord> results = new ArrayList<>();

while (true) {
    List<QueryResultsWrapper.RowRecord> batchResults = searchIterator.next();
    if (batchResults.isEmpty()) {
        searchIterator.close();
        break;
    }
    for (QueryResultsWrapper.RowRecord rowRecord : batchResults) {
        results.add(rowRecord);
    }
}

System.out.println(results.size());

Parámetro	Descripción
data	Una lista de incrustaciones vectoriales. Milvus busca las incrustaciones vectoriales más similares a las especificadas.
anns_field	El nombre del campo vectorial en la colección actual.
batch_size	El número de entidades a devolver cada vez que se llama a next() sobre el iterador actual. El valor por defecto es 1000. Ajústelo a un valor adecuado para controlar el número de entidades a devolver por iteración.
param	Los parámetros específicos de esta operación. metric_type: El tipo de métrica aplicado a esta operación. Debe ser el mismo que el utilizado al indexar el campo vectorial especificado anteriormente. Los valores posibles son L2, IP, COSINE, JACCARD, HAMMING. params: Parámetros adicionales. Para más detalles, consulte search_iterator().
output_fields	Una lista de nombres de campo para incluir en cada entidad devuelta. El valor por defecto es None. Si no se especifica, sólo se incluye el campo principal.
limit	El número total de entidades que se devolverán. El valor predeterminado es -1, lo que indica que se devolverán todas las entidades coincidentes.

Parámetro	Descripción
withCollectionName	Establece el nombre de la colección. El nombre de la colección no puede estar vacío ni ser nulo.
withVectorFieldName	Establezca el nombre del campo del vector de destino. El nombre del campo no puede estar vacío ni ser nulo.
withVectors	Defina los vectores de destino. Se permiten hasta 16384 vectores.
withBatchSize	El número de entidades a devolver cada vez que se llama a next() en el iterador actual. El valor por defecto es 1000. Establézcalo a un valor apropiado para controlar el número de entidades a devolver por iteración.
withParams	Especifica los parámetros de búsqueda en formato JSON. Para más información, consulte searchIterator().

Consulta con un iterador

# 6. Query with iterator
iterator = collection.query_iterator(
    batch_size=10, # Controls the size of the return each time you call next()
    expr="color_tag like \"brown_8\"",
    output_fields=["color_tag"]
)

results = []

while True:
    result = iterator.next()
    if not result:
        iterator.close()
        break
        
    results.extend(result)
    
# 8. Check the search results
print(len(results))

print(results[:3])

# Output
#
# [
#     {
#         "color_tag": "brown_8785",
#         "id": 94
#     },
#     {
#         "color_tag": "brown_8568",
#         "id": 176
#     },
#     {
#         "color_tag": "brown_8721",
#         "id": 289
#     }
# ]

import io.milvus.param.dml.QueryIteratorParam;
import io.milvus.orm.iterator.QueryIterator;

// 5. Query with iterators

try {
    Files.write(Path.of("results.json"), JSON.toJSONString(new ArrayList<>()).getBytes(), StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
} catch (Exception e) {
    // TODO: handle exception
    e.printStackTrace();
}

QueryIteratorParam queryIteratorParam = QueryIteratorParam.newBuilder()
    .withCollectionName("quick_setup")
    .withExpr("color_tag like \"brown_8%\"")
    .withBatchSize(50L)
    .addOutField("vector")
    .addOutField("color_tag")
    .build();

R<QueryIterator> queryIteratRes = client.queryIterator(queryIteratorParam);

if (queryIteratRes.getStatus() != R.Status.Success.getCode()) {
    System.err.println(queryIteratRes.getMessage());
}

QueryIterator queryIterator = queryIteratRes.getData();

while (true) {
    List<QueryResultsWrapper.RowRecord> batchResults = queryIterator.next();
    if (batchResults.isEmpty()) {
        queryIterator.close();
        break;
    }

    String jsonString = "";
    List<JSONObject> jsonObject = new ArrayList<>();
    try {
        jsonString = Files.readString(Path.of("results.json"));
        jsonObject = JSON.parseArray(jsonString).toJavaList(null);
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

    for (QueryResultsWrapper.RowRecord queryResult : batchResults) {
        JSONObject row = new JSONObject();
        row.put("id", queryResult.get("id"));
        row.put("vector", queryResult.get("vector"));
        row.put("color_tag", queryResult.get("color_tag"));
        jsonObject.add(row);
    }

    try {
        Files.write(Path.of("results.json"), JSON.toJSONString(jsonObject).getBytes(), StandardOpenOption.WRITE);
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
}

Parámetro	Descripción
batch_size	El número de entidades a devolver cada vez que se llama a next() en el iterador actual. El valor por defecto es 1000. Establézcalo a un valor apropiado para controlar el número de entidades a devolver por iteración.
expr	Una condición de filtrado escalar para filtrar las entidades coincidentes. El valor predeterminado es None, lo que indica que se ignora el filtrado escalar. Para crear una condición de filtrado escalar, consulte Reglas de expresión booleana.
output_fields	Una lista de nombres de campo para incluir en cada entidad devuelta. El valor por defecto es Ninguno. Si no se especifica, sólo se incluye el campo principal.
limit	El número total de entidades que se devolverán. El valor predeterminado es -1, lo que indica que se devolverán todas las entidades coincidentes.

Parámetro	Descripción
withCollectionName	Establece el nombre de la colección. El nombre de la colección no puede estar vacío ni ser nulo.
withExpr	Establezca la expresión para consultar las entidades. Para crear una condición de filtrado escalar, consulte Reglas de expresión booleana.
withBatchSize	El número de entidades a devolver cada vez que se llama a next() en el iterador actual. El valor por defecto es 1000. Establézcalo a un valor apropiado para controlar el número de entidades a devolver por iteración.
addOutField	Especifica un campo escalar de salida (Opcional).