添加milvus代码示例

shikeying

2024-03-31 bde8258f7de3dfc5cbf0fc59f2b74ed0fd073df7

添加milvus代码示例

 deploy-jar-template/pom.xml

@@ -17,6 +17,13 @@
    </properties>

    <dependencies>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>com.iplatform</groupId>
            <artifactId>iplatform-base</artifactId>
@@ -198,6 +205,12 @@
<!--            <artifactId>iplatform-report</artifactId>-->
<!--        </dependency>-->

        <!-- 2024-03-28 æµ‹è¯•ä¸€é”®å¡«å·¥å•ï¼Œéœ€è¦èŠå¤©åˆ†æžç›¸ä¼¼åº¦ -->
        <dependency>
            <groupId>com.walkersoft</groupId>
            <artifactId>walker-support-milvus</artifactId>
        </dependency>

    </dependencies>

    <build>

 deploy-jar-template/src/main/java/com/iplatform/api/MilvusChatApi.java

New file
@@ -0,0 +1,217 @@
package com.iplatform.api;

import com.iplatform.base.SystemController;
import com.iplatform.milvus.EventVo;
import com.iplatform.milvus.MilvusEngine;
import com.iplatform.milvus.ParamList;
import com.iplatform.milvus.service.EventServiceImpl;
import com.walker.infrastructure.utils.StringUtils;
import com.walker.web.ResponseValue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.RestTemplate;

import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.concurrent.TimeUnit;

@RestController
@RequestMapping("/test/milvus")
public class MilvusChatApi extends SystemController {

    private EventServiceImpl eventService;
    private RestTemplate restTemplate;
    private MilvusEngine milvusEngine;

    private Long existId = 4048L;
    private boolean isBreak = false;

    private static final String URL_EMBEDDING = "http://120.26.128.84:7003/ai/text/embedding";
    private static final String URL_MILVUS = "120.26.128.84";

    @Autowired
    public MilvusChatApi(EventServiceImpl eventService, RestTemplate restTemplate){
        this.eventService = eventService;
        this.restTemplate = restTemplate;
        if(this.milvusEngine == null){
            MilvusEngine engine = new MilvusEngine(URL_MILVUS, 19530);
            this.milvusEngine = engine;
            logger.info("milvus engine ok!");
        }
    }

    @RequestMapping("/embedding")
    public ResponseValue testHttpEmbedding(){
        List<EventVo> data = new ArrayList(8);
        EventVo eventVo = new EventVo();
        eventVo.setContent("第一句");
        data.add(eventVo);

        eventVo = new EventVo();
        eventVo.setContent("第二句");
        data.add(eventVo);

        boolean success = this.acquireEmbedding(data);
        this.milvusEngine.insertTestData();

        return ResponseValue.success("结果是：" + success);
    }

    @RequestMapping("/write")
    public ResponseValue testWriteMilvus(){
        Collection<EventVo> eventVoList = this.acquireEventVoList();
        if(eventVoList == null){
            return ResponseValue.error("没有加载到数据");
        }
        logger.info("加载了 event vo: {}个", eventVoList.size());

        new Thread(new WriteTask(eventVoList)).start();
        return ResponseValue.success();
    }

    @RequestMapping("/query")
    public ResponseValue testQueryMilvus(String text){

        return ResponseValue.success();
    }

    private boolean acquireEmbedding(List<EventVo> batchData){
        ParamList paramList = new ParamList();
        for(EventVo eventVo : batchData){
            paramList.add(eventVo.getContent());
        }
        try{
            ResponseEntity<ResponseValue> responseEntity = this.restTemplate.postForEntity(URL_EMBEDDING, paramList, ResponseValue.class);
            if(responseEntity.getStatusCode() == HttpStatus.OK){
                List<List<Float>> data = (List<List<Float>>)responseEntity.getBody().getData();
//                double[] one = null;
                List<Float> vector = null;
                for(int i=0; i<data.size(); i++){
                    vector = data.get(i);
//                    one = new double[vector.size()];
//                    for(int j=0; j< vector.size(); j++){
//                        one[j] = vector.get(j);
//                    }

//                    logger.debug("data = {}", vector);
//                    logger.debug("class type = {}", vector.getClass().getName());
                    batchData.get(i).setEmbedding(this.transfer2FloatList(vector));
                }
                return true;
            } else {
                logger.error("http è¿”回错误：{}", responseEntity.getBody());
            }
            return false;
        } catch (Exception cause){
            logger.error("获取向量出现错误：{}" + cause.getMessage(), cause);
            return false;
        }
    }

