diff --git a/agentscope-extensions/agentscope-extensions-rag/agentscope-extensions-rag-ragflow/src/main/java/io/agentscope/core/rag/integration/ragflow/RAGFlowClient.java b/agentscope-extensions/agentscope-extensions-rag/agentscope-extensions-rag-ragflow/src/main/java/io/agentscope/core/rag/integration/ragflow/RAGFlowClient.java index ccdac26d7e..84ea8796bc 100644 --- a/agentscope-extensions/agentscope-extensions-rag/agentscope-extensions-rag-ragflow/src/main/java/io/agentscope/core/rag/integration/ragflow/RAGFlowClient.java +++ b/agentscope-extensions/agentscope-extensions-rag/agentscope-extensions-rag-ragflow/src/main/java/io/agentscope/core/rag/integration/ragflow/RAGFlowClient.java @@ -20,7 +20,9 @@ import io.agentscope.core.rag.integration.ragflow.model.RAGFlowResponse; import io.agentscope.core.util.JsonUtils; import java.io.IOException; +import java.util.ArrayList; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; import okhttp3.Interceptor; @@ -102,6 +104,42 @@ public Mono retrieve( Double similarityThreshold, Map metadataCondition) { + return retrieve(question, topK, similarityThreshold, null, metadataCondition); + } + + /** + * Retrieve documents with request-specific dataset IDs and metadata conditions. + * + *

Non-empty request parameters take precedence over values from {@link RAGFlowConfig}. + * When a request parameter is {@code null} or empty, the corresponding config value is used. + * Parameter values are copied when this method is called so that later caller mutations cannot + * affect the asynchronous request. + * + * @param question the query text (required) + * @param topK the number of documents to retrieve (optional, defaults to config value) + * @param similarityThreshold the minimum similarity threshold (optional, defaults to config + * value) + * @param datasetIds dataset IDs for this request (optional, defaults to config value) + * @param metadataCondition metadata filtering conditions for this request (optional, defaults + * to config value) + * @return a Mono emitting the retrieval response + */ + public Mono retrieve( + String question, + Integer topK, + Double similarityThreshold, + List datasetIds, + Map metadataCondition) { + + List effectiveDatasetIds = + datasetIds != null && !datasetIds.isEmpty() + ? new ArrayList<>(datasetIds) + : copyList(config.getDatasetIds()); + Map effectiveMetadataCondition = + metadataCondition != null && !metadataCondition.isEmpty() + ? new HashMap<>(metadataCondition) + : copyMap(config.getMetadataCondition()); + return Mono.fromCallable( () -> { if (question == null || question.trim().isEmpty()) { @@ -113,8 +151,10 @@ public Mono retrieve( // Required: question text requestBody.put("question", question); - // Required: dataset_ids (array) - requestBody.put("dataset_ids", config.getDatasetIds()); + // Required unless document_ids is provided: dataset_ids (array) + if (!effectiveDatasetIds.isEmpty()) { + requestBody.put("dataset_ids", effectiveDatasetIds); + } // Optional: document_ids (filter to specific documents) if (config.getDocumentIds() != null && !config.getDocumentIds().isEmpty()) { @@ -185,11 +225,8 @@ public Mono retrieve( } // Optional: metadata_condition for filtering - if (metadataCondition != null && !metadataCondition.isEmpty()) { - requestBody.put("metadata_condition", metadataCondition); - } else if (config.getMetadataCondition() != null - && !config.getMetadataCondition().isEmpty()) { - requestBody.put("metadata_condition", config.getMetadataCondition()); + if (!effectiveMetadataCondition.isEmpty()) { + requestBody.put("metadata_condition", effectiveMetadataCondition); } String jsonBody = JsonUtils.getJsonCodec().toJson(requestBody); @@ -262,6 +299,14 @@ public Mono retrieve( }); } + private static List copyList(List values) { + return values == null ? new ArrayList<>() : new ArrayList<>(values); + } + + private static Map copyMap(Map values) { + return values == null ? new HashMap<>() : new HashMap<>(values); + } + private void handleErrorResponse(int statusCode, String responseBody) { logger.error("RAGFlow API error: status={}, body={}", statusCode, responseBody); diff --git a/agentscope-extensions/agentscope-extensions-rag/agentscope-extensions-rag-ragflow/src/main/java/io/agentscope/core/rag/integration/ragflow/RAGFlowKnowledge.java b/agentscope-extensions/agentscope-extensions-rag/agentscope-extensions-rag-ragflow/src/main/java/io/agentscope/core/rag/integration/ragflow/RAGFlowKnowledge.java index dd233955a5..382b276e94 100644 --- a/agentscope-extensions/agentscope-extensions-rag/agentscope-extensions-rag-ragflow/src/main/java/io/agentscope/core/rag/integration/ragflow/RAGFlowKnowledge.java +++ b/agentscope-extensions/agentscope-extensions-rag/agentscope-extensions-rag-ragflow/src/main/java/io/agentscope/core/rag/integration/ragflow/RAGFlowKnowledge.java @@ -20,6 +20,7 @@ import io.agentscope.core.rag.model.RetrieveConfig; import java.util.ArrayList; import java.util.List; +import java.util.Map; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import reactor.core.publisher.Mono; @@ -124,6 +125,30 @@ public static Builder builder() { */ @Override public Mono> retrieve(String query, RetrieveConfig config) { + return retrieve(query, config, null, null); + } + + /** + * Retrieve documents with request-specific RAGFlow filters. + * + *

Non-empty {@code datasetIds} and {@code metadataCondition} values override the defaults in + * {@link RAGFlowConfig} for this request only. If either parameter is {@code null} or empty, the + * corresponding config value is used. This allows one {@code RAGFlowKnowledge} instance to + * serve requests with different filters without mutating shared configuration or rebuilding an + * agent. + * + * @param query the query text (required) + * @param config the retrieval configuration (limit, score threshold) + * @param datasetIds dataset IDs for this request, or {@code null} to use the configured IDs + * @param metadataCondition metadata filtering conditions for this request, or {@code null} to + * use the configured condition + * @return a Mono emitting the list of retrieved documents, sorted by relevance + */ + public Mono> retrieve( + String query, + RetrieveConfig config, + List datasetIds, + Map metadataCondition) { if (query == null || query.trim().isEmpty()) { logger.warn("Empty query provided, returning empty result"); return Mono.just(new ArrayList<>()); @@ -135,8 +160,7 @@ public Mono> retrieve(String query, RetrieveConfig config) { Integer topK = config != null ? config.getLimit() : null; Double similarityThreshold = config != null ? config.getScoreThreshold() : null; - // Call RAGFlow API (metadata condition from config) - return client.retrieve(query, topK, similarityThreshold, null) + return client.retrieve(query, topK, similarityThreshold, datasetIds, metadataCondition) .map( response -> { if (response.getData() == null diff --git a/agentscope-extensions/agentscope-extensions-rag/agentscope-extensions-rag-ragflow/src/test/java/io/agentscope/core/rag/integration/ragflow/RAGFlowKnowledgeTest.java b/agentscope-extensions/agentscope-extensions-rag/agentscope-extensions-rag-ragflow/src/test/java/io/agentscope/core/rag/integration/ragflow/RAGFlowKnowledgeTest.java index d146ab1392..cd26888f50 100644 --- a/agentscope-extensions/agentscope-extensions-rag/agentscope-extensions-rag-ragflow/src/test/java/io/agentscope/core/rag/integration/ragflow/RAGFlowKnowledgeTest.java +++ b/agentscope-extensions/agentscope-extensions-rag/agentscope-extensions-rag-ragflow/src/test/java/io/agentscope/core/rag/integration/ragflow/RAGFlowKnowledgeTest.java @@ -20,12 +20,16 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import com.fasterxml.jackson.core.type.TypeReference; import io.agentscope.core.rag.model.Document; import io.agentscope.core.rag.model.RetrieveConfig; +import io.agentscope.core.util.JsonUtils; import java.io.IOException; import java.util.List; +import java.util.Map; import okhttp3.mockwebserver.MockResponse; import okhttp3.mockwebserver.MockWebServer; +import okhttp3.mockwebserver.RecordedRequest; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -239,6 +243,58 @@ void testRetrieveWithNullChunksResponse() throws Exception { assertTrue(documents.isEmpty()); } + @Test + void testRetrieveWithDynamicFiltersAndConfigFallback() throws Exception { + mockWebServer.enqueue(createSuccessResponse()); + mockWebServer.enqueue(createSuccessResponse()); + mockWebServer.enqueue(createSuccessResponse()); + + RAGFlowConfig config = + RAGFlowConfig.builder() + .apiKey("test-api-key") + .baseUrl(mockWebServer.url("").toString().replaceAll("/$", "")) + .addDatasetId("default-dataset") + .metadataCondition(Map.of("source", "default")) + .maxRetries(0) + .build(); + RAGFlowKnowledge knowledge = RAGFlowKnowledge.builder().config(config).build(); + RetrieveConfig retrieveConfig = + RetrieveConfig.builder().limit(10).scoreThreshold(0.5).build(); + + knowledge + .retrieve( + "first query", + retrieveConfig, + List.of("dataset-a"), + Map.of("source", "first")) + .block(); + knowledge + .retrieve( + "second query", + retrieveConfig, + List.of("dataset-b"), + Map.of("source", "second")) + .block(); + knowledge.retrieve("fallback query", retrieveConfig, List.of(), Map.of()).block(); + + Map firstBody = readRequestBody(mockWebServer.takeRequest()); + Map secondBody = readRequestBody(mockWebServer.takeRequest()); + Map fallbackBody = readRequestBody(mockWebServer.takeRequest()); + + assertEquals(List.of("dataset-a"), firstBody.get("dataset_ids")); + assertEquals(Map.of("source", "first"), firstBody.get("metadata_condition")); + assertEquals(List.of("dataset-b"), secondBody.get("dataset_ids")); + assertEquals(Map.of("source", "second"), secondBody.get("metadata_condition")); + assertEquals(List.of("default-dataset"), fallbackBody.get("dataset_ids")); + assertEquals(Map.of("source", "default"), fallbackBody.get("metadata_condition")); + } + + private Map readRequestBody(RecordedRequest request) { + return JsonUtils.getJsonCodec() + .fromJson( + request.getBody().readUtf8(), new TypeReference>() {}); + } + // === AddDocuments Tests === @Test