feat(knowledge): S6b DifyKnowledgeRuntimeClient adapter(5 操作)+ provider 互斥装配
S6 第二子步(不改检索/上传/轮询流程、不加 DDL、不删 RAGFlow):新增 DifyKnowledgeRuntimeClient 实现 KnowledgeRuntimeClient 5 操作,HTTP 范式对齐现有 Dify/RagFlow client(JDK HttpClient+直连 ProxySelector+Bearer(dataset-api-key)+RagFlowKnowledgeRedactor 脱敏+可测 send() 缝+fail-closed)。 API 映射以 S1 P1rDifyDatasetsContractLiveIT 为准:createDataset POST /v1/datasets(顶层 id)、 uploadDocuments create-by-file(嵌套 document.id + 顶层 batch)、startParse 降级 no-op(Dify 上传即 索引)、pollDocumentStatuses indexing-status(batch 主键、data 数组/对象两态归一)、retrieveChunks retrieve(单库,records[].segment.content/score 归一成下游已消费的 data.chunks[].content/similarity)。 装配互斥:新增共享选择器 muse.knowledge.runtime-provider——Dify 仅 =dify 装配,RAGFlow =ragflow 或缺省(matchIfMissing=true)装配,确保单人只装一个 bean(取代非确定的 @ConditionalOnMissingBean 扫描顺序)。**默认保持 RAGFlow、既有部署零变更;激活 Dify 需 S9 solo 配置设 runtime-provider=dify**。 新增 UnavailableKnowledgeRuntimeClient 通用 fail-closed 桩。未摘 RAGFlow(S6d)。 过渡缺口(属 S6c):adapter 单库检索只取首个 datasetId,多 KB fan-out+合并归 S6c 的 RetrievalApiImpl; poll batch 依赖 S6c 把 worker 轮询主键改喂 batch;归一成 data.chunks[] 形态使 S6c 只需加 fan-out。 验证:knowledge 模块 315 用例全绿(DifyClientTest 22 + ConfigTest 8 + 旧 RAGFlow 未回归),BUILD SUCCESS;装配 ApplicationContextRunner 6 组场景实证任何组合只装 1 bean。真连冒烟归 S6e。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
lili
parent
6c68b26c6f
commit
09ff354e2f
@ -0,0 +1,615 @@
|
||||
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 com.fasterxml.jackson.databind.node.ArrayNode;
|
||||
import com.fasterxml.jackson.databind.node.ObjectNode;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.net.ProxySelector;
|
||||
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.ArrayList;
|
||||
import java.util.HexFormat;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
|
||||
/**
|
||||
* Dify Datasets API 的 Knowledge 运行时 adapter。
|
||||
*
|
||||
* <p>HTTP 范式对齐 {@code RealDifyMuseAiRuntimeClient} / {@code HttpRagFlowKnowledgeRuntimeClient}:
|
||||
* JDK {@link HttpClient} + 直连 {@link ProxySelector}(内网 Dify 必须绕过系统代理,否则 live 调用被污染成假 502)+
|
||||
* Bearer(Datasets API key)+ 脱敏审计 + fail-closed 错误映射。契约以 S1 实测的 {@code P1rDifyDatasetsContractLiveIT} 为准。</p>
|
||||
*
|
||||
* <p>5 操作与 Dify Datasets API 映射(base-url 形如 {@code http://host:port/v1},路径按 {@code /v1} 归一去重):
|
||||
* <ul>
|
||||
* <li>createDataset → {@code POST /v1/datasets},响应顶层 {@code id}。</li>
|
||||
* <li>uploadDocuments → {@code POST /v1/datasets/{id}/document/create-by-file}(multipart:data+file),
|
||||
* 响应嵌套 {@code document.id} + 顶层 {@code batch}(batch 是后续轮询主键,透出到 summary 供上传链落库)。</li>
|
||||
* <li>startParseDocuments → Dify 上传即索引,无独立触发端点:降级 no-op(不外呼),仅保留 ACCEPTED 审计记录。</li>
|
||||
* <li>pollDocumentStatuses → {@code GET /v1/datasets/{id}/documents/{batch}/indexing-status},轮询主键=batch;
|
||||
* 响应 {@code data} 可能是数组或对象,两态都处理,取 {@code indexing_status}(回退 {@code status})。</li>
|
||||
* <li>retrieveChunks → {@code POST /v1/datasets/{id}/retrieve}(单库端点,本 adapter 只管单 dataset;
|
||||
* 多 dataset 并行合并归 RetrievalApiImpl),响应 {@code records[].segment.content} + {@code records[].score}。</li>
|
||||
* </ul>
|
||||
* </p>
|
||||
*/
|
||||
public class DifyKnowledgeRuntimeClient implements KnowledgeRuntimeClient {
|
||||
|
||||
private static final ProxySelector DIRECT_PROXY_SELECTOR = ProxySelector.of(null);
|
||||
|
||||
private final String baseUrl;
|
||||
private final String apiKey;
|
||||
private final Duration timeout;
|
||||
private final int retryBudget;
|
||||
private final HttpClient httpClient;
|
||||
|
||||
public DifyKnowledgeRuntimeClient(String baseUrl, String apiKey, Duration timeout, int retryBudget) {
|
||||
this.baseUrl = normalizeBaseUrl(baseUrl);
|
||||
this.apiKey = apiKey;
|
||||
this.timeout = timeout == null ? Duration.ofSeconds(5) : timeout;
|
||||
this.retryBudget = Math.max(retryBudget, 0);
|
||||
this.httpClient = HttpClient.newBuilder()
|
||||
.connectTimeout(this.timeout)
|
||||
// Dify 部署在 tailnet 内网,显式直连避免系统代理污染 live 调用与验收。
|
||||
.proxy(DIRECT_PROXY_SELECTOR)
|
||||
.build();
|
||||
}
|
||||
|
||||
// ==================== 5 操作 ====================
|
||||
|
||||
@Override
|
||||
public RuntimeResult createDataset(CreateDatasetCommand command) {
|
||||
// S1 契约:仅提交 name + permission=only_me + indexing_technique=high_quality;禁止把公共资料/作品私有资料
|
||||
// 混进单一 dataset 后仅靠 metadata 当边界,因此固定 only_me(作品/公共 dataset 隔离由上层拓扑负责)。
|
||||
Map<String, Object> body = new LinkedHashMap<>();
|
||||
body.put("name", command.name());
|
||||
body.put("permission", "only_me");
|
||||
body.put("indexing_technique", "high_quality");
|
||||
Exchange exchange = exchange(Operation.CREATE_DATASET, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), "POST", "/v1/datasets", jsonBody(body), "application/json", null);
|
||||
if (exchange instanceof Exchange.Failure failure) {
|
||||
return failure.result();
|
||||
}
|
||||
JsonNode responseBody = ((Exchange.Success) exchange).body();
|
||||
// Dify 建库响应无 data 信封,id 在顶层。
|
||||
String datasetId = text(responseBody, "id");
|
||||
if (!StringUtils.hasText(datasetId)) {
|
||||
return failure(Operation.CREATE_DATASET, command.correlationId(), command.requestHash(), command.attempt(),
|
||||
null, FailureClass.RESPONSE_SCHEMA_CHANGED, exchange.durationMillis(), "dataset id missing");
|
||||
}
|
||||
Map<String, Object> metadata = metadata(Operation.CREATE_DATASET);
|
||||
metadata.put("externalId", datasetId);
|
||||
return succeeded(datasetId, Operation.CREATE_DATASET, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), null, exchange.durationMillis(), metadata, null, List.of());
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeResult uploadDocuments(UploadDocumentsCommand command) {
|
||||
// Dify create-by-file 是单文件端点;当前上传链每次只投一份已材料化资源。
|
||||
MultipartPayload payload = buildSingleFileMultipart(command.documents());
|
||||
if (payload == null) {
|
||||
return failure(Operation.UPLOAD_DOCUMENTS, command.correlationId(), command.requestHash(), command.attempt(),
|
||||
null, FailureClass.FILE_REJECTED, 0L, "single materialized document required for dify upload");
|
||||
}
|
||||
Exchange exchange = exchange(Operation.UPLOAD_DOCUMENTS, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), "POST",
|
||||
"/v1/datasets/" + command.ragflowDatasetId() + "/document/create-by-file",
|
||||
payload.body(), payload.contentType(), null);
|
||||
if (exchange instanceof Exchange.Failure failure) {
|
||||
return failure.result();
|
||||
}
|
||||
JsonNode responseBody = ((Exchange.Success) exchange).body();
|
||||
// 契约:document.id 嵌套、batch 在顶层;batch 是 Dify 索引批次,也是后续轮询主键,二者缺一都无法闭环。
|
||||
String documentId = text(responseBody.path("document"), "id");
|
||||
String batch = text(responseBody, "batch");
|
||||
if (!StringUtils.hasText(documentId) || !StringUtils.hasText(batch)) {
|
||||
return failure(Operation.UPLOAD_DOCUMENTS, command.correlationId(), command.requestHash(), command.attempt(),
|
||||
null, FailureClass.RESPONSE_SCHEMA_CHANGED, exchange.durationMillis(), "document.id or batch missing");
|
||||
}
|
||||
Map<String, Object> metadata = metadata(Operation.UPLOAD_DOCUMENTS);
|
||||
metadata.put("externalId", documentId);
|
||||
// batch 透出到 summary:上传链据此落库 runtime_batch_id(轮询主键),externalId 仍保持 document id 以兼容既有绑定持久化。
|
||||
metadata.put("batch", batch);
|
||||
return succeeded(documentId, Operation.UPLOAD_DOCUMENTS, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), null, exchange.durationMillis(), metadata, null, List.of());
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeResult startParseDocuments(StartParseDocumentsCommand command) {
|
||||
// Dify 上传即触发索引,无独立解析触发端点:降级 no-op,不外呼;仅当运行时确实配置就绪时返回 ACCEPTED。
|
||||
FailureClass configFailure = validateConfiguration();
|
||||
if (configFailure != null) {
|
||||
return failure(Operation.START_PARSE_DOCUMENTS, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), command.processingTaskId(), configFailure, 0L, configFailure.name());
|
||||
}
|
||||
Map<String, Object> summary = metadata(Operation.START_PARSE_DOCUMENTS);
|
||||
summary.put("status", "accepted");
|
||||
// no-op 语义显式记账:Dify 无独立 parse 触发端点,索引在 upload 阶段已启动。
|
||||
summary.put("mode", "dify_auto_index_on_upload");
|
||||
summary.put("processingTaskId", command.processingTaskId());
|
||||
return new RuntimeResult(null, Operation.START_PARSE_DOCUMENTS, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), 0L, Status.ACCEPTED, null, command.processingTaskId(),
|
||||
sanitize(summary), List.of(), summary);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeResult pollDocumentStatuses(PollDocumentStatusesCommand command) {
|
||||
// 轮询主键=batch(命名债:command 沿用 ragflowDocumentIds 字段名承载 Dify batch,列注释在 S6c 订正)。
|
||||
List<String> pollKeys = command.ragflowDocumentIds();
|
||||
if (pollKeys == null || pollKeys.isEmpty()) {
|
||||
return failure(Operation.POLL_DOCUMENT_STATUSES, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), command.processingTaskId(), FailureClass.VALIDATION_ERROR, 0L,
|
||||
"batch is required for dify indexing-status polling");
|
||||
}
|
||||
if (pollKeys.size() > 1) {
|
||||
// Dify indexing-status 按单个 batch 查询;多 batch 批量轮询应在调用方拆分。
|
||||
return failure(Operation.POLL_DOCUMENT_STATUSES, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), command.processingTaskId(), FailureClass.VALIDATION_ERROR, 0L,
|
||||
"only support single batch status polling");
|
||||
}
|
||||
String batch = pollKeys.getFirst();
|
||||
Exchange exchange = exchange(Operation.POLL_DOCUMENT_STATUSES, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), "GET",
|
||||
"/v1/datasets/" + command.ragflowDatasetId() + "/documents/" + encode(batch) + "/indexing-status",
|
||||
new byte[0], null, command.processingTaskId());
|
||||
if (exchange instanceof Exchange.Failure failure) {
|
||||
return failure.result();
|
||||
}
|
||||
JsonNode responseBody = ((Exchange.Success) exchange).body();
|
||||
List<DocumentStatus> statuses = normalizeIndexingStatuses(responseBody);
|
||||
Map<String, Object> summary = metadata(Operation.POLL_DOCUMENT_STATUSES);
|
||||
summary.put("documents", statuses);
|
||||
return new RuntimeResult(null, Operation.POLL_DOCUMENT_STATUSES, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), exchange.durationMillis(), Status.SUCCEEDED, null, command.processingTaskId(),
|
||||
sanitize(summary), statuses, summary);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeResult retrieveChunks(RetrieveChunksCommand command) {
|
||||
// Dify retrieve 是单库端点,本 adapter 只检索单个 dataset;多授权 dataset 并行 + 按 score 合并归 RetrievalApiImpl。
|
||||
String datasetId = firstNonBlank(command.ragflowDatasetIds());
|
||||
if (datasetId == null) {
|
||||
return failure(Operation.RETRIEVE_CHUNKS, command.correlationId(), command.requestHash(), command.attempt(),
|
||||
null, FailureClass.VALIDATION_ERROR, 0L, "dataset id is required for dify retrieve");
|
||||
}
|
||||
Map<String, Object> retrievalModel = new LinkedHashMap<>();
|
||||
retrievalModel.put("search_method", "semantic_search");
|
||||
retrievalModel.put("reranking_enable", false);
|
||||
if (command.topK() != null && command.topK() > 0) {
|
||||
retrievalModel.put("top_k", command.topK());
|
||||
}
|
||||
// 忠实承接调用方阈值:threshold 非空即启用 Dify score_threshold(与 RAGFlow similarity_threshold 语义一致),
|
||||
// 不静默丢弃调用方意图;契约 IT 固定 false 仅为钉响应形态,不约束阈值策略。
|
||||
boolean thresholdEnabled = command.threshold() != null;
|
||||
retrievalModel.put("score_threshold_enabled", thresholdEnabled);
|
||||
if (thresholdEnabled) {
|
||||
retrievalModel.put("score_threshold", command.threshold());
|
||||
}
|
||||
Map<String, Object> body = new LinkedHashMap<>();
|
||||
body.put("query", command.question());
|
||||
body.put("retrieval_model", retrievalModel);
|
||||
Exchange exchange = exchange(Operation.RETRIEVE_CHUNKS, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), "POST", "/v1/datasets/" + datasetId + "/retrieve",
|
||||
jsonBody(body), "application/json", null);
|
||||
if (exchange instanceof Exchange.Failure failure) {
|
||||
return failure.result();
|
||||
}
|
||||
JsonNode responseBody = ((Exchange.Success) exchange).body();
|
||||
// 把 Dify records[].segment.content/score 归一成下游已消费的 data.chunks[](content/similarity)结构;
|
||||
// Dify 单库响应不含 dataset_id,归属由 RetrievalApiImpl 按本次检索的 dataset 补齐。
|
||||
ObjectNode normalized = normalizeRetrieveResponse(responseBody);
|
||||
int chunkCount = normalized.path("data").path("chunks").size();
|
||||
Map<String, Object> metadata = metadata(Operation.RETRIEVE_CHUNKS);
|
||||
metadata.put("chunksCount", chunkCount);
|
||||
Map<String, Object> summary = new LinkedHashMap<>(metadata);
|
||||
// responseBody 仅进程内承载归一后的检索结果供 RetrievalApiImpl 解析;redactedSummary 只落元数据,不落正文。
|
||||
summary.put("responseBody", normalized);
|
||||
return new RuntimeResult(null, Operation.RETRIEVE_CHUNKS, command.correlationId(), command.requestHash(),
|
||||
command.attempt(), exchange.durationMillis(), Status.SUCCEEDED, null, null,
|
||||
sanitize(metadata), List.of(), summary);
|
||||
}
|
||||
|
||||
// ==================== HTTP 交换(可测试 send 缝) ====================
|
||||
|
||||
/**
|
||||
* 底层 HTTP 交换:配置校验 → 重试循环 → 发送 → HTTP 状态分类 → JSON 解析。
|
||||
* 任何失败返回 {@link Exchange.Failure}(已是 fail-closed 的 RuntimeResult);成功返回解析后的响应树。
|
||||
*/
|
||||
private Exchange exchange(Operation operation, String correlationId, String requestHash, int attempt,
|
||||
String method, String path, byte[] requestBody, String contentType,
|
||||
String processingTaskId) {
|
||||
Instant startedAt = Instant.now();
|
||||
FailureClass configFailure = validateConfiguration();
|
||||
if (configFailure != null) {
|
||||
return new Exchange.Failure(failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
configFailure, elapsedMillis(startedAt), configFailure.name()), elapsedMillis(startedAt));
|
||||
}
|
||||
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 new Exchange.Failure(lastFailure, duration);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
JsonNode body = parseBody(response.body());
|
||||
if (body == null) {
|
||||
return new Exchange.Failure(failure(operation, correlationId, requestHash, attempt,
|
||||
processingTaskId, FailureClass.RESPONSE_SCHEMA_CHANGED, duration,
|
||||
"response is not valid json"), duration);
|
||||
}
|
||||
return new Exchange.Success(body, duration);
|
||||
} catch (java.net.http.HttpTimeoutException ex) {
|
||||
lastFailure = failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
FailureClass.TIMEOUT, elapsedMillis(startedAt), ex.getMessage());
|
||||
} catch (IOException ex) {
|
||||
// Dify 不可达按运行时不可用处理(沿用既有 RAGFLOW_UNAVAILABLE 语义,命名债 S6d 订正)。
|
||||
lastFailure = failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
FailureClass.RAGFLOW_UNAVAILABLE, elapsedMillis(startedAt), ex.getMessage());
|
||||
} catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
return new Exchange.Failure(failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
FailureClass.RAGFLOW_UNAVAILABLE, elapsedMillis(startedAt), ex.getMessage()),
|
||||
elapsedMillis(startedAt));
|
||||
} catch (RuntimeException ex) {
|
||||
return new Exchange.Failure(failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
FailureClass.UNKNOWN_FAILURE, elapsedMillis(startedAt), ex.getMessage()),
|
||||
elapsedMillis(startedAt));
|
||||
}
|
||||
if (currentAttempt == maxAttempts) {
|
||||
return new Exchange.Failure(lastFailure, elapsedMillis(startedAt));
|
||||
}
|
||||
}
|
||||
return new Exchange.Failure(lastFailure == null
|
||||
? failure(operation, correlationId, requestHash, attempt, processingTaskId,
|
||||
FailureClass.UNKNOWN_FAILURE, elapsedMillis(startedAt), "unknown failure")
|
||||
: lastFailure, elapsedMillis(startedAt));
|
||||
}
|
||||
|
||||
/** HTTP 发送缝:单测子类覆盖此方法以桩化/捕获请求,不真发网络。 */
|
||||
protected HttpResponsePayload send(HttpRequestPayload request) throws IOException, InterruptedException {
|
||||
HttpRequest.Builder builder = HttpRequest.newBuilder()
|
||||
.uri(endpoint(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());
|
||||
}
|
||||
|
||||
// ==================== 响应归一 ====================
|
||||
|
||||
/** Dify indexing-status 响应 data 可能是数组或对象,两态都归一为 DocumentStatus 列表。 */
|
||||
private List<DocumentStatus> normalizeIndexingStatuses(JsonNode responseBody) {
|
||||
List<DocumentStatus> statuses = new ArrayList<>();
|
||||
JsonNode data = responseBody.path("data");
|
||||
if (data.isArray()) {
|
||||
for (JsonNode entry : data) {
|
||||
statuses.add(toDocumentStatus(entry));
|
||||
}
|
||||
} else if (data.isObject()) {
|
||||
statuses.add(toDocumentStatus(data));
|
||||
} else {
|
||||
// 兜底:部分形态直接把状态放在顶层。
|
||||
statuses.add(toDocumentStatus(responseBody));
|
||||
}
|
||||
return statuses;
|
||||
}
|
||||
|
||||
private DocumentStatus toDocumentStatus(JsonNode entry) {
|
||||
String indexingStatus = entry.path("indexing_status").asText(entry.path("status").asText(""));
|
||||
String documentId = text(entry, "id", "document_id");
|
||||
Integer progress = computeProgress(entry);
|
||||
return new DocumentStatus(documentId, normalizeMuseStatus(indexingStatus), indexingStatus, progress,
|
||||
text(entry, "error"));
|
||||
}
|
||||
|
||||
/** Dify 提供 completed_segments / total_segments,可折算百分比进度供轮询痕迹;缺失则不填。 */
|
||||
private Integer computeProgress(JsonNode entry) {
|
||||
JsonNode completed = entry.path("completed_segments");
|
||||
JsonNode total = entry.path("total_segments");
|
||||
if (completed.isNumber() && total.isNumber() && total.asInt() > 0) {
|
||||
return Math.min(100, (int) Math.round(completed.asDouble() * 100.0 / total.asDouble()));
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/** Dify 索引状态枚举 → Muse 处理状态。paused 归 processing(非终态、非失败,保持轮询)。 */
|
||||
private String normalizeMuseStatus(String indexingStatus) {
|
||||
String normalized = indexingStatus == null ? "" : indexingStatus.toLowerCase(Locale.ROOT);
|
||||
return switch (normalized) {
|
||||
case "completed" -> "searchable";
|
||||
case "error" -> "failed";
|
||||
case "waiting", "parsing", "cleaning", "splitting", "indexing", "paused" -> "processing";
|
||||
default -> "pending_process";
|
||||
};
|
||||
}
|
||||
|
||||
/** Dify retrieve 的 records[].segment.content/score 归一为下游 data.chunks[](content/similarity/document_id)。 */
|
||||
private ObjectNode normalizeRetrieveResponse(JsonNode responseBody) {
|
||||
ObjectNode root = JsonUtils.getObjectMapper().createObjectNode();
|
||||
ObjectNode data = root.putObject("data");
|
||||
ArrayNode chunks = data.putArray("chunks");
|
||||
JsonNode records = responseBody.path("records");
|
||||
if (records.isArray()) {
|
||||
for (JsonNode record : records) {
|
||||
JsonNode segment = record.path("segment");
|
||||
String content = text(segment, "content");
|
||||
if (!StringUtils.hasText(content)) {
|
||||
continue;
|
||||
}
|
||||
ObjectNode chunk = chunks.addObject();
|
||||
chunk.put("content", content);
|
||||
if (record.path("score").isNumber()) {
|
||||
chunk.put("similarity", record.path("score").asDouble());
|
||||
}
|
||||
String documentId = text(segment, "document_id", "doc_id");
|
||||
if (StringUtils.hasText(documentId)) {
|
||||
chunk.put("document_id", documentId);
|
||||
}
|
||||
}
|
||||
}
|
||||
return root;
|
||||
}
|
||||
|
||||
// ==================== 结果构造与错误映射 ====================
|
||||
|
||||
private RuntimeResult succeeded(String externalId, Operation operation, String correlationId, String requestHash,
|
||||
int attempt, String processingTaskId, long durationMillis,
|
||||
Map<String, Object> metadata, JsonNode responseBody,
|
||||
List<DocumentStatus> documentStatuses) {
|
||||
Map<String, Object> summary = new LinkedHashMap<>(metadata);
|
||||
if (responseBody != null) {
|
||||
summary.put("responseBody", responseBody);
|
||||
}
|
||||
return new RuntimeResult(externalId, operation, correlationId, requestHash, attempt, durationMillis,
|
||||
Status.SUCCEEDED, null, processingTaskId, sanitize(metadata), documentStatuses, 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()) {
|
||||
// provider 原始 body 一律脱敏成指纹,绝不落原文,避免泄漏 key / 正文。
|
||||
summary.put("providerMessageSummary", safeSummaryMap(message));
|
||||
}
|
||||
return new RuntimeResult(null, operation, correlationId, requestHash, attempt, durationMillis,
|
||||
Status.FAILED, failureClass, processingTaskId, sanitize(summary), List.of(), summary);
|
||||
}
|
||||
|
||||
private FailureClass validateConfiguration() {
|
||||
if (!StringUtils.hasText(baseUrl)) {
|
||||
return FailureClass.CONFIG_MISSING;
|
||||
}
|
||||
if (!StringUtils.hasText(apiKey)) {
|
||||
return FailureClass.AUTH_MISSING;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/** Dify Datasets API 用 HTTP 状态码承载错误;映射到既有 FailureClass 语义。 */
|
||||
private FailureClass classifyHttpFailure(int statusCode) {
|
||||
if (statusCode >= 200 && statusCode < 300) {
|
||||
return null;
|
||||
}
|
||||
return switch (statusCode) {
|
||||
case 400, 422 -> 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;
|
||||
}
|
||||
|
||||
// ==================== multipart / 工具 ====================
|
||||
|
||||
/** 构造 Dify create-by-file 的 multipart:data 段(JSON 索引参数)+ file 段(文件字节)。单文件、须已材料化。 */
|
||||
private MultipartPayload buildSingleFileMultipart(List<DocumentUpload> documents) {
|
||||
if (documents == null || documents.size() != 1) {
|
||||
return null;
|
||||
}
|
||||
DocumentUpload document = documents.getFirst();
|
||||
if (document == null || document.content() == null || document.content().length == 0
|
||||
|| !StringUtils.hasText(document.fileName())) {
|
||||
return null;
|
||||
}
|
||||
Map<String, Object> data = new LinkedHashMap<>();
|
||||
data.put("indexing_technique", "high_quality");
|
||||
data.put("process_rule", Map.of("mode", "automatic"));
|
||||
String dataJson = JsonUtils.toJsonString(data);
|
||||
String contentType = StringUtils.hasText(document.contentType()) ? document.contentType() : "application/octet-stream";
|
||||
String boundary = "----MuseDify" + UUID.randomUUID().toString().replace("-", "");
|
||||
try {
|
||||
ByteArrayOutputStream body = new ByteArrayOutputStream();
|
||||
writeAscii(body, "--" + boundary + "\r\n");
|
||||
writeAscii(body, "Content-Disposition: form-data; name=\"data\"\r\n");
|
||||
writeAscii(body, "Content-Type: application/json; charset=UTF-8\r\n\r\n");
|
||||
body.write(dataJson.getBytes(StandardCharsets.UTF_8));
|
||||
writeAscii(body, "\r\n");
|
||||
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");
|
||||
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> metadata(Operation operation) {
|
||||
Map<String, Object> metadata = new LinkedHashMap<>();
|
||||
metadata.put("operation", operation.wireName());
|
||||
metadata.put("status", Status.SUCCEEDED.name().toLowerCase(Locale.ROOT));
|
||||
return metadata;
|
||||
}
|
||||
|
||||
private Map<String, Object> safeSummaryMap(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 summary;
|
||||
}
|
||||
|
||||
private String sanitize(Object value) {
|
||||
return RagFlowKnowledgeRedactor.sanitizeObject(value);
|
||||
}
|
||||
|
||||
private byte[] jsonBody(Object body) {
|
||||
return JsonUtils.toJsonString(body == null ? Map.of() : body).getBytes(StandardCharsets.UTF_8);
|
||||
}
|
||||
|
||||
private JsonNode parseBody(String responseBody) {
|
||||
try {
|
||||
return JsonUtils.parseTree(responseBody == null || responseBody.isBlank() ? "{}" : responseBody);
|
||||
} catch (RuntimeException ex) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private String text(JsonNode node, String... fieldNames) {
|
||||
if (node == null || node.isMissingNode() || node.isNull()) {
|
||||
return null;
|
||||
}
|
||||
for (String fieldName : fieldNames) {
|
||||
JsonNode value = node.path(fieldName);
|
||||
if (value.isTextual() && StringUtils.hasText(value.asText())) {
|
||||
return value.asText();
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private String firstNonBlank(List<String> values) {
|
||||
if (values == null) {
|
||||
return null;
|
||||
}
|
||||
for (String value : values) {
|
||||
if (StringUtils.hasText(value)) {
|
||||
return value;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/** base-url 形如 http://host:port/v1;路径按 /v1 归一去重(参 P1rDifyDatasetsContractLiveIT.endpoint)。 */
|
||||
private URI endpoint(String path) {
|
||||
String normalizedPath = path.startsWith("/") ? path : "/" + path;
|
||||
if (baseUrl != null && baseUrl.endsWith("/v1") && normalizedPath.startsWith("/v1/")) {
|
||||
normalizedPath = normalizedPath.substring("/v1".length());
|
||||
}
|
||||
return URI.create(baseUrl + normalizedPath);
|
||||
}
|
||||
|
||||
private long elapsedMillis(Instant startedAt) {
|
||||
return Math.max(0L, Duration.between(startedAt, Instant.now()).toMillis());
|
||||
}
|
||||
|
||||
private String escapeMultipart(String value) {
|
||||
return value == null ? "" : value.replace("\\", "\\\\").replace("\"", "\\\"").replace("\r", "").replace("\n", "");
|
||||
}
|
||||
|
||||
private String encode(String value) {
|
||||
// path segment 场景保守编码:batch 由 Dify 生成(通常为 uuid),仍防御非法字符。
|
||||
return java.net.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;
|
||||
}
|
||||
String trimmed = rawBaseUrl.trim();
|
||||
return trimmed.endsWith("/") ? trimmed.substring(0, trimmed.length() - 1) : trimmed;
|
||||
}
|
||||
|
||||
/** HTTP 交换结果:失败(已封 fail-closed RuntimeResult)或成功(解析后的响应树)。 */
|
||||
private sealed interface Exchange {
|
||||
|
||||
long durationMillis();
|
||||
|
||||
record Failure(RuntimeResult result, long durationMillis) implements Exchange {
|
||||
}
|
||||
|
||||
record Success(JsonNode body, long durationMillis) implements Exchange {
|
||||
}
|
||||
}
|
||||
|
||||
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) {
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,56 @@
|
||||
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;
|
||||
|
||||
/**
|
||||
* Dify Datasets runtime client 的 Spring 装配配置。
|
||||
*
|
||||
* <p>与 {@link RagFlowKnowledgeRuntimeClientConfiguration} 共存(RAGFlow 装配 S6d 才摘除)。为确保单人只装一个
|
||||
* {@link KnowledgeRuntimeClient},两个装配类按 provider 选择器 {@code muse.knowledge.runtime-provider} 互斥:
|
||||
* 本类仅在 {@code =dify} 时参与,RAGFlow 装配在 {@code =ragflow} 或缺省(matchIfMissing)时参与。这样
|
||||
* {@code @ConditionalOnMissingBean} 的求值顺序不再影响结果,避免组件扫描顺序带来的非确定性冲突。</p>
|
||||
*
|
||||
* <p>装配范式对齐 RAGFlow(无 {@code .enable} 开关):读 {@code muse.knowledge.dify.*};缺 base-url 或
|
||||
* dataset-api-key 一律装 {@link UnavailableKnowledgeRuntimeClient}(fail-closed,即"选了 Dify 但配错=拒绝",
|
||||
* 绝不静默回退 RAGFlow)。</p>
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
public class DifyKnowledgeRuntimeClientConfiguration {
|
||||
|
||||
private static final String PROPERTY_PREFIX = "muse.knowledge.dify.";
|
||||
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 = "dify")
|
||||
public KnowledgeRuntimeClient difyKnowledgeRuntimeClient(Environment environment) {
|
||||
String baseUrl = environment.getProperty(PROPERTY_PREFIX + "base-url");
|
||||
String apiKey = environment.getProperty(PROPERTY_PREFIX + "dataset-api-key");
|
||||
if (!StringUtils.hasText(baseUrl) || !StringUtils.hasText(apiKey)) {
|
||||
return new UnavailableKnowledgeRuntimeClient();
|
||||
}
|
||||
|
||||
int timeoutSeconds = positiveInt(environment, "timeout-seconds", DEFAULT_TIMEOUT_SECONDS);
|
||||
int retryBudget = nonNegativeInt(environment, "retry-budget", DEFAULT_RETRY_BUDGET);
|
||||
return new DifyKnowledgeRuntimeClient(baseUrl, apiKey, Duration.ofSeconds(timeoutSeconds), retryBudget);
|
||||
}
|
||||
|
||||
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,6 +1,7 @@
|
||||
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;
|
||||
@ -10,6 +11,10 @@ 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 {
|
||||
@ -20,6 +25,7 @@ public class RagFlowKnowledgeRuntimeClientConfiguration {
|
||||
|
||||
@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");
|
||||
|
||||
@ -0,0 +1,49 @@
|
||||
package cn.iocoder.muse.module.knowledge.application.muse.facade;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Knowledge 运行时未配置时的通用 fail-closed client。
|
||||
*
|
||||
* <p>与 provider 无关:base-url / api-key 缺失时装配此桩,所有操作直接返回 {@link FailureClass#CONFIG_MISSING},
|
||||
* 绝不发起任何外部调用,保证"未配置=拒绝"而非静默降级。S6d 摘除 RAGFlow 后,由本桩统一承担 Unavailable 语义。</p>
|
||||
*/
|
||||
public class UnavailableKnowledgeRuntimeClient 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()));
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,123 @@
|
||||
package cn.iocoder.muse.module.knowledge.application.muse.facade;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
||||
import org.springframework.mock.env.MockEnvironment;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
|
||||
|
||||
/**
|
||||
* Dify Knowledge 运行时 client 装配测试。
|
||||
*
|
||||
* <p>两层验证:①装配方法本身(配 dify.* 全→Dify、缺→Unavailable);②与 RAGFlow 装配共存时按
|
||||
* {@code muse.knowledge.runtime-provider} 选择器确定性互斥——任何组合下只装一个 {@link KnowledgeRuntimeClient},
|
||||
* 不依赖组件扫描/Bean 求值顺序。</p>
|
||||
*/
|
||||
class DifyKnowledgeRuntimeClientConfigurationTest {
|
||||
|
||||
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
|
||||
.withUserConfiguration(DifyKnowledgeRuntimeClientConfiguration.class,
|
||||
RagFlowKnowledgeRuntimeClientConfiguration.class);
|
||||
|
||||
// ==================== 装配方法本身 ====================
|
||||
|
||||
@Test
|
||||
void should_registerDifyClientWhenBaseUrlAndKeyConfigured() {
|
||||
MockEnvironment environment = new MockEnvironment()
|
||||
.withProperty("muse.knowledge.dify.base-url", "http://dify.example/v1")
|
||||
.withProperty("muse.knowledge.dify.dataset-api-key", "dataset-key")
|
||||
.withProperty("muse.knowledge.dify.timeout-seconds", "7")
|
||||
.withProperty("muse.knowledge.dify.retry-budget", "2");
|
||||
DifyKnowledgeRuntimeClientConfiguration configuration = new DifyKnowledgeRuntimeClientConfiguration();
|
||||
|
||||
KnowledgeRuntimeClient client = configuration.difyKnowledgeRuntimeClient(environment);
|
||||
|
||||
assertInstanceOf(DifyKnowledgeRuntimeClient.class, client);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_registerUnavailableWhenBaseUrlMissing() {
|
||||
MockEnvironment environment = new MockEnvironment()
|
||||
.withProperty("muse.knowledge.dify.dataset-api-key", "dataset-key");
|
||||
DifyKnowledgeRuntimeClientConfiguration configuration = new DifyKnowledgeRuntimeClientConfiguration();
|
||||
|
||||
KnowledgeRuntimeClient client = configuration.difyKnowledgeRuntimeClient(environment);
|
||||
|
||||
assertInstanceOf(UnavailableKnowledgeRuntimeClient.class, client);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_registerUnavailableWhenKeyMissing() {
|
||||
MockEnvironment environment = new MockEnvironment()
|
||||
.withProperty("muse.knowledge.dify.base-url", "http://dify.example/v1");
|
||||
DifyKnowledgeRuntimeClientConfiguration configuration = new DifyKnowledgeRuntimeClientConfiguration();
|
||||
|
||||
KnowledgeRuntimeClient client = configuration.difyKnowledgeRuntimeClient(environment);
|
||||
|
||||
assertInstanceOf(UnavailableKnowledgeRuntimeClient.class, client);
|
||||
}
|
||||
|
||||
// ==================== 与 RAGFlow 装配共存:确定性互斥 ====================
|
||||
|
||||
@Test
|
||||
void should_installDifyOnlyWhenProviderIsDify() {
|
||||
contextRunner.withPropertyValues(
|
||||
"muse.knowledge.runtime-provider=dify",
|
||||
"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(DifyKnowledgeRuntimeClient.class,
|
||||
context.getBean(KnowledgeRuntimeClient.class));
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_failClosedWhenProviderIsDifyButConfigMissing() {
|
||||
contextRunner.withPropertyValues("muse.knowledge.runtime-provider=dify")
|
||||
.run(context -> {
|
||||
assertEquals(1, context.getBeansOfType(KnowledgeRuntimeClient.class).size());
|
||||
// 选了 Dify 但没配 → fail-closed 桩,不静默回退 RAGFlow。
|
||||
assertInstanceOf(UnavailableKnowledgeRuntimeClient.class,
|
||||
context.getBean(KnowledgeRuntimeClient.class));
|
||||
});
|
||||
}
|
||||
|
||||
@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")
|
||||
.run(context -> {
|
||||
assertEquals(1, context.getBeansOfType(KnowledgeRuntimeClient.class).size());
|
||||
assertInstanceOf(HttpRagFlowKnowledgeRuntimeClient.class,
|
||||
context.getBean(KnowledgeRuntimeClient.class));
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_defaultToRagFlowWhenProviderUnset() {
|
||||
// 缺省 provider(既有部署零变更):matchIfMissing 让 RAGFlow 装配继续生效,Dify 让位。
|
||||
contextRunner.withPropertyValues(
|
||||
"muse.knowledge.ragflow.base-url=https://ragflow.example",
|
||||
"muse.knowledge.ragflow.api-key=ragflow-key")
|
||||
.run(context -> {
|
||||
assertEquals(1, context.getBeansOfType(KnowledgeRuntimeClient.class).size());
|
||||
assertInstanceOf(HttpRagFlowKnowledgeRuntimeClient.class,
|
||||
context.getBean(KnowledgeRuntimeClient.class));
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_installSingleUnavailableWhenProviderUnsetAndNothingConfigured() {
|
||||
contextRunner.run(context -> {
|
||||
// 仍确保只有一个 bean(RAGFlow 的 Unavailable 桩),不因两装配类共存而重复注册。
|
||||
assertEquals(1, context.getBeansOfType(KnowledgeRuntimeClient.class).size());
|
||||
assertInstanceOf(UnavailableRagFlowKnowledgeRuntimeClient.class,
|
||||
context.getBean(KnowledgeRuntimeClient.class));
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,447 @@
|
||||
package cn.iocoder.muse.module.knowledge.application.muse.facade;
|
||||
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.http.HttpHeaders;
|
||||
import java.net.http.HttpTimeoutException;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.time.Duration;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
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.assertInstanceOf;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* Dify Knowledge 运行时 adapter 单测。
|
||||
*
|
||||
* <p>mock HTTP 层(覆盖 {@code send} 缝),断言 5 操作的请求形态(方法/路径/body/multipart)与响应解析
|
||||
* (顶层 id、嵌套 document.id + 顶层 batch、data 数组/对象两态、records[].segment.content/score)与 S1 契约一致,
|
||||
* 以及 fail-closed(缺配置 / 401 / 超时)错误映射。</p>
|
||||
*/
|
||||
class DifyKnowledgeRuntimeClientTest {
|
||||
|
||||
private static final String BASE_URL = "http://dify.example/v1";
|
||||
private static final String API_KEY = "dataset-configured-key";
|
||||
|
||||
// ==================== createDataset ====================
|
||||
|
||||
@Test
|
||||
void createDatasetShouldPostDatasetsWithContractBodyAndReadTopLevelId() {
|
||||
CapturingDifyClient client = new CapturingDifyClient(BASE_URL, API_KEY,
|
||||
"{\"id\":\"dataset-1\",\"name\":\"KB\",\"indexing_technique\":\"high_quality\"}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, "KB",
|
||||
Map.of(), "corr-create", "hash-create", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status());
|
||||
assertEquals("dataset-1", result.externalId());
|
||||
DifyKnowledgeRuntimeClient.HttpRequestPayload request = client.requests.getFirst();
|
||||
assertEquals("POST", request.method());
|
||||
assertEquals("/v1/datasets", request.path());
|
||||
String body = new String(request.body(), StandardCharsets.UTF_8);
|
||||
assertTrue(body.contains("\"name\":\"KB\""));
|
||||
assertTrue(body.contains("\"permission\":\"only_me\""));
|
||||
assertTrue(body.contains("\"indexing_technique\":\"high_quality\""));
|
||||
}
|
||||
|
||||
@Test
|
||||
void createDatasetShouldFailClosedWhenTopLevelIdMissing() {
|
||||
StubDifyClient client = new StubDifyClient(BASE_URL, API_KEY, 200, "{\"name\":\"KB\"}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, "KB",
|
||||
Map.of(), "corr", "hash", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.RESPONSE_SCHEMA_CHANGED, result.failureClass());
|
||||
}
|
||||
|
||||
// ==================== uploadDocuments ====================
|
||||
|
||||
@Test
|
||||
void uploadDocumentsShouldPostMultipartAndReadNestedDocumentIdWithTopLevelBatch() {
|
||||
CapturingDifyClient client = new CapturingDifyClient(BASE_URL, API_KEY,
|
||||
"{\"document\":{\"id\":\"doc-1\",\"name\":\"novel.txt\"},\"batch\":\"batch-1\"}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.uploadDocuments(
|
||||
new KnowledgeRuntimeClient.UploadDocumentsCommand(100L, 2001L, 4001L, "dataset-1",
|
||||
List.of(new KnowledgeRuntimeClient.DocumentUpload(5001L, 6001L, "file-ref", "novel.txt",
|
||||
"text/plain", 5L, "hello".getBytes(StandardCharsets.UTF_8))),
|
||||
"corr-upload", "hash-upload", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status());
|
||||
// 嵌套 document.id 作为 externalId(兼容既有绑定持久化);顶层 batch 透出 summary 供上传链落库轮询主键。
|
||||
assertEquals("doc-1", result.externalId());
|
||||
assertEquals("batch-1", result.summary().get("batch"));
|
||||
DifyKnowledgeRuntimeClient.HttpRequestPayload request = client.requests.getFirst();
|
||||
assertEquals("POST", request.method());
|
||||
assertEquals("/v1/datasets/dataset-1/document/create-by-file", request.path());
|
||||
assertTrue(request.contentType().startsWith("multipart/form-data; boundary="));
|
||||
String body = new String(request.body(), StandardCharsets.ISO_8859_1);
|
||||
assertTrue(body.contains("name=\"data\""));
|
||||
assertTrue(body.contains("\"indexing_technique\":\"high_quality\""));
|
||||
assertTrue(body.contains("\"process_rule\":{\"mode\":\"automatic\"}"));
|
||||
assertTrue(body.contains("name=\"file\"; filename=\"novel.txt\""));
|
||||
assertTrue(body.contains("hello"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void uploadDocumentsShouldRejectMissingMaterializedBytesBeforeCallingDify() {
|
||||
CapturingDifyClient client = new CapturingDifyClient(BASE_URL, API_KEY, "{}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.uploadDocuments(
|
||||
new KnowledgeRuntimeClient.UploadDocumentsCommand(100L, 2001L, 4001L, "dataset-1",
|
||||
List.of(new KnowledgeRuntimeClient.DocumentUpload(5001L, 6001L, "file-ref", "novel.txt",
|
||||
"text/plain", 10L, null)),
|
||||
"corr", "hash", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.FILE_REJECTED, result.failureClass());
|
||||
assertTrue(client.requests.isEmpty());
|
||||
}
|
||||
|
||||
@Test
|
||||
void uploadDocumentsShouldFailClosedWhenBatchMissing() {
|
||||
StubDifyClient client = new StubDifyClient(BASE_URL, API_KEY, 200,
|
||||
"{\"document\":{\"id\":\"doc-1\"}}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.uploadDocuments(
|
||||
new KnowledgeRuntimeClient.UploadDocumentsCommand(100L, 2001L, 4001L, "dataset-1",
|
||||
List.of(new KnowledgeRuntimeClient.DocumentUpload(5001L, 6001L, "file-ref", "novel.txt",
|
||||
"text/plain", 5L, "hello".getBytes(StandardCharsets.UTF_8))),
|
||||
"corr", "hash", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.RESPONSE_SCHEMA_CHANGED, result.failureClass());
|
||||
}
|
||||
|
||||
@Test
|
||||
void uploadDocumentsShouldPreserveBinaryBytesInsideMultipart() {
|
||||
CapturingDifyClient client = new CapturingDifyClient(BASE_URL, API_KEY,
|
||||
"{\"document\":{\"id\":\"doc-1\"},\"batch\":\"batch-1\"}");
|
||||
byte[] binary = new byte[]{0x25, 0x50, 0x44, 0x46, 0x00, (byte) 0x80, (byte) 0xFF, 0x01};
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.uploadDocuments(
|
||||
new KnowledgeRuntimeClient.UploadDocumentsCommand(100L, 2001L, 4001L, "dataset-1",
|
||||
List.of(new KnowledgeRuntimeClient.DocumentUpload(5001L, 6001L, "file-ref", "binary.pdf",
|
||||
"application/pdf", (long) binary.length, binary)),
|
||||
"corr", "hash", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status());
|
||||
assertTrue(containsSubsequence(client.requests.getFirst().body(), binary));
|
||||
}
|
||||
|
||||
// ==================== startParseDocuments ====================
|
||||
|
||||
@Test
|
||||
void startParseShouldBeNoOpAcceptedWithoutCallingDify() {
|
||||
CapturingDifyClient client = new CapturingDifyClient(BASE_URL, API_KEY, "{}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.startParseDocuments(
|
||||
new KnowledgeRuntimeClient.StartParseDocumentsCommand(100L, 2001L, 4001L, "dataset-1",
|
||||
List.of("doc-1"), "muse-task-1", "corr-parse", "hash-parse", 1));
|
||||
|
||||
// Dify 上传即索引,无独立触发端点:降级 no-op,不外呼。
|
||||
assertEquals(KnowledgeRuntimeClient.Status.ACCEPTED, result.status());
|
||||
assertNull(result.externalId());
|
||||
assertEquals("muse-task-1", result.processingTaskId());
|
||||
assertTrue(client.requests.isEmpty());
|
||||
assertTrue(result.redactedSummary().contains("dify_auto_index_on_upload"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void startParseShouldFailClosedWhenConfigMissing() {
|
||||
CapturingDifyClient client = new CapturingDifyClient(BASE_URL, "", "{}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.startParseDocuments(
|
||||
new KnowledgeRuntimeClient.StartParseDocumentsCommand(100L, 2001L, 4001L, "dataset-1",
|
||||
List.of("doc-1"), "muse-task-1", "corr-parse", "hash-parse", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.AUTH_MISSING, result.failureClass());
|
||||
}
|
||||
|
||||
// ==================== pollDocumentStatuses ====================
|
||||
|
||||
@Test
|
||||
void pollShouldGetIndexingStatusByBatchAndNormalizeArrayCompleted() {
|
||||
CapturingDifyClient client = new CapturingDifyClient(BASE_URL, API_KEY, """
|
||||
{"data":[{"id":"doc-1","indexing_status":"completed","completed_segments":10,"total_segments":10}]}
|
||||
""");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.pollDocumentStatuses(
|
||||
new KnowledgeRuntimeClient.PollDocumentStatusesCommand(100L, 2001L, 4001L, "dataset-1",
|
||||
List.of("batch-1"), "muse-task-1", "corr-poll", "hash-poll", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status());
|
||||
assertEquals("GET", client.requests.getFirst().method());
|
||||
assertEquals("/v1/datasets/dataset-1/documents/batch-1/indexing-status", client.requests.getFirst().path());
|
||||
assertEquals(1, result.documentStatuses().size());
|
||||
KnowledgeRuntimeClient.DocumentStatus status = result.documentStatuses().getFirst();
|
||||
assertEquals("doc-1", status.ragflowDocumentId());
|
||||
assertEquals("searchable", status.museStatus());
|
||||
assertEquals("completed", status.ragflowRun());
|
||||
assertEquals(Integer.valueOf(100), status.progress());
|
||||
}
|
||||
|
||||
@Test
|
||||
void pollShouldNormalizeObjectDataStillIndexing() {
|
||||
StubDifyClient client = new StubDifyClient(BASE_URL, API_KEY, 200,
|
||||
"{\"data\":{\"id\":\"doc-1\",\"indexing_status\":\"indexing\",\"completed_segments\":1,\"total_segments\":4}}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.pollDocumentStatuses(
|
||||
new KnowledgeRuntimeClient.PollDocumentStatusesCommand(100L, 2001L, 4001L, "dataset-1",
|
||||
List.of("batch-1"), "muse-task-1", "corr-poll", "hash-poll", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status());
|
||||
assertEquals(1, result.documentStatuses().size());
|
||||
assertEquals("processing", result.documentStatuses().getFirst().museStatus());
|
||||
assertEquals(Integer.valueOf(25), result.documentStatuses().getFirst().progress());
|
||||
}
|
||||
|
||||
@Test
|
||||
void pollShouldMapErrorStatusToFailed() {
|
||||
StubDifyClient client = new StubDifyClient(BASE_URL, API_KEY, 200,
|
||||
"{\"data\":{\"id\":\"doc-1\",\"indexing_status\":\"error\",\"error\":\"boom\"}}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.pollDocumentStatuses(
|
||||
new KnowledgeRuntimeClient.PollDocumentStatusesCommand(100L, 2001L, 4001L, "dataset-1",
|
||||
List.of("batch-1"), "muse-task-1", "corr-poll", "hash-poll", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status());
|
||||
assertEquals("failed", result.documentStatuses().getFirst().museStatus());
|
||||
}
|
||||
|
||||
@Test
|
||||
void pollShouldRejectMultipleBatchesBeforeCallingDify() {
|
||||
CapturingDifyClient client = new CapturingDifyClient(BASE_URL, API_KEY, "{\"data\":{}}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.pollDocumentStatuses(
|
||||
new KnowledgeRuntimeClient.PollDocumentStatusesCommand(100L, 2001L, 4001L, "dataset-1",
|
||||
List.of("batch-1", "batch-2"), "muse-task-1", "corr-poll", "hash-poll", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.VALIDATION_ERROR, result.failureClass());
|
||||
assertTrue(client.requests.isEmpty());
|
||||
}
|
||||
|
||||
// ==================== retrieveChunks ====================
|
||||
|
||||
@Test
|
||||
void retrieveShouldPostSingleDatasetAndNormalizeRecordsToChunks() {
|
||||
CapturingDifyClient client = new CapturingDifyClient(BASE_URL, API_KEY, """
|
||||
{"records":[
|
||||
{"segment":{"content":"chunk alpha","document_id":"doc-1"},"score":0.91},
|
||||
{"segment":{"content":"chunk beta"},"score":0.72}
|
||||
]}
|
||||
""");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.retrieveChunks(
|
||||
new KnowledgeRuntimeClient.RetrieveChunksCommand(100L, 2001L, 4001L,
|
||||
List.of("dataset-1"), null, "who is alpha?", 3, 0.2, null, "corr-retrieve", "hash-retrieve", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status());
|
||||
DifyKnowledgeRuntimeClient.HttpRequestPayload request = client.requests.getFirst();
|
||||
assertEquals("POST", request.method());
|
||||
assertEquals("/v1/datasets/dataset-1/retrieve", request.path());
|
||||
String body = new String(request.body(), StandardCharsets.UTF_8);
|
||||
assertTrue(body.contains("\"query\":\"who is alpha?\""));
|
||||
assertTrue(body.contains("\"search_method\":\"semantic_search\""));
|
||||
assertTrue(body.contains("\"top_k\":3"));
|
||||
assertTrue(body.contains("\"score_threshold_enabled\":true"));
|
||||
// 归一:records[].segment.content → data.chunks[].content;records[].score → similarity。
|
||||
JsonNode responseBody = (JsonNode) result.summary().get("responseBody");
|
||||
JsonNode chunks = responseBody.path("data").path("chunks");
|
||||
assertEquals(2, chunks.size());
|
||||
assertEquals("chunk alpha", chunks.get(0).path("content").asText());
|
||||
assertEquals(0.91, chunks.get(0).path("similarity").asDouble(), 1e-9);
|
||||
assertEquals("doc-1", chunks.get(0).path("document_id").asText());
|
||||
assertEquals(2, result.summary().get("chunksCount"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void retrieveShouldOmitScoreThresholdWhenThresholdNull() {
|
||||
CapturingDifyClient client = new CapturingDifyClient(BASE_URL, API_KEY,
|
||||
"{\"records\":[{\"segment\":{\"content\":\"c\"},\"score\":0.5}]}");
|
||||
|
||||
client.retrieveChunks(new KnowledgeRuntimeClient.RetrieveChunksCommand(100L, 2001L, 4001L,
|
||||
List.of("dataset-1"), null, "q", 5, null, null, "corr", "hash", 1));
|
||||
|
||||
String body = new String(client.requests.getFirst().body(), StandardCharsets.UTF_8);
|
||||
assertTrue(body.contains("\"score_threshold_enabled\":false"));
|
||||
assertFalse(body.contains("score_threshold\""));
|
||||
}
|
||||
|
||||
@Test
|
||||
void retrieveShouldUseFirstDatasetOnlyAsSingleLibraryEndpoint() {
|
||||
CapturingDifyClient client = new CapturingDifyClient(BASE_URL, API_KEY,
|
||||
"{\"records\":[{\"segment\":{\"content\":\"c\"},\"score\":0.5}]}");
|
||||
|
||||
client.retrieveChunks(new KnowledgeRuntimeClient.RetrieveChunksCommand(100L, 2001L, 4001L,
|
||||
List.of("dataset-1", "dataset-2"), null, "q", 5, 0.2, null, "corr", "hash", 1));
|
||||
|
||||
// 单库端点:只检索首个 dataset;多授权 dataset 并行合并归 RetrievalApiImpl(S6c)。
|
||||
assertEquals("/v1/datasets/dataset-1/retrieve", client.requests.getFirst().path());
|
||||
}
|
||||
|
||||
@Test
|
||||
void retrieveShouldFailClosedWhenNoDatasetProvided() {
|
||||
CapturingDifyClient client = new CapturingDifyClient(BASE_URL, API_KEY, "{}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.retrieveChunks(
|
||||
new KnowledgeRuntimeClient.RetrieveChunksCommand(100L, 2001L, 4001L,
|
||||
List.of(), null, "q", 5, 0.2, null, "corr", "hash", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.VALIDATION_ERROR, result.failureClass());
|
||||
assertTrue(client.requests.isEmpty());
|
||||
}
|
||||
|
||||
@Test
|
||||
void retrieveRedactedSummaryShouldNotLeakChunkContent() {
|
||||
StubDifyClient client = new StubDifyClient(BASE_URL, API_KEY, 200,
|
||||
"{\"records\":[{\"segment\":{\"content\":\"secret chunk content\"},\"score\":0.9}]}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.retrieveChunks(
|
||||
new KnowledgeRuntimeClient.RetrieveChunksCommand(100L, 2001L, 4001L,
|
||||
List.of("dataset-1"), null, "private question", 3, 0.2, null, "corr", "hash", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.SUCCEEDED, result.status());
|
||||
assertFalse(result.redactedSummary().contains("secret chunk content"));
|
||||
assertFalse(result.redactedSummary().contains("private question"));
|
||||
}
|
||||
|
||||
// ==================== fail-closed 横切 ====================
|
||||
|
||||
@Test
|
||||
void shouldFailClosedWhenBaseUrlMissing() {
|
||||
StubDifyClient client = new StubDifyClient("", API_KEY, 200, "{\"id\":\"x\"}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, "KB",
|
||||
Map.of(), "corr", "hash", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.CONFIG_MISSING, result.failureClass());
|
||||
}
|
||||
|
||||
@Test
|
||||
void shouldMapUnauthorizedToAuthFailedAndRedactBody() {
|
||||
StubDifyClient client = new StubDifyClient(BASE_URL, API_KEY, 401,
|
||||
"{\"code\":\"unauthorized\",\"message\":\"invalid api key value abcdef\"}");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, "KB",
|
||||
Map.of(), "corr", "hash", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.AUTH_FAILED, result.failureClass());
|
||||
// provider 原始 body 一律脱敏成指纹,不落原文。
|
||||
assertFalse(result.redactedSummary().contains("invalid api key value abcdef"));
|
||||
assertTrue(result.redactedSummary().contains("\"redacted\":true"));
|
||||
assertTrue(result.redactedSummary().contains("\"sha256\""));
|
||||
}
|
||||
|
||||
@Test
|
||||
void shouldMapTimeoutToTimeoutFailure() {
|
||||
TimeoutDifyClient client = new TimeoutDifyClient(BASE_URL, API_KEY);
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, "KB",
|
||||
Map.of(), "corr", "hash", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.TIMEOUT, result.failureClass());
|
||||
}
|
||||
|
||||
@Test
|
||||
void shouldMapServerErrorToUnavailable() {
|
||||
StubDifyClient client = new StubDifyClient(BASE_URL, API_KEY, 503, "service unavailable");
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.createDataset(
|
||||
new KnowledgeRuntimeClient.CreateDatasetCommand(100L, 2001L, 4001L, "KB",
|
||||
Map.of(), "corr", "hash", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.RAGFLOW_UNAVAILABLE, result.failureClass());
|
||||
}
|
||||
|
||||
@Test
|
||||
void unavailableClientShouldFailClosed() {
|
||||
KnowledgeRuntimeClient client = new UnavailableKnowledgeRuntimeClient();
|
||||
|
||||
KnowledgeRuntimeClient.RuntimeResult result = client.retrieveChunks(
|
||||
new KnowledgeRuntimeClient.RetrieveChunksCommand(100L, 2001L, 4001L,
|
||||
List.of("dataset-1"), null, "q", 3, 0.2, null, "corr", "hash", 1));
|
||||
|
||||
assertEquals(KnowledgeRuntimeClient.Status.FAILED, result.status());
|
||||
assertEquals(KnowledgeRuntimeClient.FailureClass.CONFIG_MISSING, result.failureClass());
|
||||
}
|
||||
|
||||
// ==================== 测试桩 ====================
|
||||
|
||||
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 StubDifyClient extends DifyKnowledgeRuntimeClient {
|
||||
|
||||
private final int statusCode;
|
||||
private final String body;
|
||||
|
||||
private StubDifyClient(String baseUrl, String apiKey, int statusCode, String body) {
|
||||
super(baseUrl, apiKey, Duration.ofSeconds(1), 0);
|
||||
this.statusCode = statusCode;
|
||||
this.body = body;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected HttpResponsePayload send(HttpRequestPayload request) {
|
||||
return new HttpResponsePayload(statusCode, body, HttpHeaders.of(Map.of(), (name, value) -> true));
|
||||
}
|
||||
}
|
||||
|
||||
private static final class CapturingDifyClient extends DifyKnowledgeRuntimeClient {
|
||||
|
||||
private final String body;
|
||||
private final List<HttpRequestPayload> requests = new ArrayList<>();
|
||||
|
||||
private CapturingDifyClient(String baseUrl, String apiKey, String body) {
|
||||
super(baseUrl, apiKey, Duration.ofSeconds(1), 0);
|
||||
this.body = body;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected HttpResponsePayload send(HttpRequestPayload request) {
|
||||
requests.add(request);
|
||||
return new HttpResponsePayload(200, body, HttpHeaders.of(Map.of(), (name, value) -> true));
|
||||
}
|
||||
}
|
||||
|
||||
private static final class TimeoutDifyClient extends DifyKnowledgeRuntimeClient {
|
||||
|
||||
private TimeoutDifyClient(String baseUrl, String apiKey) {
|
||||
super(baseUrl, apiKey, Duration.ofSeconds(1), 0);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected HttpResponsePayload send(HttpRequestPayload request) throws IOException {
|
||||
throw new HttpTimeoutException("dify request timed out");
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user