    private List<Float> transfer2FloatList(List<?> list){
        List<Float> vector = new ArrayList<>(768);
        for(int i=0; i<list.size(); i++){
            vector.add(Float.parseFloat(list.get(i).toString()));
        }
        return vector;
    }

    private class WriteTask implements Runnable{

        private Collection<EventVo> eventVoList;

        public WriteTask(Collection<EventVo> eventVoList){
            this.eventVoList = eventVoList;
        }

        @Override
        public void run() {
            logger.info(".......... start task ...");

            // å¤ä½çŠ¶æ€
            isBreak = false;

            int count = 0;
            List<EventVo> batchData = new ArrayList<>();
            for(EventVo eventVo : this.eventVoList){
                // å·²å­˜åœ¨ä¸Šæ¬¡åŠ è½½è®°å½•ï¼Œå·²å†™å…¥çš„不在重新处理
                if(existId != null && eventVo.getId() < existId.longValue()){
                    continue;
                }

                if(isBreak){
                    break;
                }

                if(count == 0){
                    logger.info("1) å¼€å§‹ï¼ˆæˆ–继续）采集第 {} è®°å½•", eventVo.getId());
                }

                if(count >= 8){
                    // è§¦å‘一次批量写入
                    logger.info("2) è§¦å‘一次调用：{}", batchData.get(7).getId());
                    try {
                        boolean successEmbedding = acquireEmbedding(batchData);
                        if(!successEmbedding){
                            logger.error("获取向量失败，任务结束");
                            break;
                        }

                        // èŽ·å–向量，写入数据库
                        milvusEngine.insertEventVoList(batchData);

                    } catch (Exception ex){
                        logger.error("error = " + ex.getMessage(), ex);
//                        existId = eventVo.getId();
                        isBreak = true;
                        logger.error("3) é‡‡é›†ä»»åŠ¡å¼‚常， å½“前 id = {}", existId == null ? "" : existId);
                        break;

                    } finally {
                        try {
                            TimeUnit.SECONDS.sleep(1);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                    // æ¸…理本次数据
                    batchData.clear();
                    count = 0;
                }
                batchData.add(eventVo);
                count ++;
            }
            logger.info(".......... end task ...");
        }
    }

    private Collection<EventVo> acquireEventVoList(){
        List<EventVo> data = null;
//        if(this.isBreak){
//            data = this.eventService.queryEventAll(existId);
//        } else {
//            data = this.eventService.queryEventAll(null);
//        }
        data = this.eventService.queryEventAll(null);
        if(StringUtils.isEmptyList(data)){
            return null;
        }

        // è¿‡æ»¤æŽ‰ content é‡å¤çš„数据记录
        LinkedHashMap<String, EventVo> cache = new LinkedHashMap();
        EventVo temp = null;
        for(EventVo e : data){
            temp = cache.get(e.getContent());
            if(temp == null){
                cache.put(e.getContent(), e);
            }
        }
        return cache.values();
    }
}

 deploy-jar-template/src/main/java/com/iplatform/milvus/EventVo.java

New file
@@ -0,0 +1,54 @@
package com.iplatform.milvus;

import java.util.List;

public class EventVo {

    public long getId() {
        return id;
    }

    public void setId(long id) {
        this.id = id;
    }

    public String getTitle() {
        return title;
    }

    public void setTitle(String title) {
        this.title = title;
    }

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }

    public List<Float> getEmbedding() {
        return embedding;
    }

    public void setEmbedding(List<Float> embedding) {
        this.embedding = embedding;
    }

    public String getAnswer() {
        return answer;
    }

    public void setAnswer(String answer) {
        this.answer = answer;
    }

