refactor(knowledge): S6d 删 RAGFlow 运行时,Dify 转默认装配
S6 Dify grounding 已 live 证成(新 KB→Dify dataset 索引→生成 contextAssembly chunkCount=1→输出复现 KB-only 专名),RAGFlow 运行时退役。删 HttpRagFlow/ UnavailableRagFlow 两 client + RagFlowKnowledgeRuntimeClientConfiguration 装配 + 其配置测试 + KnowledgeRuntimeClientTest(整体为 RAGFlow 契约测试)+ RAGFlow live IT(knowledge 侧 P1rRagFlowLiveAcceptanceIT + muse-server 侧 P1rKnowledge RuntimeEndToEndLiveAcceptanceIT,§9.3 明示退役;Dify e2e 已活体证成、按 content- filter 记忆不新造 P1r IT 顶替)。 装配转默认:DifyKnowledgeRuntimeClientConfiguration 主 bean 加 matchIfMissing=true (provider 未设→Dify);新增 @ConditionalOnMissingBean 兜底 unavailableKnowledge RuntimeClient(原无通用兜底,provider=ragflow 显式值下主 bean havingValue=dify 不 匹配会致无 bean 启动失败,兜底补齐后落 Unavailable)。RagFlowKnowledgeRedactor 被 Dify 复用保留;ragflow 属性块/muse_knowledge_ragflow_binding 表(遗留命名存 Dify ID)不动。market-assembled 双排除的 P1rMarketKbForkMaterializationIT 保留(market+ RAGFlow 双休眠,归 market 恢复路径,符合 S3「排除不删」)。 验证:knowledge-server 单测 300/0/0/0 BUILD SUCCESS;knowledge+muse-server 全分支 test-compile BUILD SUCCESS(MVN_EXIT=0);DifyKnowledgeRuntimeClientConfigurationTest 8 全绿验三态(缺省→Dify/ragflow→Unavailable/缺省未配→fail-closed Unavailable)。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
lili
parent
9057dc8df0
commit
302603d9d9
@ -30,7 +30,7 @@ import java.util.UUID;
|
||||
/**
|
||||
* Dify Datasets API 的 Knowledge 运行时 adapter。
|
||||
*
|
||||
* <p>HTTP 范式对齐 {@code RealDifyMuseAiRuntimeClient} / {@code HttpRagFlowKnowledgeRuntimeClient}:
|
||||
* <p>HTTP 范式对齐 {@code RealDifyMuseAiRuntimeClient}:
|
||||
* JDK {@link HttpClient} + 直连 {@link ProxySelector}(内网 Dify 必须绕过系统代理,否则 live 调用被污染成假 502)+
|
||||
* Bearer(Datasets API key)+ 脱敏审计 + fail-closed 错误映射。契约以 S1 实测的 {@code P1rDifyDatasetsContractLiveIT} 为准。</p>
|
||||
*
|
||||
|
||||
@ -10,16 +10,17 @@ import org.springframework.util.StringUtils;
|
||||
import java.time.Duration;
|
||||
|
||||
/**
|
||||
* Dify Datasets runtime client 的 Spring 装配配置。
|
||||
* Dify Datasets runtime client 的 Spring 装配配置(S6d 起为 Knowledge 缺省运行时)。
|
||||
*
|
||||
* <p>与 {@link RagFlowKnowledgeRuntimeClientConfiguration} 共存(RAGFlow 装配 S6d 才摘除)。为确保单人只装一个
|
||||
* {@link KnowledgeRuntimeClient},两个装配类按 provider 选择器 {@code muse.knowledge.runtime-provider} 互斥:
|
||||
* 本类仅在 {@code =dify} 时参与,RAGFlow 装配在 {@code =ragflow} 或缺省(matchIfMissing)时参与。这样
|
||||
* {@code @ConditionalOnMissingBean} 的求值顺序不再影响结果,避免组件扫描顺序带来的非确定性冲突。</p>
|
||||
* <p>S6d 摘除 RAGFlow 运行时后,本类是唯一的具体 {@link KnowledgeRuntimeClient} 装配来源,按 provider 选择器
|
||||
* {@code muse.knowledge.runtime-provider} 生效:取 {@code dify} 或缺省({@code matchIfMissing=true},即未显式配置时
|
||||
* 默认 Dify)时装 Dify 主 bean;显式配成其它值(如遗留的 {@code ragflow},其运行时已删)时主 bean 让位,由下方
|
||||
* {@code @ConditionalOnMissingBean} 兜底装通用的 {@link UnavailableKnowledgeRuntimeClient}。两个 @Bean 同处一类、
|
||||
* 兜底声明在主 bean 之后,{@code @ConditionalOnMissingBean} 按声明顺序求值,确保任何 provider 取值下都有且仅有
|
||||
* 一个 bean,绝不因无 bean 致启动失败。</p>
|
||||
*
|
||||
* <p>装配范式对齐 RAGFlow(无 {@code .enable} 开关):读 {@code muse.knowledge.dify.*};缺 base-url 或
|
||||
* dataset-api-key 一律装 {@link UnavailableKnowledgeRuntimeClient}(fail-closed,即"选了 Dify 但配错=拒绝",
|
||||
* 绝不静默回退 RAGFlow)。</p>
|
||||
* <p>读 {@code muse.knowledge.dify.*};缺 base-url 或 dataset-api-key 一律装 {@link UnavailableKnowledgeRuntimeClient}
|
||||
* (fail-closed,即"选了 Dify 但配错=拒绝",绝不静默降级)。</p>
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
public class DifyKnowledgeRuntimeClientConfiguration {
|
||||
@ -30,7 +31,7 @@ public class DifyKnowledgeRuntimeClientConfiguration {
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean(KnowledgeRuntimeClient.class)
|
||||
@ConditionalOnProperty(name = "muse.knowledge.runtime-provider", havingValue = "dify")
|
||||
@ConditionalOnProperty(name = "muse.knowledge.runtime-provider", havingValue = "dify", matchIfMissing = true)
|
||||
public KnowledgeRuntimeClient difyKnowledgeRuntimeClient(Environment environment) {
|
||||
String baseUrl = environment.getProperty(PROPERTY_PREFIX + "base-url");
|
||||
String apiKey = environment.getProperty(PROPERTY_PREFIX + "dataset-api-key");
|
||||
@ -43,6 +44,17 @@ public class DifyKnowledgeRuntimeClientConfiguration {
|
||||
return new DifyKnowledgeRuntimeClient(baseUrl, apiKey, Duration.ofSeconds(timeoutSeconds), retryBudget);
|
||||
}
|
||||
|
||||
/**
|
||||
* 通用不可用兜底 bean:仅当上面 Dify 主 bean 未装成(provider 显式配成非 dify 值,如遗留 {@code ragflow})时生效。
|
||||
* 声明在主 bean 之后,{@code @ConditionalOnMissingBean} 按同类内 @Bean 声明顺序求值,保证 provider 任意取值下
|
||||
* 始终恰好一个 {@link KnowledgeRuntimeClient},既不因无 bean 启动失败,也不静默降级(桩返回 CONFIG_MISSING)。
|
||||
*/
|
||||
@Bean
|
||||
@ConditionalOnMissingBean(KnowledgeRuntimeClient.class)
|
||||
public KnowledgeRuntimeClient unavailableKnowledgeRuntimeClient() {
|
||||
return new UnavailableKnowledgeRuntimeClient();
|
||||
}
|
||||
|
||||
private int positiveInt(Environment environment, String propertyName, int defaultValue) {
|
||||
Integer value = environment.getProperty(PROPERTY_PREFIX + propertyName, Integer.class, defaultValue);
|
||||
return value == null || value <= 0 ? defaultValue : value;
|
||||
|
||||
@ -1,582 +0,0 @@
|
||||
package cn.iocoder.muse.module.knowledge.application.muse.facade;
|
||||
|
||||
import cn.iocoder.muse.framework.common.util.json.JsonUtils;
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.net.ProxySelector;
|
||||
import java.net.URLEncoder;
|
||||
import java.net.URI;
|
||||
import java.net.http.HttpClient;
|
||||
import java.net.http.HttpHeaders;
|
||||
import java.net.http.HttpRequest;
|
||||
import java.net.http.HttpResponse;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.security.MessageDigest;
|
||||
import java.security.NoSuchAlgorithmException;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.HexFormat;
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
|
||||
/**
|
||||
* RAGFlow HTTP runtime client。
|
||||
*/
|
||||
public class HttpRagFlowKnowledgeRuntimeClient implements KnowledgeRuntimeClient {
|
||||
|
||||
private final String baseUrl;
|
||||
private final String apiKey;
|
||||
private final Duration timeout;
|
||||
private final int retryBudget;
|
||||
private final boolean graphRagAttributionReady;
|
||||
private final HttpClient httpClient;
|
||||
private static final ProxySelector DIRECT_PROXY_SELECTOR = ProxySelector.of(null);
|
||||
|
||||
public HttpRagFlowKnowledgeRuntimeClient(String baseUrl, String apiKey, Duration timeout,
|
||||
int retryBudget, boolean graphRagAttributionReady) {
|
||||
this.baseUrl = normalizeBaseUrl(baseUrl);
|
||||
this.apiKey = apiKey;
|
||||
this.timeout = timeout == null ? Duration.ofSeconds(5) : timeout;
|
||||
this.retryBudget = Math.max(retryBudget, 0);
|
||||
this.graphRagAttributionReady = graphRagAttributionReady;
|
||||
this.httpClient = HttpClient.newBuilder()
|
||||
.connectTimeout(this.timeout)
|
||||
// RAGFlow 部署在 tailnet 内网，显式直连可避免系统代理污染 live IT 与运行时调用。
|
||||
.proxy(DIRECT_PROXY_SELECTOR)
|
||||
.build();
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeResult createDataset(CreateDatasetCommand command) {
|
||||
Map<String, Object> body = new LinkedHashMap<>();
|
||||
body.put("name", command.name());
|
||||
// RAGFlow 当前 create dataset 接口只接受 name；config 由 updateDatasetConfig 单独提交，避免创建阶段被外部 API 拒绝。
|
||||
return executeJson(Operation.CREATE_DATASET, command.correlationId(), command.requestHash(), command.attempt(),
|
||||
"POST", "/api/v1/datasets", body, "id", null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeResult uploadDocuments(UploadDocumentsCommand command) {
|
||||
MultipartPayload multipartPayload = buildMultipartPayload(command.documents());
|
||||
if (multipartPayload == null) {
|
||||
return failure(Operation.UPLOAD_DOCUMENTS, command.correlationId(), command.requestHash(), command.attempt(),
|
||||
null, FailureClass.FILE_REJECTED, 0L, "document content is not materialized");
|
||||
}
|
||||
return execute(Operation.UPLOAD_DOCUMENTS, command.correlationId(), command.requestHash(), command.attempt(),
|
||||
"POST", "/api/v1/datasets/" + command.ragflowDatasetId() + "/documents",
|
||||
multipartPayload.body(), multipartPayload.contentType(), "id", null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeResult startParseDocuments(StartParseDocumentsCommand command) {
|
||||
Map<String, Object> body = Map.of("document_ids", command.ragflowDocumentIds());
|
||||
RuntimeResult rawResult = executeJson(Operation.START_PARSE_DOCUMENTS, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), "POST", "/api/v1/datasets/" + command.ragflowDatasetId() + "/chunks",
|
||||
body, null, command.processingTaskId());
|
||||
if (rawResult.status() == Status.FAILED) {
|
||||
return rawResult;
|
||||
}
|
||||
// RAGFlow parse 没有稳定外部 task id；Muse 只向上返回内部 processingTaskId/correlationId。
|
||||
return new RuntimeResult(null, Operation.START_PARSE_DOCUMENTS, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), rawResult.durationMillis(), Status.ACCEPTED, null, command.processingTaskId(),
|
||||
"{\"status\":\"accepted\",\"processingTaskId\":\"" + escape(command.processingTaskId()) + "\"}",
|
||||
List.of(), Map.of("processingTaskId", command.processingTaskId()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeResult pollDocumentStatuses(PollDocumentStatusesCommand command) {
|
||||
String path = "/api/v1/datasets/" + command.ragflowDatasetId() + "/documents";
|
||||
if (command.ragflowDocumentIds() != null && !command.ragflowDocumentIds().isEmpty()) {
|
||||
if (command.ragflowDocumentIds().size() > 1) {
|
||||
return singleDocumentPollingValidationFailure(command);
|
||||
}
|
||||
// RAGFlow list documents 的真实查询参数是 id=<docId>，不是批量 document_ids。
|
||||
// 当前 Muse 上传链路按单文档 smoke 验收；多文档批量轮询后续应在调用方拆分，避免生成 RAGFlow 不支持的查询。
|
||||
path += "?id=" + encode(command.ragflowDocumentIds().getFirst());
|
||||
}
|
||||
RuntimeResult rawResult = execute(Operation.POLL_DOCUMENT_STATUSES, command.correlationId(),
|
||||
command.requestHash(), command.attempt(), "GET", path, new byte[0], null, null, command.processingTaskId());
|
||||
if (rawResult.status() == Status.FAILED) {
|
||||
return rawResult;
|
||||
}
|
||||
List<DocumentStatus> statuses = normalizeDocumentStatuses(rawResult.summary().get("responseBody"),
|
||||
command.ragflowDocumentIds());
|
||||
return new RuntimeResult(null, Operation.POLL_DOCUMENT_STATUSES, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), rawResult.durationMillis(), Status.SUCCEEDED, null, command.processingTaskId(),
|
||||
sanitizeSummary(Map.of("documents", statuses)), statuses, Map.of("documents", statuses));
|
||||
}
|
||||
|
||||
private RuntimeResult singleDocumentPollingValidationFailure(PollDocumentStatusesCommand command) {
|
||||
Map<String, Object> summary = new LinkedHashMap<>();
|
||||
summary.put("status", Status.FAILED.name().toLowerCase(Locale.ROOT));
|
||||
summary.put("failureClass", FailureClass.VALIDATION_ERROR.name());
|
||||
summary.put("operation", Operation.POLL_DOCUMENT_STATUSES.wireName());
|
||||
summary.put("reason", "only support single document status polling");
|
||||
return new RuntimeResult(null, Operation.POLL_DOCUMENT_STATUSES, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), 0L, Status.FAILED, FailureClass.VALIDATION_ERROR, command.processingTaskId(),
|
||||
sanitizeSummary(summary), List.of(), summary);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeResult retrieveChunks(RetrieveChunksCommand command) {
|
||||
Map<String, Object> body = new LinkedHashMap<>();
|
||||
body.put("dataset_ids", command.ragflowDatasetIds());
|
||||
// RAGFlow 要求 document_ids 为 list;为 null(不限定具体文档、按授权 dataset 全量检索)时不放入 body,
|
||||
// 否则 RAGFlow 拒绝 code 102 "`documents` should be a list" → 检索 fail-closed,P-A 上下文检索永不命中。
|
||||
if (command.ragflowDocumentIds() != null) {
|
||||
body.put("document_ids", command.ragflowDocumentIds());
|
||||
}
|
||||
body.put("question", command.question());
|
||||
body.put("top_k", command.topK());
|
||||
body.put("similarity_threshold", command.threshold());
|
||||
// RAGFlow 官方 HTTP 契约字段名是 metadata_condition；旧字段 metadata_filter 会被静默忽略。
|
||||
if (command.metadataCondition() != null) {
|
||||
body.put("metadata_condition", command.metadataCondition());
|
||||
}
|
||||
return executeJson(Operation.RETRIEVE_CHUNKS, command.correlationId(), command.requestHash(), command.attempt(),
|
||||
"POST", "/api/v1/retrieval", body, null, null);
|
||||
}
|
||||
|
||||
protected HttpResponsePayload send(HttpRequestPayload request) throws IOException, InterruptedException {
|
||||
HttpRequest.Builder builder = HttpRequest.newBuilder()
|
||||
.uri(URI.create(baseUrl + request.path()))
|
||||
.timeout(timeout)
|
||||
.header("Accept", "application/json");
|
||||
if (StringUtils.hasText(apiKey)) {
|
||||
builder.header("Authorization", "Bearer " + apiKey);
|
||||
}
|
||||
if ("GET".equals(request.method())) {
|
||||
builder.GET();
|
||||
} else {
|
||||
builder.header("Content-Type", request.contentType() == null ? "application/json" : request.contentType())
|
||||
.method(request.method(), HttpRequest.BodyPublishers.ofByteArray(request.body()));
|
||||
}
|
||||
HttpResponse<String> response = httpClient.send(builder.build(), HttpResponse.BodyHandlers.ofString());
|
||||
return new HttpResponsePayload(response.statusCode(), response.body(), response.headers());
|
||||
}
|
||||
|
||||
private RuntimeResult executeJson(Operation operation, String correlationId, String requestHash, int attempt,
|
||||
String method, String path, Object body, String externalIdField,
|
||||
String processingTaskId) {
|
||||
byte[] requestBody = "GET".equals(method) ? new byte[0]
|
||||
: JsonUtils.toJsonString(body == null ? Map.of() : body).getBytes(StandardCharsets.UTF_8);
|
||||
return execute(operation, correlationId, requestHash, attempt, method, path, requestBody,
|
||||
"application/json", externalIdField, processingTaskId);
|
||||
}
|
||||
|
||||
private RuntimeResult execute(Operation operation, String correlationId, String requestHash, int attempt,
|
||||
String method, String path, byte[] requestBody, String contentType,
|
||||
String externalIdField, String processingTaskId) {
|
||||
Instant startedAt = Instant.now();
|
||||
FailureClass configurationFailure = validateConfiguration();
|
||||
if (configurationFailure != null) {
|
||||
return failure(operation, correlationId, requestHash, attempt, processingTaskId, configurationFailure,
|
||||
elapsedMillis(startedAt), configurationFailure.name());
|
||||
}
|
||||
int maxAttempts = Math.max(1, retryBudget + 1);
|
||||
RuntimeResult lastFailure = null;
|
||||
for (int currentAttempt = 1; currentAttempt <= maxAttempts; currentAttempt++) {
|
||||
try {
|
||||
HttpResponsePayload response = send(new HttpRequestPayload(method, path,
|
||||
requestBody == null ? new byte[0] : requestBody, contentType));
|
||||
long duration = elapsedMillis(startedAt);
|
||||
FailureClass httpFailure = classifyHttpFailure(response.statusCode());
|
||||
if (httpFailure != null) {
|
||||
lastFailure = failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
httpFailure, duration, response.body());
|
||||
if (!isRetryable(httpFailure) || currentAttempt == maxAttempts) {
|
||||
return lastFailure;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
return success(operation, correlationId, requestHash, attempt, processingTaskId, externalIdField,
|
||||
response.body(), duration);
|
||||
} catch (java.net.http.HttpTimeoutException ex) {
|
||||
lastFailure = failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
FailureClass.TIMEOUT, elapsedMillis(startedAt), ex.getMessage());
|
||||
} catch (IOException ex) {
|
||||
lastFailure = failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
FailureClass.RAGFLOW_UNAVAILABLE, elapsedMillis(startedAt), ex.getMessage());
|
||||
} catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
return failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
FailureClass.RAGFLOW_UNAVAILABLE, elapsedMillis(startedAt), ex.getMessage());
|
||||
} catch (RuntimeException ex) {
|
||||
return failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
FailureClass.UNKNOWN_FAILURE, elapsedMillis(startedAt), ex.getMessage());
|
||||
}
|
||||
if (currentAttempt == maxAttempts) {
|
||||
return lastFailure;
|
||||
}
|
||||
}
|
||||
return lastFailure == null
|
||||
? failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
FailureClass.UNKNOWN_FAILURE, elapsedMillis(startedAt), "unknown failure")
|
||||
: lastFailure;
|
||||
}
|
||||
|
||||
private RuntimeResult success(Operation operation, String correlationId, String requestHash, int attempt,
|
||||
String processingTaskId, String externalIdField, String responseBody,
|
||||
long durationMillis) {
|
||||
JsonNode bodyTree = parseBody(responseBody);
|
||||
if (bodyTree == null) {
|
||||
return failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
FailureClass.RESPONSE_SCHEMA_CHANGED, durationMillis, responseBody);
|
||||
}
|
||||
FailureClass envelopeFailure = classifyEnvelopeFailure(bodyTree);
|
||||
if (envelopeFailure != null) {
|
||||
return failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
envelopeFailure, durationMillis, firstText(bodyTree, "message", "msg", "error"));
|
||||
}
|
||||
JsonNode data = bodyTree.has("data") ? bodyTree.path("data") : bodyTree;
|
||||
String externalId = externalIdField == null ? null : findText(data, externalIdField);
|
||||
if (externalIdField != null && !StringUtils.hasText(externalId)) {
|
||||
return failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
FailureClass.RESPONSE_SCHEMA_CHANGED, durationMillis, "required externalId is missing");
|
||||
}
|
||||
Map<String, Object> summary = buildMetadataSummary(operation, bodyTree, externalId, null);
|
||||
// 后续状态归一仍需要原始响应树，审计摘要只输出 buildMetadataSummary 的保守元数据。
|
||||
summary.put("responseBody", bodyTree);
|
||||
if (externalId != null) {
|
||||
summary.put("externalId", externalId);
|
||||
}
|
||||
return new RuntimeResult(externalId, operation, correlationId, requestHash, attempt, durationMillis,
|
||||
Status.SUCCEEDED, null, processingTaskId,
|
||||
sanitizeSummary(buildMetadataSummary(operation, bodyTree, externalId, null)), List.of(), summary);
|
||||
}
|
||||
|
||||
private RuntimeResult failure(Operation operation, String correlationId, String requestHash, int attempt,
|
||||
String processingTaskId, FailureClass failureClass, long durationMillis,
|
||||
String message) {
|
||||
Map<String, Object> summary = new LinkedHashMap<>();
|
||||
summary.put("status", Status.FAILED.name().toLowerCase(Locale.ROOT));
|
||||
summary.put("failureClass", failureClass.name());
|
||||
summary.put("operation", operation.wireName());
|
||||
if (message != null && !message.isBlank()) {
|
||||
summary.put("providerMessageSummary", safeSummaryNode(message));
|
||||
}
|
||||
return new RuntimeResult(null, operation, correlationId, requestHash, attempt, durationMillis,
|
||||
Status.FAILED, failureClass, processingTaskId, sanitizeSummary(summary), List.of(), summary);
|
||||
}
|
||||
|
||||
private JsonNode safeSummaryNode(String message) {
|
||||
Map<String, Object> summary = new LinkedHashMap<>();
|
||||
summary.put("type", "provider_body");
|
||||
summary.put("length", message == null ? 0 : message.length());
|
||||
summary.put("sha256", sha256(message == null ? "" : message));
|
||||
summary.put("redacted", true);
|
||||
summary.put("json", looksLikeJson(message));
|
||||
return JsonUtils.getObjectMapper().valueToTree(summary);
|
||||
}
|
||||
|
||||
private FailureClass validateConfiguration() {
|
||||
if (!StringUtils.hasText(baseUrl)) {
|
||||
return FailureClass.CONFIG_MISSING;
|
||||
}
|
||||
if (!StringUtils.hasText(apiKey)) {
|
||||
return FailureClass.AUTH_MISSING;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private FailureClass classifyHttpFailure(int statusCode) {
|
||||
if (statusCode >= 200 && statusCode < 300) {
|
||||
return null;
|
||||
}
|
||||
return switch (statusCode) {
|
||||
case 400 -> FailureClass.VALIDATION_ERROR;
|
||||
case 401, 403 -> FailureClass.AUTH_FAILED;
|
||||
case 404 -> FailureClass.TASK_NOT_FOUND;
|
||||
case 409 -> FailureClass.CONFLICT;
|
||||
case 413, 415 -> FailureClass.FILE_REJECTED;
|
||||
case 429 -> FailureClass.RATE_LIMITED;
|
||||
default -> statusCode >= 500 ? FailureClass.RAGFLOW_UNAVAILABLE : FailureClass.UNKNOWN_FAILURE;
|
||||
};
|
||||
}
|
||||
|
||||
private boolean isRetryable(FailureClass failureClass) {
|
||||
return failureClass == FailureClass.RAGFLOW_UNAVAILABLE
|
||||
|| failureClass == FailureClass.TIMEOUT
|
||||
|| failureClass == FailureClass.RATE_LIMITED;
|
||||
}
|
||||
|
||||
private FailureClass classifyEnvelopeFailure(JsonNode bodyTree) {
|
||||
if (bodyTree == null || bodyTree.isMissingNode() || bodyTree.isNull()) {
|
||||
return FailureClass.RESPONSE_SCHEMA_CHANGED;
|
||||
}
|
||||
if (bodyTree.has("code") && bodyTree.path("code").canConvertToInt() && bodyTree.path("code").asInt() != 0) {
|
||||
return classifyRagflowCode(bodyTree.path("code").asInt(), firstText(bodyTree, "message", "msg", "error"));
|
||||
}
|
||||
if (bodyTree.has("error") || bodyTree.has("errors")) {
|
||||
return classifyRagflowMessage(firstText(bodyTree, "error", "message", "msg"));
|
||||
}
|
||||
String status = firstText(bodyTree, "status");
|
||||
if (status != null && Set.of("failed", "error", "unauthorized", "forbidden").contains(status.toLowerCase(Locale.ROOT))) {
|
||||
return classifyRagflowMessage(firstText(bodyTree, "message", "msg", "error", "status"));
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private FailureClass classifyRagflowCode(int code, String message) {
|
||||
return switch (code) {
|
||||
case 401, 403, 1001, 1002 -> FailureClass.AUTH_FAILED;
|
||||
case 400, 422 -> FailureClass.VALIDATION_ERROR;
|
||||
case 404 -> FailureClass.TASK_NOT_FOUND;
|
||||
case 409 -> FailureClass.CONFLICT;
|
||||
case 429 -> FailureClass.RATE_LIMITED;
|
||||
default -> classifyRagflowMessage(message);
|
||||
};
|
||||
}
|
||||
|
||||
private FailureClass classifyRagflowMessage(String message) {
|
||||
String normalized = message == null ? "" : message.toLowerCase(Locale.ROOT);
|
||||
if (normalized.contains("auth") || normalized.contains("unauthor") || normalized.contains("forbidden")) {
|
||||
return FailureClass.AUTH_FAILED;
|
||||
}
|
||||
if (normalized.contains("validation") || normalized.contains("invalid") || normalized.contains("param")) {
|
||||
return FailureClass.VALIDATION_ERROR;
|
||||
}
|
||||
if (normalized.contains("index") && normalized.contains("ready")) {
|
||||
return FailureClass.INDEX_NOT_READY;
|
||||
}
|
||||
if (normalized.contains("chunk") && (normalized.contains("empty") || normalized.contains("no "))) {
|
||||
return FailureClass.NO_CHUNK;
|
||||
}
|
||||
if (normalized.contains("graph") && normalized.contains("ready")) {
|
||||
return FailureClass.GRAPH_NOT_READY;
|
||||
}
|
||||
if (normalized.contains("already") && normalized.contains("running")) {
|
||||
return FailureClass.TASK_ALREADY_RUNNING;
|
||||
}
|
||||
if (normalized.contains("not found")) {
|
||||
return FailureClass.TASK_NOT_FOUND;
|
||||
}
|
||||
return FailureClass.RESPONSE_SCHEMA_CHANGED;
|
||||
}
|
||||
|
||||
private List<DocumentStatus> normalizeDocumentStatuses(Object responseBody, List<String> requestedDocumentIds) {
|
||||
JsonNode root = responseBody instanceof JsonNode jsonNode ? jsonNode : JsonUtils.getObjectMapper().valueToTree(responseBody);
|
||||
JsonNode documents = root.path("data");
|
||||
if (documents.isObject() && documents.has("docs")) {
|
||||
documents = documents.path("docs");
|
||||
}
|
||||
if (documents.isObject() && documents.has("documents")) {
|
||||
documents = documents.path("documents");
|
||||
}
|
||||
List<DocumentStatus> statuses = new ArrayList<>();
|
||||
if (!documents.isArray()) {
|
||||
return statuses;
|
||||
}
|
||||
Set<String> requestedIds = requestedDocumentIds == null ? Set.of() : new LinkedHashSet<>(requestedDocumentIds);
|
||||
for (JsonNode document : documents) {
|
||||
String documentId = firstText(document, "id", "document_id");
|
||||
if (!requestedIds.isEmpty() && !requestedIds.contains(documentId)) {
|
||||
continue;
|
||||
}
|
||||
String run = document.path("run").asText("");
|
||||
Integer progress = document.has("progress") && document.path("progress").canConvertToInt()
|
||||
? document.path("progress").asInt() : null;
|
||||
statuses.add(new DocumentStatus(
|
||||
documentId,
|
||||
normalizeMuseStatus(run, progress),
|
||||
run,
|
||||
progress,
|
||||
firstText(document, "progress_msg", "progressMsg")
|
||||
));
|
||||
}
|
||||
return statuses;
|
||||
}
|
||||
|
||||
private String normalizeMuseStatus(String run, Integer progress) {
|
||||
String normalizedRun = run == null ? "" : run.toLowerCase(Locale.ROOT);
|
||||
if (Set.of("done", "completed", "success").contains(normalizedRun) || Integer.valueOf(100).equals(progress)) {
|
||||
return "searchable";
|
||||
}
|
||||
if (Set.of("running", "parsing", "indexing", "queued").contains(normalizedRun)) {
|
||||
return "processing";
|
||||
}
|
||||
if (Set.of("failed", "error", "cancelled").contains(normalizedRun)) {
|
||||
return "failed";
|
||||
}
|
||||
return "pending_process";
|
||||
}
|
||||
|
||||
private JsonNode parseBody(String responseBody) {
|
||||
try {
|
||||
return JsonUtils.parseTree(responseBody == null || responseBody.isBlank() ? "{}" : responseBody);
|
||||
} catch (RuntimeException ex) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private String findText(JsonNode node, String fieldName) {
|
||||
if (node == null || node.isMissingNode() || node.isNull()) {
|
||||
return null;
|
||||
}
|
||||
if (node.has(fieldName) && node.path(fieldName).isTextual()) {
|
||||
return node.path(fieldName).asText();
|
||||
}
|
||||
if (node.isArray() && !node.isEmpty()) {
|
||||
return findText(node.get(0), fieldName);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private String firstText(JsonNode node, String... fieldNames) {
|
||||
for (String fieldName : fieldNames) {
|
||||
if (node.has(fieldName) && node.path(fieldName).isTextual()) {
|
||||
return node.path(fieldName).asText();
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private MultipartPayload buildMultipartPayload(List<DocumentUpload> documents) {
|
||||
if (documents == null || documents.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
String boundary = "----MuseRagFlow" + UUID.randomUUID().toString().replace("-", "");
|
||||
try {
|
||||
ByteArrayOutputStream body = new ByteArrayOutputStream();
|
||||
for (DocumentUpload document : documents) {
|
||||
if (document == null || document.content() == null || document.content().length == 0
|
||||
|| !StringUtils.hasText(document.fileName())) {
|
||||
return null;
|
||||
}
|
||||
String contentType = StringUtils.hasText(document.contentType()) ? document.contentType() : "application/octet-stream";
|
||||
writeAscii(body, "--" + boundary + "\r\n");
|
||||
writeAscii(body, "Content-Disposition: form-data; name=\"file\"; filename=\""
|
||||
+ escapeMultipart(document.fileName()) + "\"\r\n");
|
||||
writeAscii(body, "Content-Type: " + contentType + "\r\n\r\n");
|
||||
// Task 3 只承接小型已材料化资源；Task 5 再接对象存储/流式文件边界。
|
||||
body.write(document.content());
|
||||
writeAscii(body, "\r\n");
|
||||
}
|
||||
writeAscii(body, "--" + boundary + "--\r\n");
|
||||
return new MultipartPayload(body.toByteArray(), "multipart/form-data; boundary=" + boundary);
|
||||
} catch (IOException ex) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private Map<String, Object> buildMetadataSummary(Operation operation, JsonNode bodyTree, String externalId,
|
||||
FailureClass failureClass) {
|
||||
Map<String, Object> summary = new LinkedHashMap<>();
|
||||
summary.put("operation", operation.wireName());
|
||||
summary.put("status", failureClass == null ? Status.SUCCEEDED.name().toLowerCase(Locale.ROOT)
|
||||
: Status.FAILED.name().toLowerCase(Locale.ROOT));
|
||||
if (bodyTree != null && bodyTree.has("code")) {
|
||||
summary.put("code", bodyTree.path("code"));
|
||||
}
|
||||
if (StringUtils.hasText(externalId)) {
|
||||
summary.put("externalId", externalId);
|
||||
}
|
||||
if (failureClass != null) {
|
||||
summary.put("failureClass", failureClass.name());
|
||||
}
|
||||
JsonNode data = bodyTree == null ? null : bodyTree.path("data");
|
||||
if (data != null && data.isArray()) {
|
||||
summary.put("count", data.size());
|
||||
} else if (data != null && data.isObject()) {
|
||||
putCountIfPresent(summary, data, "docs");
|
||||
putCountIfPresent(summary, data, "documents");
|
||||
putCountIfPresent(summary, data, "chunks");
|
||||
putCountIfPresent(summary, data, "nodes");
|
||||
putCountIfPresent(summary, data, "edges");
|
||||
JsonNode graph = data.path("graph");
|
||||
if (graph.isObject()) {
|
||||
putCountIfPresent(summary, graph, "nodes");
|
||||
putCountIfPresent(summary, graph, "edges");
|
||||
}
|
||||
if (data.has("mind_map")) {
|
||||
summary.put("mindMapPresent", !data.path("mind_map").isNull());
|
||||
if (data.path("mind_map").isArray()) {
|
||||
summary.put("mindMapCount", data.path("mind_map").size());
|
||||
}
|
||||
} else if (data.has("mindMap")) {
|
||||
summary.put("mindMapPresent", !data.path("mindMap").isNull());
|
||||
if (data.path("mindMap").isArray()) {
|
||||
summary.put("mindMapCount", data.path("mindMap").size());
|
||||
}
|
||||
}
|
||||
if (data.has("progress")) {
|
||||
summary.put("progress", data.path("progress"));
|
||||
}
|
||||
}
|
||||
return summary;
|
||||
}
|
||||
|
||||
private void putCountIfPresent(Map<String, Object> summary, JsonNode data, String fieldName) {
|
||||
if (data.has(fieldName) && data.path(fieldName).isArray()) {
|
||||
summary.put(fieldName + "Count", data.path(fieldName).size());
|
||||
}
|
||||
}
|
||||
|
||||
private String sanitizeSummary(Object value) {
|
||||
return RagFlowKnowledgeRedactor.sanitizeObject(value);
|
||||
}
|
||||
|
||||
private long elapsedMillis(Instant startedAt) {
|
||||
return Math.max(0L, Duration.between(startedAt, Instant.now()).toMillis());
|
||||
}
|
||||
|
||||
private String escape(String value) {
|
||||
return value == null ? "" : value.replace("\\", "\\\\").replace("\"", "\\\"");
|
||||
}
|
||||
|
||||
private String escapeMultipart(String value) {
|
||||
return escape(value).replace("\r", "").replace("\n", "");
|
||||
}
|
||||
|
||||
private String encode(String value) {
|
||||
return URLEncoder.encode(value == null ? "" : value, StandardCharsets.UTF_8);
|
||||
}
|
||||
|
||||
private boolean looksLikeJson(String value) {
|
||||
if (value == null) {
|
||||
return false;
|
||||
}
|
||||
String trimmed = value.trim();
|
||||
return (trimmed.startsWith("{") && trimmed.endsWith("}"))
|
||||
|| (trimmed.startsWith("[") && trimmed.endsWith("]"));
|
||||
}
|
||||
|
||||
private String sha256(String payload) {
|
||||
try {
|
||||
MessageDigest digest = MessageDigest.getInstance("SHA-256");
|
||||
return HexFormat.of().formatHex(digest.digest(payload.getBytes(StandardCharsets.UTF_8)));
|
||||
} catch (NoSuchAlgorithmException e) {
|
||||
throw new IllegalStateException("JDK 缺少 SHA-256 摘要算法", e);
|
||||
}
|
||||
}
|
||||
|
||||
private void writeAscii(ByteArrayOutputStream outputStream, String text) throws IOException {
|
||||
outputStream.write(text.getBytes(StandardCharsets.US_ASCII));
|
||||
}
|
||||
|
||||
private static String normalizeBaseUrl(String rawBaseUrl) {
|
||||
if (rawBaseUrl == null || rawBaseUrl.isBlank()) {
|
||||
return null;
|
||||
}
|
||||
return rawBaseUrl.endsWith("/") ? rawBaseUrl.substring(0, rawBaseUrl.length() - 1) : rawBaseUrl;
|
||||
}
|
||||
|
||||
protected record HttpRequestPayload(String method, String path, byte[] body, String contentType) {
|
||||
}
|
||||
|
||||
protected record HttpResponsePayload(int statusCode, String body, HttpHeaders headers) {
|
||||
}
|
||||
|
||||
private record MultipartPayload(byte[] body, String contentType) {
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,54 +0,0 @@
|
||||
package cn.iocoder.muse.module.knowledge.application.muse.facade;
|
||||
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.core.env.Environment;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
/**
|
||||
* RAGFlow runtime client 的 Spring 装配配置。
|
||||
*
|
||||
* <p>与 {@link DifyKnowledgeRuntimeClientConfiguration} 按 provider 选择器 {@code muse.knowledge.runtime-provider}
|
||||
* 互斥:仅在 {@code =ragflow} 或缺省(matchIfMissing=true,即保持既有部署零变更)时参与,{@code =dify} 时让位给 Dify 装配。
|
||||
* S6d 摘除 RAGFlow 时连同本类一并删除。</p>
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
public class RagFlowKnowledgeRuntimeClientConfiguration {
|
||||
|
||||
private static final String PROPERTY_PREFIX = "muse.knowledge.ragflow.";
|
||||
private static final int DEFAULT_TIMEOUT_SECONDS = 5;
|
||||
private static final int DEFAULT_RETRY_BUDGET = 0;
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean(KnowledgeRuntimeClient.class)
|
||||
@ConditionalOnProperty(name = "muse.knowledge.runtime-provider", havingValue = "ragflow", matchIfMissing = true)
|
||||
public KnowledgeRuntimeClient ragFlowKnowledgeRuntimeClient(Environment environment) {
|
||||
String baseUrl = environment.getProperty(PROPERTY_PREFIX + "base-url");
|
||||
String apiKey = environment.getProperty(PROPERTY_PREFIX + "api-key");
|
||||
if (!StringUtils.hasText(baseUrl) || !StringUtils.hasText(apiKey)) {
|
||||
return new UnavailableRagFlowKnowledgeRuntimeClient();
|
||||
}
|
||||
|
||||
int timeoutSeconds = positiveInt(environment, "timeout-seconds", DEFAULT_TIMEOUT_SECONDS);
|
||||
int retryBudget = nonNegativeInt(environment, "retry-budget", DEFAULT_RETRY_BUDGET);
|
||||
boolean graphRagAttributionReady = Boolean.TRUE.equals(environment.getProperty(
|
||||
PROPERTY_PREFIX + "graphrag-attribution-ready", Boolean.class, false));
|
||||
return new HttpRagFlowKnowledgeRuntimeClient(baseUrl, apiKey, Duration.ofSeconds(timeoutSeconds),
|
||||
retryBudget, graphRagAttributionReady);
|
||||
}
|
||||
|
||||
private int positiveInt(Environment environment, String propertyName, int defaultValue) {
|
||||
Integer value = environment.getProperty(PROPERTY_PREFIX + propertyName, Integer.class, defaultValue);
|
||||
return value == null || value <= 0 ? defaultValue : value;
|
||||
}
|
||||
|
||||
private int nonNegativeInt(Environment environment, String propertyName, int defaultValue) {
|
||||
Integer value = environment.getProperty(PROPERTY_PREFIX + propertyName, Integer.class, defaultValue);
|
||||
return value == null || value < 0 ? defaultValue : value;
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,46 +0,0 @@
|
||||
package cn.iocoder.muse.module.knowledge.application.muse.facade;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* RAGFlow 未配置时的 fail closed client。
|
||||
*/
|
||||
public class UnavailableRagFlowKnowledgeRuntimeClient implements KnowledgeRuntimeClient {
|
||||
|
||||
@Override
|
||||
public RuntimeResult createDataset(CreateDatasetCommand command) {
|
||||
return unavailable(Operation.CREATE_DATASET, command.correlationId(), command.requestHash(), command.attempt(), null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeResult uploadDocuments(UploadDocumentsCommand command) {
|
||||
return unavailable(Operation.UPLOAD_DOCUMENTS, command.correlationId(), command.requestHash(), command.attempt(), null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeResult startParseDocuments(StartParseDocumentsCommand command) {
|
||||
return unavailable(Operation.START_PARSE_DOCUMENTS, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), command.processingTaskId());
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeResult pollDocumentStatuses(PollDocumentStatusesCommand command) {
|
||||
return unavailable(Operation.POLL_DOCUMENT_STATUSES, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), command.processingTaskId());
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeResult retrieveChunks(RetrieveChunksCommand command) {
|
||||
return unavailable(Operation.RETRIEVE_CHUNKS, command.correlationId(), command.requestHash(), command.attempt(), null);
|
||||
}
|
||||
|
||||
private RuntimeResult unavailable(Operation operation, String correlationId, String requestHash,
|
||||
int attempt, String processingTaskId) {
|
||||
return new RuntimeResult(null, operation, correlationId, requestHash, attempt, 0L,
|
||||
Status.FAILED, FailureClass.CONFIG_MISSING, processingTaskId,
|
||||
"{\"status\":\"failed\",\"failureClass\":\"CONFIG_MISSING\"}",
|
||||
List.of(), Map.of("failureClass", FailureClass.CONFIG_MISSING.name()));
|
||||
}
|
||||
|
||||
}
|
||||
@ -8,17 +8,17 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
|
||||
|
||||
/**
|
||||
* Dify Knowledge 运行时 client 装配测试。
|
||||
* Dify Knowledge 运行时 client 装配测试(S6d 起 Dify 为缺省运行时)。
|
||||
*
|
||||
* <p>两层验证:①装配方法本身(配 dify.* 全→Dify、缺→Unavailable);②与 RAGFlow 装配共存时按
|
||||
* {@code muse.knowledge.runtime-provider} 选择器确定性互斥——任何组合下只装一个 {@link KnowledgeRuntimeClient},
|
||||
* 不依赖组件扫描/Bean 求值顺序。</p>
|
||||
* <p>两层验证:①装配方法本身(配 dify.* 全→Dify、缺→Unavailable);②provider 选择器
|
||||
* {@code muse.knowledge.runtime-provider} 语义——取 {@code dify} 或缺省装 Dify,显式配成其它值(如遗留
|
||||
* {@code ragflow},运行时已删)由兜底装通用 {@link UnavailableKnowledgeRuntimeClient};任何取值下都有且仅有
|
||||
* 一个 {@link KnowledgeRuntimeClient}。</p>
|
||||
*/
|
||||
class DifyKnowledgeRuntimeClientConfigurationTest {
|
||||
|
||||
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
|
||||
.withUserConfiguration(DifyKnowledgeRuntimeClientConfiguration.class,
|
||||
RagFlowKnowledgeRuntimeClientConfiguration.class);
|
||||
.withUserConfiguration(DifyKnowledgeRuntimeClientConfiguration.class);
|
||||
|
||||
// ==================== 装配方法本身 ====================
|
||||
|
||||
@ -58,7 +58,7 @@ class DifyKnowledgeRuntimeClientConfigurationTest {
|
||||
assertInstanceOf(UnavailableKnowledgeRuntimeClient.class, client);
|
||||
}
|
||||
|
||||
// ==================== 与 RAGFlow 装配共存:确定性互斥 ====================
|
||||
// ==================== provider 选择器:缺省 Dify + 非 dify 值兜底 Unavailable ====================
|
||||
|
||||
@Test
|
||||
void should_installDifyOnlyWhenProviderIsDify() {
|
||||
@ -85,37 +85,36 @@ class DifyKnowledgeRuntimeClientConfigurationTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_installRagFlowWhenProviderIsRagflow() {
|
||||
contextRunner.withPropertyValues(
|
||||
"muse.knowledge.runtime-provider=ragflow",
|
||||
"muse.knowledge.ragflow.base-url=https://ragflow.example",
|
||||
"muse.knowledge.ragflow.api-key=ragflow-key")
|
||||
void should_installUnavailableWhenProviderIsRagflow() {
|
||||
// 遗留 provider=ragflow(运行时 S6d 已删):Dify 主 bean 让位,兜底装通用 UnavailableKnowledgeRuntimeClient,
|
||||
// 既不因无 bean 启动失败,也不静默降级。
|
||||
contextRunner.withPropertyValues("muse.knowledge.runtime-provider=ragflow")
|
||||
.run(context -> {
|
||||
assertEquals(1, context.getBeansOfType(KnowledgeRuntimeClient.class).size());
|
||||
assertInstanceOf(HttpRagFlowKnowledgeRuntimeClient.class,
|
||||
assertInstanceOf(UnavailableKnowledgeRuntimeClient.class,
|
||||
context.getBean(KnowledgeRuntimeClient.class));
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_defaultToRagFlowWhenProviderUnset() {
|
||||
// 缺省 provider(既有部署零变更):matchIfMissing 让 RAGFlow 装配继续生效,Dify 让位。
|
||||
void should_defaultToDifyWhenProviderUnset() {
|
||||
// 缺省 provider(未显式配置):matchIfMissing 让 Dify 成缺省运行时。
|
||||
contextRunner.withPropertyValues(
|
||||
"muse.knowledge.ragflow.base-url=https://ragflow.example",
|
||||
"muse.knowledge.ragflow.api-key=ragflow-key")
|
||||
"muse.knowledge.dify.base-url=http://dify.example/v1",
|
||||
"muse.knowledge.dify.dataset-api-key=dataset-key")
|
||||
.run(context -> {
|
||||
assertEquals(1, context.getBeansOfType(KnowledgeRuntimeClient.class).size());
|
||||
assertInstanceOf(HttpRagFlowKnowledgeRuntimeClient.class,
|
||||
assertInstanceOf(DifyKnowledgeRuntimeClient.class,
|
||||
context.getBean(KnowledgeRuntimeClient.class));
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_installSingleUnavailableWhenProviderUnsetAndNothingConfigured() {
|
||||
// 缺省 provider + 未配 dify.*:缺省进 Dify 装配但配缺 → fail-closed 通用 Unavailable 桩,仍确保只有一个 bean。
|
||||
contextRunner.run(context -> {
|
||||
// 仍确保只有一个 bean(RAGFlow 的 Unavailable 桩),不因两装配类共存而重复注册。
|
||||
assertEquals(1, context.getBeansOfType(KnowledgeRuntimeClient.class).size());
|
||||
assertInstanceOf(UnavailableRagFlowKnowledgeRuntimeClient.class,
|
||||
assertInstanceOf(UnavailableKnowledgeRuntimeClient.class,
|
||||
context.getBean(KnowledgeRuntimeClient.class));
|
||||
});
|
||||
}
|
||||
|
||||
@ -1,398 +0,0 @@
|
||||
package cn.iocoder.muse.module.knowledge.application.muse.facade;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.http.HttpHeaders;
|
||||
import java.time.Duration;
|
||||
import java.util.Arrays;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* Knowledge 运行时 client 基础行为测试。
|
||||
*/
|
||||
class KnowledgeRuntimeClientTest {
|
||||
|
||||
@Test
|
||||
void unavailableClientShouldFailClosedWhenConfigMissing() {
|
||||
KnowledgeRuntimeClient client = new UnavailableRagFlowKnowledgeRuntimeClient();
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L,
|
||||
"KB", Map.of(), "corr-1", "hash-1", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.CONFIG_MISSING, result.failureClass());
|
||||
assertEquals("createDataset", result.operation().wireName());
|
||||
assertFalse(result.redactedSummary().isBlank());
|
||||
}
|
||||
|
||||
@Test
|
||||
void startParseShouldExposeOnlyMuseProcessingTaskAndCorrelation() {
|
||||
HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key", """
|
||||
{"code":0,"data":{"task_id":"external-task-should-not-leak","id":"external-task-should-not-leak"}}
|
||||
""");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.startParseDocuments(
|
||||
new KnowledgeRuntimeClient.StartParseDocumentsCommand(100L, 2001L, 4001L,
|
||||
"dataset-1", List.of("doc-1"), "muse-task-1", "corr-parse-1", "hash-parse-1", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.ACCEPTED, result.status());
|
||||
assertNull(result.externalId());
|
||||
assertEquals("muse-task-1", result.processingTaskId());
|
||||
assertEquals("corr-parse-1", result.correlationId());
|
||||
assertFalse(result.redactedSummary().contains("external-task-should-not-leak"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void redactionShouldNotLeakCredentialsFromSummaries() {
|
||||
HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key", "{\"code\":0,\"data\":{\"id\":\"dataset-1\"}}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L,
|
||||
"KB", Map.of("apiKey", "secret-value", "authorization", "Bearer abcdefgh"),
|
||||
"corr-1", "hash-1", 1));
|
||||
|
||||
assertFalse(result.redactedSummary().contains("secret-value"));
|
||||
assertFalse(result.redactedSummary().toLowerCase().contains("bearer abcdefgh"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void createDatasetShouldSendOnlyNameWithoutConfigPayload() {
|
||||
CapturingHttpRagFlowKnowledgeRuntimeClient client = new CapturingHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key", "{\"code\":0,\"data\":{\"id\":\"dataset-1\"}}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L,
|
||||
"KB", Map.of("chunk_method", "naive", "apiKey", "secret-value"),
|
||||
"corr-create-name-only", "hash-create-name-only", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status());
|
||||
String requestBody = new String(client.requests.get(0).body(), StandardCharsets.UTF_8);
|
||||
assertTrue(requestBody.contains("KB"));
|
||||
assertFalse(requestBody.contains("config"));
|
||||
assertFalse(requestBody.contains("chunk_method"));
|
||||
assertFalse(requestBody.contains("secret-value"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void ragflowEnvelopeNonZeroCodeShouldFailClosedInsteadOfSucceeded() {
|
||||
HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key", "{\"code\":401,\"message\":\"auth failed\"}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L,
|
||||
"KB", Map.of(), "corr-envelope-1", "hash-envelope-1", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.AUTH_FAILED, result.failureClass());
|
||||
assertNull(result.externalId());
|
||||
}
|
||||
|
||||
@Test
|
||||
void createDatasetShouldFailClosedWhenRequiredExternalIdMissing() {
|
||||
HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key", "{\"code\":0,\"data\":{\"name\":\"KB\"}}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L,
|
||||
"KB", Map.of(), "corr-envelope-2", "hash-envelope-2", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.RESPONSE_SCHEMA_CHANGED, result.failureClass());
|
||||
}
|
||||
|
||||
@Test
|
||||
void failureSummaryShouldNotContainPlainProviderErrorBody() {
|
||||
String providerError = "plain provider unavailable body";
|
||||
HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key", 500, providerError);
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L,
|
||||
"KB", Map.of(), "corr-fail-body-1", "hash-fail-body-1", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.RAGFLOW_UNAVAILABLE, result.failureClass());
|
||||
assertFalse(result.redactedSummary().contains(providerError));
|
||||
assertTrue(result.redactedSummary().contains("\"redacted\":true"));
|
||||
assertTrue(result.redactedSummary().contains("\"sha256\""));
|
||||
}
|
||||
|
||||
@Test
|
||||
void failureSummaryShouldNotContainJsonProviderMessageOrErrorFields() {
|
||||
String providerMessage = "plain provider raw error";
|
||||
String providerError = "raw error field";
|
||||
HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key", 500,
|
||||
"{\"message\":\"" + providerMessage + "\",\"error\":\"" + providerError + "\"}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L,
|
||||
"KB", Map.of(), "corr-fail-body-json-1", "hash-fail-body-json-1", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.RAGFLOW_UNAVAILABLE, result.failureClass());
|
||||
assertFalse(result.redactedSummary().contains(providerMessage));
|
||||
assertFalse(result.redactedSummary().contains(providerError));
|
||||
assertTrue(result.redactedSummary().contains("\"redacted\":true"));
|
||||
assertTrue(result.redactedSummary().contains("\"sha256\""));
|
||||
}
|
||||
|
||||
@Test
|
||||
void failureSummaryShouldNotContainJsonProviderErrorsArray() {
|
||||
String providerError = "array raw error";
|
||||
HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key", 400,
|
||||
"{\"errors\":[{\"message\":\"" + providerError + "\"}]}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L,
|
||||
"KB", Map.of(), "corr-fail-body-json-2", "hash-fail-body-json-2", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.VALIDATION_ERROR, result.failureClass());
|
||||
assertFalse(result.redactedSummary().contains(providerError));
|
||||
assertTrue(result.redactedSummary().contains("\"redacted\":true"));
|
||||
assertTrue(result.redactedSummary().contains("\"sha256\""));
|
||||
}
|
||||
|
||||
@Test
|
||||
void nonZeroEnvelopeFailureSummaryShouldNotContainPlainMessage() {
|
||||
String providerMessage = "business rule rejected this dataset";
|
||||
HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key",
|
||||
"{\"code\":422,\"message\":\"" + providerMessage + "\"}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L,
|
||||
"KB", Map.of(), "corr-fail-body-2", "hash-fail-body-2", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.VALIDATION_ERROR, result.failureClass());
|
||||
assertFalse(result.redactedSummary().contains(providerMessage));
|
||||
assertTrue(result.redactedSummary().contains("\"redacted\":true"));
|
||||
assertTrue(result.redactedSummary().contains("\"sha256\""));
|
||||
}
|
||||
|
||||
@Test
|
||||
void uploadDocumentsShouldUseMultipartAndRejectMissingMaterializedBytes() {
|
||||
CapturingHttpRagFlowKnowledgeRuntimeClient client = new CapturingHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key", "{\"code\":0,\"data\":[{\"id\":\"doc-1\"}]}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult rejected = client.uploadDocuments(
|
||||
new KnowledgeRuntimeClient.UploadDocumentsCommand(100L, 2001L, 4001L,
|
||||
"dataset-1", List.of(new KnowledgeRuntimeClient.DocumentUpload(
|
||||
5001L, 6001L, "file-ref-1", "novel.txt", "text/plain", 10L, null)),
|
||||
"corr-upload-1", "hash-upload-1", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, rejected.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.FILE_REJECTED, rejected.failureClass());
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult uploaded = client.uploadDocuments(
|
||||
new KnowledgeRuntimeClient.UploadDocumentsCommand(100L, 2001L, 4001L,
|
||||
"dataset-1", List.of(new KnowledgeRuntimeClient.DocumentUpload(
|
||||
5001L, 6001L, "file-ref-1", "novel.txt", "text/plain", 5L,
|
||||
"hello".getBytes(StandardCharsets.UTF_8))),
|
||||
"corr-upload-2", "hash-upload-2", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, uploaded.status());
|
||||
assertEquals("POST", client.requests.get(0).method());
|
||||
assertTrue(client.requests.get(0).contentType().startsWith("multipart/form-data; boundary="));
|
||||
assertTrue(new String(client.requests.get(0).body(), StandardCharsets.ISO_8859_1).contains("filename=\"novel.txt\""));
|
||||
assertFalse(new String(client.requests.get(0).body(), StandardCharsets.ISO_8859_1).contains("\"documents\""));
|
||||
}
|
||||
|
||||
@Test
|
||||
void uploadDocumentsShouldPreserveBinaryBytesInsideMultipartPayload() {
|
||||
CapturingHttpRagFlowKnowledgeRuntimeClient client = new CapturingHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key", "{\"code\":0,\"data\":[{\"id\":\"doc-1\"}]}");
|
||||
byte[] binaryContent = new byte[]{0x25, 0x50, 0x44, 0x46, 0x00, (byte) 0x80, (byte) 0xFF, 0x01};
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult uploaded = client.uploadDocuments(
|
||||
new KnowledgeRuntimeClient.UploadDocumentsCommand(100L, 2001L, 4001L,
|
||||
"dataset-1", List.of(new KnowledgeRuntimeClient.DocumentUpload(
|
||||
5001L, 6001L, "file-ref-1", "binary.pdf", "application/pdf",
|
||||
(long) binaryContent.length, binaryContent)),
|
||||
"corr-upload-bin", "hash-upload-bin", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, uploaded.status());
|
||||
assertTrue(containsSubsequence(client.requests.get(0).body(), binaryContent));
|
||||
}
|
||||
|
||||
@Test
|
||||
void redactedSummaryShouldNotLeakChunkQuestionPromptOrResponseContent() {
|
||||
String responseBody = """
|
||||
{"code":0,"data":{"chunks":[{"id":"chunk-1","content":"secret chunk content"}],"response":"raw answer text"}}
|
||||
""";
|
||||
HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key", responseBody);
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.retrieveChunks(
|
||||
new KnowledgeRuntimeClient.RetrieveChunksCommand(100L, 2001L, 4001L,
|
||||
List.of("dataset-1"), List.of("doc-1"), "private retrieval question",
|
||||
3, 0.5, Map.of("prompt", "private prompt"), "corr-redact-1", "hash-redact-1", 1));
|
||||
|
||||
assertFalse(result.redactedSummary().contains("secret chunk content"));
|
||||
assertFalse(result.redactedSummary().contains("raw answer text"));
|
||||
assertFalse(result.redactedSummary().contains("private retrieval question"));
|
||||
assertFalse(result.redactedSummary().contains("private prompt"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void retrieveChunksShouldSucceedWithoutTopLevelIdAndKeepOnlyMetadataSummary() {
|
||||
HttpRagFlowKnowledgeRuntimeClient client = new StubHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key", """
|
||||
{"code":0,"data":{"chunks":[{"content":"secret chunk content","score":0.91}],"response":"raw answer text"}}
|
||||
""");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.retrieveChunks(
|
||||
new KnowledgeRuntimeClient.RetrieveChunksCommand(100L, 2001L, 4001L,
|
||||
List.of("dataset-1"), List.of("doc-1"), "private retrieval question",
|
||||
3, 0.5, Map.of("prompt", "private prompt"), "corr-retrieve-1", "hash-retrieve-1", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status());
|
||||
assertNull(result.externalId());
|
||||
assertTrue(result.redactedSummary().contains("\"chunksCount\":1"));
|
||||
assertFalse(result.redactedSummary().contains("secret chunk content"));
|
||||
assertFalse(result.redactedSummary().contains("raw answer text"));
|
||||
assertFalse(result.redactedSummary().contains("private retrieval question"));
|
||||
assertFalse(result.redactedSummary().contains("private prompt"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void retrieveChunksShouldOmitDocumentIdsFromBodyWhenNull() {
|
||||
// 回归:document_ids 为 null(P-A 按授权 dataset 全量检索、不限定具体文档)时不得放入 body;
|
||||
// RAGFlow 要求其为 list,传 null 会被拒 code 102 "`documents` should be a list" 致检索 fail-closed → P-A 永不命中。
|
||||
CapturingHttpRagFlowKnowledgeRuntimeClient client = new CapturingHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key",
|
||||
"{\"code\":0,\"data\":{\"chunks\":[{\"content\":\"c\",\"score\":0.9}]}}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.retrieveChunks(
|
||||
new KnowledgeRuntimeClient.RetrieveChunksCommand(100L, 2001L, 4001L,
|
||||
List.of("dataset-1"), null, "q", 5, 0.2, null, "corr-null-doc", "hash-null-doc", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status());
|
||||
String requestBody = new String(client.requests.get(0).body(), StandardCharsets.UTF_8);
|
||||
assertFalse(requestBody.contains("document_ids"), "document_ids 为 null 时不应出现在请求 body");
|
||||
assertTrue(requestBody.contains("dataset_ids"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void retrieveChunksShouldSendMetadataConditionWithOfficialRagflowFieldName() {
|
||||
// 回归:RAGFlow 检索元数据过滤的官方字段是 metadata_condition；
|
||||
// metadata_filter 会被外部服务静默忽略，不能作为任何隔离或筛选证据。
|
||||
CapturingHttpRagFlowKnowledgeRuntimeClient client = new CapturingHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key",
|
||||
"{\"code\":0,\"data\":{\"chunks\":[{\"content\":\"c\",\"score\":0.9}]}}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.retrieveChunks(
|
||||
new KnowledgeRuntimeClient.RetrieveChunksCommand(100L, 2001L, 4001L,
|
||||
List.of("dataset-1"), List.of("doc-1"), "q", 5, 0.2,
|
||||
Map.of("logic", "and", "conditions", List.of(Map.of(
|
||||
"name", "source", "comparison_operator", "=", "value", "public"))),
|
||||
"corr-metadata-condition", "hash-metadata-condition", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status());
|
||||
String requestBody = new String(client.requests.get(0).body(), StandardCharsets.UTF_8);
|
||||
assertTrue(requestBody.contains("\"metadata_condition\""));
|
||||
assertFalse(requestBody.contains("\"metadata_filter\""));
|
||||
}
|
||||
|
||||
@Test
|
||||
void pollDocumentStatusesShouldUseRagflowIdQueryForSingleDocumentSmoke() {
|
||||
CapturingHttpRagFlowKnowledgeRuntimeClient client = new CapturingHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key", """
|
||||
{"code":0,"data":{"docs":[
|
||||
{"id":"doc-1","run":"DONE","progress":100,"progress_msg":"done"},
|
||||
{"id":"doc-2","run":"RUNNING","progress":50,"progress_msg":"running"},
|
||||
{"id":"doc-3","run":"DONE","progress":100,"progress_msg":"done"}
|
||||
]}}
|
||||
""");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.pollDocumentStatuses(
|
||||
new KnowledgeRuntimeClient.PollDocumentStatusesCommand(100L, 2001L, 4001L,
|
||||
"dataset-1", List.of("doc-1"), "task-1", "corr-poll-1", "hash-poll-1", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status());
|
||||
assertEquals(List.of("doc-1"), result.documentStatuses().stream()
|
||||
.map(KnowledgeRuntimeClient.DocumentStatus::ragflowDocumentId).toList());
|
||||
assertTrue(client.requests.get(0).path().contains("id=doc-1"));
|
||||
assertFalse(client.requests.get(0).path().contains("document_ids="));
|
||||
}
|
||||
|
||||
@Test
|
||||
void pollDocumentStatusesShouldRejectMultipleDocumentIdsBeforeCallingRagflow() {
|
||||
CapturingHttpRagFlowKnowledgeRuntimeClient client = new CapturingHttpRagFlowKnowledgeRuntimeClient(
|
||||
"https://ragflow.example", "configured-api-key", "{\"code\":0,\"data\":{\"docs\":[]}}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.pollDocumentStatuses(
|
||||
new KnowledgeRuntimeClient.PollDocumentStatusesCommand(100L, 2001L, 4001L,
|
||||
"dataset-1", List.of("doc-1", "doc-2"), "task-1", "corr-poll-many", "hash-poll-many", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.VALIDATION_ERROR, result.failureClass());
|
||||
assertTrue(result.redactedSummary().contains("single document status polling"));
|
||||
assertEquals(0, client.requests.size());
|
||||
}
|
||||
|
||||
private static boolean containsSubsequence(byte[] payload, byte[] expected) {
|
||||
for (int start = 0; start <= payload.length - expected.length; start++) {
|
||||
if (Arrays.equals(Arrays.copyOfRange(payload, start, start + expected.length), expected)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private static final class StubHttpRagFlowKnowledgeRuntimeClient extends HttpRagFlowKnowledgeRuntimeClient {
|
||||
|
||||
private final String body;
|
||||
private final int statusCode;
|
||||
|
||||
private StubHttpRagFlowKnowledgeRuntimeClient(String baseUrl, String apiKey, String body) {
|
||||
this(baseUrl, apiKey, 200, body);
|
||||
}
|
||||
|
||||
private StubHttpRagFlowKnowledgeRuntimeClient(String baseUrl, String apiKey, int statusCode, String body) {
|
||||
super(baseUrl, apiKey, Duration.ofSeconds(1), 0, true);
|
||||
this.statusCode = statusCode;
|
||||
this.body = body;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected HttpResponsePayload send(HttpRequestPayload request) throws IOException, InterruptedException {
|
||||
return new HttpResponsePayload(statusCode, body, HttpHeaders.of(Map.of(), (name, value) -> true));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static final class CapturingHttpRagFlowKnowledgeRuntimeClient extends HttpRagFlowKnowledgeRuntimeClient {
|
||||
|
||||
private final String body;
|
||||
private final List<HttpRequestPayload> requests = new ArrayList<>();
|
||||
|
||||
private CapturingHttpRagFlowKnowledgeRuntimeClient(String baseUrl, String apiKey, String body) {
|
||||
super(baseUrl, apiKey, Duration.ofSeconds(1), 0, true);
|
||||
this.body = body;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected HttpResponsePayload send(HttpRequestPayload request) throws IOException, InterruptedException {
|
||||
requests.add(request);
|
||||
return new HttpResponsePayload(200, body, HttpHeaders.of(Map.of(), (name, value) -> true));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,237 +0,0 @@
|
||||
package cn.iocoder.muse.module.knowledge.application.muse.facade;
|
||||
|
||||
import cn.iocoder.muse.framework.common.util.json.JsonUtils;
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.net.URI;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.security.MessageDigest;
|
||||
import java.security.NoSuchAlgorithmException;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.HexFormat;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.junit.jupiter.api.Assumptions.assumeTrue;
|
||||
|
||||
/**
|
||||
* P1R-5 RAGFlow 外部 live acceptance 入口。
|
||||
*
|
||||
* <p>默认必须跳过，只有显式设置 MUSE_P1R_EXTERNAL_ACCEPTANCE=true 才会真实调用外部 RAGFlow。
|
||||
* 本测试会保留新建 dataset 作为验收证据，不做删除；输出证据只打印 endpoint 与 key 指纹。</p>
|
||||
*/
|
||||
class P1rRagFlowLiveAcceptanceIT {
|
||||
|
||||
private static final String ACCEPTANCE_ENV = "MUSE_P1R_EXTERNAL_ACCEPTANCE";
|
||||
private static final String BASE_URL_ENV = "MUSE_KNOWLEDGE_RAGFLOW_BASE_URL";
|
||||
private static final String API_KEY_ENV = "MUSE_KNOWLEDGE_RAGFLOW_API_KEY";
|
||||
private static final String GRAPHRAG_READY_ENV = "MUSE_KNOWLEDGE_RAGFLOW_GRAPHRAG_ATTRIBUTION_READY";
|
||||
|
||||
@Test
|
||||
void shouldCallRagFlowThroughMuseRuntimeClientAndPrintRedactedEvidence() throws Exception {
|
||||
assumeTrue(externalAcceptanceEnabled(), "P1R 外部验收未启用，设置 MUSE_P1R_EXTERNAL_ACCEPTANCE=true 后才运行");
|
||||
|
||||
String baseUrl = requiredEnv(BASE_URL_ENV);
|
||||
String apiKey = requiredEnv(API_KEY_ENV);
|
||||
boolean graphRagReady = Boolean.parseBoolean(envOrDefault(GRAPHRAG_READY_ENV, "false"));
|
||||
HttpRagFlowKnowledgeRuntimeClient client = client(baseUrl, apiKey, graphRagReady);
|
||||
String suffix = String.valueOf(Instant.now().toEpochMilli());
|
||||
String datasetName = "p1r-live-" + suffix;
|
||||
String correlationPrefix = "p1r5-ragflow-live-" + suffix;
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult createDataset = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(1L, 1L, 1L, datasetName,
|
||||
Map.of(), correlationPrefix + "-dataset", hash(datasetName), 1));
|
||||
assertSucceeded(createDataset, "createDataset");
|
||||
assertNotNull(createDataset.externalId(), "RAGFlow createDataset 必须返回 dataset id");
|
||||
String datasetId = createDataset.externalId();
|
||||
|
||||
byte[] content = liveDocumentContent(suffix);
|
||||
KnowledgeRuntimeClient.RuntimeResult upload = client.uploadDocuments(
|
||||
new KnowledgeRuntimeClient.UploadDocumentsCommand(1L, 1L, 1L, datasetId,
|
||||
List.of(new KnowledgeRuntimeClient.DocumentUpload(1L, 1L,
|
||||
"p1r-live-" + suffix, datasetName + ".txt", "text/plain",
|
||||
(long) content.length, content)),
|
||||
correlationPrefix + "-upload", hash(datasetId + "-upload"), 1));
|
||||
assertSucceeded(upload, "uploadDocuments");
|
||||
assertNotNull(upload.externalId(), "RAGFlow uploadDocuments 必须返回 document id");
|
||||
String documentId = upload.externalId();
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult parse = client.startParseDocuments(
|
||||
new KnowledgeRuntimeClient.StartParseDocumentsCommand(1L, 1L, 1L, datasetId,
|
||||
List.of(documentId), correlationPrefix + "-parse-task",
|
||||
correlationPrefix + "-parse", hash(documentId + "-parse"), 1));
|
||||
assertEquals(KnowledgeRuntimeClient.Status.ACCEPTED, parse.status(),
|
||||
() -> "startParseDocuments 未进入 accepted：" + parse.redactedSummary());
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult poll = waitUntilSearchable(client, datasetId, documentId,
|
||||
correlationPrefix);
|
||||
KnowledgeRuntimeClient.RuntimeResult retrieval = client.retrieveChunks(
|
||||
new KnowledgeRuntimeClient.RetrieveChunksCommand(1L, 1L, 1L,
|
||||
List.of(datasetId), List.of(documentId), "P1R live acceptance marker 是什么？",
|
||||
3, 0.0, Map.of(), correlationPrefix + "-retrieval",
|
||||
hash(documentId + "-retrieval"), 1));
|
||||
assertSucceeded(retrieval, "retrieveChunks");
|
||||
assertTrue(countFromRedactedSummary(retrieval.redactedSummary(), "chunksCount") > 0,
|
||||
() -> "retrieveChunks 没有返回 chunk，不能作为 retrieval 正向证据：" + retrieval.redactedSummary());
|
||||
|
||||
Map<String, Object> evidence = redactedEvidence(baseUrl, apiKey, datasetName, datasetId, documentId,
|
||||
createDataset, upload, parse, poll, retrieval, graphRagReady);
|
||||
System.out.println(JsonUtils.toJsonString(evidence));
|
||||
}
|
||||
|
||||
private HttpRagFlowKnowledgeRuntimeClient client(String baseUrl, String apiKey, boolean graphRagReady) {
|
||||
return new HttpRagFlowKnowledgeRuntimeClient(baseUrl, apiKey,
|
||||
Duration.ofSeconds(intEnv("MUSE_KNOWLEDGE_RAGFLOW_TIMEOUT_SECONDS", 30)),
|
||||
intEnv("MUSE_KNOWLEDGE_RAGFLOW_RETRY_BUDGET", 0), graphRagReady);
|
||||
}
|
||||
|
||||
private KnowledgeRuntimeClient.RuntimeResult waitUntilSearchable(
|
||||
HttpRagFlowKnowledgeRuntimeClient client, String datasetId, String documentId, String correlationPrefix)
|
||||
throws InterruptedException {
|
||||
long deadline = System.nanoTime()
|
||||
+ Duration.ofSeconds(intEnv("MUSE_P1R_RAGFLOW_PARSE_WAIT_SECONDS", 120)).toNanos();
|
||||
KnowledgeRuntimeClient.RuntimeResult last = null;
|
||||
int attempt = 1;
|
||||
while (System.nanoTime() < deadline) {
|
||||
last = client.pollDocumentStatuses(new KnowledgeRuntimeClient.PollDocumentStatusesCommand(
|
||||
1L, 1L, 1L, datasetId, List.of(documentId), correlationPrefix + "-parse-task",
|
||||
correlationPrefix + "-poll-" + attempt, hash(documentId + "-poll-" + attempt), attempt));
|
||||
assertSucceeded(last, "pollDocumentStatuses");
|
||||
if (last.documentStatuses().stream().anyMatch(status -> "searchable".equals(status.museStatus()))) {
|
||||
return last;
|
||||
}
|
||||
Thread.sleep(2_000L);
|
||||
attempt++;
|
||||
}
|
||||
throw new AssertionError("RAGFlow document parse 未在等待窗口内进入 searchable，last="
|
||||
+ (last == null ? "none" : last.redactedSummary()));
|
||||
}
|
||||
|
||||
private void assertSucceeded(KnowledgeRuntimeClient.RuntimeResult result, String operation) {
|
||||
assertNotNull(result, operation + " result 不能为空");
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status(),
|
||||
() -> operation + " failed: " + result.redactedSummary());
|
||||
}
|
||||
|
||||
private byte[] liveDocumentContent(String suffix) {
|
||||
String text = """
|
||||
P1R live acceptance marker %s.
|
||||
This document is intentionally small and kept in RAGFlow as external acceptance evidence.
|
||||
Muse Knowledge RAGFlow smoke verifies dataset creation, document upload, parse polling, chunks listing and retrieval.
|
||||
""".formatted(suffix);
|
||||
return text.getBytes(StandardCharsets.UTF_8);
|
||||
}
|
||||
|
||||
private Map<String, Object> redactedEvidence(String baseUrl, String apiKey, String datasetName,
|
||||
String datasetId, String documentId,
|
||||
KnowledgeRuntimeClient.RuntimeResult createDataset,
|
||||
KnowledgeRuntimeClient.RuntimeResult upload,
|
||||
KnowledgeRuntimeClient.RuntimeResult parse,
|
||||
KnowledgeRuntimeClient.RuntimeResult poll,
|
||||
KnowledgeRuntimeClient.RuntimeResult retrieval,
|
||||
boolean graphRagReady) {
|
||||
Map<String, Object> evidence = new LinkedHashMap<>();
|
||||
evidence.put("acceptance", "p1r5-ragflow-live");
|
||||
evidence.put("endpoint", endpointSummary(baseUrl, "/api/v1"));
|
||||
evidence.put("apiKey", secretFingerprint(apiKey));
|
||||
evidence.put("datasetName", datasetName);
|
||||
evidence.put("datasetId", datasetId);
|
||||
evidence.put("documentId", documentId);
|
||||
evidence.put("createDataset", operationSummary(createDataset));
|
||||
evidence.put("uploadDocuments", operationSummary(upload));
|
||||
evidence.put("startParseDocuments", operationSummary(parse));
|
||||
evidence.put("pollDocumentStatuses", operationSummary(poll));
|
||||
evidence.put("documentStatuses", poll.documentStatuses());
|
||||
evidence.put("retrieveChunks", operationSummary(retrieval));
|
||||
evidence.put("retrievalChunksCount", countFromRedactedSummary(retrieval.redactedSummary(), "chunksCount"));
|
||||
evidence.put("graphRagAttributionReady", graphRagReady);
|
||||
evidence.put("datasetRetained", true);
|
||||
return evidence;
|
||||
}
|
||||
|
||||
private Map<String, Object> operationSummary(KnowledgeRuntimeClient.RuntimeResult result) {
|
||||
Map<String, Object> summary = new LinkedHashMap<>();
|
||||
summary.put("operation", result.operation().wireName());
|
||||
summary.put("status", result.status());
|
||||
summary.put("failureClass", result.failureClass());
|
||||
summary.put("durationMillis", result.durationMillis());
|
||||
summary.put("externalIdPresent", result.externalId() != null && !result.externalId().isBlank());
|
||||
return summary;
|
||||
}
|
||||
|
||||
private int countFromRedactedSummary(String redactedSummary, String fieldName) {
|
||||
JsonNode node = JsonUtils.parseTree(redactedSummary == null || redactedSummary.isBlank() ? "{}" : redactedSummary);
|
||||
return node.has(fieldName) && node.path(fieldName).canConvertToInt() ? node.path(fieldName).asInt() : 0;
|
||||
}
|
||||
|
||||
private Map<String, Object> endpointSummary(String baseUrl, String path) {
|
||||
URI uri = URI.create(trimRight(baseUrl, "/") + path);
|
||||
Map<String, Object> endpoint = new LinkedHashMap<>();
|
||||
endpoint.put("scheme", uri.getScheme());
|
||||
endpoint.put("host", uri.getHost());
|
||||
endpoint.put("port", uri.getPort());
|
||||
endpoint.put("path", uri.getPath());
|
||||
return endpoint;
|
||||
}
|
||||
|
||||
private Map<String, Object> secretFingerprint(String secret) {
|
||||
Map<String, Object> fingerprint = new LinkedHashMap<>();
|
||||
fingerprint.put("length", secret == null ? 0 : secret.length());
|
||||
fingerprint.put("sha256Prefix", sha256Prefix(secret == null ? "" : secret, 12));
|
||||
return fingerprint;
|
||||
}
|
||||
|
||||
private boolean externalAcceptanceEnabled() {
|
||||
return "true".equalsIgnoreCase(System.getenv(ACCEPTANCE_ENV));
|
||||
}
|
||||
|
||||
private String requiredEnv(String name) {
|
||||
String value = System.getenv(name);
|
||||
assertTrue(value != null && !value.isBlank(), "缺少外部验收环境变量：" + name);
|
||||
return value;
|
||||
}
|
||||
|
||||
private String envOrDefault(String name, String defaultValue) {
|
||||
String value = System.getenv(name);
|
||||
return value == null || value.isBlank() ? defaultValue : value;
|
||||
}
|
||||
|
||||
private int intEnv(String name, int defaultValue) {
|
||||
String value = System.getenv(name);
|
||||
if (value == null || value.isBlank()) {
|
||||
return defaultValue;
|
||||
}
|
||||
return Integer.parseInt(value);
|
||||
}
|
||||
|
||||
private String hash(String value) {
|
||||
return sha256Prefix(value == null ? "" : value, 64);
|
||||
}
|
||||
|
||||
private String sha256Prefix(String value, int length) {
|
||||
try {
|
||||
MessageDigest digest = MessageDigest.getInstance("SHA-256");
|
||||
String hex = HexFormat.of().formatHex(digest.digest(value.getBytes(StandardCharsets.UTF_8)));
|
||||
return hex.substring(0, Math.min(length, hex.length()));
|
||||
} catch (NoSuchAlgorithmException e) {
|
||||
throw new IllegalStateException("JDK 缺少 SHA-256 摘要算法", e);
|
||||
}
|
||||
}
|
||||
|
||||
private String trimRight(String value, String suffix) {
|
||||
String result = value == null ? "" : value.trim();
|
||||
while (result.endsWith(suffix)) {
|
||||
result = result.substring(0, result.length() - suffix.length());
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,37 +0,0 @@
|
||||
package cn.iocoder.muse.module.knowledge.application.muse.facade;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.mock.env.MockEnvironment;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
|
||||
|
||||
/**
|
||||
* RAGFlow runtime client Spring 装配测试。
|
||||
*/
|
||||
class RagFlowKnowledgeRuntimeClientConfigurationTest {
|
||||
|
||||
@Test
|
||||
void should_registerUnavailableClientWhenRequiredConfigMissing() {
|
||||
RagFlowKnowledgeRuntimeClientConfiguration configuration = new RagFlowKnowledgeRuntimeClientConfiguration();
|
||||
|
||||
KnowledgeRuntimeClient client = configuration.ragFlowKnowledgeRuntimeClient(new MockEnvironment());
|
||||
|
||||
assertInstanceOf(UnavailableRagFlowKnowledgeRuntimeClient.class, client);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_registerHttpClientWhenBaseUrlAndApiKeyAreConfigured() {
|
||||
MockEnvironment environment = new MockEnvironment()
|
||||
.withProperty("muse.knowledge.ragflow.base-url", "https://ragflow.example")
|
||||
.withProperty("muse.knowledge.ragflow.api-key", "test-api-key")
|
||||
.withProperty("muse.knowledge.ragflow.timeout-seconds", "7")
|
||||
.withProperty("muse.knowledge.ragflow.retry-budget", "2")
|
||||
.withProperty("muse.knowledge.ragflow.graphrag-attribution-ready", "true");
|
||||
RagFlowKnowledgeRuntimeClientConfiguration configuration = new RagFlowKnowledgeRuntimeClientConfiguration();
|
||||
|
||||
KnowledgeRuntimeClient client = configuration.ragFlowKnowledgeRuntimeClient(environment);
|
||||
|
||||
assertInstanceOf(HttpRagFlowKnowledgeRuntimeClient.class, client);
|
||||
}
|
||||
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
Loading…
x
Reference in New Issue
Block a user