feat: 支持工作流知识库多库检索

This commit is contained in:
2026-09-04 17:40:55 +08:00
parent 386cebf342
commit 9722bea701
56 changed files with 5539 additions and 154 deletions

View File

@@ -24,6 +24,7 @@ public class AiMilvusConfig extends MilvusVectorStoreConfig {
config.setPoolMaxWaitMillis(getPoolMaxWaitMillis());
config.setPoolEvictionIntervalMillis(getPoolEvictionIntervalMillis());
config.setPoolMinEvictableIdleMillis(getPoolMinEvictableIdleMillis());
config.setSearchTimeoutMillis(getSearchTimeoutMillis());
return config;
}
}

View File

@@ -14,7 +14,8 @@ import tech.easyflow.ai.documentimport.task.DocumentImportStatusBroadcastPropert
DocumentImportBulkProperties.class,
DocumentImportParseMonitorProperties.class,
DocumentImportStatusBroadcastProperties.class,
RagHealthProperties.class
RagHealthProperties.class,
MultiKnowledgeRetrievalProperties.class
})
@AutoConfiguration
public class AiModuleConfig {

View File

@@ -11,6 +11,7 @@ public class EasyFlowThreadPoolProperties {
private Pool sse = new Pool(4, 16, 2000, 30, true);
private Pool documentImport = new Pool(2, 4, 200, 60, true);
private Pool agentAsyncTool = new Pool(2, 8, 200, 60, true);
private Pool knowledgeRetrieval = new Pool(4, 8, 64, 30, true);
/**
* 获取 SSE 线程池配置。
@@ -66,6 +67,14 @@ public class EasyFlowThreadPoolProperties {
this.agentAsyncTool = agentAsyncTool;
}
public Pool getKnowledgeRetrieval() {
return knowledgeRetrieval;
}
public void setKnowledgeRetrieval(Pool knowledgeRetrieval) {
this.knowledgeRetrieval = knowledgeRetrieval;
}
/**
* 线程池配置项。
*/

View File

@@ -0,0 +1,108 @@
package tech.easyflow.ai.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
import java.time.Duration;
/**
* 工作流多知识库向量检索的资源与时限配置。
*/
@ConfigurationProperties(prefix = "easyflow.ai.knowledge.multi-retrieval")
public class MultiKnowledgeRetrievalProperties {
private int maxSources = 8;
private int candidateMultiplier = 5;
private int perSourceCandidateLimit = 50;
private int totalCandidateLimit = 400;
private double minVectorScore = 0.6D;
private Duration perSourceTimeout = Duration.ofSeconds(10);
private Duration totalTimeout = Duration.ofSeconds(20);
public int getMaxSources() {
return maxSources;
}
public void setMaxSources(int maxSources) {
this.maxSources = maxSources;
}
public int getCandidateMultiplier() {
return candidateMultiplier;
}
public void setCandidateMultiplier(int candidateMultiplier) {
this.candidateMultiplier = candidateMultiplier;
}
public int getPerSourceCandidateLimit() {
return perSourceCandidateLimit;
}
public void setPerSourceCandidateLimit(int perSourceCandidateLimit) {
this.perSourceCandidateLimit = perSourceCandidateLimit;
}
public int getTotalCandidateLimit() {
return totalCandidateLimit;
}
public void setTotalCandidateLimit(int totalCandidateLimit) {
this.totalCandidateLimit = totalCandidateLimit;
}
public double getMinVectorScore() {
return minVectorScore;
}
public void setMinVectorScore(double minVectorScore) {
this.minVectorScore = minVectorScore;
}
public Duration getPerSourceTimeout() {
return perSourceTimeout;
}
public void setPerSourceTimeout(Duration perSourceTimeout) {
this.perSourceTimeout = perSourceTimeout;
}
public Duration getTotalTimeout() {
return totalTimeout;
}
public void setTotalTimeout(Duration totalTimeout) {
this.totalTimeout = totalTimeout;
}
/**
* 启动期校验全部有界配置。
*/
public void validate() {
if (maxSources < 2 || maxSources > 64) {
throw new IllegalArgumentException("多知识库最大来源数必须在 2 到 64 之间");
}
if (candidateMultiplier < 1 || candidateMultiplier > 20) {
throw new IllegalArgumentException("多知识库候选倍率必须在 1 到 20 之间");
}
if (perSourceCandidateLimit < 1 || perSourceCandidateLimit > 1000) {
throw new IllegalArgumentException("单知识库候选上限必须在 1 到 1000 之间");
}
long derivedCandidateLimit = (long) maxSources
* perSourceCandidateLimit;
if (totalCandidateLimit < derivedCandidateLimit
|| totalCandidateLimit > 10000) {
throw new IllegalArgumentException(
"多知识库总候选上限不能小于最大来源数与单库候选上限的乘积");
}
if (!Double.isFinite(minVectorScore) || minVectorScore < 0D || minVectorScore > 1D) {
throw new IllegalArgumentException("多知识库向量阈值必须在 0 到 1 之间");
}
if (perSourceTimeout == null || perSourceTimeout.isZero() || perSourceTimeout.isNegative()) {
throw new IllegalArgumentException("单知识库超时时间必须大于 0");
}
if (totalTimeout == null || totalTimeout.isZero() || totalTimeout.isNegative()
|| totalTimeout.compareTo(perSourceTimeout) < 0) {
throw new IllegalArgumentException("节点总超时时间不能小于单知识库超时时间");
}
}
}

View File

@@ -104,4 +104,28 @@ public class ThreadPoolConfig {
executor.initialize();
return executor;
}
/**
* 创建工作流多知识库检索线程池。
*
* @return 多知识库检索线程池
*/
@Bean(name = "knowledgeRetrievalExecutor")
public ThreadPoolTaskExecutor knowledgeRetrievalExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
EasyFlowThreadPoolProperties.Pool pool = properties.getKnowledgeRetrieval();
executor.setCorePoolSize(pool.getCoreSize());
executor.setMaxPoolSize(pool.getMaxSize());
executor.setQueueCapacity(pool.getQueueCapacity());
executor.setKeepAliveSeconds(pool.getKeepAliveSeconds());
executor.setAllowCoreThreadTimeOut(pool.isAllowCoreThreadTimeout());
executor.setThreadNamePrefix("knowledge-retrieval-");
executor.setRejectedExecutionHandler((runnable, executorService) -> {
log.error("多知识库检索线程池过载active={}, queue={}",
executorService.getActiveCount(), executorService.getQueue().size());
throw new BusinessException("知识库检索繁忙,请稍后重试");
});
executor.initialize();
return executor;
}
}

View File