    private long id;
    private String title;
    private String content;
//    private double[] embedding;
//    private List<Double> embedding;
    private List<Float> embedding;
    private String answer;  // ä¸å­˜å…¥milvus，仅在缓存中用于返回问答结果
}

 deploy-jar-template/src/main/java/com/iplatform/milvus/MilvusEngine.java

New file
@@ -0,0 +1,152 @@
package com.iplatform.milvus;

import com.walker.infrastructure.utils.StringUtils;
import com.walker.support.milvus.DataSet;
import com.walker.support.milvus.DataType;
import com.walker.support.milvus.FieldType;
import com.walker.support.milvus.MetricType;
import com.walker.support.milvus.OperateService;
import com.walker.support.milvus.Table;
import com.walker.support.milvus.engine.DefaultOperateService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class MilvusEngine {

    protected final transient Logger logger = LoggerFactory.getLogger(this.getClass());

    public MilvusEngine(String ip, int port){
        DefaultOperateService service = new DefaultOperateService();
        service.connect(ip, port);
        this.operateService = service;
        logger.info("connect milvus: {}:{}", ip, port);
    }

    public void close(){
        if(this.operateService != null){
            this.operateService.close();
        }
    }

    /**
     * åˆ›å»ºè¡¨ï¼šæµ‹è¯•ä»ŽèŠå¤©ä¸€é”®æå–工单内容使用。
     * <pre>
     *     1) ä»ŽåŽ†å²å·¥å•æ•°æ®ä¸­ï¼Œæ”¶é›†ç”¨æˆ·æé—®å†…容，整理到表中
     *     2) æŠŠè¿™äº›æ•°æ®é€šè¿‡å‘量转化，写入milvus数据库。
     * </pre>
     * @date 2024-03-28
     */
    public void createChatSimilarTable(){
        Table chatSimilarTable = new Table();
        chatSimilarTable.setCollectionName("chat_similar");
        chatSimilarTable.setDescription("聊天提取工单摘要历史数据");
        chatSimilarTable.setShardsNum(1);
        chatSimilarTable.setDimension(768); // è¿™ä¸ªæ˜¯æ ¹æ®ä½¿ç”¨å‘量模型维度定的

        // è®¾ç½®å­—段
        FieldType id = FieldType.newBuilder()
                .withName("id").withPrimaryKey(true).withMaxLength(18).withDataType(DataType.Long).build();
        FieldType title = FieldType.newBuilder()
                .withName("title").withPrimaryKey(false).withMaxLength(180).withDataType(DataType.VarChar).build();
        FieldType content = FieldType.newBuilder()
                .withName("content").withPrimaryKey(false).withMaxLength(255).withDataType(DataType.VarChar).build();
        FieldType answer = FieldType.newBuilder()
                .withName("answer").withPrimaryKey(false).withMaxLength(255).withDataType(DataType.VarChar).build();
        FieldType embedding = FieldType.newBuilder()
                .withName("embedding").withPrimaryKey(false).withDataType(DataType.FloatVector).withDimension(768).build();

        List<FieldType> fieldTypeList = new ArrayList<>(8);
        fieldTypeList.add(id);
        fieldTypeList.add(title);
        fieldTypeList.add(content);
        fieldTypeList.add(answer);
        fieldTypeList.add(embedding);
        chatSimilarTable.setFieldTypes(fieldTypeList);

        this.operateService.createTable(chatSimilarTable);
        logger.info("创建了 table = {}", chatSimilarTable.getCollectionName());

        // åˆ›å»ºç´¢å¼•
        this.operateService.createIndex(chatSimilarTable.getCollectionName()
                , "embedding", "HNSW", "{\"nlist\":16384, \"efConstruction\":128, \"M\":8}", MetricType.NLP);
        logger.info("创建了 index = {}", chatSimilarTable.getCollectionName() + "_index");
    }

    public void dropChatSimilarTable(){
        this.operateService.dropTable("chat_similar");
    }

    public void insertTestData(){
        DataSet dataSet = new DataSet();
        dataSet.setTableName("chat_similar");

        List<List<Float>> vectorList = new ArrayList<>();
        vectorList.add(Arrays.asList(mockVector));
        vectorList.add(Arrays.asList(mockVector));

        Map<String, List<?>> fieldMap = new HashMap();
        fieldMap.put("id", Arrays.asList(new Long[]{1L, 2L}));
        fieldMap.put("title", Arrays.asList(new String[]{"第一个标题", "第二个标题"}));
        fieldMap.put("content", Arrays.asList(new String[]{"第一个内容", "2222"}));
        fieldMap.put("answer", Arrays.asList(new String[]{"第一个答案", "22222222"}));
        fieldMap.put("embedding", vectorList);
        dataSet.setFields(fieldMap);
        this.operateService.insertDataSet(dataSet);
        logger.info("写入了测试数据: {}", dataSet);
    }

    public void insertEventVoList(List<EventVo> batchData){
        if(StringUtils.isEmptyList(batchData)){
            return;
        }
        List<Long> ids = new ArrayList<>(8);
        List<String> titles = new ArrayList<>(8);
        List<String> contents = new ArrayList<>(8);
        List<String> answers = new ArrayList<>(8);
        List<List<Float>> vectorSet = new ArrayList<>(8);

        for(EventVo vo : batchData){
            ids.add(vo.getId());
            vectorSet.add(vo.getEmbedding());
            if(StringUtils.isNotEmpty(vo.getTitle())){
                titles.add(vo.getTitle());
            } else {
                titles.add("none");
            }
            if(StringUtils.isNotEmpty(vo.getAnswer())){
                answers.add(vo.getAnswer());
            } else {
                answers.add("none");
            }
            if(vo.getContent().length() > 200){
                contents.add(vo.getContent().substring(0, 200));
            } else {
                contents.add(vo.getContent());
            }
        }

        DataSet dataSet = new DataSet();
        dataSet.setTableName("chat_similar");

        Map<String, List<?>> fieldMap = new HashMap();
        fieldMap.put("id", ids);
        fieldMap.put("title", titles);
        fieldMap.put("content", contents);
        fieldMap.put("answer", answers);
        fieldMap.put("embedding", vectorSet);
        dataSet.setFields(fieldMap);
        this.operateService.insertDataSet(dataSet);
        logger.info("写入了: {}", ids);
    }

    private OperateService operateService;

//    private Double[] mockVector = new Double[]{-0.051114246249198914, 0.889954432};
    private Float[] mockVector = new Float[]{-0.051114246249198914f, 0.889954432f};
}

