init

.gitignore

@@ -0,0 +1,73 @@
+# 忽略匹配下列规则的Git 提交 V2.1.0
+### gradle ###
+.gradle
+/build/
+!gradle/wrapper/gradle-wrapper.jar
+
+### STS ###
+.settings/
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+bin/
+
+### IntelliJ IDEA ###
+.idea
+.qoder
+*.iws
+*.iml
+*.ipr
+*.lock
+rebel.xml
+
+### NetBeans ###
+nbproject/private/
+build/
+nbbuild/
+nbdist/
+.nb-gradle/
+
+### maven ###
+target/
+*.war
+*.ear
+*.zip
+*.tar
+*.tar.gz
+
+### logs ####
+/logs/
+*.log
+
+### temp ignore ###
+*.cache
+*.diff
+*.patch
+*.tmp
+*.pdf
+*.java~
+*.properties~
+*.xml~
+
+### system ignore ###
+.DS_Store
+Thumbs.db
+Servers
+.metadata
+upload
+gen_code
+
+### node ###
+node_modules
+
+### oneself ###
+.idea/
+.vscode/
+.claude/
+hnqz-upms/hnqz-upms-biz/logs/
+
+### Protobuf ###
+**/src/main/**/com/qunzhixinxi/hnqz/admin/message/

fn-common/pom.xml

@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>net.yaoyi.pipeline</groupId>
+        <artifactId>yyc-pipeline</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>fn-common</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+
+    <properties>
+        <maven.compiler.source>21</maven.compiler.source>
+        <maven.compiler.target>21</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jsr310</artifactId>
+        </dependency>
+    </dependencies>
+</project>

fn-common/src/main/java/net/yaoyi/pipeline/common/JacksonUtils.java

