|
|
|
@ -17,6 +17,8 @@ import org.springframework.ai.document.Document; |
|
|
|
import org.springframework.ai.vectorstore.SearchRequest; |
|
|
|
import org.springframework.ai.vectorstore.VectorStore; |
|
|
|
import org.springframework.ai.vectorstore.filter.Filter; |
|
|
|
import org.springframework.beans.factory.annotation.Value; |
|
|
|
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; |
|
|
|
import org.springframework.stereotype.Component; |
|
|
|
import org.springframework.util.StringUtils; |
|
|
|
|
|
|
|
@ -26,6 +28,7 @@ import java.util.LinkedHashMap; |
|
|
|
import java.util.List; |
|
|
|
import java.util.Map; |
|
|
|
import java.util.Optional; |
|
|
|
import java.util.concurrent.CompletableFuture; |
|
|
|
import java.util.stream.Collectors; |
|
|
|
|
|
|
|
/** |
|
|
|
@ -93,6 +96,14 @@ public class RagPipeline { |
|
|
|
@Resource |
|
|
|
private MultiQueryExpanderRewriter multiQueryExpanderRewriter; |
|
|
|
|
|
|
|
/** MULTI_QUERY 多路检索扇出线程池：并行执行各路向量检索，饱和时退回请求线程串行（CallerRuns） */ |
|
|
|
@Resource(name = "ragRetrieveExecutor") |
|
|
|
private ThreadPoolTaskExecutor ragRetrieveExecutor; |
|
|
|
|
|
|
|
/** MULTI_QUERY 多路检索是否并行执行（false 回退串行，供高并发/embedding 侧限流时降级） */ |
|
|
|
@Value("${knowledge.rag.multiquery.parallel:true}") |
|
|
|
private boolean multiQueryParallel; |
|
|
|
|
|
|
|
private final DatabaseChatMemory chatMemory; |
|
|
|
|
|
|
|
public RagPipeline(DatabaseChatMemory chatMemory) { |
|
|
|
@ -108,7 +119,22 @@ public class RagPipeline { |
|
|
|
* @return 检索结果；FAQ 命中时 documents 与 contextText 为空，rewrittenQuery 为原始 message |
|
|
|
*/ |
|
|
|
public RagContext retrieve(ChatContext ctx) { |
|
|
|
// 1. FAQ 优先匹配：命中则直接返回标准答案，跳过检索与生成 |
|
|
|
return retrieve(ctx, false); |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* 执行一次统一的 RAG 检索（可跳过前序已做过的 FAQ 匹配）。 |
|
|
|
* <p> |
|
|
|
* 流程：FAQ 优先（未匹配过时）→ 查询重写/扩展 → 统一检索 → 拼接资料文本。 |
|
|
|
* |
|
|
|
* @param ctx 对话上下文（使用 {@code message / chatId / rewriteStrategy / categoryIds}） |
|
|
|
* @param faqAlreadyMatched 编排层是否已在前序阶段（FAQ 高置信未命中降级）干净跑过完整 FAQ 三级匹配； |
|
|
|
* true 时跳过 retrieve 内重复的 FAQ 匹配，避免同一请求重复做 FAQ 语义 embedding |
|
|
|
* @return 检索结果；FAQ 命中时 documents 与 contextText 为空，rewrittenQuery 为原始 message |
|
|
|
*/ |
|
|
|
public RagContext retrieve(ChatContext ctx, boolean faqAlreadyMatched) { |
|
|
|
// 1. FAQ 优先匹配：命中则直接返回标准答案，跳过检索与生成（已匹配过则跳过，防止重复 embedding） |
|
|
|
if (!faqAlreadyMatched) { |
|
|
|
Optional<FaqMatchResult> faqMatch = tryFaqMatchResult(ctx.message(), ctx.categoryIds()); |
|
|
|
if (faqMatch.isPresent()) { |
|
|
|
log.info("FAQ 命中，跳过知识库检索: chatId={}, matchType={}", ctx.chatId(), faqMatch.get().getMatchType()); |
|
|
|
@ -116,6 +142,7 @@ public class RagPipeline { |
|
|
|
return new RagContext(Optional.ofNullable(answer), Collections.emptyList(), "", ctx.message(), |
|
|
|
currentSearchMode(), faqMatch.get()); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
// 2. 统一检索 + 组装结果 |
|
|
|
return retrieveDocumentsAndContext(ctx); |
|
|
|
@ -190,6 +217,27 @@ public class RagPipeline { |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* FAQ 匹配结果 + 是否「完整跑完三级匹配后仍未命中」（区别于异常降级）。 |
|
|
|
* 供编排层判定：前序已干净跑过 FAQ 三级匹配且未命中时，可在后续检索中跳过重复的 FAQ 匹配。 |
|
|
|
*/ |
|
|
|
public record FaqMatchOutcome(Optional<FaqMatchResult> result, boolean completedCleanly) { |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* 尝试 FAQ 三级匹配并携带「是否干净完成」标记。 |
|
|
|
* match 内部各子级已吞掉各自异常并返回 empty，此处仅在 match 整体抛出未捕获异常时置 completedCleanly=false， |
|
|
|
* 避免异常降级被误当作「已匹配未命中」而在后续检索中跳过第二次 FAQ（可能本可命中）。 |
|
|
|
*/ |
|
|
|
public FaqMatchOutcome tryFaqMatchClean(String message, List<Long> categoryIds) { |
|
|
|
try { |
|
|
|
return new FaqMatchOutcome(faqMatchEngine.match(message, categoryIds), true); |
|
|
|
} catch (Exception e) { |
|
|
|
log.warn("FAQ 匹配异常(clean 标记)，降级到 RAG: {}", e.getMessage()); |
|
|
|
return new FaqMatchOutcome(Optional.empty(), false); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* 尝试 FAQ 三级匹配，命中返回标准答案（仅答案文本）。 |
|
|
|
* 异常时降级为未命中，供仅需答案的调用方使用。 |
|
|
|
@ -246,6 +294,7 @@ public class RagPipeline { |
|
|
|
/** |
|
|
|
* 多路查询扩展 + 分别检索 + 按文档 ID 去重合并。 |
|
|
|
* 扩展失败时退回原问题检索，避免整次 RAG 不可用。 |
|
|
|
* 多路检索默认并行执行（每路 embedding HTTP + PG 查询相互独立），关闭或异常时回退串行。 |
|
|
|
*/ |
|
|
|
private List<Document> retrieveMultiQueryDocs(String message, List<Long> categoryIds) { |
|
|
|
List<String> expandedQueries; |
|
|
|
@ -260,6 +309,18 @@ public class RagPipeline { |
|
|
|
} |
|
|
|
log.info("多路查询扩展结果: {}", expandedQueries); |
|
|
|
|
|
|
|
// 多路检索默认并行执行（每路 embedding HTTP + PG 查询相互独立，可把 N 路串行降为 1 段最慢路）； |
|
|
|
// 单路或关闭并行开关时回退串行，保证结果与旧版一致 |
|
|
|
boolean useParallel = multiQueryParallel && expandedQueries.size() > 1; |
|
|
|
return useParallel |
|
|
|
? retrieveMultiQueryParallel(expandedQueries, categoryIds) |
|
|
|
: retrieveMultiQuerySequential(expandedQueries, categoryIds); |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* 串行执行多路检索（并行开关关闭 / 单路 / 异常回退路径，行为与旧版一致）。 |
|
|
|
*/ |
|
|
|
private List<Document> retrieveMultiQuerySequential(List<String> expandedQueries, List<Long> categoryIds) { |
|
|
|
Map<String, Document> merged = new LinkedHashMap<>(); |
|
|
|
for (String query : expandedQueries) { |
|
|
|
if (!StringUtils.hasText(query) || merged.size() >= MAX_DOCS) { |
|
|
|
@ -276,6 +337,43 @@ public class RagPipeline { |
|
|
|
return new ArrayList<>(merged.values()); |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* 并行执行多路检索：各路相互独立（embedding HTTP + PG 查询），可把 N 路串行降为 1 段最慢路。 |
|
|
|
* join 后仍按原 query 顺序合并去重（LinkedHashMap + MAX_DOCS 封顶），保证文档集合与串行结果一致； |
|
|
|
* 任一路异常降级为空列表，不拖垮整次检索。 |
|
|
|
*/ |
|
|
|
private List<Document> retrieveMultiQueryParallel(List<String> expandedQueries, List<Long> categoryIds) { |
|
|
|
List<CompletableFuture<List<Document>>> futures = new ArrayList<>(expandedQueries.size()); |
|
|
|
for (String query : expandedQueries) { |
|
|
|
if (!StringUtils.hasText(query)) { |
|
|
|
futures.add(CompletableFuture.completedFuture(Collections.emptyList())); |
|
|
|
continue; |
|
|
|
} |
|
|
|
futures.add(CompletableFuture |
|
|
|
.supplyAsync(() -> similaritySearch(query, categoryIds), ragRetrieveExecutor) |
|
|
|
.exceptionally(ex -> { |
|
|
|
log.warn("多路并行检索单路失败，该路降级为空: error={}", ex.getMessage()); |
|
|
|
return Collections.emptyList(); |
|
|
|
})); |
|
|
|
} |
|
|
|
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); |
|
|
|
|
|
|
|
// 按原 query 顺序合并，去重/封顶规则与串行路径一致（不按完成先后，保证结果确定） |
|
|
|
Map<String, Document> merged = new LinkedHashMap<>(); |
|
|
|
for (int i = 0; i < expandedQueries.size() && merged.size() < MAX_DOCS; i++) { |
|
|
|
if (!StringUtils.hasText(expandedQueries.get(i))) { |
|
|
|
continue; |
|
|
|
} |
|
|
|
for (Document doc : futures.get(i).join()) { |
|
|
|
if (merged.size() >= MAX_DOCS) { |
|
|
|
break; |
|
|
|
} |
|
|
|
merged.putIfAbsent(doc.getId(), doc); |
|
|
|
} |
|
|
|
} |
|
|
|
return new ArrayList<>(merged.values()); |
|
|
|
} |
|
|
|
|
|
|
|
// ==================== 底层检索 ==================== |
|
|
|
|
|
|
|
/** |
|
|
|
|
wanghanlin