 deploy-jar-template/src/main/java/com/iplatform/milvus/ParamList.java

New file
@@ -0,0 +1,35 @@
package com.iplatform.milvus;

import java.util.ArrayList;
import java.util.List;

/**
 * è¯·æ±‚参数，调用向量生成服务。
 * @date 2024-03-29
 */
public class ParamList {

    public void add(String text){
        if(this.list == null){
            this.list = new ArrayList<>(8);
        }
        this.list.add(text);
    }

    public List<String> getList() {
        return list;
    }

    public void setList(List<String> list) {
        this.list = list;
    }

    private List<String> list;

    @Override
    public String toString() {
        return "ParamList{" +
                "list=" + list +
                '}';
    }
}

 deploy-jar-template/src/main/java/com/iplatform/milvus/service/EventServiceImpl.java

New file
@@ -0,0 +1,46 @@
package com.iplatform.milvus.service;

import com.iplatform.milvus.EventVo;
import com.walker.jdbc.service.BaseServiceImpl;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.stereotype.Service;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;

@Service
public class EventServiceImpl extends BaseServiceImpl {

    private final EventVoMapper eventVoMapper = new EventVoMapper();

    public List<EventVo> queryEventAll(Long existId){
        if(existId == null){
            return this.select("select * from event_history order by id asc", new Object[]{}, this.eventVoMapper);
        } else {
            return this.select("select * from event_history where id > ? order by id asc", new Object[]{existId}, this.eventVoMapper);
        }
    }