@@ -5,6 +5,7 @@ import com.alibaba.fastjson2.JSONObject;
import com.easyagents.flow.core.chain.Chain;
import com.easyagents.flow.core.knowledge.Knowledge;
import com.easyagents.flow.core.knowledge.KnowledgeProvider;
import com.easyagents.flow.core.knowledge.KnowledgeSearchRequest;
import com.easyagents.flow.core.node.KnowledgeNode;
import org.springframework.stereotype.Component;
import tech.easyflow.ai.rag.KnowledgeRetrievalRequest;
@@ -14,6 +15,7 @@ import tech.easyflow.ai.service.DocumentCollectionService;
import javax.annotation.Resource;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -26,6 +28,9 @@ public class KnowledgeProviderImpl implements KnowledgeProvider {
@Resource
private DocumentCollectionService documentCollectionService;
@Resource
private WorkflowMultiKnowledgeRetrievalService multiKnowledgeRetrievalService;
/**
* 获取知识库检索器。
*
@@ -44,26 +49,88 @@ public class KnowledgeProviderImpl implements KnowledgeProvider {
int limit,
KnowledgeNode knowledgeNode,
Chain chain) {
KnowledgeRetrievalRequest request = new KnowledgeRetrievalRequest();
request.setKnowledgeId(new BigInteger(id.toString()));
request.setQuery(keyword);
request.setLimit(limit);
request.setRetrievalMode(KnowledgeRetrievalModes.parse(knowledgeNode.getRetrievalMode()));
request.setCallerType("WORKFLOW");
request.setCallerId(knowledgeNode.getId());
List<Document> documents = documentCollectionService.search(request);
if (limit > 0 && documents.size() > limit) {
documents = new ArrayList<>(documents.subList(0, limit));
}
List<Map<String, Object>> res = new ArrayList<>();
for (Document document : documents) {
res.add(toWorkflowDocument(document, id));
}
return res;
return searchSingle(
new BigInteger(id.toString()),
keyword,
limit,
knowledgeNode);
}
};
}
@Override
public Map<String, Object> search(KnowledgeSearchRequest request) {
if (request == null || request.getKnowledgeIds().isEmpty()) {
return null;
}
List<BigInteger> knowledgeIds = new ArrayList<>();
for (Object id : request.getKnowledgeIds()) {
try {
knowledgeIds.add(new BigInteger(String.valueOf(id)));
} catch (RuntimeException exception) {
throw new IllegalArgumentException("知识库 ID 无效: " + id, exception);
}
}
if (knowledgeIds.size() == 1) {
List<Map<String, Object>> documents = searchSingle(
knowledgeIds.get(0),
request.getKeyword(),
request.getLimit(),
request.getKnowledgeNode());
return buildOutputs(documents);
}
if (!"VECTOR".equalsIgnoreCase(request.getRetrievalMode())) {
throw new IllegalArgumentException("多知识库检索仅支持 VECTOR 模式");
}
MultiKnowledgeRetrievalResult result = multiKnowledgeRetrievalService.search(
knowledgeIds,
request.getKeyword(),
request.getLimit(),
request.getKnowledgeNode() == null
? null
: request.getKnowledgeNode().getId(),
request.getChain());
List<Map<String, Object>> documents = new ArrayList<>();
for (Document document : result.getDocuments()) {
documents.add(toWorkflowDocument(
document,
document.getMetadata("knowledgeId", null)));
}
return buildOutputs(documents);
}
private List<Map<String, Object>> searchSingle(
BigInteger knowledgeId,
String keyword,
int limit,
KnowledgeNode knowledgeNode) {
KnowledgeRetrievalRequest request = new KnowledgeRetrievalRequest();
request.setKnowledgeId(knowledgeId);
request.setQuery(keyword);
request.setLimit(limit);
request.setRetrievalMode(KnowledgeRetrievalModes.parse(
knowledgeNode == null
? null
: knowledgeNode.getRetrievalMode()));
request.setCallerType("WORKFLOW");
request.setCallerId(knowledgeNode == null ? null : knowledgeNode.getId());
List<Document> documents = documentCollectionService.search(request);
if (limit > 0 && documents.size() > limit) {
documents = new ArrayList<>(documents.subList(0, limit));
}
List<Map<String, Object>> result = new ArrayList<>();
for (Document document : documents) {
result.add(toWorkflowDocument(document, knowledgeId));
}
return result;
}
private Map<String, Object> buildOutputs(List<Map<String, Object>> documents) {
Map<String, Object> outputs = new LinkedHashMap<>();
outputs.put("documents", documents);
return outputs;
}
/**
* 将检索文档转换为工作流稳定对象,并保留旧序列化字段。
*

View File

@@ -0,0 +1,46 @@
package tech.easyflow.ai.easyagentsflow.knowledge;
import com.easyagents.core.document.Document;
import java.util.Collections;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
/**
* 多知识库检索结果及可观察状态。
*/
public class MultiKnowledgeRetrievalResult {
private final List<Document> documents;
private final Map<String, Object> summary;
private final List<Map<String, Object>> sourceStatuses;
public MultiKnowledgeRetrievalResult(
List<Document> documents,
Map<String, Object> summary,
List<Map<String, Object>> sourceStatuses) {
this.documents = documents == null
? Collections.emptyList()
: Collections.unmodifiableList(new ArrayList<>(documents));
this.summary = summary == null
? Collections.emptyMap()
: Collections.unmodifiableMap(new LinkedHashMap<>(summary));
this.sourceStatuses = sourceStatuses == null
? Collections.emptyList()
: Collections.unmodifiableList(new ArrayList<>(sourceStatuses));
}
public List<Document> getDocuments() {
return documents;
}
public Map<String, Object> getSummary() {
return summary;
}
public List<Map<String, Object>> getSourceStatuses() {
return sourceStatuses;
}
}

View File

@@ -0,0 +1,579 @@
package tech.easyflow.ai.easyagentsflow.knowledge;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import org.springframework.stereotype.Service;
import tech.easyflow.ai.entity.DocumentCollection;
import tech.easyflow.ai.entity.Model;
import tech.easyflow.ai.service.DocumentCollectionService;
import tech.easyflow.ai.service.ModelService;
import tech.easyflow.common.web.exceptions.BusinessException;
import java.math.BigInteger;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HexFormat;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
/**
* 工作流多知识库 Embedding 契约校验与发布快照服务。
*/
@Service
public class WorkflowKnowledgeContractService {
public static final String SNAPSHOT_KEY = "knowledgeContracts";
private final DocumentCollectionService documentCollectionService;
private final ModelService modelService;
public WorkflowKnowledgeContractService(
DocumentCollectionService documentCollectionService,
ModelService modelService) {
this.documentCollectionService = documentCollectionService;
this.modelService = modelService;
}
/**
* 批量计算设计器中真正具备向量检索条件的知识库。
*
* @param collections 候选知识库
* @param tenantId 当前租户
* @return 可用知识库 ID
*/
public Set<BigInteger> findVectorReadyKnowledgeIds(
List<DocumentCollection> collections,
BigInteger tenantId) {
if (collections == null || collections.isEmpty()) {
return Collections.emptySet();
}
Map<BigInteger, Model> models = loadModels(collections);
Set<BigInteger> result = new LinkedHashSet<>();
for (DocumentCollection collection : collections) {
if (isVectorReady(collection, models, tenantId)) {
result.add(collection.getId());
}
}
return result;
}
/**
* 校验保存阶段全部多知识库节点的 Embedding 契约。
*
* @param knowledgeGroups 各知识库节点的有序引用
* @param tenantId 当前租户
*/
public void assertMultiKnowledgeContracts(
List<List<BigInteger>> knowledgeGroups,
BigInteger tenantId) {
resolveContext(onlyMultiGroups(knowledgeGroups), tenantId);
}
/**
* 解析工作流引用的全部知识库,并校验存在性与租户归属。
*
* @param content 工作流内容
* @param tenantId 工作流租户
* @return 按工作流首次引用顺序排列的知识库
*/
public List<DocumentCollection> resolveReferencedCollections(
String content,
BigInteger tenantId) {
List<List<BigInteger>> groups = readKnowledgeGroups(content);
if (groups.isEmpty()) {
return List.of();
}
Set<BigInteger> ids = new LinkedHashSet<>();
groups.forEach(ids::addAll);
Map<BigInteger, DocumentCollection> collections = loadCollections(
ids, tenantId);
return ids.stream().map(collections::get).toList();
}
/**
* 校验工作流内容并生成不包含凭据的多知识库发布契约。
*
* @param content 工作流内容
* @param tenantId 工作流租户
* @return 稳定的发布契约列表
*/
public List<Map<String, Object>> buildSnapshotContracts(
String content,
BigInteger tenantId) {
List<List<BigInteger>> groups = readKnowledgeGroups(content).stream()
.filter(group -> group.size() > 1)
.toList();
ContractContext context = resolveContext(groups, tenantId);
if (groups.isEmpty()) {
return List.of();
}
Set<BigInteger> orderedIds = new LinkedHashSet<>();
groups.forEach(orderedIds::addAll);
List<Map<String, Object>> result = new ArrayList<>();
for (BigInteger id : orderedIds) {
result.add(toContract(
context.collections.get(id),
context.models,
tenantId));
}
return List.copyOf(result);
}
/**
* 审批真正发布前重新校验提交快照与当前知识库契约是否一致。
*
* @param resourceSnapshot 待发布工作流快照
*/
public void assertSnapshotCurrent(Map<String, Object> resourceSnapshot) {
if (resourceSnapshot == null) {
throw new BusinessException("工作流发布快照不能为空");
}
String content = text(resourceSnapshot.get("content"));
BigInteger tenantId = bigInteger(resourceSnapshot.get("tenantId"));
List<Map<String, Object>> current = buildSnapshotContracts(
content, tenantId);
List<Map<String, Object>> frozen = readSnapshotContracts(
resourceSnapshot.get(SNAPSHOT_KEY));
if (!current.equals(frozen)) {
throw new BusinessException("工作流引用的知识库 Embedding 配置已变化,请重新提交发布");
}
}
/**
* 已发布工作流运行时校验本节点引用仍与发布契约一致。
*
* @param publishedSnapshot 已发布工作流快照
* @param collections 当前节点知识库
*/
public Model assertPublishedContracts(
Map<String, Object> publishedSnapshot,
List<DocumentCollection> collections) {
if (collections == null || collections.size() < 2) {
throw new BusinessException("多知识库检索至少需要两个知识库");
}
if (publishedSnapshot == null || publishedSnapshot.isEmpty()) {
throw new BusinessException("已发布工作流缺少知识库契约");
}
BigInteger tenantId = bigInteger(publishedSnapshot.get("tenantId"));
ContractContext context = resolveLoadedContext(collections, tenantId);
Model embeddingModel = requireCompatibleEmbeddingModel(
collections, context, tenantId);
Map<String, Map<String, Object>> frozenById = new LinkedHashMap<>();
for (Map<String, Object> contract : readSnapshotContracts(
publishedSnapshot.get(SNAPSHOT_KEY))) {
frozenById.put(text(contract.get("knowledgeId")), contract);
}
for (DocumentCollection collection : collections) {
Map<String, Object> frozen = frozenById.get(
String.valueOf(collection.getId()));
Map<String, Object> current = toContract(
collection, context.models, tenantId);
if (!current.equals(frozen)) {
throw new BusinessException("已发布工作流的知识库 Embedding 配置已变化,请重新发布");
}
}
return embeddingModel;
}
/**
* 校验 Agent 冻结定义中的契约指纹与当前有效配置一致。
*
* @param expectedFingerprint 冻结定义中的契约指纹
* @param collections 定义引用的全部多知识库
* @param tenantId 冻结定义租户
*/
public Model assertFrozenContractFingerprint(
String expectedFingerprint,
List<DocumentCollection> allCollections,
List<DocumentCollection> currentCollections,
BigInteger tenantId) {
ContractContext context = resolveLoadedContext(
allCollections, tenantId);
Model embeddingModel = requireCompatibleEmbeddingModel(
currentCollections, context, tenantId);
List<Map<String, Object>> current = new ArrayList<>();
for (DocumentCollection collection : allCollections) {
current.add(toContract(collection, context.models, tenantId));
}
if (!Objects.equals(expectedFingerprint, fingerprint(current))) {
throw new BusinessException("Agent 冻结工作流的知识库 Embedding 配置已变化,请重新发布 Agent");
}
return embeddingModel;
}
/**
* 使用已加载的知识库快照校验当前多库契约,并返回同一次批量读取的模型快照。
*
* @param collections 当前节点知识库快照
* @param tenantId 执行租户
* @return 已校验的 Embedding 模型快照
*/
public Model requireCompatibleEmbeddingModel(
List<DocumentCollection> collections,
BigInteger tenantId) {
ContractContext context = resolveLoadedContext(collections, tenantId);
return requireCompatibleEmbeddingModel(
collections, context, tenantId);
}
/**
* 计算发布快照中规范知识库契约的稳定指纹。
*
* @param contracts 发布快照契约
* @return SHA-256 十六进制指纹
*/
public String fingerprintSnapshotContracts(Object contracts) {
return fingerprint(readSnapshotContracts(contracts));
}
private ContractContext resolveContext(
List<List<BigInteger>> groups,
BigInteger tenantId) {
if (groups == null || groups.isEmpty()) {
return ContractContext.empty();
}
if (tenantId == null) {
throw new BusinessException("工作流租户不能为空");
}
Set<BigInteger> ids = new LinkedHashSet<>();
groups.forEach(ids::addAll);
Map<BigInteger, DocumentCollection> collections = loadCollections(
ids, tenantId);
Map<BigInteger, Model> models = loadModels(collections.values());
for (List<BigInteger> group : groups) {
assertCompatibleGroup(group, collections, models, tenantId);
}
return new ContractContext(collections, models);
}
private ContractContext resolveLoadedContext(
List<DocumentCollection> collections,
BigInteger tenantId) {
if (collections == null || collections.isEmpty()) {
throw new BusinessException("工作流引用的知识库不存在或已失效");
}
if (tenantId == null) {
throw new BusinessException("工作流租户不能为空");
}
Map<BigInteger, DocumentCollection> byId = new LinkedHashMap<>();
for (DocumentCollection collection : collections) {
if (collection == null || collection.getId() == null) {
throw new BusinessException("工作流引用的知识库不存在或已失效");
}
if (!Objects.equals(tenantId, collection.getTenantId())) {
throw new BusinessException("工作流引用了其他租户的知识库");
}
if (byId.put(collection.getId(), collection) != null) {
throw new BusinessException("知识库节点不能重复选择同一知识库");
}
}
return new ContractContext(byId, loadModels(collections));
}
private Model requireCompatibleEmbeddingModel(
List<DocumentCollection> collections,
ContractContext context,
BigInteger tenantId) {
if (collections == null || collections.size() < 2) {
throw new BusinessException("多知识库检索至少需要两个知识库");
}
List<BigInteger> ids = collections.stream()
.map(DocumentCollection::getId)
.toList();
assertCompatibleGroup(
ids, context.collections, context.models, tenantId);
Model embeddingModel = context.models.get(
collections.get(0).getVectorEmbedModelId());
if (embeddingModel == null) {
throw new BusinessException("知识库 Embedding 模型不存在或已失效");
}
return embeddingModel;
}
private void assertCompatibleGroup(
List<BigInteger> group,
Map<BigInteger, DocumentCollection> collections,
Map<BigInteger, Model> models,
BigInteger tenantId) {
DocumentCollection first = collections.get(group.get(0));
if (!isVectorReady(first, models, tenantId)) {
throw new BusinessException("知识库未完成有效的向量检索配置: " + first.getTitle());
}
for (BigInteger id : group) {
DocumentCollection current = collections.get(id);
if (!isVectorReady(current, models, tenantId)) {
throw new BusinessException("知识库未完成有效的向量检索配置: " + current.getTitle());
}
if (!Objects.equals(
first.getVectorEmbedModelId(),
current.getVectorEmbedModelId())
|| !Objects.equals(
first.getDimensionOfVectorModel(),
current.getDimensionOfVectorModel())
|| !Objects.equals(
first.getTenantId(), current.getTenantId())) {
throw new BusinessException("多知识库检索要求使用相同的 Embedding 模型和向量维度");
}
}
}
private boolean isVectorReady(
DocumentCollection collection,
Map<BigInteger, Model> models,
BigInteger tenantId) {
if (collection == null
|| collection.getId() == null
|| !Objects.equals(collection.getTenantId(), tenantId)
|| !Boolean.TRUE.equals(collection.getVectorStoreEnable())
|| collection.getVectorEmbedModelId() == null
|| collection.getDimensionOfVectorModel() == null
|| collection.getDimensionOfVectorModel() <= 0
|| collection.getVectorStoreCollection() == null
|| collection.getVectorStoreCollection().isBlank()) {
return false;
}
Model model = models.get(collection.getVectorEmbedModelId());
return model != null
&& Model.MODEL_TYPES[1].equals(model.getModelType())
&& model.getModelProvider() != null
&& text(model.getModelProvider().getProviderType()) != null
&& Objects.equals(model.getTenantId(), tenantId);
}
private Map<BigInteger, Model> loadModels(
Collection<DocumentCollection> collections) {
Set<BigInteger> modelIds = new LinkedHashSet<>();
for (DocumentCollection collection : collections) {
if (collection != null && collection.getVectorEmbedModelId() != null) {
modelIds.add(collection.getVectorEmbedModelId());
}
}
if (modelIds.isEmpty()) {
return Collections.emptyMap();
}
List<Model> loaded = modelService.listModelInstances(modelIds);
Map<BigInteger, Model> result = new LinkedHashMap<>();
if (loaded != null) {
for (Model model : loaded) {
if (model != null && model.getId() != null) {
result.put(model.getId(), model);
}
}
}
return result;
}
private Map<String, Object> toContract(
DocumentCollection collection,
Map<BigInteger, Model> models,
BigInteger tenantId) {
if (!isVectorReady(collection, models, tenantId)) {
throw new BusinessException("知识库未完成有效的向量检索配置: "
+ (collection == null ? "" : collection.getTitle()));
}
Model model = models.get(collection.getVectorEmbedModelId());
Map<String, Object> contract = new LinkedHashMap<>();
contract.put("knowledgeId", String.valueOf(collection.getId()));
contract.put("tenantId", String.valueOf(collection.getTenantId()));
contract.put("embeddingModelId", String.valueOf(model.getId()));
contract.put("embeddingDimension", collection.getDimensionOfVectorModel());
contract.put("vectorStoreCollection", collection.getVectorStoreCollection());
contract.put("vectorStoreType", nullableText(collection.getVectorStoreType()));
contract.put("modelProviderId", nullableString(model.getProviderId()));
contract.put(
"modelProviderType",
nullableText(model.getModelProvider().getProviderType()));
contract.put("modelType", model.getModelType());
contract.put("modelName", nullableText(model.getModelName()));
contract.put("modelEndpoint", nullableText(model.getEndpoint()));
contract.put("modelRequestPath", nullableText(model.getRequestPath()));
return contract;
}
private Map<BigInteger, DocumentCollection> loadCollections(
Set<BigInteger> ids,
BigInteger tenantId) {
if (tenantId == null) {
throw new BusinessException("工作流租户不能为空");
}
List<DocumentCollection> loaded = documentCollectionService.listByIds(ids);
Map<BigInteger, DocumentCollection> collections = new LinkedHashMap<>();
if (loaded != null) {
for (DocumentCollection collection : loaded) {
if (collection != null && collection.getId() != null) {
collections.put(collection.getId(), collection);
}
}
}
if (collections.size() != ids.size()) {
throw new BusinessException("工作流引用的知识库不存在或已失效");
}
for (DocumentCollection collection : collections.values()) {
if (!Objects.equals(tenantId, collection.getTenantId())) {
throw new BusinessException("工作流引用了其他租户的知识库");
}
}
return collections;
}
private List<List<BigInteger>> readKnowledgeGroups(String content) {
if (content == null || content.isBlank()) {
return List.of();
}
JSONObject root;
try {
root = JSON.parseObject(content);
} catch (RuntimeException exception) {
throw new BusinessException("工作流内容不是合法JSON");
}
JSONArray nodes = root.getJSONArray("nodes");
if (nodes == null || nodes.isEmpty()) {
return List.of();
}
List<List<BigInteger>> groups = new ArrayList<>();
for (int index = 0; index < nodes.size(); index++) {
JSONObject node = nodes.getJSONObject(index);
JSONObject data = node == null ? null : node.getJSONObject("data");
if (data == null || !"knowledgeNode".equals(nodeType(node, data))) {
continue;
}
List<BigInteger> group = new ArrayList<>();
Set<BigInteger> unique = new LinkedHashSet<>();
Object rawIds = data.get("knowledgeIds");
if (rawIds instanceof JSONArray ids && !ids.isEmpty()) {
for (Object rawId : ids) {
BigInteger id = bigInteger(rawId);
if (!unique.add(id)) {
throw new BusinessException("知识库节点不能重复选择同一知识库");
}
group.add(id);
}
} else {
group.add(bigInteger(data.get("knowledgeId")));
}
groups.add(List.copyOf(group));
}
return List.copyOf(groups);
}
private List<List<BigInteger>> onlyMultiGroups(
List<List<BigInteger>> knowledgeGroups) {
if (knowledgeGroups == null || knowledgeGroups.isEmpty()) {
return List.of();
}
return knowledgeGroups.stream()
.filter(Objects::nonNull)
.filter(group -> group.size() > 1)
.map(List::copyOf)
.toList();
}
private List<Map<String, Object>> readSnapshotContracts(Object value) {
if (value == null) {
return List.of();
}
JSONArray array;
try {
array = value instanceof JSONArray jsonArray
? jsonArray
: JSON.parseArray(JSON.toJSONString(value));
} catch (RuntimeException exception) {
throw new BusinessException("工作流知识库发布契约无效");
}
List<Map<String, Object>> result = new ArrayList<>();
for (int index = 0; index < array.size(); index++) {
JSONObject object = array.getJSONObject(index);
if (object == null) {
throw new BusinessException("工作流知识库发布契约无效");
}
Map<String, Object> contract = new LinkedHashMap<>();
contract.put("knowledgeId", text(object.get("knowledgeId")));
contract.put("tenantId", text(object.get("tenantId")));
contract.put("embeddingModelId", text(object.get("embeddingModelId")));
contract.put("embeddingDimension", object.getInteger("embeddingDimension"));
contract.put("vectorStoreCollection", nullableText(object.getString("vectorStoreCollection")));
contract.put("vectorStoreType", nullableText(object.getString("vectorStoreType")));
contract.put("modelProviderId", nullableText(object.getString("modelProviderId")));
contract.put("modelProviderType", nullableText(object.getString("modelProviderType")));
contract.put("modelType", nullableText(object.getString("modelType")));
contract.put("modelName", nullableText(object.getString("modelName")));
contract.put("modelEndpoint", nullableText(object.getString("modelEndpoint")));
contract.put("modelRequestPath", nullableText(object.getString("modelRequestPath")));
result.add(contract);
}
return List.copyOf(result);
}
private String nodeType(JSONObject node, JSONObject data) {
String rootType = text(node.getString("type"));
String dataType = text(data.getString("type"));
if (rootType != null && dataType != null
&& !Objects.equals(rootType, dataType)) {
throw new BusinessException("工作流节点类型与节点数据类型不一致");
}
return rootType == null ? dataType : rootType;
}
private BigInteger bigInteger(Object value) {
String normalized = text(value);
if (normalized == null) {
throw new BusinessException("知识库或租户 ID 无效");
}
try {
BigInteger result = new BigInteger(normalized);
if (result.signum() <= 0) {
throw new NumberFormatException("non-positive");
}
return result;
} catch (NumberFormatException exception) {
throw new BusinessException("知识库或租户 ID 无效");
}
}
private String text(Object value) {
if (value == null) {
return null;
}
String result = String.valueOf(value).trim();
return result.isEmpty() ? null : result;
}
private String nullableText(String value) {
return text(value);
}
private String nullableString(Object value) {
return value == null ? null : String.valueOf(value);
}
private String fingerprint(Object value) {
try {
byte[] bytes = JSON.toJSONString(value)
.getBytes(StandardCharsets.UTF_8);
return HexFormat.of().formatHex(
MessageDigest.getInstance("SHA-256").digest(bytes));
} catch (NoSuchAlgorithmException exception) {
throw new IllegalStateException("SHA-256 is unavailable", exception);
}
}
private record ContractContext(
Map<BigInteger, DocumentCollection> collections,
Map<BigInteger, Model> models) {
private static ContractContext empty() {
return new ContractContext(
Collections.emptyMap(),
Collections.emptyMap());
}
}
}

View File

@@ -1,5 +1,6 @@
package tech.easyflow.ai.easyagentsflow.repository;
import com.alibaba.fastjson2.JSON;
import com.easyagents.flow.core.chain.ChainDefinition;
import com.easyagents.flow.core.node.ConfirmNode;
import com.easyagents.flow.core.parser.ChainParser;
@@ -84,9 +85,35 @@ public class AgentWorkflowSnapshotFactory {
snapshot.put("englishName", workflow.getEnglishName());
snapshot.put("revision", workflow.getRevision());
snapshot.put("content", prepared.content());
snapshot.put("tenantId", workflow.getTenantId());
snapshot.put(
"publishedSnapshotJson",
knowledgeRuntimeSnapshot(workflow));
return snapshot;
}
/**
* 提取冻结执行所需的知识库契约,不携带工作流快照中的其他字段。
*
* @param workflow 已发布工作流
* @return 租户与知识库契约白名单
*/
Map<String, Object> knowledgeRuntimeSnapshot(Workflow workflow) {
Map<String, Object> runtimeSnapshot = new LinkedHashMap<>();
runtimeSnapshot.put("tenantId", workflow.getTenantId());
Map<String, Object> publishedSnapshot =
workflow.getPublishedSnapshotJson();
Object contracts = publishedSnapshot == null
? null
: publishedSnapshot.get("knowledgeContracts");
runtimeSnapshot.put(
"knowledgeContracts",
contracts == null
? java.util.List.of()
: JSON.parse(JSON.toJSONString(contracts)));
return runtimeSnapshot;
}
/**
* Agent 可执行 Workflow 的准备结果。
*

View File

@@ -2,8 +2,11 @@ package tech.easyflow.ai.easyagentsflow.repository;
import com.easyagents.flow.core.chain.ChainDefinition;
import org.springframework.stereotype.Component;
import tech.easyflow.ai.easyagentsflow.knowledge.WorkflowKnowledgeContractService;
import tech.easyflow.ai.entity.Workflow;
import tech.easyflow.common.web.exceptions.BusinessException;
import java.math.BigInteger;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
@@ -24,6 +27,7 @@ public class FrozenWorkflowDefinitionRegistry {
private static final int MAX_ENTRIES = 512;
private final AgentWorkflowSnapshotFactory snapshotFactory;
private final WorkflowKnowledgeContractService workflowKnowledgeContractService;
private final Map<String, ChainDefinition> definitions =
new LinkedHashMap<>(32, 0.75F, true);
private final Map<String, Workflow> workflows =
@@ -33,9 +37,13 @@ public class FrozenWorkflowDefinitionRegistry {
* 创建冻结定义注册表。
*
* @param snapshotFactory Agent Workflow 冻结快照工厂
* @param workflowKnowledgeContractService 知识库契约服务
*/
public FrozenWorkflowDefinitionRegistry(AgentWorkflowSnapshotFactory snapshotFactory) {
public FrozenWorkflowDefinitionRegistry(
AgentWorkflowSnapshotFactory snapshotFactory,
WorkflowKnowledgeContractService workflowKnowledgeContractService) {
this.snapshotFactory = snapshotFactory;
this.workflowKnowledgeContractService = workflowKnowledgeContractService;
}
/**
@@ -48,7 +56,19 @@ public class FrozenWorkflowDefinitionRegistry {
public String register(Workflow workflow) {
AgentWorkflowSnapshotFactory.PreparedWorkflow prepared = snapshotFactory.prepare(workflow);
String preparedContent = prepared.content();
String id = PREFIX + workflow.getId() + ":" + sha256(preparedContent);
Map<String, Object> runtimeSnapshot =
snapshotFactory.knowledgeRuntimeSnapshot(workflow);
if (workflow.getTenantId() == null
|| workflow.getTenantId().signum() <= 0) {
throw new BusinessException("绑定工作流租户快照不完整,请重新发布工作流");
}
String contractFingerprint = workflowKnowledgeContractService
.fingerprintSnapshotContracts(
runtimeSnapshot.get("knowledgeContracts"));
String id = PREFIX + workflow.getId()
+ ":" + workflow.getTenantId()
+ ":" + contractFingerprint
+ ":" + sha256(preparedContent);
synchronized (definitions) {
if (definitions.containsKey(id)) {
definitions.get(id);
@@ -59,7 +79,7 @@ public class FrozenWorkflowDefinitionRegistry {
definition.setName(workflow.getEnglishName());
definition.setDescription(workflow.getDescription());
definitions.put(id, definition);
workflows.put(id, workflow);
workflows.put(id, frozenWorkflow(workflow, runtimeSnapshot));
while (definitions.size() > MAX_ENTRIES) {
String eldest = definitions.keySet().iterator().next();
definitions.remove(eldest);
@@ -100,9 +120,66 @@ public class FrozenWorkflowDefinitionRegistry {
* @return 是否冻结定义
*/
public boolean isFrozen(String definitionId) {
return isFrozenDefinitionId(definitionId);
}
/**
* 判断定义 ID 是否属于冻结 Agent 工作流,不触发注册表实例创建。
*
* @param definitionId 定义 ID
* @return 是否冻结定义
*/
public static boolean isFrozenDefinitionId(String definitionId) {
return definitionId != null && definitionId.startsWith(PREFIX);
}
/**
* 解析冻结定义中可独立校验的租户和知识库契约指纹。
*
* @param definitionId 冻结定义 ID
* @return 冻结定义身份
*/
public static FrozenDefinitionIdentity parseIdentity(String definitionId) {
if (!isFrozenDefinitionId(definitionId)) {
throw new IllegalArgumentException("Not a frozen workflow definition");
}
String[] parts = definitionId.substring(PREFIX.length())
.split(":", 4);
if (parts.length != 4
|| !parts[2].matches("[0-9a-f]{64}")
|| !parts[3].matches("[0-9a-f]{64}")) {
throw new IllegalArgumentException("Invalid frozen workflow definition");
}
try {
BigInteger workflowId = new BigInteger(parts[0]);
BigInteger tenantId = new BigInteger(parts[1]);
if (workflowId.signum() <= 0 || tenantId.signum() <= 0) {
throw new NumberFormatException("non-positive");
}
return new FrozenDefinitionIdentity(
workflowId, tenantId, parts[2]);
} catch (NumberFormatException exception) {
throw new IllegalArgumentException(
"Invalid frozen workflow definition", exception);
}
}
private Workflow frozenWorkflow(
Workflow source,
Map<String, Object> runtimeSnapshot) {
Workflow frozen = new Workflow();
frozen.setId(source.getId());
frozen.setTenantId(source.getTenantId());
frozen.setTitle(source.getTitle());
frozen.setDescription(source.getDescription());
frozen.setEnglishName(source.getEnglishName());
frozen.setRevision(source.getRevision());
frozen.setContent(source.getContent());
frozen.setPublishedSnapshotJson(
new LinkedHashMap<>(runtimeSnapshot));
return frozen;
}
private String sha256(String content) {
try {
return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256")
@@ -111,4 +188,17 @@ public class FrozenWorkflowDefinitionRegistry {
throw new IllegalStateException("SHA-256 is unavailable", exception);
}
}
/**
* 冻结定义中不依赖进程内 LRU 状态的执行身份。
*
* @param workflowId 工作流 ID
* @param tenantId 工作流租户
* @param knowledgeContractFingerprint 知识库契约指纹
*/
public record FrozenDefinitionIdentity(
BigInteger workflowId,
BigInteger tenantId,
String knowledgeContractFingerprint) {
}
}

View File

@@ -6,8 +6,11 @@ import com.alibaba.fastjson2.JSONObject;
import com.easyagents.flow.core.chain.DataType;
import com.easyagents.flow.core.node.ConfirmNode;
import com.easyagents.flow.core.parser.ChainParser;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import tech.easyflow.ai.config.MultiKnowledgeRetrievalProperties;
import tech.easyflow.ai.easyagentsflow.entity.WorkflowCheckIssue;
import tech.easyflow.ai.easyagentsflow.entity.WorkflowCheckResult;
import tech.easyflow.ai.easyagentsflow.entity.WorkflowCheckStage;
@@ -42,6 +45,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
/**
@@ -49,6 +53,9 @@ import java.util.stream.Collectors;
*/
@Service
public class WorkflowCheckService {
private static final Logger LOGGER = LoggerFactory.getLogger(
WorkflowCheckService.class);
private static final long SLOW_CHECK_THRESHOLD_MS = 500L;
private static final String LEVEL_ERROR = "ERROR";
private static final String LEVEL_WARNING = "WARNING";
private static final String TYPE_START = "startNode";
@@ -56,6 +63,7 @@ public class WorkflowCheckService {
private static final String TYPE_LOOP = "loopNode";
private static final String TYPE_CONDITION = "conditionNode";
private static final String TYPE_CONFIRM = "confirmNode";
private static final String TYPE_KNOWLEDGE = "knowledgeNode";
private static final Set<String> CONFIRM_ARRAY_LEFT_OPERATORS = Set.of(
"contains", "notContains", "isEmpty", "isNotEmpty");
private static final String TYPE_WORKFLOW = "workflow-node";
@@ -64,6 +72,10 @@ public class WorkflowCheckService {
private static final String SYSTEM_START_PARAM_NAME = "user_input";
private static final int MIN_LOOP_COUNT = 1;
private static final int MAX_LOOP_COUNT = 300;
private static final int DEFAULT_MAX_KNOWLEDGE_SOURCES = 8;
private static final int DEFAULT_MAX_MULTI_KNOWLEDGE_LIMIT = 200;
private static final Pattern COMPLETE_VARIABLE_REFERENCE =
Pattern.compile("^\\{\\{\\s*[^\\s{}][^{}]*?\\s*}}$");
private static final String JOIN_MODE_ANY = "any";
private static final String JOIN_MODE_ALL = "all";
@@ -79,6 +91,8 @@ public class WorkflowCheckService {
private PluginItemService pluginItemService;
@Resource
private WorkflowPluginSnapshotResolver workflowPluginSnapshotResolver;
@Resource
private MultiKnowledgeRetrievalProperties multiKnowledgeRetrievalProperties;
public WorkflowCheckResult checkWorkflow(BigInteger workflowId, WorkflowCheckStage stage) {
if (workflowId == null) {
@@ -95,9 +109,11 @@ public class WorkflowCheckService {
if (stage == null) {
throw new BusinessException("校验阶段不能为空");
}
long checkStartedAt = System.nanoTime();
List<WorkflowCheckIssue> issues = new ArrayList<>();
Set<String> issueKeys = new LinkedHashSet<>();
ParsedWorkflow parsedWorkflow = parseAndCheckBase(content, issues, issueKeys);
long baseFinishedAt = System.nanoTime();
if (parsedWorkflow != null) {
List<NodeView> startNodes = parsedWorkflow.nodes.stream()
.filter(node -> TYPE_START.equals(node.type))
@@ -105,11 +121,22 @@ public class WorkflowCheckService {
checkStartFormSchema(startNodes, issues, issueKeys);
checkPluginSchemaHashes(parsedWorkflow, issues, issueKeys);
}
long schemaFinishedAt = System.nanoTime();
if (stage == WorkflowCheckStage.PRE_EXECUTE && parsedWorkflow != null) {
runStrictChecks(content, parsedWorkflow, currentWorkflowId, issues, issueKeys);
}
return buildResult(stage, issues);
WorkflowCheckResult result = buildResult(stage, issues);
logSlowCheck(
stage,
currentWorkflowId,
parsedWorkflow,
issues.size(),
checkStartedAt,
baseFinishedAt,
schemaFinishedAt,
System.nanoTime());
return result;
}
public void checkOrThrow(String content, WorkflowCheckStage stage, BigInteger currentWorkflowId) {
@@ -174,6 +201,21 @@ public class WorkflowCheckService {
if (!StringUtils.hasText(node.type) || !parserMap.containsKey(node.type)) {
addIssue(issues, issueKeys, "NODE_TYPE_UNKNOWN", "节点类型无法识别: " + safe(node.type), node.id, null, node.name);
}
String dataType = node.data == null
? null
: trimToNull(node.data.getString("type"));
if (StringUtils.hasText(node.type)
&& StringUtils.hasText(dataType)
&& !Objects.equals(node.type, dataType)) {
addIssue(
issues,
issueKeys,
"NODE_TYPE_MISMATCH",
"节点类型与节点数据类型不一致",
node.id,
null,
node.name);
}
if (StringUtils.hasText(node.parentId) && node.parentId.equals(node.id)) {
addIssue(issues, issueKeys, "NODE_PARENT_SELF", "节点不能引用自己作为父节点", node.id, null, node.name);
}
@@ -189,6 +231,7 @@ public class WorkflowCheckService {
}
checkLoopConfigurations(nodes, nodeMap, issues, issueKeys);
checkConditionConfigurations(nodes, issues, issueKeys);
checkKnowledgeConfigurations(nodes, issues, issueKeys);
checkConfirmConfigurations(nodes, issues, issueKeys);
checkConfirmOutputReferences(nodes, issues, issueKeys);
@@ -246,6 +289,195 @@ public class WorkflowCheckService {
return parsedWorkflow;
}
/**
* 校验知识库节点的新旧引用字段和多库向量模式约束。
*/
private void checkKnowledgeConfigurations(
List<NodeView> nodes,
List<WorkflowCheckIssue> issues,
Set<String> issueKeys) {
for (NodeView node : nodes) {
if (!TYPE_KNOWLEDGE.equals(node.type) || node.data == null) {
continue;
}
checkKnowledgeOutputContract(node, issues, issueKeys);
List<String> knowledgeIds = new ArrayList<>();
if (node.data.containsKey("knowledgeIds")) {
Object rawIds = node.data.get("knowledgeIds");
if (!(rawIds instanceof JSONArray ids) || ids.isEmpty()) {
addIssue(
issues,
issueKeys,
"KNOWLEDGE_IDS_INVALID",
"知识库节点至少需要选择一个知识库",
node.id,
null,
node.name);
continue;
}
Set<String> unique = new LinkedHashSet<>();
for (Object rawId : ids) {
String id = trimToNull(rawId == null
? null
: String.valueOf(rawId));
if (id == null || !id.matches("[0-9]+")) {
addIssue(
issues,
issueKeys,
"KNOWLEDGE_IDS_INVALID",
"知识库节点包含无效的知识库ID",
node.id,
null,
node.name);
continue;
}
if (!unique.add(id)) {
addIssue(
issues,
issueKeys,
"KNOWLEDGE_IDS_DUPLICATE",
"知识库节点不能重复选择同一知识库",
node.id,
null,
node.name);
}
}
knowledgeIds.addAll(unique);
if (knowledgeIds.size() > maxKnowledgeSources()) {
addIssue(
issues,
issueKeys,
"KNOWLEDGE_SOURCE_LIMIT_EXCEEDED",
"知识库节点选择数量超过平台上限",
node.id,
null,
node.name);
}
} else {
String legacyId = trimToNull(node.data.getString("knowledgeId"));
if (legacyId == null || !legacyId.matches("[0-9]+")) {
addIssue(
issues,
issueKeys,
"KNOWLEDGE_ID_INVALID",
"知识库节点需要选择知识库",
node.id,
null,
node.name);
continue;
}
knowledgeIds.add(legacyId);
}
String retrievalMode = trimToNull(
node.data.getString("retrievalMode"));
if (knowledgeIds.size() > 1
&& !"VECTOR".equalsIgnoreCase(retrievalMode)) {
addIssue(
issues,
issueKeys,
"MULTI_KNOWLEDGE_MODE_INVALID",
"多知识库检索仅支持向量检索",
node.id,
null,
node.name);
}
String limit = trimToNull(node.data.getString("limit"));
if (limit != null
&& !COMPLETE_VARIABLE_REFERENCE.matcher(limit).matches()) {
try {
int parsedLimit = Integer.parseInt(limit);
if (parsedLimit <= 0
|| (knowledgeIds.size() > 1
&& parsedLimit > maxMultiKnowledgeLimit())) {
throw new NumberFormatException("non-positive");
}
} catch (NumberFormatException exception) {
addIssue(
issues,
issueKeys,
"KNOWLEDGE_LIMIT_INVALID",
"知识库节点最终返回条数必须为正整数或有效变量引用",
node.id,
null,
node.name);
}
}
}
}
/**
* 知识库节点只允许暴露稳定的 documents 输出及四个历史子字段。
*/
private void checkKnowledgeOutputContract(
NodeView node,
List<WorkflowCheckIssue> issues,
Set<String> issueKeys) {
if (!node.data.containsKey("outputDefs")) {
return;
}
Object rawOutputDefs = node.data.get("outputDefs");
if (rawOutputDefs instanceof JSONArray outputDefs
&& outputDefs.size() == 1
&& isCanonicalKnowledgeDocumentsOutput(
outputDefs.getJSONObject(0))) {
return;
}
addIssue(
issues,
issueKeys,
"KNOWLEDGE_OUTPUT_SCHEMA_INVALID",
"知识库节点输出参数必须为 documents并仅包含 title、content、documentId、knowledgeId",
node.id,
null,
node.name);
}
private boolean isCanonicalKnowledgeDocumentsOutput(JSONObject output) {
if (output == null
|| !"documents".equals(output.getString("name"))
|| !"Array".equalsIgnoreCase(output.getString("dataType"))) {
return false;
}
JSONArray children = output.getJSONArray("children");
if (children == null || children.size() != 4) {
return false;
}
Map<String, String> expectedTypes = Map.of(
"title", "String",
"content", "String",
"documentId", "Number",
"knowledgeId", "Number");
Set<String> names = new LinkedHashSet<>();
for (int index = 0; index < children.size(); index++) {
JSONObject child = children.getJSONObject(index);
if (child == null) {
return false;
}
String name = trimToNull(child.getString("name"));
String expectedType = expectedTypes.get(name);
if (expectedType == null
|| !names.add(name)
|| !expectedType.equalsIgnoreCase(
child.getString("dataType"))) {
return false;
}
}
return names.equals(expectedTypes.keySet());
}
private int maxKnowledgeSources() {
return multiKnowledgeRetrievalProperties == null
? DEFAULT_MAX_KNOWLEDGE_SOURCES
: multiKnowledgeRetrievalProperties.getMaxSources();
}
private int maxMultiKnowledgeLimit() {
return multiKnowledgeRetrievalProperties == null
? DEFAULT_MAX_MULTI_KNOWLEDGE_LIMIT
: multiKnowledgeRetrievalProperties.getTotalCandidateLimit();
}
/**
* 校验节点汇聚模式及其静态可证明的到达安全性。
*
@@ -1963,6 +2195,36 @@ public class WorkflowCheckService {
return result;
}
private void logSlowCheck(
WorkflowCheckStage stage,
BigInteger workflowId,
ParsedWorkflow parsedWorkflow,
int issueCount,
long checkStartedAt,
long baseFinishedAt,
long schemaFinishedAt,
long checkFinishedAt) {
long totalMs = elapsedMillis(checkStartedAt, checkFinishedAt);
if (totalMs < SLOW_CHECK_THRESHOLD_MS) {
return;
}
LOGGER.warn(
"Workflow check is slow: stage={}, workflowId={}, nodes={}, "
+ "issues={}, baseMs={}, schemaMs={}, strictMs={}, totalMs={}",
stage,
workflowId,
parsedWorkflow == null ? 0 : parsedWorkflow.nodes.size(),
issueCount,
elapsedMillis(checkStartedAt, baseFinishedAt),
elapsedMillis(baseFinishedAt, schemaFinishedAt),
elapsedMillis(schemaFinishedAt, checkFinishedAt),
totalMs);
}
private long elapsedMillis(long startedAt, long finishedAt) {
return Math.max(0L, (finishedAt - startedAt) / 1_000_000L);
}
private static class ConfirmOutputIndex {
private final Map<String, String> outputTypes = new HashMap<>();
private final Set<String> nodeIds = new HashSet<>();

View File

@@ -157,6 +157,15 @@ public abstract class AbstractAiResourceLifecycleHandler<T> implements ApprovalS
protected void enrichOfflineSnapshot(T resource, Map<String, Object> snapshot) {
}
/**
* 在发布快照参与重复发布比较前补充资源专属契约。
*
* @param resource 资源
* @param snapshot 当前发布快照
*/
protected void enrichPublishSnapshot(T resource, Map<String, Object> snapshot) {
}
/**
* 删除前额外校验。
*
@@ -287,6 +296,7 @@ public abstract class AbstractAiResourceLifecycleHandler<T> implements ApprovalS
throw new BusinessException("当前" + resourceLabel() + "状态不允许发布");
}
Map<String, Object> snapshot = buildResourceSnapshot(resource);
enrichPublishSnapshot(resource, snapshot);
if (currentStatus == PublishStatus.PUBLISHED && isSameSnapshot(snapshot, getPublishedSnapshot(resource))) {
throw new BusinessException("当前内容与已发布版本一致,无需重新发布");
}

View File

@@ -64,6 +64,30 @@ public interface AiResourceLifecycleHandler {
applyApprovedAction(actionType, resourceId, resourceSnapshot, operatorId);
}
/**
* 执行带审批申请人身份的通过回调。
*
* @param actionType 动作类型
* @param resourceId 资源 ID
* @param resourceSnapshot 审批冻结快照
* @param operatorId 审批操作人 ID
* @param approvalInstanceId 审批实例 ID
* @param applicantId 审批申请人 ID
*/
default void applyApprovedAction(String actionType,
BigInteger resourceId,
Map<String, Object> resourceSnapshot,
BigInteger operatorId,
BigInteger approvalInstanceId,
BigInteger applicantId) {
applyApprovedAction(
actionType,
resourceId,
resourceSnapshot,
operatorId,
approvalInstanceId);
}
/**
* 在实际提交或直接执行前持有冻结快照所需资源。
*

View File

@@ -110,7 +110,8 @@ public class AiResourceLifecycleServiceImpl implements AiResourceLifecycleServic
instance.getResourceId(),
readResourceSnapshot(instance.getSnapshotJson()),
operatorId,
instance.getId()
instance.getId(),
instance.getApplicantId()
);
}

View File

@@ -3,6 +3,9 @@ package tech.easyflow.ai.publish;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.stereotype.Component;
import tech.easyflow.ai.entity.Workflow;
import tech.easyflow.ai.easyagentsflow.entity.WorkflowCheckStage;
import tech.easyflow.ai.easyagentsflow.knowledge.WorkflowKnowledgeContractService;
import tech.easyflow.ai.easyagentsflow.service.WorkflowCheckService;
import tech.easyflow.ai.enums.PublishStatus;
import tech.easyflow.ai.plugin.workflow.binding.WorkflowPluginBindingService;
import tech.easyflow.ai.plugin.workflow.snapshot.WorkflowPluginSnapshotResolver;
@@ -13,16 +16,22 @@ import tech.easyflow.ai.service.WorkflowScheduleReferenceProvider;
import tech.easyflow.ai.vo.OfflineImpactCheckVo;
import tech.easyflow.ai.vo.OfflineImpactBindingVo;
import tech.easyflow.approval.service.ApprovalInstanceService;
import tech.easyflow.approval.enums.ApprovalActionType;
import tech.easyflow.approval.enums.ApprovalResourceType;
import tech.easyflow.common.constant.enums.EnumDataStatus;
import tech.easyflow.common.entity.LoginAccount;
import tech.easyflow.common.web.exceptions.BusinessException;
import tech.easyflow.system.entity.SysAccount;
import tech.easyflow.system.enums.CategoryResourceType;
import tech.easyflow.system.enums.ResourceAction;
import tech.easyflow.system.service.ResourceAccessService;
import tech.easyflow.system.service.SysAccountService;
import java.math.BigInteger;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
/**
* 工作流生命周期处理器。
@@ -37,6 +46,9 @@ public class WorkflowApprovalSubjectHandler extends AbstractAiResourceLifecycleH
private final WorkflowPluginSnapshotResolver workflowPluginSnapshotResolver;
private final AgentResourceReferenceService agentResourceReferenceService;
private final List<WorkflowScheduleReferenceProvider> workflowScheduleReferenceProviders;
private final WorkflowCheckService workflowCheckService;
private final WorkflowKnowledgeContractService workflowKnowledgeContractService;
private final SysAccountService sysAccountService;
public WorkflowApprovalSubjectHandler(WorkflowService workflowService,
ResourceAccessService resourceAccessService,
@@ -46,7 +58,10 @@ public class WorkflowApprovalSubjectHandler extends AbstractAiResourceLifecycleH
WorkflowPluginSnapshotResolver workflowPluginSnapshotResolver,
AgentResourceReferenceService agentResourceReferenceService,
ObjectMapper objectMapper,
List<WorkflowScheduleReferenceProvider> workflowScheduleReferenceProviders) {
List<WorkflowScheduleReferenceProvider> workflowScheduleReferenceProviders,
WorkflowCheckService workflowCheckService,
WorkflowKnowledgeContractService workflowKnowledgeContractService,
SysAccountService sysAccountService) {
super(approvalInstanceService, objectMapper);
this.workflowService = workflowService;
this.resourceAccessService = resourceAccessService;
@@ -57,6 +72,9 @@ public class WorkflowApprovalSubjectHandler extends AbstractAiResourceLifecycleH
this.workflowScheduleReferenceProviders = workflowScheduleReferenceProviders == null
? List.of()
: List.copyOf(workflowScheduleReferenceProviders);
this.workflowCheckService = workflowCheckService;
this.workflowKnowledgeContractService = workflowKnowledgeContractService;
this.sysAccountService = sysAccountService;
}
@Override
@@ -132,6 +150,12 @@ public class WorkflowApprovalSubjectHandler extends AbstractAiResourceLifecycleH
@Override
protected Map<String, Object> buildPublishSnapshot(Workflow resource, PublishStatus currentStatus) {
workflowCheckService.checkOrThrow(
resource.getContent(),
WorkflowCheckStage.SAVE,
resource.getId());
assertKnowledgeUseAccess(
resource.getContent(), resource.getTenantId(), null);
Map<String, Object> snapshot = super.buildPublishSnapshot(resource, currentStatus);
OfflineImpactCheckVo impact = resourceOfflineImpactService.checkWorkflowImpact(resource.getId());
if (impact.isHasPluginBindings()) {
@@ -140,6 +164,53 @@ public class WorkflowApprovalSubjectHandler extends AbstractAiResourceLifecycleH
return snapshot;
}
@Override
protected void enrichPublishSnapshot(
Workflow resource,
Map<String, Object> snapshot) {
snapshot.put(
WorkflowKnowledgeContractService.SNAPSHOT_KEY,
workflowKnowledgeContractService.buildSnapshotContracts(
resource.getContent(), resource.getTenantId()));
}
@Override
public void applyApprovedAction(
String actionType,
BigInteger resourceId,
Map<String, Object> resourceSnapshot,
BigInteger operatorId) {
if (ApprovalActionType.PUBLISH == ApprovalActionType.from(actionType)) {
assertKnowledgeUseAccess(
String.valueOf(resourceSnapshot.get("content")),
snapshotTenantId(resourceSnapshot),
null);
}
super.applyApprovedAction(
actionType, resourceId, resourceSnapshot, operatorId);
}
@Override
public void applyApprovedAction(
String actionType,
BigInteger resourceId,
Map<String, Object> resourceSnapshot,
BigInteger operatorId,
BigInteger approvalInstanceId,
BigInteger applicantId) {
if (ApprovalActionType.PUBLISH == ApprovalActionType.from(actionType)) {
BigInteger tenantId = snapshotTenantId(resourceSnapshot);
LoginAccount applicant = requireCurrentApplicant(
applicantId, tenantId);
assertKnowledgeUseAccess(
String.valueOf(resourceSnapshot.get("content")),
tenantId,
applicant);
}
super.applyApprovedAction(
actionType, resourceId, resourceSnapshot, operatorId);
}
@Override
protected void persistResourceState(BigInteger resourceId, PublishStatus publishStatus, BigInteger currentApprovalInstanceId) {
Workflow update = new Workflow();
@@ -151,6 +222,7 @@ public class WorkflowApprovalSubjectHandler extends AbstractAiResourceLifecycleH
@Override
protected void publishResource(BigInteger resourceId, Map<String, Object> resourceSnapshot, BigInteger operatorId) {
workflowKnowledgeContractService.assertSnapshotCurrent(resourceSnapshot);
Workflow update = new Workflow();
update.setId(resourceId);
update.setPublishStatus(PublishStatus.PUBLISHED.getCode());
@@ -162,6 +234,62 @@ public class WorkflowApprovalSubjectHandler extends AbstractAiResourceLifecycleH
workflowPluginBindingService.syncByWorkflowId(resourceId);
}
private void assertKnowledgeUseAccess(
String content,
BigInteger tenantId,
LoginAccount account) {
workflowKnowledgeContractService.resolveReferencedCollections(
content, tenantId)
.forEach(collection -> {
if (account == null) {
resourceAccessService.assertAccess(
CategoryResourceType.KNOWLEDGE,
collection,
ResourceAction.USE,
"无权限使用工作流知识库");
return;
}
if (!resourceAccessService.canAccess(
account,
CategoryResourceType.KNOWLEDGE,
collection,
ResourceAction.USE)) {
throw new BusinessException(
403, 403, "无权限使用工作流知识库");
}
});
}
private BigInteger snapshotTenantId(Map<String, Object> resourceSnapshot) {
Object value = resourceSnapshot == null
? null
: resourceSnapshot.get("tenantId");
try {
BigInteger tenantId = new BigInteger(String.valueOf(value));
if (tenantId.signum() <= 0) {
throw new NumberFormatException("non-positive");
}
return tenantId;
} catch (RuntimeException exception) {
throw new BusinessException("工作流发布快照租户无效");
}
}
private LoginAccount requireCurrentApplicant(
BigInteger applicantId,
BigInteger tenantId) {
SysAccount account = applicantId == null
? null
: sysAccountService.getById(applicantId);
if (account == null
|| !EnumDataStatus.AVAILABLE.getCode().equals(account.getStatus())
|| !Objects.equals(tenantId, account.getTenantId())) {
throw new BusinessException(
403, 403, "审批申请人账号已失效或不属于当前租户");
}
return account.toLoginAccount();
}
@Override
protected void markResourceOffline(BigInteger resourceId) {
Workflow update = new Workflow();

View File

@@ -0,0 +1,93 @@
package tech.easyflow.ai.rag;
import tech.easyflow.ai.entity.DocumentCollection;
import java.math.BigInteger;
/**
* 单知识库原始向量候选请求,仅供内部跨库编排使用。
*/
public class KnowledgeVectorCandidateRequest {
private BigInteger knowledgeId;
private DocumentCollection collection;
private String query;
private int limit;
private double minVectorScore;
private float[] queryVector;
private Long timeoutMillis;
private String callerType;
private String callerId;
public BigInteger getKnowledgeId() {
return knowledgeId;
}
public void setKnowledgeId(BigInteger knowledgeId) {
this.knowledgeId = knowledgeId;
}
public DocumentCollection getCollection() {
return collection;
}
public void setCollection(DocumentCollection collection) {
this.collection = collection;
}
public String getQuery() {
return query;
}
public void setQuery(String query) {
this.query = query;
}
public int getLimit() {
return limit;
}
public void setLimit(int limit) {
this.limit = limit;
}
public double getMinVectorScore() {
return minVectorScore;
}
public void setMinVectorScore(double minVectorScore) {
this.minVectorScore = minVectorScore;
}
public float[] getQueryVector() {
return queryVector;
}
public void setQueryVector(float[] queryVector) {
this.queryVector = queryVector;
}
public Long getTimeoutMillis() {
return timeoutMillis;
}
public void setTimeoutMillis(Long timeoutMillis) {
this.timeoutMillis = timeoutMillis;
}
public String getCallerType() {
return callerType;
}
public void setCallerType(String callerType) {
this.callerType = callerType;
}
public String getCallerId() {
return callerId;
}
public void setCallerId(String callerId) {
this.callerId = callerId;
}
}

View File

@@ -3,6 +3,7 @@ package tech.easyflow.ai.service;
import com.easyagents.core.document.Document;
import tech.easyflow.ai.entity.DocumentCollection;
import tech.easyflow.ai.rag.KnowledgeRetrievalRequest;
import tech.easyflow.ai.rag.KnowledgeVectorCandidateRequest;
import com.mybatisflex.core.service.IService;
import java.math.BigInteger;
@@ -20,6 +21,14 @@ public interface DocumentCollectionService extends IService<DocumentCollection>
List<Document> search(KnowledgeRetrievalRequest request);
/**
* 查询未归一化、未取整、未重排的原始向量候选。
*
* @param request 原始向量候选请求
* @return 保留向量存储分数原始精度的候选
*/
List<Document> searchVectorCandidates(KnowledgeVectorCandidateRequest request);
DocumentCollection getDetail(String idOrAlias);
DocumentCollection getByAlias(String idOrAlias);

View File

@@ -5,6 +5,7 @@ import tech.easyflow.ai.entity.Model;
import tech.easyflow.ai.service.capability.ModelCapabilityResolution;
import java.math.BigInteger;
import java.util.Collection;
import java.util.List;
import java.util.Map;
@@ -35,6 +36,14 @@ public interface ModelService extends IService<Model> {
Model getModelInstance(BigInteger modelId);
/**
* 批量读取已关联供应商并补齐供应商默认配置的模型。
*
* @param modelIds 模型 ID
* @return 实际运行配置模型
*/
List<Model> listModelInstances(Collection<BigInteger> modelIds);
Model getModelInstanceByInvokeCode(String invokeCode);
void validateForSaveOrUpdate(Model entity, boolean isSave);

View File

@@ -29,6 +29,7 @@ import tech.easyflow.ai.mapper.DocumentCollectionMapper;
import tech.easyflow.ai.mapper.DocumentMapper;
import tech.easyflow.ai.mapper.FaqItemMapper;
import tech.easyflow.ai.rag.KnowledgeRetrievalRequest;
import tech.easyflow.ai.rag.KnowledgeVectorCandidateRequest;
import tech.easyflow.ai.service.DocumentCollectionService;
import tech.easyflow.ai.service.ModelService;
import tech.easyflow.ai.support.DocumentStoreLifecycleSupport;
@@ -60,7 +61,6 @@ public class DocumentCollectionServiceImpl extends ServiceImpl<DocumentCollectio
private static final int MAX_FAQ_IMAGES_IN_PROMPT = 3;
private static final int INTERNAL_RECALL_MULTIPLIER = 5;
private static final int MAX_INTERNAL_RECALL_LIMIT = 100;
private static final int LOG_TEXT_MAX_LENGTH = 300;
@Resource
private ModelService llmService;
@@ -204,6 +204,54 @@ public class DocumentCollectionServiceImpl extends ServiceImpl<DocumentCollectio
return formattedDocuments;
}
@Override
public List<Document> searchVectorCandidates(
KnowledgeVectorCandidateRequest request) {
if (request == null || request.getKnowledgeId() == null) {
throw new BusinessException("知识库ID不能为空");
}
if (StringUtil.noText(request.getQuery())) {
return Collections.emptyList();
}
if (request.getLimit() <= 0) {
throw new BusinessException("向量候选数量必须大于0");
}
if (!Double.isFinite(request.getMinVectorScore())
|| request.getMinVectorScore() < 0D
|| request.getMinVectorScore() > 1D) {
throw new BusinessException("向量相似度阈值无效");
}
DocumentCollection collection = request.getCollection();
if (collection == null
|| !Objects.equals(
request.getKnowledgeId(), collection.getId())) {
throw new BusinessException("知识库检索快照无效");
}
List<Document> documents = prepareSearchDocuments(
collection,
searchVectorDocuments(
collection,
request.getQuery(),
request.getLimit(),
request.getMinVectorScore(),
request.getQueryVector(),
request.getTimeoutMillis()));
for (Document document : documents) {
document.addMetadata("knowledgeId", collection.getId());
document.addMetadata("knowledgeName", collection.getTitle());
document.addMetadata("vectorScore", document.getScore());
}
LOG.info(
"Knowledge raw vector candidates completed, callerType={}, callerId={}, knowledgeId={}, limit={}, minVectorScore={}, hitCount={}",
request.getCallerType(),
request.getCallerId(),
request.getKnowledgeId(),
request.getLimit(),
request.getMinVectorScore(),
documents.size());
return documents;
}
/**
* {@inheritDoc}
*/
@@ -279,38 +327,60 @@ public class DocumentCollectionServiceImpl extends ServiceImpl<DocumentCollectio
String keyword,
int docRecallMaxNum,
Float minSimilarity) {
return searchVectorDocuments(
documentCollection,
keyword,
docRecallMaxNum,
minSimilarity == null
? null
: minSimilarity.doubleValue(),
null,
null);
}
private List<Document> searchVectorDocuments(DocumentCollection documentCollection,
String keyword,
int docRecallMaxNum,
Double minSimilarity,
float[] queryVector,
Long timeoutMillis) {
DocumentStore documentStore = documentCollection.toDocumentStore();
if (documentStore == null) {
throw new BusinessException("知识库没有配置向量库");
}
try {
Model model = llmService.getModelInstance(documentCollection.getVectorEmbedModelId());
if (model == null) {
throw new BusinessException("知识库没有配置向量模型");
if (queryVector == null || queryVector.length == 0) {
Model model = llmService.getModelInstance(documentCollection.getVectorEmbedModelId());
if (model == null) {
throw new BusinessException("知识库没有配置向量模型");
}
documentStore.setEmbeddingModel(model.toEmbeddingModel());
}
documentStore.setEmbeddingModel(model.toEmbeddingModel());
SearchWrapper wrapper = new SearchWrapper();
wrapper.setMaxResults(docRecallMaxNum);
if (queryVector != null && queryVector.length > 0) {
wrapper.setVector(queryVector);
}
if (minSimilarity != null) {
wrapper.setMinScore((double) minSimilarity);
wrapper.setMinScore(minSimilarity);
}
wrapper.setText(keyword);
StoreOptions options = StoreOptions.ofCollectionName(documentCollection.getVectorStoreCollection());
options.setIndexName(documentCollection.getVectorStoreCollection());
if (timeoutMillis != null) {
options.setTimeoutMillis(timeoutMillis);
}
List<Document> documents = documentStore.search(wrapper, options);
List<Document> result = documents == null ? Collections.<Document>emptyList() : documents;
LOG.info(
"Knowledge vector search completed, knowledgeId={}, collectionName={}, query={}, limit={}, minSimilarity={}, hitCount={}, hits={}",
"Knowledge vector search completed, knowledgeId={}, collectionName={}, limit={}, minSimilarity={}, hitCount={}",
documentCollection.getId(),
documentCollection.getVectorStoreCollection(),
keyword,
docRecallMaxNum,
minSimilarity,
result.size(),
summarizeDocuments(result)
result.size()
);
return result;
} finally {
@@ -854,7 +924,7 @@ public class DocumentCollectionServiceImpl extends ServiceImpl<DocumentCollectio
}
/**
* 构建 RAG 原始命中摘要,便于排查向量、关键词与融合阶段的召回情况
* 构建不包含知识正文和元数据的 RAG 原始命中摘要。
*
* @param hits RAG 命中列表
* @return 命中摘要
@@ -868,21 +938,18 @@ public class DocumentCollectionServiceImpl extends ServiceImpl<DocumentCollectio
.map(hit -> {
Map<String, Object> summary = new LinkedHashMap<>();
summary.put("id", hit.getDocumentId());
summary.put("title", hit.getTitle());
summary.put("source", hit.getHitSource());
summary.put("score", hit.getScore());
summary.put("vectorScore", hit.getVectorScore());
summary.put("keywordScore", hit.getKeywordScore());
summary.put("rank", hit.getRank());
summary.put("content", truncate(hit.getContent()));
summary.put("metadata", hit.getMetadata());
return summary;
})
.collect(Collectors.toList());
}
/**
* 构建文档命中摘要,避免完整知识库内容撑爆日志
* 构建不包含知识正文、标题和元数据的文档命中摘要。
*
* @param documents 文档命中列表
* @return 文档摘要
@@ -896,25 +963,10 @@ public class DocumentCollectionServiceImpl extends ServiceImpl<DocumentCollectio
.map(document -> {
Map<String, Object> summary = new LinkedHashMap<>();
summary.put("id", document.getId());
summary.put("title", document.getTitle());
summary.put("score", document.getScore());
summary.put("content", truncate(document.getContent()));
summary.put("metadata", document.getMetadataMap());
return summary;
})
.collect(Collectors.toList());
}
/**
* 截断日志文本,保留足够排查上下文。
*
* @param text 原始文本
* @return 截断文本
*/
private String truncate(String text) {
if (text == null || text.length() <= LOG_TEXT_MAX_LENGTH) {
return text;
}
return text.substring(0, LOG_TEXT_MAX_LENGTH) + "...";
}
}

View File

@@ -216,6 +216,20 @@ public class ModelServiceImpl extends ServiceImpl<ModelMapper, Model> implements
return fillProviderDefaults(model);
}
@Override
public List<Model> listModelInstances(Collection<BigInteger> modelIds) {
if (modelIds == null || modelIds.isEmpty()) {
return List.of();
}
return modelMapper.selectListWithRelationsByQuery(
QueryWrapper.create().in(Model::getId, modelIds))
.stream()
.map(model -> model.getModelProvider() == null
? model
: fillProviderDefaults(model))
.toList();
}
@Override
public Model getModelInstanceByInvokeCode(String invokeCode) {
if (StrUtil.isBlank(invokeCode)) {

View File

@@ -2,6 +2,7 @@ package tech.easyflow.ai.easyagentsflow.knowledge;
import com.easyagents.core.document.Document;
import com.easyagents.flow.core.knowledge.Knowledge;
import com.easyagents.flow.core.knowledge.KnowledgeSearchRequest;
import com.easyagents.flow.core.node.KnowledgeNode;
import org.junit.Assert;
import org.junit.Test;
@@ -13,6 +14,7 @@ import java.math.BigInteger;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
@@ -35,6 +37,7 @@ public class KnowledgeProviderImplTest {
document.setId(BigInteger.valueOf(42));
document.setTitle("文档标题");
document.setContent("文档内容");
document.setScore(0.9D);
document.addMetadata("documentId", BigInteger.valueOf(420));
document.addMetadata("sourceFileName", "元数据标题");
document.addMetadata("legacyKey", "legacy-value");
@@ -59,6 +62,7 @@ public class KnowledgeProviderImplTest {
Assert.assertEquals("文档内容", item.get("content"));
Assert.assertEquals(BigInteger.valueOf(420), item.get("documentId"));
Assert.assertEquals(BigInteger.valueOf(88), item.get("knowledgeId"));
Assert.assertNull(item.get("vectorScore"));
Assert.assertEquals(42L, ((Number) item.get("id")).longValue());
Assert.assertTrue(item.containsKey("metadataMap"));
Assert.assertEquals(
@@ -112,6 +116,66 @@ public class KnowledgeProviderImplTest {
.get("question"));
}
@Test
public void shouldKeepLegacyOutputContractForMultiKnowledge()
throws Exception {
Document document = new Document();
document.setId(BigInteger.valueOf(45));
document.setTitle("来源文档");
document.setContent("跨库内容");
document.setScore(0.923456D);
document.addMetadata("knowledgeId", BigInteger.valueOf(88));
document.addMetadata("knowledgeName", "业务知识库");
document.addMetadata("documentId", BigInteger.valueOf(450));
document.addMetadata("chunkId", BigInteger.valueOf(45));
document.addMetadata("vectorScore", 0.923456D);
document.addMetadata("globalRank", 1);
document.addMetadata("sourceReferences", List.of(Map.of(
"knowledgeId", BigInteger.valueOf(88),
"documentId", BigInteger.valueOf(450),
"chunkId", BigInteger.valueOf(45))));
MultiKnowledgeRetrievalResult retrievalResult =
new MultiKnowledgeRetrievalResult(
List.of(document),
Map.of("requestedSourceCount", 2, "resultCount", 1),
List.of(Map.of("status", "SUCCEEDED")));
WorkflowMultiKnowledgeRetrievalService multiService =
mock(WorkflowMultiKnowledgeRetrievalService.class);
when(multiService.search(
any(), any(), any(Integer.class), any(), any()))
.thenReturn(retrievalResult);
KnowledgeProviderImpl provider = new KnowledgeProviderImpl();
setField(provider, "documentCollectionService",
mock(DocumentCollectionService.class));
setField(provider, "multiKnowledgeRetrievalService", multiService);
KnowledgeNode node = new KnowledgeNode();
node.setId("knowledge-node");
node.setKnowledgeIds(List.of("88", "99"));
node.setRetrievalMode("VECTOR");
Map<String, Object> output = provider.search(
new KnowledgeSearchRequest(
node.getKnowledgeIds(),
"问题",
3,
"VECTOR",
node,
null));
Assert.assertEquals(Set.of("documents"), output.keySet());
List<?> documents = (List<?>) output.get("documents");
Map<?, ?> item = (Map<?, ?>) documents.get(0);
Assert.assertEquals("来源文档", item.get("title"));
Assert.assertEquals("跨库内容", item.get("content"));
Assert.assertEquals(BigInteger.valueOf(450), item.get("documentId"));
Assert.assertEquals(BigInteger.valueOf(88), item.get("knowledgeId"));
Assert.assertFalse(item.containsKey("knowledgeName"));
Assert.assertFalse(item.containsKey("vectorScore"));
Assert.assertFalse(item.containsKey("globalRank"));
}
/**
* 注入测试依赖。
*

View File

@@ -0,0 +1,209 @@
package tech.easyflow.ai.easyagentsflow.knowledge;
import org.junit.Assert;
import org.junit.Test;
import tech.easyflow.ai.entity.DocumentCollection;
import tech.easyflow.ai.entity.Model;
import tech.easyflow.ai.entity.ModelProvider;
import tech.easyflow.ai.service.DocumentCollectionService;
import tech.easyflow.ai.service.ModelService;
import tech.easyflow.common.web.exceptions.BusinessException;
import java.math.BigInteger;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* {@link WorkflowKnowledgeContractService} 契约回归测试。
*/
public class WorkflowKnowledgeContractServiceTest {
@Test
public void shouldRequireExistingEmbeddingModelForVectorReadiness() {
DocumentCollectionService collectionService =
mock(DocumentCollectionService.class);
ModelService modelService = mock(ModelService.class);
DocumentCollection ready = knowledge(1, 7, 3);
DocumentCollection orphan = knowledge(2, 8, 3);
when(modelService.listModelInstances(any()))
.thenReturn(List.of(embeddingModel(7)));
WorkflowKnowledgeContractService service =
new WorkflowKnowledgeContractService(
collectionService, modelService);
Set<BigInteger> result = service.findVectorReadyKnowledgeIds(
List.of(ready, orphan), BigInteger.TEN);
Assert.assertEquals(Set.of(BigInteger.ONE), result);
}
@Test
public void shouldRejectEmbeddingContractChangedAfterSubmission() {
DocumentCollectionService collectionService =
mock(DocumentCollectionService.class);
ModelService modelService = mock(ModelService.class);
DocumentCollection first = knowledge(1, 7, 3);
DocumentCollection second = knowledge(2, 7, 3);
when(collectionService.listByIds(any()))
.thenReturn(List.of(first, second));
when(modelService.listModelInstances(any()))
.thenReturn(List.of(embeddingModel(7)));
WorkflowKnowledgeContractService service =
new WorkflowKnowledgeContractService(
collectionService, modelService);
String content = """
{"nodes":[{"type":"knowledgeNode","data":{
"knowledgeIds":["1","2"],"retrievalMode":"VECTOR"
}}]}
""";
List<Map<String, Object>> contracts =
service.buildSnapshotContracts(content, BigInteger.TEN);
Map<String, Object> snapshot = new LinkedHashMap<>();
snapshot.put("tenantId", BigInteger.TEN);
snapshot.put("content", content);
snapshot.put(WorkflowKnowledgeContractService.SNAPSHOT_KEY, contracts);
second.setDimensionOfVectorModel(4);
BusinessException exception = Assert.assertThrows(
BusinessException.class,
() -> service.assertSnapshotCurrent(snapshot));
Assert.assertTrue(exception.getMessage().contains("Embedding"));
}
@Test
public void shouldRejectMultiKnowledgeWhenEmbeddingModelIsMissing() {
DocumentCollectionService collectionService =
mock(DocumentCollectionService.class);
ModelService modelService = mock(ModelService.class);
when(collectionService.listByIds(any()))
.thenReturn(List.of(
knowledge(1, 7, 3),
knowledge(2, 7, 3)));
when(modelService.listModelInstances(any())).thenReturn(List.of());
WorkflowKnowledgeContractService service =
new WorkflowKnowledgeContractService(
collectionService, modelService);
BusinessException exception = Assert.assertThrows(
BusinessException.class,
() -> service.assertMultiKnowledgeContracts(
List.of(List.of(BigInteger.ONE, BigInteger.TWO)),
BigInteger.TEN));
Assert.assertTrue(exception.getMessage().contains("向量检索配置"));
}
@Test
public void shouldRejectEffectiveProviderEndpointDrift() {
DocumentCollectionService collectionService =
mock(DocumentCollectionService.class);
ModelService modelService = mock(ModelService.class);
DocumentCollection first = knowledge(1, 7, 3);
DocumentCollection second = knowledge(2, 7, 3);
Model effectiveModel = embeddingModel(7);
when(collectionService.listByIds(any()))
.thenReturn(List.of(first, second));
when(modelService.listModelInstances(any()))
.thenReturn(List.of(effectiveModel));
WorkflowKnowledgeContractService service =
new WorkflowKnowledgeContractService(
collectionService, modelService);
String content = """
{"nodes":[{"type":"knowledgeNode","data":{
"knowledgeIds":["1","2"],"retrievalMode":"VECTOR"
}}]}
""";
Map<String, Object> snapshot = new LinkedHashMap<>();
snapshot.put("tenantId", BigInteger.TEN);
snapshot.put("content", content);
snapshot.put(
WorkflowKnowledgeContractService.SNAPSHOT_KEY,
service.buildSnapshotContracts(content, BigInteger.TEN));
effectiveModel.setEndpoint("https://embedding.changed.example");
BusinessException exception = Assert.assertThrows(
BusinessException.class,
() -> service.assertSnapshotCurrent(snapshot));
Assert.assertTrue(exception.getMessage().contains("Embedding"));
}
@Test
public void shouldResolveSingleKnowledgeReferenceForPublishAccessCheck() {
DocumentCollectionService collectionService =
mock(DocumentCollectionService.class);
DocumentCollection collection = knowledge(1, 7, 3);
when(collectionService.listByIds(any()))
.thenReturn(List.of(collection));
WorkflowKnowledgeContractService service =
new WorkflowKnowledgeContractService(
collectionService, mock(ModelService.class));
List<DocumentCollection> result = service.resolveReferencedCollections(
"""
{"nodes":[{"type":"knowledgeNode","data":{
"knowledgeId":"1"
}}]}
""",
BigInteger.TEN);
Assert.assertEquals(List.of(collection), result);
}
@Test
public void shouldRejectConflictingRootAndDataNodeTypes() {
WorkflowKnowledgeContractService service =
new WorkflowKnowledgeContractService(
mock(DocumentCollectionService.class),
mock(ModelService.class));
BusinessException exception = Assert.assertThrows(
BusinessException.class,
() -> service.buildSnapshotContracts(
"""
{"nodes":[{"type":"knowledgeNode","data":{
"type":"llmNode","knowledgeIds":["1","2"]
}}]}
""",
BigInteger.TEN));
Assert.assertTrue(exception.getMessage().contains("类型"));
}
private DocumentCollection knowledge(
long id, long embeddingModelId, int dimension) {
DocumentCollection collection = new DocumentCollection();
collection.setId(BigInteger.valueOf(id));
collection.setTitle("知识库" + id);
collection.setTenantId(BigInteger.TEN);
collection.setVectorStoreEnable(true);
collection.setVectorStoreCollection("collection_" + id);
collection.setVectorStoreType("MILVUS");
collection.setVectorEmbedModelId(BigInteger.valueOf(embeddingModelId));
collection.setDimensionOfVectorModel(dimension);
return collection;
}
private Model embeddingModel(long id) {
Model model = new Model();
model.setId(BigInteger.valueOf(id));
model.setTenantId(BigInteger.TEN);
model.setProviderId(BigInteger.ONE);
model.setModelType(Model.MODEL_TYPES[1]);
model.setModelName("embedding-" + id);
model.setEndpoint("https://embedding.example");
model.setRequestPath("/v1/embeddings");
ModelProvider provider = new ModelProvider();
provider.setId(BigInteger.ONE);
provider.setProviderType("openai");
model.setModelProvider(provider);
return model;
}
}

View File

@@ -0,0 +1,911 @@
package tech.easyflow.ai.easyagentsflow.knowledge;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.read.ListAppender;
import com.easyagents.core.document.Document;
import com.easyagents.core.model.embedding.EmbeddingModel;
import com.easyagents.core.store.StoreTimeoutException;
import com.easyagents.core.store.VectorData;
import com.easyagents.flow.core.chain.Chain;
import com.easyagents.flow.core.chain.ChainDefinition;
import com.easyagents.flow.core.chain.ChainState;
import com.easyagents.flow.core.chain.ChainStatus;
import com.easyagents.flow.core.node.KnowledgeNode;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.slf4j.LoggerFactory;
import tech.easyflow.ai.config.MultiKnowledgeRetrievalProperties;
import tech.easyflow.ai.entity.DocumentCollection;
import tech.easyflow.ai.entity.Model;
import tech.easyflow.ai.entity.ModelProvider;
import tech.easyflow.ai.entity.Workflow;
import tech.easyflow.ai.enums.PublishStatus;
import tech.easyflow.ai.rag.KnowledgeVectorCandidateRequest;
import tech.easyflow.ai.service.DocumentCollectionService;
import tech.easyflow.ai.service.ModelService;
import tech.easyflow.ai.service.WorkflowService;
import tech.easyflow.common.constant.Constants;
import tech.easyflow.common.entity.LoginAccount;
import tech.easyflow.common.web.exceptions.BusinessException;
import tech.easyflow.system.enums.CategoryResourceType;
import tech.easyflow.system.enums.ResourceAction;
import tech.easyflow.system.service.ResourceAccessService;
import java.math.BigInteger;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.when;
/**
* 多知识库向量汇总核心语义测试。
*/
public class WorkflowMultiKnowledgeRetrievalServiceTest {
@Test
public void shouldEmbedOnceAndSortByFullPrecision() {
Fixture fixture = new Fixture();
Model driftedModel = mock(Model.class);
EmbeddingModel driftedEmbeddingModel = mock(EmbeddingModel.class);
when(driftedModel.toEmbeddingModel()).thenReturn(driftedEmbeddingModel);
when(fixture.modelService.listModelInstances(any()))
.thenReturn(List.of(fixture.model), List.of(driftedModel));
Document lower = document(11, "第一条", 0.8123451D);
Document higher = document(22, "第二条", 0.8123452D);
when(fixture.collectionService.searchVectorCandidates(any()))
.thenAnswer(invocation -> {
KnowledgeVectorCandidateRequest request = invocation.getArgument(0);
return request.getKnowledgeId().equals(BigInteger.ONE)
? List.of(lower)
: List.of(higher);
});
MultiKnowledgeRetrievalResult result = fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.assertEquals(2, result.getDocuments().size());
Assert.assertEquals(BigInteger.valueOf(22), result.getDocuments().get(0).getId());
Assert.assertEquals(1, result.getDocuments().get(0).getMetadataMap().get("globalRank"));
Assert.assertEquals(0.8123452D,
(Double) result.getDocuments().get(0).getMetadataMap().get("vectorScore"),
0D);
verify(fixture.embeddingModel, times(1)).embed("问题");
verify(driftedEmbeddingModel, never()).embed(any(String.class));
verify(fixture.modelService, times(1)).listModelInstances(any());
ArgumentCaptor<KnowledgeVectorCandidateRequest> captor =
ArgumentCaptor.forClass(KnowledgeVectorCandidateRequest.class);
verify(fixture.collectionService, times(2))
.searchVectorCandidates(captor.capture());
Assert.assertSame(
captor.getAllValues().get(0).getQueryVector(),
captor.getAllValues().get(1).getQueryVector());
for (KnowledgeVectorCandidateRequest request : captor.getAllValues()) {
DocumentCollection expected = request.getKnowledgeId().equals(BigInteger.ONE)
? fixture.first
: fixture.second;
Assert.assertSame(expected, request.getCollection());
Assert.assertNotNull(request.getTimeoutMillis());
Assert.assertTrue(request.getTimeoutMillis() > 0L);
Assert.assertTrue(request.getTimeoutMillis()
<= fixture.properties.getPerSourceTimeout().toMillis());
}
}
@Test
public void shouldDeduplicateExactContentAndKeepAllReferences() {
Fixture fixture = new Fixture();
Document first = document(11, "相同内容\r\n第二行", 0.91D);
Document repeatedResource = document(11, "相同内容\r\n第二行", 0.89D);
Document second = document(22, "相同内容\n第二行", 0.87D);
when(fixture.collectionService.searchVectorCandidates(any()))
.thenAnswer(invocation -> {
KnowledgeVectorCandidateRequest request = invocation.getArgument(0);
return request.getKnowledgeId().equals(BigInteger.ONE)
? List.of(first, repeatedResource)
: List.of(second);
});
MultiKnowledgeRetrievalResult result = fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
5,
"node-1",
fixture.chain);
Assert.assertEquals(1, result.getDocuments().size());
List<?> references = (List<?>) result.getDocuments().get(0)
.getMetadataMap().get("sourceReferences");
Assert.assertEquals(2, references.size());
Assert.assertEquals(3, result.getSummary().get("candidateCount"));
}
@Test
public void shouldPreferChunkIdAndFallbackToDocumentIdForResourceDedup() {
Fixture fixture = new Fixture();
Document first = document(11, "分片内容", 0.91D);
Document sameChunk = document(12, "分片内容的旧副本", 0.89D);
sameChunk.addMetadata("chunkId", BigInteger.valueOf(11));
Document firstDocument = document(13, "文档内容", 0.88D);
firstDocument.getMetadataMap().remove("chunkId");
firstDocument.addMetadata("documentId", BigInteger.valueOf(130));
Document sameDocument = document(14, "文档内容的旧副本", 0.87D);
sameDocument.getMetadataMap().remove("chunkId");
sameDocument.addMetadata("documentId", BigInteger.valueOf(130));
when(fixture.collectionService.searchVectorCandidates(any()))
.thenAnswer(invocation -> {
KnowledgeVectorCandidateRequest request = invocation.getArgument(0);
return request.getKnowledgeId().equals(BigInteger.ONE)
? List.of(first, sameChunk, firstDocument, sameDocument)
: List.of();
});
MultiKnowledgeRetrievalResult result = fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
5,
"node-1",
fixture.chain);
Assert.assertEquals(2, result.getDocuments().size());
Assert.assertEquals(BigInteger.valueOf(11), result.getDocuments().get(0).getId());
Assert.assertEquals(BigInteger.valueOf(13), result.getDocuments().get(1).getId());
}
@Test
public void shouldFilterInvalidScoresApplyThresholdAndUseStableTieOrder() {
Fixture fixture = new Fixture();
when(fixture.collectionService.searchVectorCandidates(any()))
.thenAnswer(invocation -> {
KnowledgeVectorCandidateRequest request = invocation.getArgument(0);
if (request.getKnowledgeId().equals(BigInteger.ONE)) {
return List.of(
document(11, "低分", 0.59D),
document(12, "非数字", Double.NaN),
document(13, "第一库同分", 0.8D));
}
return List.of(
document(22, "无穷值", Double.POSITIVE_INFINITY),
document(23, "第二库同分", 0.8D),
document(24, "阈值边界", 0.6D));
});
MultiKnowledgeRetrievalResult result = fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.assertEquals(2, result.getDocuments().size());
Assert.assertEquals(BigInteger.valueOf(13), result.getDocuments().get(0).getId());
Assert.assertEquals(BigInteger.valueOf(23), result.getDocuments().get(1).getId());
}
@Test
public void shouldKeepIndependentCandidateBudgetForEachSource() {
Fixture fixture = new Fixture();
fixture.properties.setPerSourceCandidateLimit(10);
fixture.properties.setMaxSources(2);
fixture.properties.setTotalCandidateLimit(20);
fixture.properties.setCandidateMultiplier(3);
WorkflowMultiKnowledgeRetrievalService service = fixture.createService();
when(fixture.collectionService.searchVectorCandidates(any()))
.thenReturn(List.of());
service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
3,
"node-1",
fixture.chain);
ArgumentCaptor<KnowledgeVectorCandidateRequest> captor =
ArgumentCaptor.forClass(KnowledgeVectorCandidateRequest.class);
verify(fixture.collectionService, times(2))
.searchVectorCandidates(captor.capture());
Assert.assertEquals(9, captor.getAllValues().get(0).getLimit());
Assert.assertEquals(9, captor.getAllValues().get(1).getLimit());
}
@Test
public void shouldAllowGlobalTopKToComeFromOneKnowledgeSource() {
Fixture fixture = new Fixture();
when(fixture.collectionService.searchVectorCandidates(any()))
.thenAnswer(invocation -> {
KnowledgeVectorCandidateRequest request = invocation.getArgument(0);
if (BigInteger.ONE.equals(request.getKnowledgeId())) {
return List.of(
document(11, "第一库第一条", 0.99D),
document(12, "第一库第二条", 0.98D),
document(13, "第一库第三条", 0.97D));
}
return List.of(document(21, "第二库", 0.8D));
});
MultiKnowledgeRetrievalResult result = fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
3,
"node-1",
fixture.chain);
Assert.assertEquals(
List.of(
BigInteger.valueOf(11),
BigInteger.valueOf(12),
BigInteger.valueOf(13)),
result.getDocuments().stream().map(Document::getId).toList());
}
@Test
public void shouldValidateContextAndExposeNamedStatusesForEmptyQuery() {
Fixture fixture = new Fixture();
MultiKnowledgeRetrievalResult result = fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
" ",
2,
"node-1",
fixture.chain);
Assert.assertTrue(result.getDocuments().isEmpty());
Assert.assertEquals("知识库一",
result.getSourceStatuses().get(0).get("knowledgeName"));
Assert.assertEquals("EMPTY",
result.getSourceStatuses().get(0).get("status"));
Assert.assertTrue(result.getSourceStatuses().get(0)
.containsKey("elapsedMillis"));
verify(fixture.embeddingModel, never()).embed(any(String.class));
verify(fixture.resourceAccessService, times(2)).canAccess(
any(LoginAccount.class),
eq(CategoryResourceType.KNOWLEDGE),
any(DocumentCollection.class),
eq(ResourceAction.USE));
}
@Test
public void shouldTimeoutOneSourceFromItsActualStartAndKeepOtherResults()
throws Exception {
Fixture fixture = new Fixture();
fixture.properties.setPerSourceTimeout(Duration.ofMillis(30));
fixture.properties.setTotalTimeout(Duration.ofMillis(300));
ExecutorService executor = Executors.newSingleThreadExecutor();
try {
WorkflowMultiKnowledgeRetrievalService service =
fixture.createService(executor);
when(fixture.collectionService.searchVectorCandidates(any()))
.thenAnswer(invocation -> {
KnowledgeVectorCandidateRequest request =
invocation.getArgument(0);
if (request.getKnowledgeId().equals(BigInteger.ONE)) {
Thread.sleep(200L);
}
return List.of(document(22, "可用内容", 0.9D));
});
MultiKnowledgeRetrievalResult result = service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.assertEquals(Boolean.TRUE,
result.getSummary().get("partialFailure"));
Assert.assertEquals("TIMED_OUT",
result.getSourceStatuses().get(0).get("status"));
Assert.assertEquals("SUCCEEDED",
result.getSourceStatuses().get(1).get("status"));
} finally {
executor.shutdownNow();
executor.awaitTermination(2, TimeUnit.SECONDS);
}
}
@Test
public void shouldKeepSourcesThatCompletedAtTheTimeoutBoundary() {
Fixture fixture = new Fixture();
fixture.properties.setPerSourceTimeout(Duration.ofMillis(80));
fixture.properties.setTotalTimeout(Duration.ofMillis(80));
AtomicInteger submissions = new AtomicInteger();
Executor boundaryExecutor = command -> {
int submission = submissions.incrementAndGet();
command.run();
if (submission > 1) {
try {
Thread.sleep(60L);
} catch (InterruptedException exception) {
Thread.currentThread().interrupt();
throw new IllegalStateException(exception);
}
}
};
when(fixture.collectionService.searchVectorCandidates(any()))
.thenReturn(List.of(document(11, "临界点结果", 0.9D)));
MultiKnowledgeRetrievalResult result = fixture
.createService(boundaryExecutor)
.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.assertEquals(2, result.getSummary().get("successfulSourceCount"));
Assert.assertEquals(Boolean.FALSE, result.getSummary().get("partialFailure"));
}
@Test
public void shouldRejectSourceCompletedAfterItsDeadlineBeforeCollection()
throws Exception {
Fixture fixture = new Fixture();
fixture.properties.setPerSourceTimeout(Duration.ofMillis(30));
fixture.properties.setTotalTimeout(Duration.ofMillis(300));
when(fixture.collectionService.searchVectorCandidates(any()))
.thenAnswer(invocation -> {
KnowledgeVectorCandidateRequest request =
invocation.getArgument(0);
if (request.getKnowledgeId().equals(BigInteger.ONE)) {
Thread.sleep(45L);
}
return List.of(document(11, "截止时间结果", 0.9D));
});
MultiKnowledgeRetrievalResult result = fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.assertEquals("TIMED_OUT",
result.getSourceStatuses().get(0).get("status"));
Assert.assertEquals("SUCCEEDED",
result.getSourceStatuses().get(1).get("status"));
}
@Test
public void shouldClassifyStoreDeadlineAsTimedOut() {
Fixture fixture = new Fixture();
when(fixture.collectionService.searchVectorCandidates(any()))
.thenAnswer(invocation -> {
KnowledgeVectorCandidateRequest request =
invocation.getArgument(0);
if (request.getKnowledgeId().equals(BigInteger.ONE)) {
throw new StoreTimeoutException(
"synthetic Milvus deadline");
}
return List.of(document(22, "可用内容", 0.9D));
});
MultiKnowledgeRetrievalResult result = fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.assertEquals("TIMED_OUT",
result.getSourceStatuses().get(0).get("status"));
Assert.assertEquals("SUCCEEDED",
result.getSourceStatuses().get(1).get("status"));
Assert.assertEquals(Boolean.TRUE,
result.getSummary().get("partialFailure"));
}
@Test
public void shouldReturnPartialSuccessAndRejectAllFailure() {
Fixture fixture = new Fixture();
when(fixture.collectionService.searchVectorCandidates(any()))
.thenAnswer(invocation -> {
KnowledgeVectorCandidateRequest request = invocation.getArgument(0);
if (request.getKnowledgeId().equals(BigInteger.ONE)) {
throw new IllegalStateException("private host detail");
}
return List.of(document(22, "可用内容", 0.9D));
});
ch.qos.logback.classic.Logger logger =
(ch.qos.logback.classic.Logger) LoggerFactory.getLogger(
WorkflowMultiKnowledgeRetrievalService.class);
ListAppender<ILoggingEvent> logAppender = new ListAppender<>();
logAppender.start();
logger.addAppender(logAppender);
MultiKnowledgeRetrievalResult result;
try {
result = fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
} finally {
logger.detachAppender(logAppender);
logAppender.stop();
}
Assert.assertEquals(1, result.getDocuments().size());
Assert.assertEquals(Boolean.TRUE, result.getSummary().get("partialFailure"));
Assert.assertEquals("FAILED", result.getSourceStatuses().get(0).get("status"));
Assert.assertFalse(String.valueOf(result.getSourceStatuses().get(0).get("error"))
.contains("private host detail"));
Assert.assertFalse(logAppender.list.stream().anyMatch(event -> {
String message = event.getFormattedMessage();
String throwableMessage = event.getThrowableProxy() == null
? ""
: event.getThrowableProxy().getMessage();
return message.contains("private host detail")
|| throwableMessage.contains("private host detail");
}));
doThrow(new IllegalStateException("unavailable"))
.when(fixture.collectionService)
.searchVectorCandidates(any());
try {
fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.fail("all source failure must fail the node");
} catch (BusinessException expected) {
Assert.assertTrue(expected.getMessage().contains("全部知识库"));
}
}
@Test
public void shouldFailFastWhenSourceConfigurationBecomesInvalid() {
Fixture fixture = new Fixture();
doThrow(new BusinessException("知识库配置已失效"))
.when(fixture.collectionService)
.searchVectorCandidates(any());
try {
fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.fail("configuration failures must fail the node");
} catch (BusinessException expected) {
Assert.assertTrue(expected.getMessage().contains("配置已失效"));
}
}
@Test
public void shouldApplyTotalTimeoutToEmbedding() throws Exception {
Fixture fixture = new Fixture();
fixture.properties.setPerSourceTimeout(Duration.ofMillis(50));
fixture.properties.setTotalTimeout(Duration.ofMillis(80));
ExecutorService executor = Executors.newSingleThreadExecutor();
try {
when(fixture.embeddingModel.embed("问题"))
.thenAnswer(invocation -> {
Thread.sleep(500L);
VectorData vectorData = new VectorData();
vectorData.setVector(new float[]{0.1F, 0.2F, 0.3F});
return vectorData;
});
WorkflowMultiKnowledgeRetrievalService service =
fixture.createService(executor);
long startedAt = System.nanoTime();
try {
service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.fail("embedding must respect the node total timeout");
} catch (BusinessException expected) {
Assert.assertTrue(expected.getMessage().contains("总耗时超时"));
Assert.assertTrue(TimeUnit.NANOSECONDS.toMillis(
System.nanoTime() - startedAt) < 400L);
}
} finally {
executor.shutdownNow();
executor.awaitTermination(2, TimeUnit.SECONDS);
}
}
@Test
public void shouldStopWhenWorkflowExecutionIsCancelled() throws Exception {
Fixture fixture = new Fixture();
AtomicInteger activeChecks = new AtomicInteger();
when(fixture.chain.isExecutionActiveNow())
.thenAnswer(invocation -> activeChecks.incrementAndGet() < 3);
when(fixture.collectionService.searchVectorCandidates(any()))
.thenAnswer(invocation -> {
Thread.sleep(5_000L);
return List.of();
});
ExecutorService executor = Executors.newFixedThreadPool(2);
try {
WorkflowMultiKnowledgeRetrievalService service =
fixture.createService(executor);
try {
service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.fail("cancelled workflow must stop knowledge retrieval");
} catch (BusinessException expected) {
Assert.assertTrue(expected.getMessage().contains("已取消"));
}
} finally {
executor.shutdownNow();
Assert.assertTrue(executor.awaitTermination(2, TimeUnit.SECONDS));
}
}
@Test
public void shouldAllowDirectNodeRunWithoutRunningChainState() {
Fixture fixture = new Fixture();
when(fixture.state.getStatus()).thenReturn(ChainStatus.READY);
when(fixture.chain.isExecutionActiveNow()).thenReturn(false);
when(fixture.collectionService.searchVectorCandidates(any()))
.thenReturn(List.of(document(11, "单节点运行结果", 0.91D)));
MultiKnowledgeRetrievalResult result = fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.assertEquals(1, result.getDocuments().size());
Assert.assertEquals(2, result.getSummary().get("successfulSourceCount"));
}
@Test
public void shouldRejectInconsistentEmbeddingContract() {
Fixture fixture = new Fixture();
fixture.second.setVectorEmbedModelId(BigInteger.valueOf(99));
Model secondModel = mock(Model.class);
when(secondModel.getModelType()).thenReturn(Model.MODEL_TYPES[1]);
when(secondModel.getId()).thenReturn(BigInteger.valueOf(99));
when(secondModel.getTenantId()).thenReturn(BigInteger.TEN);
when(secondModel.getProviderId()).thenReturn(BigInteger.ONE);
when(secondModel.getModelProvider()).thenReturn(fixture.modelProvider);
when(secondModel.getModelName()).thenReturn("embedding-other");
when(fixture.modelService.listModelInstances(any()))
.thenReturn(List.of(fixture.model, secondModel));
try {
fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.fail("inconsistent embedding contract must be rejected");
} catch (BusinessException expected) {
Assert.assertTrue(expected.getMessage().contains("Embedding"));
}
}
@Test
public void shouldRejectUnauthorizedDraftKnowledgeSource() {
Fixture fixture = new Fixture();
when(fixture.resourceAccessService.canAccess(
any(LoginAccount.class),
eq(CategoryResourceType.KNOWLEDGE),
any(DocumentCollection.class),
eq(ResourceAction.USE)))
.thenReturn(false);
try {
fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.fail("draft execution must recheck knowledge permission");
} catch (BusinessException expected) {
Assert.assertEquals(403, expected.getHttpStatus());
}
}
@Test
public void shouldTrustPublishedSnapshotBindingWithinSameTenant() {
Fixture fixture = new Fixture();
when(fixture.definition.getId()).thenReturn("published:100");
fixture.workflow.setPublishStatus(PublishStatus.PUBLISHED.getCode());
String content = """
{"nodes":[{"type":"knowledgeNode","data":{
"knowledgeIds":["1","2"],"retrievalMode":"VECTOR"
}}]}
""";
fixture.workflow.setPublishedSnapshotJson(Map.of(
"tenantId", BigInteger.TEN,
"content", content,
WorkflowKnowledgeContractService.SNAPSHOT_KEY,
fixture.knowledgeContractService.buildSnapshotContracts(
content, BigInteger.TEN)));
when(fixture.resourceAccessService.canAccess(
any(LoginAccount.class),
eq(CategoryResourceType.KNOWLEDGE),
any(DocumentCollection.class),
eq(ResourceAction.USE)))
.thenReturn(false);
when(fixture.collectionService.searchVectorCandidates(any()))
.thenReturn(List.of());
MultiKnowledgeRetrievalResult result = fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.assertEquals(2, result.getSummary().get("successfulSourceCount"));
verify(fixture.resourceAccessService, never()).canAccess(
any(LoginAccount.class),
eq(CategoryResourceType.KNOWLEDGE),
any(DocumentCollection.class),
eq(ResourceAction.USE));
}
@Test
public void shouldRejectPublishedRunAfterEffectiveModelConfigDrifts() {
Fixture fixture = new Fixture();
when(fixture.definition.getId()).thenReturn("published:100");
fixture.workflow.setPublishStatus(PublishStatus.PUBLISHED.getCode());
String content = """
{"nodes":[{"type":"knowledgeNode","data":{
"knowledgeIds":["1","2"],"retrievalMode":"VECTOR"
}}]}
""";
fixture.workflow.setPublishedSnapshotJson(Map.of(
"tenantId", BigInteger.TEN,
WorkflowKnowledgeContractService.SNAPSHOT_KEY,
fixture.knowledgeContractService.buildSnapshotContracts(
content, BigInteger.TEN)));
when(fixture.model.getEndpoint()).thenReturn("https://changed.example");
try {
fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.fail("published execution must reject embedding drift");
} catch (BusinessException expected) {
Assert.assertTrue(expected.getMessage().contains("配置已变化"));
}
}
@Test
public void shouldTrustFrozenAgentWorkflowBindingWithinSameTenant() {
Fixture fixture = new Fixture();
String content = """
{"nodes":[{"type":"knowledgeNode","data":{
"knowledgeIds":["1","2"],"retrievalMode":"VECTOR"
}}]}
""";
List<Map<String, Object>> contracts =
fixture.knowledgeContractService.buildSnapshotContracts(
content, BigInteger.TEN);
fixture.workflow.setPublishedSnapshotJson(Map.of(
"tenantId", BigInteger.TEN,
WorkflowKnowledgeContractService.SNAPSHOT_KEY,
contracts));
String definitionId = "agent-frozen:100:10:"
+ fixture.knowledgeContractService
.fingerprintSnapshotContracts(contracts)
+ ":" + "0".repeat(64);
when(fixture.definition.getId()).thenReturn(definitionId);
KnowledgeNode knowledgeNode = new KnowledgeNode();
knowledgeNode.setKnowledgeIds(List.of("1", "2"));
when(fixture.definition.getNodes())
.thenReturn(List.of(knowledgeNode));
when(fixture.resourceAccessService.canAccess(
any(LoginAccount.class),
eq(CategoryResourceType.KNOWLEDGE),
any(DocumentCollection.class),
eq(ResourceAction.USE)))
.thenReturn(false);
when(fixture.collectionService.searchVectorCandidates(any()))
.thenReturn(List.of());
MultiKnowledgeRetrievalResult result = fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.assertEquals(2, result.getSummary().get("successfulSourceCount"));
verify(fixture.workflowService, never()).getById(any());
verify(fixture.resourceAccessService, never()).canAccess(
any(LoginAccount.class),
eq(CategoryResourceType.KNOWLEDGE),
any(DocumentCollection.class),
eq(ResourceAction.USE));
}
@Test
public void shouldRejectFrozenAgentRunAfterEmbeddingContractDrifts() {
Fixture fixture = new Fixture();
String content = """
{"nodes":[{"type":"knowledgeNode","data":{
"knowledgeIds":["1","2"],"retrievalMode":"VECTOR"
}}]}
""";
List<Map<String, Object>> contracts =
fixture.knowledgeContractService.buildSnapshotContracts(
content, BigInteger.TEN);
String definitionId = "agent-frozen:100:10:"
+ fixture.knowledgeContractService
.fingerprintSnapshotContracts(contracts)
+ ":" + "0".repeat(64);
when(fixture.definition.getId()).thenReturn(definitionId);
KnowledgeNode knowledgeNode = new KnowledgeNode();
knowledgeNode.setKnowledgeIds(List.of("1", "2"));
when(fixture.definition.getNodes())
.thenReturn(List.of(knowledgeNode));
when(fixture.model.getEndpoint())
.thenReturn("https://changed.example");
try {
fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.fail("frozen Agent execution must reject embedding drift");
} catch (BusinessException expected) {
Assert.assertTrue(expected.getMessage().contains("重新发布 Agent"));
}
}
@Test
public void shouldRejectCrossTenantKnowledgeSources() {
Fixture fixture = new Fixture();
fixture.second.setTenantId(BigInteger.valueOf(11));
try {
fixture.service.search(
List.of(BigInteger.ONE, BigInteger.TWO),
"问题",
2,
"node-1",
fixture.chain);
Assert.fail("cross-tenant knowledge sources must be rejected");
} catch (BusinessException expected) {
Assert.assertTrue(expected.getMessage().contains("无权限"));
}
}
private static Document document(long id, String content, double score) {
Document document = new Document();
document.setId(BigInteger.valueOf(id));
document.setContent(content);
document.setScore(score);
document.addMetadata("chunkId", BigInteger.valueOf(id));
document.addMetadata("documentId", BigInteger.valueOf(id * 10));
return document;
}
private static DocumentCollection collection(long id, String title) {
DocumentCollection collection = new DocumentCollection();
collection.setId(BigInteger.valueOf(id));
collection.setTitle(title);
collection.setTenantId(BigInteger.TEN);
collection.setVectorStoreEnable(true);
collection.setVectorStoreCollection("collection_" + id);
collection.setVectorEmbedModelId(BigInteger.valueOf(7));
collection.setDimensionOfVectorModel(3);
return collection;
}
private static final class Fixture {
private final DocumentCollectionService collectionService =
mock(DocumentCollectionService.class);
private final ModelService modelService = mock(ModelService.class);
private final WorkflowService workflowService = mock(WorkflowService.class);
private final ResourceAccessService resourceAccessService =
mock(ResourceAccessService.class);
private final EmbeddingModel embeddingModel = mock(EmbeddingModel.class);
private final Model model = mock(Model.class);
private final ModelProvider modelProvider = mock(ModelProvider.class);
private final Chain chain = mock(Chain.class);
private final ChainDefinition definition = mock(ChainDefinition.class);
private final ChainState state = mock(ChainState.class);
private final Workflow workflow = new Workflow();
private final DocumentCollection first = collection(1, "知识库一");
private final DocumentCollection second = collection(2, "知识库二");
private final MultiKnowledgeRetrievalProperties properties =
new MultiKnowledgeRetrievalProperties();
private final WorkflowMultiKnowledgeRetrievalService service;
private final WorkflowKnowledgeContractService knowledgeContractService;
private Fixture() {
when(collectionService.listByIds(any()))
.thenReturn(List.of(first, second));
when(model.getModelType()).thenReturn(Model.MODEL_TYPES[1]);
when(model.getId()).thenReturn(BigInteger.valueOf(7));
when(model.getTenantId()).thenReturn(BigInteger.TEN);
when(model.getProviderId()).thenReturn(BigInteger.ONE);
when(model.getModelProvider()).thenReturn(modelProvider);
when(modelProvider.getProviderType()).thenReturn("openai");
when(model.getModelName()).thenReturn("embedding-test");
when(model.toEmbeddingModel()).thenReturn(embeddingModel);
when(modelService.listModelInstances(any())).thenReturn(List.of(model));
VectorData vectorData = new VectorData();
vectorData.setVector(new float[]{0.1F, 0.2F, 0.3F});
when(embeddingModel.embed("问题")).thenReturn(vectorData);
when(definition.getId()).thenReturn("100");
when(chain.getDefinition()).thenReturn(definition);
when(chain.isExecutionActiveNow()).thenReturn(true);
when(state.getStatus()).thenReturn(ChainStatus.RUNNING);
ConcurrentHashMap<String, Object> memory = new ConcurrentHashMap<>();
LoginAccount account = new LoginAccount();
account.setId(BigInteger.ONE);
account.setTenantId(BigInteger.TEN);
memory.put(Constants.LOGIN_USER_KEY, account);
when(state.getMemory()).thenReturn(memory);
when(chain.getExecutionState()).thenReturn(state);
workflow.setId(BigInteger.valueOf(100));
workflow.setTenantId(BigInteger.TEN);
when(workflowService.getById(BigInteger.valueOf(100)))
.thenReturn(workflow);
when(resourceAccessService.canAccess(
any(LoginAccount.class),
eq(CategoryResourceType.KNOWLEDGE),
any(DocumentCollection.class),
eq(ResourceAction.USE)))
.thenReturn(true);
properties.setMinVectorScore(0.6D);
knowledgeContractService = new WorkflowKnowledgeContractService(
collectionService, modelService);
service = createService();
}
private WorkflowMultiKnowledgeRetrievalService createService() {
Executor directExecutor = Runnable::run;
return createService(directExecutor);
}
private WorkflowMultiKnowledgeRetrievalService createService(
Executor executor) {
return new WorkflowMultiKnowledgeRetrievalService(
collectionService,
workflowService,
resourceAccessService,
knowledgeContractService,
properties,
executor);
}
}
}

View File

@@ -11,6 +11,7 @@ import tech.easyflow.ai.node.WorkflowNode;
import tech.easyflow.common.web.exceptions.BusinessException;
import java.math.BigInteger;
import java.util.List;
import java.util.Map;
import static org.mockito.Mockito.mock;
@@ -34,9 +35,16 @@ public class AgentWorkflowSnapshotFactoryTest {
Assert.assertEquals(workflow.getId(), snapshot.get("id"));
Assert.assertEquals("prepared-content", snapshot.get("content"));
Assert.assertEquals(6, snapshot.size());
Assert.assertFalse(snapshot.containsKey("tenantId"));
Assert.assertFalse(snapshot.containsKey("publishedSnapshotJson"));
Assert.assertEquals(8, snapshot.size());
Assert.assertEquals(BigInteger.TEN, snapshot.get("tenantId"));
@SuppressWarnings("unchecked")
Map<String, Object> publishedSnapshot =
(Map<String, Object>) snapshot.get("publishedSnapshotJson");
Assert.assertEquals(BigInteger.TEN, publishedSnapshot.get("tenantId"));
Assert.assertEquals(
List.of(Map.of("knowledgeId", "1")),
publishedSnapshot.get("knowledgeContracts"));
Assert.assertFalse(publishedSnapshot.containsKey("secret"));
}
/**
@@ -78,7 +86,9 @@ public class AgentWorkflowSnapshotFactoryTest {
workflow.setRevision(3);
workflow.setContent("raw-content");
workflow.setTenantId(BigInteger.TEN);
workflow.setPublishedSnapshotJson(Map.of("secret", "hidden"));
workflow.setPublishedSnapshotJson(Map.of(
"secret", "hidden",
"knowledgeContracts", List.of(Map.of("knowledgeId", "1"))));
return workflow;
}

View File

@@ -0,0 +1,67 @@
package tech.easyflow.ai.easyagentsflow.repository;
import com.easyagents.flow.core.chain.ChainDefinition;
import com.easyagents.flow.core.parser.ChainParser;
import org.junit.Assert;
import org.junit.Test;
import tech.easyflow.ai.easyagentsflow.service.WorkflowDatacenterContentService;
import tech.easyflow.ai.easyagentsflow.knowledge.WorkflowKnowledgeContractService;
import tech.easyflow.ai.entity.Workflow;
import tech.easyflow.ai.service.DocumentCollectionService;
import tech.easyflow.ai.service.ModelService;
import java.math.BigInteger;
import java.util.List;
import java.util.Map;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* Agent 工作流冻结定义内容寻址测试。
*/
public class FrozenWorkflowDefinitionRegistryTest {
@Test
public void shouldIncludeKnowledgeContractInDefinitionIdentity() {
ChainParser parser = mock(ChainParser.class);
WorkflowDatacenterContentService contentService =
mock(WorkflowDatacenterContentService.class);
when(contentService.prepareContent("raw-content"))
.thenReturn("prepared-content");
when(parser.parse("prepared-content"))
.thenAnswer(ignored -> new ChainDefinition());
AgentWorkflowSnapshotFactory factory =
new AgentWorkflowSnapshotFactory(parser, contentService);
FrozenWorkflowDefinitionRegistry registry =
new FrozenWorkflowDefinitionRegistry(
factory,
new WorkflowKnowledgeContractService(
mock(DocumentCollectionService.class),
mock(ModelService.class)));
Workflow first = workflow("https://one.example");
Workflow second = workflow("https://two.example");
String firstId = registry.register(first);
String secondId = registry.register(second);
Assert.assertNotEquals(firstId, secondId);
Map<String, Object> frozenSnapshot = registry.getWorkflow(firstId)
.getPublishedSnapshotJson();
Assert.assertEquals(BigInteger.TEN, frozenSnapshot.get("tenantId"));
Assert.assertFalse(frozenSnapshot.containsKey("secret"));
}
private Workflow workflow(String endpoint) {
Workflow workflow = new Workflow();
workflow.setId(BigInteger.ONE);
workflow.setTenantId(BigInteger.TEN);
workflow.setContent("raw-content");
workflow.setPublishedSnapshotJson(Map.of(
"secret", "hidden",
"knowledgeContracts", List.of(Map.of(
"knowledgeId", "1",
"modelEndpoint", endpoint))));
return workflow;
}
}

View File

@@ -1295,6 +1295,182 @@ public class WorkflowCheckServiceTest {
Assert.assertTrue(result.isPassed());
}
@Test
public void testKnowledgeNodeShouldAcceptLegacyAndMultiVectorReferences()
throws Exception {
WorkflowCheckService service = newService(new HashMap<>());
JSONObject legacy = data("历史知识库");
legacy.put("knowledgeId", "101");
legacy.put("retrievalMode", "HYBRID");
legacy.put("limit", "5");
JSONObject multi = data("多知识库");
multi.put("knowledgeIds", stringArray("201", "202"));
multi.put("retrievalMode", "VECTOR");
multi.put("limit", "{{start.limit}}");
WorkflowCheckResult result = service.checkContent(
workflowJson(
array(
node("k1", "knowledgeNode", null, legacy),
node("k2", "knowledgeNode", null, multi)),
new JSONArray()),
WorkflowCheckStage.SAVE,
null);
Assert.assertTrue(result.isPassed());
}
@Test
public void testMultiKnowledgeNodeShouldRequireVectorMode()
throws Exception {
WorkflowCheckService service = newService(new HashMap<>());
JSONObject data = data("多知识库");
data.put("knowledgeIds", stringArray("201", "202"));
data.put("retrievalMode", "HYBRID");
data.put("limit", "5");
WorkflowCheckResult result = service.checkContent(
workflowJson(
array(node("k1", "knowledgeNode", null, data)),
new JSONArray()),
WorkflowCheckStage.SAVE,
null);
assertHasCode(result, "MULTI_KNOWLEDGE_MODE_INVALID");
}
@Test
public void testKnowledgeNodeShouldRejectDuplicateIdsAndInvalidLimit()
throws Exception {
WorkflowCheckService service = newService(new HashMap<>());
JSONObject data = data("多知识库");
data.put("knowledgeIds", stringArray("201", "201"));
data.put("retrievalMode", "VECTOR");
data.put("limit", "0");
WorkflowCheckResult result = service.checkContent(
workflowJson(
array(node("k1", "knowledgeNode", null, data)),
new JSONArray()),
WorkflowCheckStage.SAVE,
null);
assertHasCode(result, "KNOWLEDGE_IDS_DUPLICATE");
assertHasCode(result, "KNOWLEDGE_LIMIT_INVALID");
}
@Test
public void testKnowledgeNodeShouldRejectMalformedVariableLimit()
throws Exception {
WorkflowCheckService service = newService(new HashMap<>());
JSONObject data = data("多知识库");
data.put("knowledgeIds", stringArray("201", "202"));
data.put("retrievalMode", "VECTOR");
data.put("limit", "abc{{");
WorkflowCheckResult result = service.checkContent(
workflowJson(
array(node("k1", "knowledgeNode", null, data)),
new JSONArray()),
WorkflowCheckStage.SAVE,
null);
assertHasCode(result, "KNOWLEDGE_LIMIT_INVALID");
}
@Test
public void testKnowledgeNodeShouldRejectBlankVariableLimit()
throws Exception {
WorkflowCheckService service = newService(new HashMap<>());
JSONObject data = data("多知识库");
data.put("knowledgeIds", stringArray("201", "202"));
data.put("retrievalMode", "VECTOR");
data.put("limit", "{{ }}");
WorkflowCheckResult result = service.checkContent(
workflowJson(
array(node("k1", "knowledgeNode", null, data)),
new JSONArray()),
WorkflowCheckStage.SAVE,
null);
assertHasCode(result, "KNOWLEDGE_LIMIT_INVALID");
}
@Test
public void testKnowledgeNodeShouldRejectMismatchedDataType()
throws Exception {
WorkflowCheckService service = newService(new HashMap<>());
JSONObject data = data("多知识库");
data.put("type", "llmNode");
data.put("knowledgeIds", stringArray("201", "202"));
data.put("retrievalMode", "VECTOR");
data.put("limit", "5");
WorkflowCheckResult result = service.checkContent(
workflowJson(
array(node("k1", "knowledgeNode", null, data)),
new JSONArray()),
WorkflowCheckStage.SAVE,
null);
assertHasCode(result, "NODE_TYPE_MISMATCH");
}
@Test
public void testMultiKnowledgeNodeShouldRejectConfiguredBounds()
throws Exception {
WorkflowCheckService service = newService(new HashMap<>());
JSONObject data = data("多知识库");
data.put("knowledgeIds", stringArray(
"201", "202", "203", "204", "205",
"206", "207", "208", "209"));
data.put("retrievalMode", "VECTOR");
data.put("limit", "201");
WorkflowCheckResult result = service.checkContent(
workflowJson(
array(node("k1", "knowledgeNode", null, data)),
new JSONArray()),
WorkflowCheckStage.SAVE,
null);
assertHasCode(result, "KNOWLEDGE_SOURCE_LIMIT_EXCEEDED");
assertHasCode(result, "KNOWLEDGE_LIMIT_INVALID");
}
@Test
public void testKnowledgeNodeShouldRejectRemovedDiagnosticOutputs()
throws Exception {
WorkflowCheckService service = newService(new HashMap<>());
JSONObject data = data("多知识库");
data.put("knowledgeIds", stringArray("201", "202"));
data.put("retrievalMode", "VECTOR");
data.put("limit", "5");
JSONArray outputDefs = new JSONArray();
JSONObject documents = new JSONObject();
documents.put("name", "documents");
documents.put("dataType", "Array");
JSONArray children = new JSONArray();
children.add(outputDef("title", "String"));
children.add(outputDef("content", "String"));
children.add(outputDef("documentId", "Number"));
children.add(outputDef("knowledgeId", "Number"));
documents.put("children", children);
outputDefs.add(documents);
outputDefs.add(outputDef("sourceStatuses", "Array"));
data.put("outputDefs", outputDefs);
WorkflowCheckResult result = service.checkContent(
workflowJson(
array(node("k1", "knowledgeNode", null, data)),
new JSONArray()),
WorkflowCheckStage.SAVE,
null);
assertHasCode(result, "KNOWLEDGE_OUTPUT_SCHEMA_INVALID");
}
private static WorkflowCheckService newService(Map<String, String> workflowStore) throws Exception {
WorkflowCheckService service = new WorkflowCheckService();
ChainParser parser = ChainParser.builder()
@@ -1387,6 +1563,13 @@ public class WorkflowCheckServiceTest {
return array;
}
private static JSONObject outputDef(String name, String dataType) {
JSONObject output = new JSONObject();
output.put("name", name);
output.put("dataType", dataType);
return output;
}
private static JSONObject node(String id, String type, String parentId, JSONObject data) {
JSONObject node = new JSONObject();
node.put("id", id);

View File

@@ -3,7 +3,11 @@ package tech.easyflow.ai.publish;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Assert;
import org.junit.Test;
import tech.easyflow.ai.entity.DocumentCollection;
import tech.easyflow.ai.entity.Workflow;
import tech.easyflow.ai.easyagentsflow.entity.WorkflowCheckStage;
import tech.easyflow.ai.easyagentsflow.knowledge.WorkflowKnowledgeContractService;
import tech.easyflow.ai.easyagentsflow.service.WorkflowCheckService;
import tech.easyflow.ai.enums.PublishStatus;
import tech.easyflow.ai.plugin.workflow.binding.WorkflowPluginBindingService;
import tech.easyflow.ai.plugin.workflow.snapshot.WorkflowPluginSnapshotResolver;
@@ -15,8 +19,13 @@ import tech.easyflow.ai.vo.OfflineImpactBindingVo;
import tech.easyflow.ai.vo.OfflineImpactCheckVo;
import tech.easyflow.approval.enums.ApprovalActionType;
import tech.easyflow.approval.service.ApprovalInstanceService;
import tech.easyflow.common.constant.enums.EnumDataStatus;
import tech.easyflow.common.web.exceptions.BusinessException;
import tech.easyflow.system.entity.SysAccount;
import tech.easyflow.system.service.ResourceAccessService;
import tech.easyflow.system.service.SysAccountService;
import tech.easyflow.system.enums.CategoryResourceType;
import tech.easyflow.system.enums.ResourceAction;
import java.math.BigInteger;
import java.util.List;
@@ -24,8 +33,13 @@ import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import static org.mockito.Mockito.mock;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
/**
@@ -53,7 +67,10 @@ public class WorkflowApprovalSubjectHandlerTest {
mock(WorkflowPluginSnapshotResolver.class),
mock(AgentResourceReferenceService.class),
new ObjectMapper(),
List.of(scheduleReferenceProvider)
List.of(scheduleReferenceProvider),
mock(WorkflowCheckService.class),
mock(WorkflowKnowledgeContractService.class),
mock(SysAccountService.class)
);
Workflow workflow = new Workflow();
workflow.setId(workflowId);
@@ -91,7 +108,10 @@ public class WorkflowApprovalSubjectHandlerTest {
mock(WorkflowPluginSnapshotResolver.class),
mock(AgentResourceReferenceService.class),
new ObjectMapper(),
List.of(scheduleReferenceProvider)
List.of(scheduleReferenceProvider),
mock(WorkflowCheckService.class),
mock(WorkflowKnowledgeContractService.class),
mock(SysAccountService.class)
);
Workflow workflow = new Workflow();
workflow.setId(workflowId);
@@ -116,6 +136,285 @@ public class WorkflowApprovalSubjectHandlerTest {
Assert.assertEquals(2, referenceChecks.get());
}
@Test
public void shouldValidateAndFreezeKnowledgeContractOnPublish() {
BigInteger workflowId = BigInteger.valueOf(103);
ResourceOfflineImpactService offlineImpactService =
mock(ResourceOfflineImpactService.class);
when(offlineImpactService.checkWorkflowImpact(workflowId))
.thenReturn(new OfflineImpactCheckVo());
WorkflowCheckService workflowCheckService =
mock(WorkflowCheckService.class);
WorkflowKnowledgeContractService contractService =
mock(WorkflowKnowledgeContractService.class);
List<Map<String, Object>> contracts = List.of(
Map.of("knowledgeId", "1"));
when(contractService.buildSnapshotContracts(
"{\"nodes\":[]}", BigInteger.TEN))
.thenReturn(contracts);
WorkflowApprovalSubjectHandler handler =
new WorkflowApprovalSubjectHandler(
mock(WorkflowService.class),
mock(ResourceAccessService.class),
mock(ApprovalInstanceService.class),
offlineImpactService,
mock(WorkflowPluginBindingService.class),
mock(WorkflowPluginSnapshotResolver.class),
mock(AgentResourceReferenceService.class),
new ObjectMapper(),
List.of(),
workflowCheckService,
contractService,
mock(SysAccountService.class));
Workflow workflow = new Workflow();
workflow.setId(workflowId);
workflow.setTenantId(BigInteger.TEN);
workflow.setContent("{\"nodes\":[]}");
workflow.setPublishStatus(PublishStatus.DRAFT.getCode());
Map<String, Object> snapshot = handler.buildPublishSnapshot(
workflow, PublishStatus.DRAFT);
Assert.assertEquals(
contracts,
snapshot.get(WorkflowKnowledgeContractService.SNAPSHOT_KEY));
verify(workflowCheckService).checkOrThrow(
workflow.getContent(), WorkflowCheckStage.SAVE, workflowId);
}
@Test
public void shouldNotRequireKnowledgeContractWhenDeletingDraft() {
WorkflowKnowledgeContractService contractService =
mock(WorkflowKnowledgeContractService.class);
WorkflowApprovalSubjectHandler handler =
new WorkflowApprovalSubjectHandler(
mock(WorkflowService.class),
mock(ResourceAccessService.class),
mock(ApprovalInstanceService.class),
mock(ResourceOfflineImpactService.class),
mock(WorkflowPluginBindingService.class),
mock(WorkflowPluginSnapshotResolver.class),
mock(AgentResourceReferenceService.class),
new ObjectMapper(),
List.of(),
mock(WorkflowCheckService.class),
contractService,
mock(SysAccountService.class));
Workflow workflow = new Workflow();
workflow.setId(BigInteger.valueOf(104));
workflow.setContent("{\"nodes\":[]}");
Map<String, Object> snapshot = handler.buildDeleteSnapshot(
workflow, PublishStatus.DRAFT);
Assert.assertFalse(snapshot.containsKey(
WorkflowKnowledgeContractService.SNAPSHOT_KEY));
verifyNoInteractions(contractService);
}
@Test
public void shouldRejectPublishWhenSingleKnowledgeAccessWasRevoked() {
ResourceAccessService accessService = mock(ResourceAccessService.class);
WorkflowKnowledgeContractService contractService =
mock(WorkflowKnowledgeContractService.class);
DocumentCollection collection = new DocumentCollection();
collection.setId(BigInteger.ONE);
collection.setTenantId(BigInteger.TEN);
when(contractService.resolveReferencedCollections(any(), eq(BigInteger.TEN)))
.thenReturn(List.of(collection));
doThrow(new BusinessException(403, 403, "无权限使用工作流知识库"))
.when(accessService)
.assertAccess(
eq(CategoryResourceType.KNOWLEDGE),
eq(collection),
eq(ResourceAction.USE),
any());
WorkflowApprovalSubjectHandler handler = handler(
mock(WorkflowService.class), accessService, contractService);
Workflow workflow = workflow(105);
BusinessException exception = Assert.assertThrows(
BusinessException.class,
() -> handler.buildPublishSnapshot(
workflow, PublishStatus.DRAFT));
Assert.assertEquals(403, exception.getHttpStatus());
}
@Test
public void shouldRecheckApplicantKnowledgeAccessBeforeApprovalTakesEffect() {
WorkflowService workflowService = mock(WorkflowService.class);
ResourceAccessService accessService = mock(ResourceAccessService.class);
SysAccountService accountService = mock(SysAccountService.class);
WorkflowKnowledgeContractService contractService =
mock(WorkflowKnowledgeContractService.class);
DocumentCollection collection = new DocumentCollection();
collection.setId(BigInteger.ONE);
collection.setTenantId(BigInteger.TEN);
when(contractService.resolveReferencedCollections(any(), eq(BigInteger.TEN)))
.thenReturn(List.of(collection));
SysAccount applicant = new SysAccount();
applicant.setId(BigInteger.ONE);
applicant.setDeptId(BigInteger.valueOf(3));
applicant.setTenantId(BigInteger.TEN);
applicant.setStatus(EnumDataStatus.AVAILABLE.getCode());
when(accountService.getById(BigInteger.ONE)).thenReturn(applicant);
WorkflowApprovalSubjectHandler handler = handler(
workflowService,
accessService,
contractService,
accountService);
Map<String, Object> snapshot = Map.of(
"content", "{\"nodes\":[]}",
"tenantId", BigInteger.TEN);
BusinessException exception = Assert.assertThrows(
BusinessException.class,
() -> handler.applyApprovedAction(
ApprovalActionType.PUBLISH.getCode(),
BigInteger.valueOf(106),
snapshot,
BigInteger.valueOf(9),
BigInteger.valueOf(99),
BigInteger.ONE));
Assert.assertEquals(403, exception.getHttpStatus());
verify(accessService).canAccess(
argThat(account -> BigInteger.valueOf(3).equals(
account.getDeptId())),
eq(CategoryResourceType.KNOWLEDGE),
eq(collection),
eq(ResourceAction.USE));
verify(workflowService, never()).updateById(any(Workflow.class));
}
@Test
public void shouldRejectApprovedPublishWhenApplicantIsDisabled() {
WorkflowService workflowService = mock(WorkflowService.class);
SysAccountService accountService = mock(SysAccountService.class);
SysAccount applicant = new SysAccount();
applicant.setId(BigInteger.ONE);
applicant.setTenantId(BigInteger.TEN);
applicant.setStatus(EnumDataStatus.UNAVAILABLE.getCode());
when(accountService.getById(BigInteger.ONE)).thenReturn(applicant);
WorkflowApprovalSubjectHandler handler = handler(
workflowService,
mock(ResourceAccessService.class),
mock(WorkflowKnowledgeContractService.class),
accountService);
Map<String, Object> snapshot = Map.of(
"content", "{\"nodes\":[]}",
"tenantId", BigInteger.TEN);
BusinessException exception = Assert.assertThrows(
BusinessException.class,
() -> handler.applyApprovedAction(
ApprovalActionType.PUBLISH.getCode(),
BigInteger.valueOf(109),
snapshot,
BigInteger.valueOf(9),
BigInteger.valueOf(99),
BigInteger.ONE));
Assert.assertEquals(403, exception.getHttpStatus());
verify(workflowService, never()).updateById(any(Workflow.class));
}
@Test
public void shouldRejectApprovedPublishWhenEmbeddingContractDrifts() {
WorkflowService workflowService = mock(WorkflowService.class);
WorkflowKnowledgeContractService contractService =
mock(WorkflowKnowledgeContractService.class);
Map<String, Object> snapshot = Map.of(
"content", "{\"nodes\":[]}",
"tenantId", BigInteger.TEN);
doThrow(new BusinessException("工作流引用的知识库 Embedding 配置已变化,请重新提交发布"))
.when(contractService)
.assertSnapshotCurrent(snapshot);
WorkflowApprovalSubjectHandler handler = handler(
workflowService,
mock(ResourceAccessService.class),
contractService);
BusinessException exception = Assert.assertThrows(
BusinessException.class,
() -> handler.applyApprovedAction(
ApprovalActionType.PUBLISH.getCode(),
BigInteger.valueOf(107),
snapshot,
BigInteger.ONE));
Assert.assertTrue(exception.getMessage().contains("配置已变化"));
verify(workflowService, never()).updateById(any(Workflow.class));
}
@Test
public void shouldRejectRepeatedPublishWithSameFullSnapshot() {
WorkflowKnowledgeContractService contractService =
mock(WorkflowKnowledgeContractService.class);
when(contractService.buildSnapshotContracts(any(), eq(BigInteger.TEN)))
.thenReturn(List.of(Map.of("knowledgeId", "1")));
WorkflowApprovalSubjectHandler handler = handler(
mock(WorkflowService.class),
mock(ResourceAccessService.class),
contractService);
Workflow workflow = workflow(108);
Map<String, Object> first = handler.buildPublishSnapshot(
workflow, PublishStatus.DRAFT);
workflow.setPublishedSnapshotJson(first);
BusinessException exception = Assert.assertThrows(
BusinessException.class,
() -> handler.buildPublishSnapshot(
workflow, PublishStatus.PUBLISHED));
Assert.assertTrue(exception.getMessage().contains("无需重新发布"));
}
private WorkflowApprovalSubjectHandler handler(
WorkflowService workflowService,
ResourceAccessService accessService,
WorkflowKnowledgeContractService contractService) {
return handler(
workflowService,
accessService,
contractService,
mock(SysAccountService.class));
}
private WorkflowApprovalSubjectHandler handler(
WorkflowService workflowService,
ResourceAccessService accessService,
WorkflowKnowledgeContractService contractService,
SysAccountService accountService) {
ResourceOfflineImpactService impactService =
mock(ResourceOfflineImpactService.class);
when(impactService.checkWorkflowImpact(any()))
.thenReturn(new OfflineImpactCheckVo());
return new WorkflowApprovalSubjectHandler(
workflowService,
accessService,
mock(ApprovalInstanceService.class),
impactService,
mock(WorkflowPluginBindingService.class),
mock(WorkflowPluginSnapshotResolver.class),
mock(AgentResourceReferenceService.class),
new ObjectMapper(),
List.of(),
mock(WorkflowCheckService.class),
contractService,
accountService);
}
private Workflow workflow(long id) {
Workflow workflow = new Workflow();
workflow.setId(BigInteger.valueOf(id));
workflow.setTenantId(BigInteger.TEN);
workflow.setContent("{\"nodes\":[]}");
workflow.setPublishStatus(PublishStatus.DRAFT.getCode());
return workflow;
}
/**
* 创建定时任务引用摘要。
*

View File

@@ -1,17 +1,22 @@
package tech.easyflow.ai.service.impl;
import com.easyagents.core.document.Document;
import com.easyagents.core.store.DocumentStore;
import com.easyagents.core.store.SearchWrapper;
import com.easyagents.core.store.StoreOptions;
import com.easyagents.search.engine.service.DocumentSearcher;
import com.easyagents.search.engine.service.KeywordSearchRequest;
import com.mybatisflex.core.query.QueryWrapper;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.ObjectProvider;
import tech.easyflow.ai.config.SearcherFactory;
import tech.easyflow.ai.enums.DocumentProcessStatus;
import tech.easyflow.ai.mapper.DocumentChunkMapper;
import tech.easyflow.ai.mapper.DocumentMapper;
import tech.easyflow.ai.mapper.FaqItemMapper;
import tech.easyflow.ai.rag.KnowledgeVectorCandidateRequest;
import java.io.Serializable;
import java.lang.reflect.Field;
@@ -23,6 +28,12 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static tech.easyflow.ai.entity.DocumentCollection.KEY_DOC_RECALL_MAX_NUM;
import static tech.easyflow.ai.entity.DocumentCollection.KEY_SIMILARITY_THRESHOLD;
@@ -34,6 +45,80 @@ import static tech.easyflow.ai.entity.DocumentCollection.KEY_SIMILARITY_THRESHOL
*/
public class DocumentCollectionServiceImplTest {
/**
* 验证多知识库统一阈值不会在向量查询前降为 float。
*/
@Test
public void searchVectorCandidatesShouldKeepDoubleThresholdPrecision() {
BigInteger knowledgeId = BigInteger.ONE;
DocumentStore documentStore = mock(DocumentStore.class);
tech.easyflow.ai.entity.DocumentCollection collection =
mock(tech.easyflow.ai.entity.DocumentCollection.class);
when(collection.getId()).thenReturn(knowledgeId);
when(collection.getTitle()).thenReturn("知识库");
when(collection.getVectorStoreCollection()).thenReturn("knowledge_1");
when(collection.toDocumentStore()).thenReturn(documentStore);
when(documentStore.search(any(), any())).thenReturn(List.of());
DocumentCollectionServiceImpl service =
new TestDocumentCollectionService(collection);
KnowledgeVectorCandidateRequest request =
new KnowledgeVectorCandidateRequest();
request.setKnowledgeId(knowledgeId);
request.setCollection(collection);
request.setQuery("问题");
request.setLimit(5);
request.setMinVectorScore(0.6D);
request.setQueryVector(new float[]{0.1F, 0.2F});
request.setTimeoutMillis(1_234L);
service.searchVectorCandidates(request);
ArgumentCaptor<SearchWrapper> wrapper =
ArgumentCaptor.forClass(SearchWrapper.class);
ArgumentCaptor<StoreOptions> options =
ArgumentCaptor.forClass(StoreOptions.class);
verify(documentStore).search(wrapper.capture(), options.capture());
Assert.assertEquals(0.6D, wrapper.getValue().getMinScore(), 0D);
Assert.assertEquals(Long.valueOf(1_234L),
options.getValue().getTimeoutMillis());
}
/**
* 验证多库检索沿用契约校验时加载的知识库快照,不在执行前重新读取可变配置。
*/
@Test
public void searchVectorCandidatesShouldUseValidatedCollectionSnapshot() {
BigInteger knowledgeId = BigInteger.ONE;
DocumentStore snapshotStore = mock(DocumentStore.class);
DocumentStore changedStore = mock(DocumentStore.class);
tech.easyflow.ai.entity.DocumentCollection snapshot =
mock(tech.easyflow.ai.entity.DocumentCollection.class);
tech.easyflow.ai.entity.DocumentCollection changed =
mock(tech.easyflow.ai.entity.DocumentCollection.class);
when(snapshot.getId()).thenReturn(knowledgeId);
when(snapshot.getVectorStoreCollection()).thenReturn("validated_collection");
when(snapshot.toDocumentStore()).thenReturn(snapshotStore);
when(changed.getId()).thenReturn(knowledgeId);
when(changed.toDocumentStore()).thenReturn(changedStore);
when(snapshotStore.search(any(), any())).thenReturn(List.of());
DocumentCollectionServiceImpl service =
new TestDocumentCollectionService(changed);
KnowledgeVectorCandidateRequest request =
new KnowledgeVectorCandidateRequest();
request.setKnowledgeId(knowledgeId);
request.setCollection(snapshot);
request.setQuery("问题");
request.setLimit(5);
request.setMinVectorScore(0.6D);
request.setQueryVector(new float[]{0.1F, 0.2F});
service.searchVectorCandidates(request);
verify(snapshotStore).search(any(), any());
verify(changedStore, never()).search(any(), any());
}
/**
* 验证最终相关度阈值会过滤所有已统一到零到一范围的检索结果。
*/

View File

@@ -0,0 +1,52 @@
package tech.easyflow.ai.service.impl;
import com.mybatisflex.core.query.QueryWrapper;
import org.junit.Assert;
import org.junit.Test;
import tech.easyflow.ai.entity.Model;
import tech.easyflow.ai.entity.ModelProvider;
import tech.easyflow.ai.mapper.ModelMapper;
import java.math.BigInteger;
import java.util.List;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* {@link ModelServiceImpl} 实际运行配置装配测试。
*/
public class ModelServiceImplTest {
@Test
public void shouldFillProviderDefaultsWhenLoadingModelInstances() {
ModelProvider provider = new ModelProvider();
provider.setId(BigInteger.ONE);
provider.setProviderType("openai");
provider.setEndpoint("https://provider.example");
provider.setEmbedPath("/v1/provider-embeddings");
Model model = new Model();
model.setId(BigInteger.TEN);
model.setProviderId(BigInteger.ONE);
model.setModelProvider(provider);
model.setModelType(Model.MODEL_TYPES[1]);
model.setModelName("embedding-model");
ModelMapper mapper = mock(ModelMapper.class);
when(mapper.selectListWithRelationsByQuery(any(QueryWrapper.class)))
.thenReturn(List.of(model));
ModelServiceImpl service = new ModelServiceImpl();
service.modelMapper = mapper;
List<Model> result = service.listModelInstances(
List.of(BigInteger.TEN));
Assert.assertEquals(1, result.size());
Assert.assertEquals(
"https://provider.example",
result.get(0).getEndpoint());
Assert.assertEquals(
"/v1/provider-embeddings",
result.get(0).getRequestPath());
}
}