向量数据库Milvus

向量数据库

工作原理

1. 通过把数据（文本和图像）通过嵌入式模型转化成向量
2. 存储这些向量化后的信息
3. 查询时通过不同算法（常用余弦相似度）计算向量之间的距离，然后得到一个分数score
4. 根据传入的topK得到一个大小为topK的列表

向量化

通过向量模型转成向量

距离计算算法

• 欧氏距离 (L2)：范围: 0(相同) → +∞(不同)
• 余弦相似度： -1(相反) → 1(相同方向)
• 内积相似度： -1(相反) → 1(相同方向)
• 曼哈顿距离 (L1)：范围: 0(相同) → +∞(不同)
• Jaccard相似度 (适合集合数据)：范围: 0(无交集) → 1(完全相同)

常用向量数据库

• Pinecone
• Milvus
• Weaviate
• Qdrant
• Chroma
• FAISS（库而非完整数据库）

milvus向量数据库

SpringBoot调用milvus

maven引入

milvus引入了protobuf可能与自己引入的protobuf冲突，因此要在外面重新导入声明

<dependency>
    <groupId>com.google.protobuf</groupId>
    <artifactId>protobuf-java</artifactId>
    <version>3.24.0</version>
</dependency>
<dependency>
    <groupId>io.milvus</groupId>
    <artifactId>milvus-sdk-java</artifactId>
    <version>2.5.2</version>
    <exclusions>
        <exclusion>
            <groupId>com.google.protobuf</groupId>
            <artifactId>protobuf-java</artifactId>
        </exclusion>
    </exclusions>
</dependency>

application.yml 配置

milvus:
  config:
    host: ${MILVUS_CONFIG_HOST:your_milvus_host}
    port: ${MILVUS_CONFIG_PORT:milvus_port}
    embeddingUrl: ${MILVUS_CONFIG_EMBEDDING_URL:your_vector_url}

config 配置

@Configuration
@Slf4j
public class MilvusConfig {
    @Value("${milvus.config.host}")
    private String host;

    @Value("${milvus.config.port}")
    private Integer port;

    @Bean("milvusClientV2")
    public MilvusClientV2 milvusClientV2() {
        // 添加try catch 防止milvus挂了影响其他业务
        try {
            ConnectConfig connectConfig = ConnectConfig.builder().uri("http://" + host + ":" + port).build();
            return new MilvusClientV2(connectConfig);
        } catch (Exception e) {
            log.error("向量数据库 milvus 连接失败",e);
            return null;
        }

    }
}

向量化操作

• dto类

  @Data
  @Accessors(chain = true)
  public class EmbeddingDTO {
      List<RespEmbedding> data;
  
      @Data
      @Accessors(chain = true)
      public static class ReqEmbedding {
          private List<String> input;
          private String model = "模型名称";
          // 提示词
          private String prompt;
      }
  
      @Data
      @Accessors(chain = true)
      public static class RespEmbedding {
          String object;
          List<Float> embedding;
          int index;
      }
  
      public List<List<Float>> getAllSortedEmbeddings() {
          return data.stream()
                  .sorted((x, y) -> x.index - y.index)
                  .map(RespEmbedding::getEmbedding)
                  .collect(Collectors.toList());
      }
  
      public static ReqEmbedding buildReqEmbedding(List<String> input) {
          return new ReqEmbedding().setInput(input);
      }
  }

