feat(ragflow): support request-scoped retrieval filters

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<RAGFlowResponse> retrieve(
             Double similarityThreshold,
             Map<String, Object> metadataCondition) {
 
+        return retrieve(question, topK, similarityThreshold, null, metadataCondition);
+    }
+
+    /**
+     * Retrieve documents with request-specific dataset IDs and metadata conditions.
+     *
+     * <p>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<RAGFlowResponse> retrieve(
+            String question,
+            Integer topK,
+            Double similarityThreshold,
+            List<String> datasetIds,
+            Map<String, Object> metadataCondition) {
+
+        List<String> effectiveDatasetIds =
+                datasetIds != null && !datasetIds.isEmpty()
+                        ? new ArrayList<>(datasetIds)
+                        : copyList(config.getDatasetIds());
+        Map<String, Object> 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<RAGFlowResponse> 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<RAGFlowResponse> 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<RAGFlowResponse> retrieve(
                 });
     }
 
+    private static <T> List<T> copyList(List<T> values) {
+        return values == null ? new ArrayList<>() : new ArrayList<>(values);
+    }
+
+    private static <K, V> Map<K, V> copyMap(Map<K, V> 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);
 

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<List<Document>> retrieve(String query, RetrieveConfig config) {
+        return retrieve(query, config, null, null);
+    }
+
+    /**
+     * Retrieve documents with request-specific RAGFlow filters.
+     *
+     * <p>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<List<Document>> retrieve(
+            String query,
+            RetrieveConfig config,
+            List<String> datasetIds,
+            Map<String, Object> 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<List<Document>> 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

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<String, Object> firstBody = readRequestBody(mockWebServer.takeRequest());
+        Map<String, Object> secondBody = readRequestBody(mockWebServer.takeRequest());
+        Map<String, Object> 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<String, Object> readRequestBody(RecordedRequest request) {
+        return JsonUtils.getJsonCodec()
+                .fromJson(
+                        request.getBody().readUtf8(), new TypeReference<Map<String, Object>>() {});
+    }
+
     // === AddDocuments Tests ===
 
     @Test