驻马店住房和城乡建设局网站,没人愿意干的68个暴利行业,列举五种常用的网站推广方法,西安房产网最新楼盘Milvus简介
全民AI的时代已经在趋势之中#xff0c;各类应用层出不穷#xff0c;而想要构建一个完善的AI应用/系统#xff0c;底层存储是不可缺少的一个组件。 与传统数据库或大数据存储不同的是#xff0c;这种场景下则需要选择向量数据库#xff0c;是专门用来存储和查…Milvus简介
全民AI的时代已经在趋势之中各类应用层出不穷而想要构建一个完善的AI应用/系统底层存储是不可缺少的一个组件。 与传统数据库或大数据存储不同的是这种场景下则需要选择向量数据库是专门用来存储和查询向量的数据库其存储的向量来自于对文本、语音、图像、视频等的向量化数据向量数据库不仅能够完成基本的CRUD添加、读取查询、更新、删除等操作还能够对向量数据进行更快速的相似性搜索。
Milvus是众多向量库中的之一适用于多个场景如Questions Answering系统、推荐系统等单节点 Milvus 可以在秒内完成十亿级的向量搜索分布式架构亦能满足用户的水平扩展需求。
参考文档 Milvus官网 为AI而生的数据库Milvus详解及实战
实践之问答系统推荐系统Mix-in
元数据定义
不论是问答还是推荐它们对上层暴露的接口仅仅是predict(...)/search(...)/query(...)模式是相同的因此可以共用一个基本的Schema固定基本的字段即可。
public class MilvusMeta {Getterprivate final RecommenderSchema defaultMetricsSchema;Getterprivate final QASchema defaultQASchema;public interface Schema {String getCollectionName();CreateCollectionParam getCreateCollectionParam();CreateIndexParam getCreateIndexParam();}AllArgsConstructorpublic abstract static class BasicSchema implements Schema {public static final String SEARCH_PARAM {\nprobe\:10}; // Paramspublic static final String INDEX_PARAM {\nlist\:1024}; // ExtraParampublic static final IndexType INDEX_TYPE_DEFAULT IndexType.IVF_FLAT;public static final MetricType METRIC_TYPE_DEFAULT MetricType.L2; // metric typepublic static final String INDEX_NAME_DEFAULT ivf_flat;public static final String PRIMARY_KEY_FIELD_NAME_DEFAULT id;public static final String PARTITION_KEY_FIELD_NAME_DEFAULT public;public static final String FIELD_NAME_DEFAULT embeddings;Getterprotected final CreateCollectionParam createCollectionParam;Getterprotected final CreateIndexParam createIndexParam;Overridepublic String getCollectionName() {return createCollectionParam.getCollectionName();}}/*** This schema is designed for storing Zen metrics.* TODO: Add more fields/features to describe a metric.*/public static class RecommenderSchema extends BasicSchema {private RecommenderSchema(CreateCollectionParam collectionParam, CreateIndexParam indexParam) {super(collectionParam, indexParam);}public static RecommenderSchema create(ZenAiConfig.Storages.MilvusConf conf) {ZenAiConfig.Storages.Collection collection conf.getActiveRecommenderCollection();return new RecommenderSchema(defaultCollectionParam(collection, collection.getEmbeddingsDimension()),MilvusUtil.createIndexParam(collection));}private static CreateCollectionParam defaultCollectionParam(ZenAiConfig.Storages.Collection collection,int dimension) {FieldType pkType FieldType.newBuilder().withName(collection.getPrimaryKey()).withDataType(DataType.VarChar).withPrimaryKey(true).withMaxLength(100).withAutoID(false).build();// 被embedding的字段FieldType embeddedFieldType FieldType.newBuilder().withName(collection.getEmbeddedFieldName()).withDataType(DataType.VarChar).withMaxLength(255).build();// embedding vector字段FieldType embeddingFieldType FieldType.newBuilder().withName(collection.getFieldName()).withDataType(DataType.FloatVector).withDimension(dimension).build();// 指定分区键字段每一个Collection都需要指定一个分区键除了能够Hive/Spark那样切分数据外还能够加速相似查询。// 虽然Milvus支持多种方案以切分数据但从管理复杂度、查询效率上来看一个Collection对应多个数据分区是最佳的方案。FieldType partitionKeyType FieldType.newBuilder().withName(collection.getPartitionKey()).withPartitionKey(true).withDataType(DataType.VarChar).withMaxLength(100).build();return CreateCollectionParam.newBuilder().withCollectionName(collection.getName()).withDescription(collection.getDescription())// .withShardsNum(2).addFieldType(pkType).addFieldType(embeddedFieldType).addFieldType(embeddingFieldType).addFieldType(partitionKeyType)// 开启动态字段添加功能.withEnableDynamicField(true).build();}}public static class QASchema extends BasicSchema {public static final String ANSWER_FIELD_NAME answer;public static final String SCORE_FIELD_NAME score;public static final float SCORE_MAX_DEFAULT 5.0f;public static final float SCORE_MIN_DEFAULT 0.0f;public static final String INTENTION_FIELD_NAME intention;public static final String QUESTION_OCCURRENCE occurrence;public QASchema(CreateCollectionParam createCollectionParam, CreateIndexParam createIndexParam) {super(createCollectionParam, createIndexParam);}public static QASchema create(ZenAiConfig.Storages.MilvusConf conf) {ZenAiConfig.Storages.Collection collection conf.getActiveQACollection();return new QASchema(defaultCollectionParam(collection, collection.getEmbeddingsDimension()),MilvusUtil.createIndexParam(collection));}private static CreateCollectionParam defaultCollectionParam(ZenAiConfig.Storages.Collection collection,int dimension) {FieldType pkType FieldType.newBuilder().withName(collection.getPrimaryKey()).withDataType(DataType.VarChar).withPrimaryKey(true).withMaxLength(100).withAutoID(false).build();FieldType embeddedFieldType FieldType.newBuilder().withName(collection.getEmbeddedFieldName()).withDataType(DataType.VarChar).withMaxLength(65535).build();FieldType embeddingFieldType FieldType.newBuilder().withName(collection.getFieldName()).withDataType(DataType.FloatVector).withDimension(dimension).build();FieldType partitionKeyType FieldType.newBuilder().withName(collection.getPartitionKey()).withPartitionKey(true).withDataType(DataType.VarChar).withMaxLength(100).build();return CreateCollectionParam.newBuilder().withCollectionName(collection.getName()).withDescription(collection.getDescription())// .withShardsNum(2).addFieldType(pkType).addFieldType(embeddedFieldType).addFieldType(embeddingFieldType).addFieldType(partitionKeyType).withEnableDynamicField(true) // enable to insert new fields without modifying the code.build();}}Milvus可行的操作接口定义
public interface IMilvusOperations {ZenAiConfig.Storages.Collection getCollection();MilvusConnection.MultiStatus delete(Filter filter);MilvusConnection.MultiStatus create(MilvusMeta.Index index);MilvusConnection.MultiStatus drop(String index);MilvusConnection.MultiStatus insert(MilvusData.Dataset dataset);MilvusConnection.MultiStatus insertAndFlush(MilvusData.Dataset dataset);/*** Query records by filter on the specified partition, which works like a normal SQL engine.** param partition which partition to query* param filter boolean expression obeys the rules of Milvus* param outputFields if empty, the result will contain all the fields, including the dynamic;* otherwise the result only contains the specified fields.* return a nonnull instance, size of which is 0 if no matched records, otherwise is positive.*/MilvusData.BasicPredictData queryByPartition(String partition, Filter filter, ListString outputFields);ListMilvusData.BasicPredictData search(ListListFloat vectors, Filter filter, int topK,ListString outputFields);}抽象系统接口定义
/*** 每个系统可能有不同的embedding的实现因此需要定义一个接口。*/
public interface IEmbedding {ImmutableListListFloat getEmbeddings(ListString messages);
}/*** 通用接口定义供应用层使用可以基于sentence返回Milvus相似性结果集。*/
public interface INlpSystem extends IMilvusOperations, IDataset, IEmbedding {default MilvusData.BasicPredictData predict(String sentence) {return predict(sentence, Filter.TRUE);}default MilvusData.BasicPredictData predict(String sentence, Filter filter) {return predict(sentence, filter, getCollection().getOutputFields());}default MilvusData.BasicPredictData predict(String sentence, Filter filter,ListString outputFields) {ImmutableListListFloat vectors getEmbeddings(Lists.newArrayList(sentence));if (vectors.isEmpty()) {return MilvusData.BasicPredictData.EMPTY;}ListString mergedOutputFields Sets.union(ImmutableSet.copyOf(outputFields),ImmutableSet.copyOf(getCollection().getOutputFields())).immutableCopy().asList();ListMilvusData.BasicPredictData res search(vectors, filter, getCollection().getTopk(), mergedOutputFields);return res.isEmpty() ? MilvusData.BasicPredictData.EMPTY : res.get(0);}default MilvusData.BasicPredictData predictByPartition(String partition, String sentence) {return predictByPartition(partition, sentence, Filter.TRUE, getCollection().getOutputFields());}default MilvusData.BasicPredictData predictByPartition(String partition, String sentence,Filter filter, ListString outputFields) {ImmutableListListFloat vectors getEmbeddings(Lists.newArrayList(sentence));if (vectors.isEmpty()) {return MilvusData.BasicPredictData.EMPTY;}ListString mergedOutputFields Sets.union(ImmutableSet.copyOf(outputFields),ImmutableSet.copyOf(getCollection().getOutputFields())).immutableCopy().asList();return searchByPartition(partition, vectors.get(0), filter, mergedOutputFields);}}/*** Q A系统接口。*/
public interface IQuestionAnswering extends INlpSystem {
}/*** 推荐系统接口。*/
public interface IRecommender extends INlpSystem, ISyncer {
}插入数据集定义
以列式格式构建插入Milvus的数据集需要注意的是Milvus JAVA SDK 2.3.1版本并不支持列式导致dynamic fields因此我对源码进行了改造以支持列式插入动态字段。 这个问题已经反馈给了社区并且已经在v2.3.2版本中支持。
public interface MilvusData {interface BasicData {/*** Return a list view of the splitted data, to avoid copy.*/BasicData[] split(int splitSize);/*** Return a view of the range [start, end) data, to avoid copy.*/BasicData subData(int groupId, int start, int end);int size();}interface EmbeddingsProducer extends FunctionListString, ImmutableListListFloat {}GetterSetterAllArgsConstructorNoArgsConstructorclass Dataset {private ListBasicInsertData inserts;}abstract class GroupedBasicData implements BasicData {Getterprivate final int groupId;GetterSetterAccessors(chain true)private GroupedBasicData parent;protected GroupedBasicData(int groupId) {this.groupId groupId;}/*** Split the data into more more sub-dataset.** param groups the number of expected groups* return an array of sub-dataset views from the original dataset*/public abstract BasicData[] grouped(int groups);/*** 每一个切分或是extract的子数据集都应该拥有一个可以唯一标识它的ID*/public String fullGroupId() {if (parent null) {return String.valueOf(groupId);}return parent.fullGroupId() - groupId;}}Getterabstract class PartitionedBasicDataT extends GroupedBasicData {// 每一个系统都需要指定一个分区键因此为了能够最小化存储这里使用一个变量// 保存整个数据集应该插入private final T partition;protected PartitionedBasicData(T partition, int groupId) {super(groupId);this.partition partition;}public abstract ListT getPartitions();}/*** 以列式的形式构建插入数据集并完成数据导入到Milvus。* Milvus*/class BasicInsertData extends PartitionedBasicDataString {private final String collection;private final ImmutableListString ids;private final ImmutableListString embeddingsInput;private final SupplierImmutableListListFloat vectorsSupplier;private final EmbeddingsProducer embeddingsProducer;private ImmutableMapString, List? dynamicFields;private final AtomicBoolean vectorsInitialized new AtomicBoolean(false);
}结果集定义
public interface MilvusData {/*** 一个通用的数据集可以保存search/query的结果行式数据结构。*/Getterclass BasicPredictData extends GroupedBasicData {GetterBuilderAllArgsConstructorpublic static class Row {JsonPropertyprivate String id;private String embeddingsInput;JsonPropertyprivate MapString, Object extensions;JsonPropertyprivate float distance;JsonIgnoreprivate ListFloat vector;public T T getAs(String key, ClassT clazz) {return getAs(key, clazz, null);}public T T getAs(String key, ClassT clazz, T defaultValue) {return clazz.cast(extensions.getOrDefault(key, defaultValue));}}}
}数据插入实例列式插入
这个代码示例展示了如何构建列式数据集并将其插入Milvus的流程。
注意到这里特别演示了使用了多线程并行 插入的功能其原因有二
一个批次的数据集过大Milvus无法一次快速且稳定地完成插入动作因此需要将原始数据集进行分组例如这里分成3个组通常LLMLarge language Model的一次API调用只能支持生成16个向量数组因此这里又对每一个分组后的子数据集进行横向切分产生多个Batch每个Batch包含一条记录。
Test
void testSyncCollections() throws ExecutionException, InterruptedException {ExecutorService executorService Executors.newFixedThreadPool(2);// 一组唯一值用于区别每一条数据记录ImmutableListString ids ImmutableList.of(1, 2, 3, 4, 5);// 一组指标名这些指标就是待检索的合法指标集。ImmutableListString metrics ImmutableList.of(m1, m2, m3, m4, m5);// 生成一组包含5个向量的列表对应于每一个指标名ImmutableListListFloat vectors generateVectors(5, TEST_COLLECTION_DIMENSION);// 构建插入数据集MilvusData.BasicInsertData data new MilvusData.BasicInsertData(config.getStorages().getMilvusConf().getActiveRecommenderCollection().getName(), ids, metrics,//这里使用Java中的Provider接口提供执行插入数据任务时对指标名列表向量化由于这里事先生成了向量数组因此直接从索引构建数据messages - messages.stream().map(metrics::indexOf).map(vectors::get).collect(toImmutableList()));ImmutableListDouble randoms ImmutableList.of(1.0d, 2.0d, 3.0d, 4.0d, 5.0d);// 添加动态字段及相应的数据data.updateDynamicFields(ImmutableMap.of(random, randoms));// 构建并行插入数据任务// 3 groups: ([1, 2]), ([3, 4]), ([5])// 1 batche: ([1],[2]),([3],[4]),([5])CompletableFutureInteger[] futures milvusService.syncMetrics(data, 3, 1, executorService);assertEquals(3, futures.length);CompletableFuture.allOf(futures).join();assertEquals(2, futures[0].get());assertEquals(2, futures[1].get());assertEquals(1, futures[2].get());
}相似性检索实例指标推荐
用户输入一个指标Metric名或是包含指标名的语句可以通过Milvus的Search接口找到最相近的TOP K指标前提是需要对输入指标名进行向量化然后以此向量来r从Milvus库中既存的指标集中计算找到最相似的。 Testvoid testLookingForMetric() {int topK config..getMilvusConf().getActiveRecommenderCollection().getTopk();OptionalMilvusData.BasicPredictData metrics milvusService.getActiveRecommenderSys().map(system - system.predict(销售总额));assertTrue(metrics.isPresent());assertEquals(metrics.get().getRows().size(), topK);metrics milvusService.getActiveRecommenderSys().map(system - system.predict(销售总额, lt(random, 0f)));assertFalse(metrics.isPresent());}相似性检索实例问答
用户输入一段描述可以通过Milvus提供的Search接口找到历史相关的问题并返回与此问题相关的上下文并辅助回答AI模型回答用户的当前问题。 Testvoid testSearchWithSimilarityOfMultiVectors() {ImmutableListString testQuestions ImmutableList.of(用柱形图展示2019年12月的总销售额, 用拆线图展示2020年12月的总净利润);ImmutableListString testIds testQuestions.stream().map(DefaultQuestionAnswering::encodeQuestion).collect(ImmutableList.toImmutableList());IEmbedding embeddingSvc aiService.getMilvusService().get().getActiveQuestionAnswering().get();DefaultQuestionAnswering.QAInsertData insertData system.getInsertDataBuilder().ids(testIds).questions(testQuestions).answers(ImmutableList.of(很好, 不错))// 用户对于此问题返回结果的评价.scores(ImmutableList.of(1.0f, 1.0f))// 定义embeddings生成器在插入时才会计算embeddings.embeddingsProducer(questions - {ImmutableListListFloat qvectors embeddingSvc.getEmbeddings(questions);ImmutableListListFloat mvectors embeddingSvc.getEmbeddings(ImmutableList.of(总销售额, 总净利润));return ImmutableList.of(merge(qvectors.get(0), mvectors.get(0)), merge(qvectors.get(1), mvectors.get(1)));}).build();system.insert(new MilvusData.Dataset(ImmutableList.of(insertData)));// Case 1:// 用柱形图展示2019年12月的总销售额: 52.0181// 用拆线图展示2020年12月的总净利润: 147.0664verifySearch(system, 用拆线图展示2020年12月的总销售额, 总销售额, 0, 销售额, this::merge);// Case 2:// 用柱形图展示2019年12月的总销售额: 313.70105// 用拆线图展示2020年12月的总净利润: 181.3783verifySearch(system, 今年5月的净利润详情, 净利润, 0, 利润, this::merge);// Case 3:// 用柱形图展示2019年12月的总销售额: 160.30568// 用拆线图展示2020年12月的总净利润: 357.7008verifySearch(system, 今年5月的销售额详情, 销售额, 0, 销售额, this::merge);}总结
Milvus对上层提供了与传统数据库相似的接口以管理Milvus数据同时提供了带有过滤功能的数据检索接口使得上层应用能够很方便地利用传统数据库思维来设计 和实现自己的系统。 但在使用中也感受到一些局限性或可能提升的点
库中的一行记录只能对应一个embedding vector只能使用相同模型生成的vector才能更好地检索向量如果想一处持编码的文本对应多个vectors是不可能的用户不得不创建新的Collection存储相同文本的不同向量。用户显示Flush/Load Collection每一次更新数据集客户端必须要显示地load collection的操作才能将新的数据加载到Server结点的内存中同时第一次加载Collection必须是全量。粗糙的表达式字符串对于API接口的使用缺少便利的表达式类定义只能传递字符串很容易出错只能在运行时才知道哪些出错了。缓存特性的支持通常Milvus被用作Cache角色被引入系统中但Milvus缺少一些缓存特性如过期自动清理、partial dataset的load/unload功能等。