• milvus模板

  @Service
  @Slf4j
  public class MilvusServiceImpl {
      @Value("${milvus.config.embeddingUrl}")
      private String embeddingUrl;
  
      private static final String COLLECTION_NAME = "collection_name";
  
      private static final String ID_FIELD = "id";
      private static final String IMAGE_URL_FIELD = "image_url";
      private static final String HASH_CODE_FIELD = "hash";
      private static final String FEATURE_FIELD = "feature";
      private static final String OPERATE_FEATURE_FIELD = "operate_feature";
      private static final String FEATURE_VECTOR_FIELD = "feature_vector";
      private static final String OPERATE_FEATURE_VECTOR_FIELD = "operate_feature_vector";
  
      private final static int VECTOR_DIM = 1024;
  
      private final static Float DEFAULT_SCORE = 0.4f;
  
      private final static Integer DEFAULT_TOP_K = 3;
  
      @Autowired(required = false)
      private MilvusClientV2 client;
  
      @Resource
      private Executor executor;
  
      @PostConstruct
      public void init() {
          // 在对象创建后执行的初始化逻辑
          if (client == null) {
              return;
          }
          if (!hasCollection()) {
              log.info("向量数据不存在 {}，进行创建", COLLECTION_NAME);
              createCollection();
          }
          log.info("加载向量数据库Collection :{}到内存",COLLECTION_NAME);
          loadCollection();
      }
  
      public List<List<Float>> getEmbeddings(List<String> input) {
          EmbeddingDTO.ReqEmbedding reqEmbedding = EmbeddingDTO.buildReqEmbedding(input);
          // 防止网络问题，重试一次
          int count = 2;
          while (count > 0) {
              try {
                  log.info("JSONObject.toJSONString(reqEmbedding):{}", reqEmbedding);
                  return TxzHttp.builder()
                          .url(embeddingUrl)
                          .json(JSONObject.toJSONString(reqEmbedding))
                          .post()
                          .build()
                          .json(EmbeddingDTO.class)
                          .getAllSortedEmbeddings();
              } catch (Exception e) {
                  count--;
              }
          }
          throw new ServiceException("请求向量化接口失败");
      }
  
      public void createCollection() {
          CreateCollectionReq.CollectionSchema schema = client.createSchema();
          schema.addField(AddFieldReq.builder()
                  .fieldName(ID_FIELD)
                  .description("唯一id")
                  .dataType(DataType.Int64)
                  .isPrimaryKey(true)
                  .autoID(false)
                  .build());
          schema.addField(AddFieldReq.builder()
                  .fieldName(IMAGE_URL_FIELD)
                  .description("图片路径")
                  .dataType(DataType.VarChar)
                  .maxLength(1024)
                  .build());
          schema.addField(AddFieldReq.builder()
                  .fieldName(HASH_CODE_FIELD)
                  .description("图片hash")
                  .dataType(DataType.VarChar)
                  .maxLength(1024)
                  .build());
          schema.addField(AddFieldReq.builder()
                  .fieldName(FEATURE_FIELD)
                  .description("大模型生成特征")
                  .dataType(DataType.VarChar)
                  .maxLength(256)
                  .build());
          schema.addField(AddFieldReq.builder()
                  .fieldName(OPERATE_FEATURE_FIELD)
                  .description("运营编辑特征")
                  .dataType(DataType.VarChar)
                  .maxLength(256)
                  .isNullable(true)
                  .build());
          schema.addField(AddFieldReq.builder()
                  .fieldName(FEATURE_VECTOR_FIELD)
                  .description("大模型生成特征向量")
                  .dataType(DataType.FloatVector)
                  .dimension(VECTOR_DIM)
                  .build());
          schema.addField(AddFieldReq.builder()
                  .fieldName(OPERATE_FEATURE_VECTOR_FIELD)
                  .description("运营编辑特征向量")
                  .dataType(DataType.FloatVector)
                  .dimension(VECTOR_DIM)
                  .build());
  
          IndexParam hashCodeIndexParam = IndexParam.builder()
                  .fieldName(HASH_CODE_FIELD)
                  .indexType(IndexParam.IndexType.INVERTED)
                  .build();
  
          IndexParam featureVectorIndexParam = IndexParam.builder()
                  .fieldName(FEATURE_VECTOR_FIELD)
                  .indexType(IndexParam.IndexType.FLAT)
                  .metricType(IndexParam.MetricType.COSINE)
                  .build();
  
          IndexParam operatFeatureVectorIndexParam = IndexParam.builder()
                  .fieldName(OPERATE_FEATURE_VECTOR_FIELD)
                  .indexType(IndexParam.IndexType.FLAT)
                  .metricType(IndexParam.MetricType.COSINE)
                  .build();
  
          client.createCollection(CreateCollectionReq.builder()
                  .collectionName(COLLECTION_NAME)
                  .collectionSchema(schema)
                  .indexParams(new ArrayList<>(List.of(hashCodeIndexParam, featureVectorIndexParam, operatFeatureVectorIndexParam)))
                  .build());
      }
  
      public void dropCollection() {
          client.dropCollection(DropCollectionReq.builder()
                  .collectionName(COLLECTION_NAME)
                  .build());
          log.info("删除:{}", COLLECTION_NAME);
      }
  
      public Boolean hasCollection() {
          return client.hasCollection(HasCollectionReq.builder()
                  .collectionName(COLLECTION_NAME)
                  .build());
      }
  
      public void loadCollection() {
          client.loadCollection(LoadCollectionReq.builder()
                  .collectionName(COLLECTION_NAME)
                  .async(false)
                  .build());
      }
  
      public Boolean insertBatchMilvus(List<ImageFeature> featureImages) {
          if (featureImages == null || featureImages.isEmpty()) {
              throw new ServiceException("可见即可说在缓存特征值时不能为空");
          }
          // 业务hash唯一
          List<String> hashCodes = featureImages.stream().map(ImageFeature::getHash).collect(Collectors.toList());
          Set<String> hashSet = new HashSet<>(getNotExistHashCodes(hashCodes));
          List<ImageFeature> records = featureImages.stream().filter(record -> hashSet.contains(record.getHash())).collect(Collectors.toList());
          if (records.isEmpty()) {
              return true;
          }
          List<JsonObject> insertDataList = new ArrayList<>();
          // 求feature_embedding
          List<String> features = records.stream().map(ImageFeature::getFeature).collect(Collectors.toList());
          List<List<Float>> embeddings = getEmbeddings(features);
          for (int i = 0; i < records.size(); i++) {
              ImageFeature record = records.get(i);
              if (ImageFeature.isEmpty(record)) {
                  throw new ServiceException("可见即可说在缓存特征值时不能为空");
              }
              JsonObject insertData = new JsonObject();
              insertData.addProperty(ID_FIELD, IdUtil.getSnowflake(1,1).nextId());
              insertData.addProperty(IMAGE_URL_FIELD, record.getImageUrl());
              insertData.addProperty(HASH_CODE_FIELD, record.getHash());
              insertData.addProperty(FEATURE_FIELD, record.getFeature());
              insertData.add(FEATURE_VECTOR_FIELD, gson.toJsonTree(embeddings.get(i)));
              if (StringUtil.isNotBlank(record.getOperateFeature())) {
                  List<List<Float>> operateEmbedding = getEmbeddings(List.of(record.getOperateFeature()));
                  insertData.addProperty(OPERATE_FEATURE_FIELD, record.getOperateFeature());
                  insertData.add(OPERATE_FEATURE_VECTOR_FIELD, gson.toJsonTree(operateEmbedding.get(0)));
              } else {
                  insertData.add(OPERATE_FEATURE_VECTOR_FIELD, gson.toJsonTree(Collections.nCopies(VECTOR_DIM, 0.0f)));
              }
              insertDataList.add(insertData);
          }
  
          InsertResp insert = client.insert(InsertReq.builder()
                  .collectionName(COLLECTION_NAME)
                  .data(insertDataList)
                  .build());
  
          return insert.getInsertCnt() == records.size();
      }
  
      public Boolean insertMilvus(ImageFeature record) {
          return insertBatchMilvus(Collections.singletonList(record));
      }
  
      public ImageFeature queryMilvus(Long id) {
          QueryResp queryResp = client.query(QueryReq.builder()
                  .collectionName(COLLECTION_NAME)
                  .outputFields(List.of(ID_FIELD, IMAGE_URL_FIELD, HASH_CODE_FIELD, FEATURE_FIELD, OPERATE_FEATURE_FIELD))
                  .ids(List.of(id))
                  .build());
          List<QueryResp.QueryResult> queryResults = queryResp.getQueryResults();
          if (queryResults == null || queryResults.isEmpty()) {
              return null;
          }
          return JSONObject.parseObject(JSONObject.toJSONString(queryResults.get(0).getEntity()), ImageFeature.class);
      }
  
      private ImageFeature queryMilvusById(Long id) {
          QueryResp queryResp = client.query(QueryReq.builder()
                  .collectionName(COLLECTION_NAME)
                  .outputFields(List.of(ID_FIELD, IMAGE_URL_FIELD, HASH_CODE_FIELD, FEATURE_FIELD, OPERATE_FEATURE_FIELD, FEATURE_VECTOR_FIELD, OPERATE_FEATURE_VECTOR_FIELD))
                  .ids(List.of(id))
                  .build());
          List<QueryResp.QueryResult> queryResults = queryResp.getQueryResults();
          if (queryResults == null || queryResults.isEmpty()) {
              return null;
          }
          return JSONObject.parseObject(JSONObject.toJSONString(queryResults.get(0).getEntity()), ImageFeature.class);
      }
  
      public List<String> getNotExistHashCodes(List<String> hashCodes) {
          if (hashCodes == null || hashCodes.isEmpty()) {
              throw new ServiceException("可见即可说进行匹配的hash是null");
          }
          QueryResp queryResp = client.query(QueryReq.builder()
                  .collectionName(COLLECTION_NAME)
                  .outputFields(List.of(HASH_CODE_FIELD))
                  .filter(inStr(HASH_CODE_FIELD))
                  .filterTemplateValues(Map.of(HASH_CODE_FIELD, hashCodes))
                  .build());
          List<QueryResp.QueryResult> queryResults = queryResp.getQueryResults();
          if (queryResults == null || queryResults.isEmpty()) {
              return hashCodes;
          }
          HashSet<String> hashSet = new HashSet<>(hashCodes);
          for (QueryResp.QueryResult queryResult : queryResults) {
              String hashcode = (String) queryResult.getEntity().get(HASH_CODE_FIELD);
              hashSet.remove(hashcode);
          }
          log.info("没有识别的hash:{}", hashSet);
          return new ArrayList<>(hashSet);
      }
  
      public static String inStr(String filedName) {
          return filedName + " IN " +"{" + filedName + "}";
      }
  
      public Boolean updateMilvus(ImageFeature record) {
          if (record == null) {
              return false;
          }
  
          JsonObject updateData = new JsonObject();
          if (record.getId() == null) {
              throw new ServiceException("可见即可说修改向量数据库时未指定id");
          }
          updateData.addProperty(ID_FIELD, record.getId());
          ImageFeature imageFeature = queryMilvusById(record.getId());
          if (imageFeature == null) {
              throw new ServiceException("不存在的id");
          }
          boolean isUpdate = false;
  
          // 3. 处理 imageUrl
          if (StringUtil.isNotBlank(record.getImageUrl())) {
              updateData.addProperty(IMAGE_URL_FIELD, record.getImageUrl());
              isUpdate = true;
          } else {
              updateData.addProperty(IMAGE_URL_FIELD, imageFeature.getImageUrl());
          }
  
          // 5. 处理 feature
          if (StringUtil.isNotBlank(record.getFeature())) {
              updateData.addProperty(FEATURE_FIELD, record.getFeature());
              updateData.add(FEATURE_VECTOR_FIELD, gson.toJsonTree(getEmbeddings(List.of(record.getFeature())).get(0)));
              isUpdate = true;
          } else {
              updateData.addProperty(FEATURE_FIELD, imageFeature.getFeature());
              updateData.add(FEATURE_VECTOR_FIELD, gson.toJsonTree(imageFeature.getFeatureVector()));
          }
  
          // 6. 处理 operateFeature (运营特征，允许覆盖为空)
          if (StringUtil.isNotBlank(record.getOperateFeature())) {  // 明确允许设置为空
              updateData.addProperty(OPERATE_FEATURE_FIELD, record.getOperateFeature());
              updateData.add(OPERATE_FEATURE_VECTOR_FIELD, gson.toJsonTree(getEmbeddings(List.of(record.getOperateFeature())).get(0)));
              isUpdate = true;
          } else {
              updateData.addProperty(OPERATE_FEATURE_FIELD, imageFeature.getOperateFeature());
              updateData.add(OPERATE_FEATURE_VECTOR_FIELD, gson.toJsonTree(imageFeature.getOperateFeatureVector()));
          }
  
          if (StringUtil.isNotBlank(record.getHash())) {
              updateData.addProperty(HASH_CODE_FIELD, record.getHash());
              isUpdate = true;
          } else if (imageFeature.getHash() != null) {
              updateData.addProperty(HASH_CODE_FIELD, imageFeature.getHash());
          }
  
          if (isUpdate) {
              UpsertResp upsert = client.upsert(UpsertReq.builder()
                      .collectionName(COLLECTION_NAME)
                      .data(Collections.singletonList(updateData))
                      .build());
              return upsert.getUpsertCnt() > 0;
          }
          return isUpdate;
      }
  
      public List<ImageFeature> list() {
          QueryResp queryResp = client.query(QueryReq.builder()
                  .collectionName(COLLECTION_NAME)
                  .outputFields(List.of(ID_FIELD, IMAGE_URL_FIELD, HASH_CODE_FIELD, FEATURE_FIELD, OPERATE_FEATURE_FIELD))
                  .filter("id > 0")
                  .build());
          List<ImageFeature> imageFeatures = new ArrayList<>();
          for (QueryResp.QueryResult result : queryResp.getQueryResults()) {
              ImageFeature imageFeature = JSONObject.parseObject(JSONObject.toJSONString(result.getEntity()), ImageFeature.class);
              imageFeatures.add(imageFeature);
          }
          return imageFeatures;
      }
  
      public boolean delete(List<Long> ids) {
          DeleteResp delete = client.delete(DeleteReq.builder()
                  .collectionName(COLLECTION_NAME)
                  .ids(new ArrayList<>(ids))
                  .build());
          return delete.getDeleteCnt() > 0;
      }
  
      public List<ImageFeature> search(ReqFeatureMatch req, List<String> hashs) {
          int topK = (req.getTopK() == null || req.getTopK() <= 0) ? DEFAULT_TOP_K : req.getTopK();
          float score = (req.getScore() == null || req.getScore() <= 0) ? DEFAULT_SCORE : req.getScore();
  
          List<List<Float>> embeddings = getEmbeddings(List.of(req.getFeature()));
          log.info("content :{} ,向量计算：{}", req.getFeature(), embeddings);
          CompletableFuture<List<SearchResp.SearchResult>> operateFeatureFuture = CompletableFuture.supplyAsync(() -> {
              SearchResp search = client.search(SearchReq.builder()
                      .collectionName(COLLECTION_NAME)
                      .annsField(OPERATE_FEATURE_VECTOR_FIELD)
                      .data(Collections.singletonList(new FloatVec(embeddings.get(0))))
                      .topK(topK)
                      .filter(inStr(HASH_CODE_FIELD))
                      .filterTemplateValues(Map.of(HASH_CODE_FIELD, hashs))
                      .outputFields(List.of(ID_FIELD, HASH_CODE_FIELD, OPERATE_FEATURE_FIELD))
                      .build());
              if (search != null) {
                  List<List<SearchResp.SearchResult>> searchResults = search.getSearchResults();
                  if (searchResults != null) {
                      List<SearchResp.SearchResult> results = searchResults.get(0);
                      log.info("milvus 运营特征匹配得到结果:{}",results);
                      return results;
                  }
              }
              return null;
          },executor);
  
          CompletableFuture<List<SearchResp.SearchResult>> featureFuture = CompletableFuture.supplyAsync(() -> {
              SearchResp search = client.search(SearchReq.builder()
                      .collectionName(COLLECTION_NAME)
                      .annsField(FEATURE_VECTOR_FIELD)
                      .data(Collections.singletonList(new FloatVec(embeddings.get(0))))
                      .topK(topK)
                      .filter(inStr(HASH_CODE_FIELD))
                      .filterTemplateValues(Map.of(HASH_CODE_FIELD, hashs))
                      .outputFields(List.of(ID_FIELD, HASH_CODE_FIELD, FEATURE_FIELD))
                      .build());
              if (search != null) {
                  List<List<SearchResp.SearchResult>> searchResults = search.getSearchResults();
                  if (searchResults != null) {
                      List<SearchResp.SearchResult> results = searchResults.get(0);
                      log.info("milvus 特征匹配得到结果:{}",results);
                      return results;
                  }
              }
              return null;
          },executor);
  
          try {
              List<ImageFeature> operateFeatures = getMatchImageFeatures(operateFeatureFuture, score, true);
              if (operateFeatures != null && !operateFeatures.isEmpty()) {
                  featureFuture.cancel(true);
                  return operateFeatures;
              }
              return getMatchImageFeatures(featureFuture, score, false);
          } catch(Exception e) {
              try {
                  return getMatchImageFeatures(featureFuture, score, false);
              } catch (Exception ex) {
                  throw new ServiceException(ex,"可见即可说，特征匹配异常");
              }
          }
      }
  
      private List<ImageFeature> getMatchImageFeatures(CompletableFuture<List<SearchResp.SearchResult>> future, Float score, boolean isOperate) throws ExecutionException, InterruptedException, TimeoutException {
          List<SearchResp.SearchResult> featureResult = future.get(3, TimeUnit.SECONDS);
          if (featureResult != null && !featureResult.isEmpty()) {
              List<ImageFeature> imageFeatures = new ArrayList<>();
              for (SearchResp.SearchResult searchResult : featureResult) {
                  if (searchResult.getScore().compareTo(score) >= 0) {
                      ImageFeature imageFeature = JSONObject.parseObject(JSONObject.toJSONString(searchResult.getEntity()), ImageFeature.class);
                      imageFeature.setScore(searchResult.getScore());
                      if (isOperate && StringUtil.isBlank(imageFeature.getOperateFeature())) {
                          continue;
                      }
                      imageFeatures.add(imageFeature);
                  }
              }
              return imageFeatures;
          }
          return null;
      }
  }

连接milvus工具 attu