@@ -0,0 +1,205 @@
+package net.yaoyi.pipeline.common;
+
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.*;
+import com.fasterxml.jackson.databind.introspect.VisibilityChecker;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import com.fasterxml.jackson.datatype.jsr310.deser.LocalDateDeserializer;
+import com.fasterxml.jackson.datatype.jsr310.deser.LocalDateTimeDeserializer;
+import com.fasterxml.jackson.datatype.jsr310.deser.LocalTimeDeserializer;
+import com.fasterxml.jackson.datatype.jsr310.ser.LocalDateSerializer;
+import com.fasterxml.jackson.datatype.jsr310.ser.LocalDateTimeSerializer;
+import com.fasterxml.jackson.datatype.jsr310.ser.LocalTimeSerializer;
+import net.yaoyi.pipeline.common.exception.JsonException;
+
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
+import java.util.List;
+import java.util.Map;
+import java.util.TimeZone;
+
+public class JacksonUtils {
+
+    private static final ObjectMapper objectMapper;
+    private static final ObjectMapper strictObjectMapper;
+
+    static {
+        objectMapper = new ObjectMapper();
+        objectMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
+        objectMapper.setTimeZone(TimeZone.getTimeZone("Asia/Shanghai"));
+        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+        objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+        JavaTimeModule javaTimeModule = new JavaTimeModule();
+        javaTimeModule.addSerializer(LocalDate.class, new LocalDateSerializer(DateTimeFormatter.ofPattern("yyyy-MM-dd")));
+        javaTimeModule.addSerializer(LocalTime.class, new LocalTimeSerializer(DateTimeFormatter.ofPattern("HH:mm:ss")));
+        javaTimeModule.addSerializer(LocalDateTime.class, new LocalDateTimeSerializer(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
+        javaTimeModule.addDeserializer(LocalDate.class, new LocalDateDeserializer(DateTimeFormatter.ofPattern("yyyy-MM-dd")));
+        javaTimeModule.addDeserializer(LocalTime.class, new LocalTimeDeserializer(DateTimeFormatter.ofPattern("HH:mm:ss")));
+        javaTimeModule.addDeserializer(LocalDateTime.class, new LocalDateTimeDeserializer(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
+        objectMapper.registerModule(javaTimeModule);
+        objectMapper.setVisibility(VisibilityChecker.Std.defaultInstance().withFieldVisibility(JsonAutoDetect.Visibility.ANY));
+
+        strictObjectMapper = new ObjectMapper();
+    }
+
+    public JacksonUtils() {
+    }
+
+    public static JsonNode toJsonNode(String json) {
+        try {
+            return objectMapper.readTree(json);
+        } catch (Exception e) {
+            throw new JsonException(e);
+        }
+    }
+
+    public static String toJsonString(Object object) {
+        try {
+            return objectMapper.writeValueAsString(object);
+        } catch (Exception e) {
+            throw new JsonException(e);
+        }
+    }
+
+    public static <T> T fromJSON(String content, Class<T> valueType) {
+        try {
+            return objectMapper.readValue(content, valueType);
+        } catch (Exception e) {
+            throw new JsonException(e);
+        }
+    }
+
+    public static <T> List<T> fromJSONArray(String contend, Class<T> type) throws JsonProcessingException {
+        return objectMapper.readValue(contend, objectMapper.getTypeFactory().constructCollectionType(List.class, type));
+    }
+
+    public static <T> T parseObject(JsonNode jsonNode, Class<T> valueType) {
+        try {
+            return objectMapper.convertValue(jsonNode, valueType);
+        } catch (Exception e) {
+            throw new JsonException(e);
+        }
+    }
+
+    public static <T> T parseObject(String content, Class<T> valueType) {
+        try {
+            return objectMapper.readValue(content, valueType);
+        } catch (Exception e) {
+            throw new JsonException(e);
+        }
+    }
+
+    public static <T> T parseObject(String content, JavaType valueType) {
+        try {
+            return objectMapper.readValue(content, valueType);
+        } catch (Exception e) {
+            throw new JsonException(e);
+        }
+    }
+
+    public static <T> T parseObject(JsonNode jsonNode, JavaType valueType) {
+        try {
+            return objectMapper.convertValue(jsonNode, valueType);
+        } catch (Exception e) {
+            throw new JsonException(e);
+        }
+    }
+
+    public static <T> T parseObject(String content, TypeReference<T> valueTypeRef) {
+        try {
+            return objectMapper.readValue(content, valueTypeRef);
+        } catch (Exception e) {
+            throw new JsonException(e);
+        }
+    }
+
+    public static <T> T parseObject(JsonNode jsonNode, TypeReference<T> valueTypeRef) {
+        try {
+            return objectMapper.convertValue(jsonNode, valueTypeRef);
+        } catch (Exception e) {
+            throw new JsonException(e);
+        }
+    }
+
+    public static <T> T parseObjectRef(String content) {
+        try {
+            return objectMapper.readValue(content, new TypeReference<T>() {
+            });
+        } catch (Exception e) {
+            throw new JsonException(e);
+        }
+    }
+
+    public static <T> T strictParseObject(String content, Class<T> valueType) {
+        try {
+            return strictObjectMapper.readValue(content, valueType);
+        } catch (Exception e) {
+            throw new JsonException(e);
+        }
+    }
+
+    public static <T> T strictParseObject(String content, TypeReference<T> valueTypeRef) {
+        try {
+            return strictObjectMapper.readValue(content, valueTypeRef);
+        } catch (Exception e) {
+            throw new JsonException(e);
+        }
+    }
+
+    public static <T> TypeReference<T> typeReferenceOf(Class<T> clazz) {
+        return new TypeReference<T>() {
+        };
+    }
+
+    public static String textValue(JsonNode node, String field) {
+        JsonNode fieldNode = findValidNode(node, field);
+        if (fieldNode == null) {
+            return null;
+        }
+        return fieldNode.asText();
+    }
+
+    public static Integer integerValue(JsonNode node, String field) {
+        JsonNode fieldNode = findValidNode(node, field);
+        if (fieldNode == null) {
+            return null;
+        }
+        return fieldNode.asInt();
+    }
+
+    public static Long longValue(JsonNode node, String field) {
+        JsonNode fieldNode = findValidNode(node, field);
+        if (fieldNode == null) {
+            return null;
+        }
+        return fieldNode.asLong();
+    }
+
+    private static JsonNode findValidNode(JsonNode node, String field) {
+        JsonNode fieldNode;
+        if (node == null
+                || (fieldNode = node.get(field)) == null
+                || fieldNode.isNull()) {
+            return null;
+        }
+        return fieldNode;
+    }
+
+    public static <T> Map<String, Object> toMap(T entity) {
+        if (entity == null) {
+            return null;
+        }
+        try {
+            return objectMapper.convertValue(entity, new TypeReference<>() {
+            });
+        } catch (Exception e) {
+            throw new JsonException(e);
+        }
+    }
+}
+

fn-common/src/main/java/net/yaoyi/pipeline/common/exception/JsonException.java

@@ -0,0 +1,8 @@
+package net.yaoyi.pipeline.common.exception;
+
+public class JsonException extends RuntimeException {
+
+    public JsonException(Throwable cause) {
+        super(cause);
+    }
+}

fn-opensearch/pom.xml

@@ -0,0 +1,49 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>net.yaoyi.pipeline</groupId>
+        <artifactId>yyc-pipeline</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>fn-opensearch</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-simple</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>net.yaoyi.pipeline</groupId>
+            <artifactId>fn-common</artifactId>
+            <version>1.0.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>net.yaoyi.pipeline</groupId>
+            <artifactId>yyc-pipeline-model</artifactId>
+            <version>1.0.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>com.aliyun</groupId>
+            <artifactId>aliyun-sdk-ha3engine-vector</artifactId>
+            <version>1.1.18</version>
+        </dependency>
+        <dependency>
+            <groupId>com.aliyun.fc.runtime</groupId>
+            <artifactId>fc-java-core</artifactId>
+            <version>1.4.1</version>
+        </dependency>
+    </dependencies>
+
+</project>

fn-opensearch/src/main/java/net/yaoyi/pipeline/opensearch/Contents.java

@@ -0,0 +1,8 @@
+package net.yaoyi.pipeline.opensearch;
+
+public class Contents {
+    public static final String ENDPOINT = "http://ha-cn-yb84tj5o101.ha.aliyuncs.com";
+    public static final String INSTANCE_ID = "ha-cn-yb84tj5o101";
+    public static final String ACCESS_USER_NAME = "opensearchadmin";
+    public static final String ACCESS_PASS_WORD = "yyc_OpenSearch_1234";
+}

fn-opensearch/src/main/java/net/yaoyi/pipeline/opensearch/OpenSearchHandler.java

@@ -0,0 +1,293 @@
+package net.yaoyi.pipeline.opensearch;
+
+import com.aliyun.fc.runtime.Context;
+import com.aliyun.fc.runtime.PojoRequestHandler;
+import com.aliyun.ha3engine.vector.models.*;
+import com.fasterxml.jackson.databind.JsonNode;
+import lombok.extern.slf4j.Slf4j;
+import net.yaoyi.pipeline.common.JacksonUtils;
+import net.yaoyi.pipeline.model.index.TaskImageIndex;
+import net.yaoyi.pipeline.model.mq.ImageMultiMsg;
+import net.yaoyi.pipeline.model.mq.TaskImageMsg;
+
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class OpenSearchHandler implements PojoRequestHandler<TaskImageMsg, List<ImageMultiMsg>> {
+    private static final String TABLE_NAME = "";
+
+    private static final int TOP_K = 10;
+    private static final float SCORE_THRESHOLD = 0.7f;
+
+    @Override
+    public List<ImageMultiMsg> handleRequest(TaskImageMsg taskImageMsg, Context context) {
+        Long taskId = taskImageMsg.getTaskId();
+        String namespace = taskImageMsg.getNamespace();
+
+        // 1. 从数据到索引：将 image 数据转换为向量并写入 OpenSearch
+        saveByTask(taskImageMsg);
+
+        try {
+            Thread.sleep(2000);
+        } catch (Exception e) {
+            log.warn("等待被中断：" + e.getMessage());
+        }
+
+        // 2. 以 namespace + taskId 维度来完成数据更新，image相同的数据保持id不变
+        List<TaskImageIndex> list = fetchByTaskId(namespace, taskId);
+        for (TaskImageIndex taskImageIndex : list) {
+            if (taskImageIndex.getVector() == null) {
+                throw new RuntimeException("尚未更新索引信息，id:" + taskImageIndex.getId());
+            }
+
+            // 维度一：namespace 维度
+            String nsFilter = "taskId!=" + taskId;
+            List<TaskImageIndex.MultiResult> nsResult = knnSearch(namespace, taskImageIndex.getVector(), nsFilter, "namespace");
+            List<TaskImageIndex.MultiResult> multiResultList = new ArrayList<>(nsResult);
+
+            // 维度二：namespace + taskId 维度
+            String nsTaskFilter = "taskId=" + taskId + " and id!=\"" + taskImageIndex.getId() + "\"";
+            List<TaskImageIndex.MultiResult> nsTaskResult = knnSearch(namespace, taskImageIndex.getVector(), nsTaskFilter, "task");
+            multiResultList.addAll(nsTaskResult);
+            taskImageIndex.setMultiResult(JacksonUtils.toJsonString(multiResultList));
+        }
+
+        //先落库
+        pushDocument(list, Collections.emptyList());
+        //构建消息体
+        List<ImageMultiMsg> msgList = new ArrayList<>();
+        for (TaskImageIndex taskImageIndex : list) {
+            msgList.add(buildImageMultiMsg(taskImageMsg, taskImageIndex));
+        }
+        return msgList;
+    }
+
+    private void saveByTask(TaskImageMsg taskImageMsg) {
+        String[] images = taskImageMsg.getImage();
+
+        List<TaskImageIndex> taskImageList = fetchByTaskId(taskImageMsg.getNamespace(), taskImageMsg.getTaskId());
+        Map<String, TaskImageIndex> map = taskImageList.stream().collect(Collectors.toMap(TaskImageIndex::getImage, Function.identity()));
+
+        List<TaskImageIndex> addList = new ArrayList<>();
+        List<TaskImageIndex> deleteList = new ArrayList<>();
+        for (String image : images) {
+            TaskImageIndex exists = map.remove(image);
+            if (exists != null) {
+                TaskImageIndex index = buildTaskImageIndex(exists.getId(), taskImageMsg, image);
+                index.setVector(exists.getVector());
+                addList.add(index);
+            } else {
+                String id = UUID.randomUUID().toString().replaceAll("-", "");
+                addList.add(buildTaskImageIndex(id, taskImageMsg, image));
+            }
+        }
+
+        if (!map.isEmpty()) {
+            deleteList = new ArrayList<>(map.values());
+        }
+
+        pushDocument(addList, deleteList);
+    }
+
+    private void pushDocument(List<TaskImageIndex> addList, List<TaskImageIndex> deleteList) {
+        if (addList.isEmpty() && deleteList.isEmpty()) {
+            return;
+        }
+
+        String fullTableName = Contents.INSTANCE_ID + "_" + TABLE_NAME;
+
+        ArrayList<Map<String, ?>> documents = new ArrayList<>();
+        for (TaskImageIndex taskImageIndex : addList) {
+            documents.add(Map.of("cmd", "add", "fields", JacksonUtils.toMap(taskImageIndex)));
+        }
+        for (TaskImageIndex taskImageIndex : deleteList) {
+            documents.add(Map.of("cmd", "delete", "id", Map.of("id", taskImageIndex.getId())));
+        }
+
+        PushDocumentsRequest request = new PushDocumentsRequest();
+        request.setBody(documents);
+        try {
+            PushDocumentsResponse response = OpenSearchUtils.getClient().pushDocuments(fullTableName, "id", request);
+            String responseBody = response.getBody();
+            JsonNode bodyNode = JacksonUtils.toJsonNode(responseBody);
+
+            if (!bodyNode.has("status") || !"OK".equals(bodyNode.get("status").asText())) {
+                log.error("push document with multiResult failed, request:{}, response:{}", JacksonUtils.toJsonString(request), responseBody);
+                throw new RuntimeException("push document failed");
+            }
+        } catch (Exception e) {
+            log.error("push document with multiResult exception, request:{}", JacksonUtils.toJsonString(request), e);
+            throw new RuntimeException(e);
+        }
+    }
+
+
+    private List<TaskImageIndex> fetchByTaskId(String namespace, Long taskId) {
+        try {
+            FetchRequest request = new FetchRequest();
+            request.setTableName(TABLE_NAME);
+            request.setFilter(String.format("taskId:%d AND namespace:\"%s\"", taskId, namespace));
+            request.setIncludeVector(true);
+
+            SearchResponse response = OpenSearchUtils.getClient().fetch(request);
+            JsonNode bodyNode = JacksonUtils.toJsonNode(response.getBody());
+            if (bodyNode.has("errorCode")) {
+                throw new RuntimeException(String.format("查询失败,namespace:[%s], taskId:%d ", namespace, taskId));
+            }
+
+            int totalCount = bodyNode.get("totalCount").asInt();
+            if (totalCount == 0) {
+                return Collections.emptyList();
+            }
+            return parseToModelList(bodyNode.get("result"));
+        } catch (Exception e) {
+            throw new RuntimeException(String.format("查询异常,namespace:[%s], taskId:%d ", namespace, taskId), e);
+        }
+    }
+
+    private List<TaskImageIndex> parseToModelList(JsonNode result) {
+        List<TaskImageIndex> list = new ArrayList<>();
+        for (JsonNode jsonNode : result) {
+            list.add(parseToModel(jsonNode));
+        }
+        return list;
+    }
+
+    private TaskImageIndex parseToModel(JsonNode result) {
+        TaskImageIndex index = new TaskImageIndex();
+        if (result.has("id")) {
+            index.setId(result.get("id").asText());
+        }
+        if (result.has("namespace")) {
+            index.setNamespace(result.get("namespace").asText());
+        }
+        if (result.has("vector")) {
+            index.setVector(extractVector(result.get("vector")));
+        }
+
+        JsonNode fields = result.get("fields");
+        if (fields != null) {
+            if (fields.has("taskId")) {
+                index.setTaskId(fields.get("taskId").asLong());
+            }
+            if (fields.has("taskTypeId")) {
+                index.setTaskTypeId(fields.get("taskTypeId").asInt());
+            }
+            if (fields.has("userId")) {
+                index.setUserId(fields.get("userId").asLong());
+            }
+            if (fields.has("status")) {
+                index.setStatus(fields.get("status").asInt());
+            }
+            if (fields.has("image")) {
+                index.setImage(fields.get("image").asText());
+            }
+            if (fields.has("taskTime")) {
+                index.setTaskTime(fields.get("taskTime").asText());
+            }
+            if (fields.has("multiResult")) {
+                index.setMultiResult(fields.get("multiResult").asText());
+            }
+        }
+        return index;
+    }
+
+
+    private List<TaskImageIndex.MultiResult> knnSearch(String namespace, List<Float> vector, String filter, String type) {
+        try {
+            QueryRequest knn = new QueryRequest();
+            knn.setNamespace(namespace);
+            knn.setVector(vector);
+            knn.setTopK(TOP_K);
+            knn.setFilter(filter);
+            knn.setScoreThreshold(SCORE_THRESHOLD);
+            knn.setOutputFields(List.of("id", "namespace", "fields"));
+
+            SearchRequest request = new SearchRequest();
+            request.setTableName(TABLE_NAME);
+            request.setKnn(knn);
+
+            SearchResponse response = OpenSearchUtils.getClient().search(request);
+            return parseToMultiResultList(response.getBody(), type);
+        } catch (Exception e) {
+            log.error("knnSearch exception, filter:{}", filter, e);
+            throw new RuntimeException("对比异常，filter:" + filter, e);
+        }
+    }
+
+    private List<TaskImageIndex.MultiResult> parseToMultiResultList(String body, String type) {
+        JsonNode bodyNode = JacksonUtils.toJsonNode(body);
+
+        JsonNode resultArray;
+        if (bodyNode == null || !bodyNode.has("result")
+                || !(resultArray = bodyNode.get("result")).isArray() || resultArray.isEmpty()) {
+            return Collections.emptyList();
+        }
+
+        List<TaskImageIndex.MultiResult> list = new ArrayList<>();
+        for (JsonNode jsonNode : resultArray) {
+            TaskImageIndex.MultiResult multiResult = new TaskImageIndex.MultiResult();
+            JsonNode fields = jsonNode.get("fields");
+            multiResult.setTaskId(fields.get("taskId").asLong());
+            multiResult.setImage(fields.get("image").asText());
+
+            if (jsonNode.has("score")) {
+                multiResult.setScore(jsonNode.get("score").asDouble());
+            } else if (jsonNode.has("similarityScore")) {
+                multiResult.setScore(jsonNode.get("similarityScore").asDouble());
+            }
+            multiResult.setType(type);
+        }
+        return list;
+    }
+
+    private TaskImageIndex buildTaskImageIndex(String id, TaskImageMsg msg, String image) {
+        TaskImageIndex index = new TaskImageIndex();
+        index.setId(id);
+        index.setNamespace(msg.getNamespace());
+        index.setTaskId(msg.getTaskId());
+        index.setTaskTypeId(msg.getTaskTypeId());
+        index.setUserId(msg.getUserId());
+        index.setStatus(msg.getStatus());
+        index.setTaskTime(msg.getTaskTime());
+        index.setImage(image);
+        return index;
+    }
+
+    private List<Float> extractVector(JsonNode vectorNode) {
+        if (vectorNode == null || !vectorNode.isArray() || vectorNode.isEmpty()) {
+            return null;
+        }
+        List<Float> vector = new ArrayList<>(vectorNode.size());
+        for (JsonNode v : vectorNode) {
+            vector.add((float) v.asDouble());
+        }
+        return vector;
+    }
+
+    private ImageMultiMsg buildImageMultiMsg(TaskImageMsg taskImageMsg, TaskImageIndex taskImageIndex) {
+        ImageMultiMsg msg = new ImageMultiMsg();
+        msg.setAlgoConfigId(taskImageMsg.getAlgoConfigId());
+        msg.setTaskId(taskImageIndex.getTaskId());
+        msg.setImage(taskImageIndex.getImage());
+
+        if (taskImageIndex.getMultiResult() != null) {
+            TaskImageIndex.MultiResult[] results = JacksonUtils.parseObject(taskImageIndex.getMultiResult(), TaskImageIndex.MultiResult[].class);
+            if (results != null) {
+                List<ImageMultiMsg.MultiInfo> multiInfoList = new ArrayList<>();
+                for (TaskImageIndex.MultiResult r : results) {
+                    ImageMultiMsg.MultiInfo info = new ImageMultiMsg.MultiInfo();
+                    info.setTaskId(r.getTaskId());
+                    info.setImage(r.getImage());
+                    info.setScore(r.getScore());
+                    multiInfoList.add(info);
+                }
+                msg.setMultiResult(multiInfoList);
+            }
+        }
+        return msg;
+    }
+
+}

fn-opensearch/src/main/java/net/yaoyi/pipeline/opensearch/OpenSearchUtils.java

@@ -0,0 +1,28 @@
+package net.yaoyi.pipeline.opensearch;
+
+import com.aliyun.ha3engine.vector.Client;
+import com.aliyun.ha3engine.vector.models.Config;
+
+public class OpenSearchUtils {
+    private OpenSearchUtils() {
+    }
+
+    private static final Client client;
+
+    static {
+        try {
+            Config config = new Config();
+            config.setInstanceId(Contents.INSTANCE_ID);
+            config.setAccessUserName(Contents.ACCESS_USER_NAME);
+            config.setAccessPassWord(Contents.ACCESS_PASS_WORD);
+            config.setEndpoint(Contents.ENDPOINT);
+            client = new Client(config);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public static Client getClient() {
+        return client;
+    }
+}

fn-rabbitmq/pom.xml

@@ -0,0 +1,54 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>net.yaoyi.pipeline</groupId>
+        <artifactId>yyc-pipeline</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+    <artifactId>fn-rabbitmq</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-simple</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.aliyun.fc.runtime</groupId>
+            <artifactId>fc-java-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>net.yaoyi.pipeline</groupId>
+            <artifactId>yyc-pipeline-model</artifactId>
+            <version>1.0.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>com.rabbitmq</groupId>
+            <artifactId>amqp-client</artifactId>
+            <version>5.31.0</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.squareup.okhttp3</groupId>
+            <artifactId>okhttp</artifactId>
+            <version>4.12.0</version>
+        </dependency>
+    </dependencies>
+
+</project>

fn-rabbitmq/src/main/java/net/yao/pipeline/rabbitmq/RabbitConsumer.java

@@ -0,0 +1,123 @@
+package net.yao.pipeline.rabbitmq;
+
+import com.aliyun.fc.runtime.Context;
+import com.aliyun.fc.runtime.PojoRequestHandler;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.DeliverCallback;
+import lombok.extern.slf4j.Slf4j;
+import okhttp3.*;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * RabbitMQ 消费者 - 阿里云函数计算实现
+ */
+@Slf4j
+public class RabbitConsumer implements PojoRequestHandler<String, Void> {
+
+    private static final String RABBITMQ_HOST = "rabbitmq-cn-39y4twfud01-cn-beijing-vpc.mq.amqp.aliyuncs.com";
+    private static final int RABBITMQ_PORT = 5672;
+    private static final String RABBITMQ_USERNAME = "MjpyYWJiaXRtcS1jbi0zOXk0dHdmdWQwMTpMVEFJNXQ2SDR1OGFxb3A3cGViNkRSOFo=";
+    private static final String RABBITMQ_PASSWORD = "MDdCQ0FFNjFEMTQyMEZDN0UyMThDNkM4MDgwMkI5QzgyNjAzQTdGQzoxNzgyMTk2NzQxNjcy";
+    private static final String QUEUE_NAME = "vector.search.task.img.dul.check.queue";
+    private static final String V_HOST = "test";
+
+    private static final String PIPELINE_API = "http://1861991840278303.eventbridge.cn-beijing.aliyuncs.com/webhook/putEvents?token=63b99b24a3d545d880fa86f57c5ae56cae00c81d2fc74d058e9ace260515163132f6e27deeaa4bd4bc8e2a21cf6b6d6517b8d1e6d33c4de89f00c3524a54095b";
+
+    private static ConnectionFactory factory;
+    private static Connection connection;
+    private static Channel channel;
+    private static final OkHttpClient httpClient = new OkHttpClient.Builder()
+            .connectTimeout(10, TimeUnit.SECONDS)
+            .readTimeout(30, TimeUnit.SECONDS)
+            .writeTimeout(30, TimeUnit.SECONDS)
+            .build();
+    private static final MediaType JSON = MediaType.get("application/json; charset=utf-8");
+
+    @Override
+    public Void handleRequest(String input, Context context) {
+        log.info("函数计算启动, Request ID: {}", context.getRequestId());
+
+        try {
+            // 初始化 RabbitMQ 连接
+            initRabbitMQ();
+            // 启动消费者
+            startConsumer();
+            log.info("RabbitMQ 消费者启动成功");
+        } catch (Exception e) {
+            log.error("启动 RabbitMQ 消费者失败", e);
+            throw new RuntimeException("启动失败: " + e.getMessage(), e);
+        }
+        return null;
+    }
+
+    /**
+     * 初始化 RabbitMQ 连接
+     */
+    private void initRabbitMQ() throws IOException, TimeoutException {
+        if (factory == null) {
+            factory = new ConnectionFactory();
+            factory.setHost(RABBITMQ_HOST);
+            factory.setPort(RABBITMQ_PORT);
+            factory.setUsername(RABBITMQ_USERNAME);
+            factory.setPassword(RABBITMQ_PASSWORD);
+            factory.setVirtualHost(V_HOST);
+
+            log.info("连接 RabbitMQ: {}:{}", RABBITMQ_HOST, RABBITMQ_PORT);
+            connection = factory.newConnection();
+            channel = connection.createChannel();
+
+            // 声明队列
+            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
+            log.info("队列声明成功: {}", QUEUE_NAME);
+        }
+    }
+
+    /**
+     * 启动消费者
+     */
+    private void startConsumer() throws IOException {
+        channel.basicQos(1);
+
+        log.info("开始消费队列: {}", QUEUE_NAME);
+        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
+            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
+            log.info("收到消息: {}", message);
+
+            try {
+                processMessage(message);
+                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
+                log.info("消息处理成功并已确认");
+            } catch (Exception e) {
+                log.error("处理消息失败", e);
+                channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
+                log.warn("消息已重新入队");
+            }
+        };
+
+        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {
+            log.info("消费者被取消: {}", consumerTag);
+        });
+    }
+
+    private void processMessage(String message) {
+        RequestBody body = RequestBody.create(message, JSON);
+        Request request = new Request.Builder().url(PIPELINE_API).put(body).build();
+
+        try (Response response = httpClient.newCall(request).execute()) {
+            if (response.isSuccessful()) {
+                String responseBody = response.body() != null ? response.body().string() : "";
+                log.info("Pipline API 请求成功:{}", responseBody);
+            } else {
+                log.info("Pipline API 请求失败:{}", response.code());
+            }
+        } catch (Exception e) {
+            log.error("Pipline API 请求异常", e);
+        }
+    }
+}

pom.xml

@@ -0,0 +1,117 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>net.yaoyi.pipeline</groupId>
+    <artifactId>yyc-pipeline</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <packaging>pom</packaging>
+
+    <modules>
+        <module>fn-rabbitmq</module>
+        <module>yyc-pipeline-model</module>
+        <module>fn-opensearch</module>
+        <module>fn-common</module>
+    </modules>
+
+    <properties>
+        <maven.compiler.source>11</maven.compiler.source>
+        <maven.compiler.target>11</maven.compiler.target>
+        <lombok.version>1.18.30</lombok.version>
+        <org.mapstruct.version>1.4.2.Final</org.mapstruct.version>
+        <guava.version>33.0.0-jre</guava.version>
+        <lang3.version>3.12.0</lang3.version>
+        <slf4j.version>1.7.30</slf4j.version>
+    </properties>
+
+    <distributionManagement>
+        <repository>
+            <id>iqiyi-maven-release</id>
+            <name>iqiyi-maven-release</name>
+            <url>http://jfrog.cloud.qiyi.domain/iqiyi-maven-release</url>
+        </repository>
+        <snapshotRepository>
+            <id>iqiyi-maven-snapshot</id>
+            <name>iqiyi-maven-snapshot</name>
+            <url>http://jfrog.cloud.qiyi.domain/iqiyi-maven-snapshot</url>
+        </snapshotRepository>
+    </distributionManagement>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.slf4j</groupId>
+                <artifactId>slf4j-api</artifactId>
+                <version>${slf4j.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.slf4j</groupId>
+                <artifactId>slf4j-simple</artifactId>
+                <version>${slf4j.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.projectlombok</groupId>
+                <artifactId>lombok</artifactId>
+                <version>${lombok.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>com.google.guava</groupId>
+                <artifactId>guava</artifactId>
+                <version>${guava.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.commons</groupId>
+                <artifactId>commons-lang3</artifactId>
+                <version>${lang3.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.commons</groupId>
+                <artifactId>commons-collections4</artifactId>
+                <version>4.4</version>
+            </dependency>
+
+            <dependency>
+                <groupId>com.fasterxml.jackson.core</groupId>
+                <artifactId>jackson-databind</artifactId>
+                <version>2.15.2</version>
+            </dependency>
+            <dependency>
+                <groupId>com.fasterxml.jackson.datatype</groupId>
+                <artifactId>jackson-datatype-jsr310</artifactId>
+                <version>2.15.2</version>
+            </dependency>
+            <dependency>
+                <groupId>org.projectlombok</groupId>
+                <artifactId>lombok</artifactId>
+                <version>${lombok.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>com.aliyun</groupId>
+                <artifactId>aliyun-sdk-ha3engine-vector</artifactId>
+                <version>1.1.18</version>
+            </dependency>
+            <dependency>
+                <groupId>com.aliyun.fc.runtime</groupId>
+                <artifactId>fc-java-core</artifactId>
+                <version>1.4.1</version>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>3.8.1</version>
+                <configuration>
+                    <source>${maven.compiler.source}</source>
+                    <target>${maven.compiler.target}</target>
+                    <encoding>UTF-8</encoding>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>

yyc-pipeline-model/pom.xml

@@ -0,0 +1,20 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>net.yaoyi.pipeline</groupId>
+        <artifactId>yyc-pipeline</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>yyc-pipeline-model</artifactId>
+    <dependencies>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

yyc-pipeline-model/src/main/java/net/yaoyi/pipeline/model/AlgoBaseInfo.java

@@ -0,0 +1,8 @@
+package net.yaoyi.pipeline.model;
+
+import lombok.Data;
+
+@Data
+public class AlgoBaseInfo {
+    private Integer algoConfigId;
+}

yyc-pipeline-model/src/main/java/net/yaoyi/pipeline/model/index/TaskImageIndex.java

@@ -0,0 +1,27 @@
+package net.yaoyi.pipeline.model.index;
+
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class TaskImageIndex {
+    private String id;
+    private String namespace;
+    private Long taskId;
+    private Integer taskTypeId;
+    private Long userId;
+    private Integer status;
+    private String taskTime;
+    private String image;
+    private List<Float> vector;
+    private String multiResult;
+
+    @Data
+    public static class MultiResult {
+        private Long taskId;
+        private String image;
+        private double score;
+        private String type;
+    }
+}

yyc-pipeline-model/src/main/java/net/yaoyi/pipeline/model/mq/ImageMultiMsg.java

@@ -0,0 +1,20 @@
+package net.yaoyi.pipeline.model.mq;
+
+import lombok.Data;
+import net.yaoyi.pipeline.model.AlgoBaseInfo;
+
+import java.util.List;
+
+@Data
+public class ImageMultiMsg extends AlgoBaseInfo {
+    private Long taskId;
+    private String image;
+    private List<MultiInfo> multiResult;
+
+    @Data
+    public static class MultiInfo {
+        private Long taskId;
+        private String image;
+        private double score;
+    }
+}

yyc-pipeline-model/src/main/java/net/yaoyi/pipeline/model/mq/TaskImageMsg.java

@@ -0,0 +1,15 @@
+package net.yaoyi.pipeline.model.mq;
+
+import lombok.Data;
+import net.yaoyi.pipeline.model.AlgoBaseInfo;
+
+@Data
+public class TaskImageMsg extends AlgoBaseInfo {
+    private String namespace;
+    private Long taskId;
+    private Integer taskTypeId;
+    private Long userId;
+    private Integer status;
+    private String taskTime;
+    private String[] image;
+}