    private static class EventVoMapper implements RowMapper<EventVo> {
        @Override
        public EventVo mapRow(ResultSet rs, int rowNum) throws SQLException {
            EventVo e = new EventVo();
            e.setId(rs.getLong("id"));
            e.setContent(rs.getString("content"));
            String title = rs.getString("title");
            if(title.length() > 200){
                title = title.substring(0, 200);
            }
            e.setTitle(title);
            if(rs.getObject("answer") != null){
                String answer = rs.getString("answer");
                if(answer.length() > 180){
                    answer = answer.substring(0,180);
                }
                e.setAnswer(answer);
            }
            return e;
        }
    }
}

 deploy-jar-template/src/main/resources/application-dev.yml

@@ -424,11 +424,11 @@
    # ç©ºé—²è¿žæŽ¥æ•°é‡ï¼Œé»˜è®¤ï¼š200个
    max-idle-connections: 200
    # è¿žæŽ¥è¶…时，默认：2秒
    connect-timeout-seconds: 2
    connect-timeout-seconds: 20
    # è¯»è¶…时，默认3秒
    read-timeout-seconds: 3
    read-timeout-seconds: 30
    # å†™è¶…时，默认3秒
    write-timeout-seconds: 3
    write-timeout-seconds: 30

  # é›†ç¾¤æ”¯æŒç›¸å…³é…ç½®ï¼Œ2023-09-29
  lb:

 deploy-jar-template/src/test/java/com/iplatform/milvus/TestMilvus.java

New file
@@ -0,0 +1,37 @@
package com.iplatform.milvus;

import org.junit.Test;
import org.springframework.web.client.RestTemplate;

public class TestMilvus {

    /**
     * åˆ›å»ºèŠå¤©ä¸­æå–工单摘要功能，milvus测试表创建。
     * <p>第一步：创建表和索引</p>
     */
//    @Test
    public void createTable(){
        this.acquireMilvusEngine();
        this.milvusEngine.createChatSimilarTable();
    }

    /**
     * æµ‹è¯•ï¼šå†™å…¥ä¸€ä¸ªæ¨¡æ‹Ÿæ•°æ®
     */
//    @Test
    public void insertTestMockData(){
        this.acquireMilvusEngine();
        this.milvusEngine.insertTestData();
    }


    private void acquireMilvusEngine(){
        if(this.milvusEngine == null){
            MilvusEngine engine = new MilvusEngine("120.26.128.84", 19530);
            this.milvusEngine = engine;
        }
    }

    private RestTemplate restTemplate = new RestTemplate();
    private MilvusEngine milvusEngine;
}

 iplatform-file-server/src/main/java/com/iplatform/file/FileEngineFactory.java

@@ -256,6 +256,11 @@
        if(StringUtils.isNotEmpty(fileUrlPrefixConfig)){
            FileStoreUtils.fileStoreTypeUrlPrefix.put(FileStoreType.OssAli.getIndex(), fileUrlPrefixConfig);
        }
        // 2024-02-23 s3对象上传
        fileUrlPrefixConfig = this.getFileUrlPrefix(FileStoreType.OssAws.getIndex());
        if(StringUtils.isNotEmpty(fileUrlPrefixConfig)){
            FileStoreUtils.fileStoreTypeUrlPrefix.put(FileStoreType.OssAws.getIndex(), fileUrlPrefixConfig);
        }
    }

    /**

 iplatform-file-server/src/main/java/com/iplatform/file/util/FileStoreUtils.java

@@ -38,6 +38,8 @@
            uploadUrl = ArgumentsConstants.CONFIG_AL_UPLOAD_URL;
        } else if(fileStoreType.equals("4") || fileStoreType.equals(FileStoreType.INDEX_OSS_TX)){
            uploadUrl = ArgumentsConstants.CONFIG_TX_UPLOAD_URL;
        } else if(fileStoreType.equals("5") || fileStoreType.equals(FileStoreType.INDEX_OSS_AWS)){
            uploadUrl = ArgumentsConstants.CONFIG_S3_UPLOAD_URL;
        } else if(fileStoreType.equals(FileStoreType.INDEX_FTP)){
//            throw new UnsupportedOperationException("ftp存储方式未设置参数！");
            uploadUrl = ArgumentsConstants.CONFIG_FTP_UPLOAD_URL;

 iplatform-pay-server/doc/Ö§¸¶ÖÐ̨Éè¼Æ.EAP

Binary files differ