【Spring AI实战】第6章 RAG检索增强生成完整实战
·
1. RAG整体架构：文档解析→切片→向量化→存储→检索→生成
Spring AI RAG（检索增强生成）的整体架构是一个典型的多阶段流程，旨在将外部知识库与大型语言模型结合，以生成更准确、更具上下文的回答。以下是其核心环节的详细解析：
文档解析
目标 ：将各种格式的原始文档（PDF、Word、HTML、Markdown等）转换为机器可读的纯文本或结构化数据。
关键技术与组件 ：
- Spring AI 文档加载器 ：如
WebPdfDocumentReader、TextDocumentReader等，支持多种格式。 - 解析策略 ：提取文本、表格、代码块、元数据（作者、标题等），并处理特殊字符和编码。
- 输出 ：统一的
Document对象（包含文本内容和元数据）。
切片
目标 ：将长文档分割成适合向量化和检索的小片段，平衡上下文完整性与检索精度。
常见策略 ：
- 固定大小重叠切片 ：按字符/Token数分割，保留重叠部分以维持上下文连贯性。
- 语义切片 ：基于句子或段落边界，利用自然语言处理（NLP）模型识别语义边界。
- 递归切片 ：结合固定大小与语义边界，动态调整分割点。
- Spring AI 支持 ：通过
TokenTextSplitter、RecursiveCharacterTextSplitter等实现。
向量化
目标 ：将文本切片转换为高维向量（嵌入），捕捉语义信息，便于相似度计算。
关键组件 ：
- 嵌入模型 ：如 OpenAI
text-embedding-ada-002、本地模型（Sentence Transformers）。 - Spring AI 抽象 ：通过
EmbeddingClient接口调用不同模型，统一向量生成。 - 输出 ：每个文本切片对应一个浮点数向量（维度通常为 384、768、1536 等）。
存储
目标 ：高效存储向量及其关联的文本片段，支持快速相似度检索。
技术选型 ：
- 向量数据库 ：如 Pinecone、Weaviate、Milvus、Chroma、Redis Stack 等。
- Spring AI 支持 ：通过
VectorStore接口抽象化操作，支持多种后端。 - 存储内容 ：
- 向量数据（嵌入向量）。
- 原始文本片段。
- 元数据（来源、页码、时间戳等）。
检索
目标 ：根据用户查询，从向量数据库中找出最相关的文本片段。
检索策略 ：
- 相似度检索 ：计算查询向量与存储向量的余弦相似度，返回 Top-K 结果。
- 混合检索 ：结合关键词搜索（BM25）与向量检索，提升召回率。
- 过滤 ：基于元数据（如文档来源、日期）缩小检索范围。
- Spring AI 支持 ：通过
VectorStore.similaritySearch()等方法实现。
生成
目标 ：将检索到的上下文与用户查询结合，生成自然语言回答。
流程 ：
- 提示工程 ：构建包含以下内容的提示模板：
基于以下上下文：
{检索到的文本片段}
回答这个问题：
{用户查询}
如果上下文不包含答案，请说明“根据已知信息无法回答”。- 调用 LLM ：通过
ChatClient（如 OpenAI GPT、Azure OpenAI、本地模型）生成回答。 - 后处理 ：过滤无关内容，添加引用来源（可选项）。
Spring AI 中的核心抽象
表格 还在加载中，请等待加载完成后再尝试复制
典型代码流程
// 1. 加载文档
List<Document> documents = new WebPdfDocumentReader("url").read();
// 2. 分割文本
TextSplitter splitter = new TokenTextSplitter();
List<Document> chunks = splitter.split(documents);
// 3. 向量化并存储
VectorStore vectorStore = new InMemoryVectorStore(embeddingClient);
vectorStore.add(chunks);
// 4. 检索
List<Document> retrieved = vectorStore.similaritySearch(query);
// 5. 生成
String prompt = """
基于以下上下文：
{context}
回答：{question}
""";
PromptTemplate template = new PromptTemplate(prompt);
String context = retrieved.stream().map(Document::getContent).collect(Collectors.joining("\n"));
ChatResponse response = chatClient.call(
template.create(Map.of("context", context, "question", query))
);优化方向
- 检索优化 ：重排序（Re-ranking）、多查询扩展、HyDE（假设性文档嵌入）。
- 切片策略 ：自适应切片、基于实体的分割。
- 缓存机制 ：缓存常见查询的嵌入或检索结果。
- 评估 ：使用 RAGAS 等框架评估检索相关性、回答准确性。
通过 Spring AI 的模块化设计，开发者可以灵活替换各阶段组件（如切换向量数据库或嵌入模型），快速构建高效的 RAG 系统。
2. 文档处理实战：PDF/Word/TXT 文档解析、清洗、分块策略
概述
Spring AI 提供了强大的文档处理能力，支持多种格式文档的解析、清洗和分块，为后续的向量化存储和AI应用打下基础。
环境准备
2.1 添加依赖
<dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-document-readers</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-pdf-document-reader</artifactId>
</dependency>
<dependency>
<groupId>org.apache.poi</groupId>
<artifactId>poi-ooxml</artifactId>
<version>5.2.3</version>
</dependency>
<dependency>
<groupId>org.apache.tika</groupId>
<artifactId>tika-core</artifactId>
<version>2.9.1</version>
</dependency>文档解析
3.1 PDF文档解析
import org.springframework.ai.reader.ExtractedTextFormatter;
import org.springframework.ai.reader.pdf.PagePdfDocumentReader;
import org.springframework.ai.reader.pdf.config.PdfDocumentReaderConfig;
import org.springframework.core.io.Resource;
@Component
public class PdfDocumentProcessor {
public List<Document> parsePdf(Resource pdfResource) {
// 配置PDF读取器
PdfDocumentReaderConfig config = PdfDocumentReaderConfig.builder()
.withPageExtractedTextFormatter(
ExtractedTextFormatter.builder()
.withNumberOfTopTextLinesToDelete(0)
.withNumberOfBottomTextLinesToDelete(0)
.withNumberOfLeftMostCharactersToDelete(0)
.withNumberOfRightMostCharactersToDelete(0)
.build()
)
.withPagesPerDocument(1) // 每页作为一个文档
.build();
// 创建PDF阅读器
PagePdfDocumentReader pdfReader = new PagePdfDocumentReader(
pdfResource,
config
);
// 读取文档
return pdfReader.get();
}
// 带OCR的PDF解析（处理扫描件）
public List<Document> parsePdfWithOcr(Resource pdfResource) {
// 需要额外的OCR配置
// 可以使用TesseractOCR
}
}3.2 Word文档解析
import org.springframework.ai.reader.tika.TikaDocumentReader;
@Component
public class WordDocumentProcessor {
public List<Document> parseWordDocument(Resource wordResource) {
TikaDocumentReader reader = new TikaDocumentReader(wordResource);
return reader.get();
}
// 解析DOCX文件（使用POI）
public List<Document> parseDocxDocument(Resource docxResource) {
TikaDocumentReader reader = new TikaDocumentReader(docxResource);
// Tika会自动检测文件类型并解析
return reader.get();
}
}3.3 纯文本文件解析
import org.springframework.ai.reader.TextReader;
@Component
public class TextDocumentProcessor {
public List<Document> parseTextFile(Resource textResource) {
TextReader textReader = new TextReader(textResource);
textReader.setCharset(StandardCharsets.UTF_8);
return textReader.get();
}
// 批量处理文本文件
public List<Document> parseTextFiles(List<Resource> textResources) {
List<Document> allDocuments = new ArrayList<>();
for (Resource resource : textResources) {
TextReader reader = new TextReader(resource);
allDocuments.addAll(reader.get());
}
return allDocuments;
}
}文档清洗
4.1 基础清洗策略
import org.springframework.ai.transformer.splitter.TokenTextSplitter;
import org.springframework.ai.document.Document;
import org.springframework.ai.document.DocumentTransformer;
@Component
public class DocumentCleaner {
// 移除特殊字符和空白
public String cleanText(String text) {
if (text == null) return "";
// 1. 移除多余空白
text = text.replaceAll("\\s+", " ");
// 2. 移除不可见字符
text = text.replaceAll("[\\u0000-\\u001F\\u007F-\\u009F]", "");
// 3. 移除特殊符号（保留中文标点和基本英文标点）
text = text.replaceAll("[^\\p{L}\\p{N}\\p{P}\\p{Z}\\p{Sm}\\p{Sc}\\p{Sk}\\p{So}\\u4e00-\\u9fff，。！？；：""''、（）《》【】]", " ");
// 4. 标准化换行符
text = text.replaceAll("\\r\\n|\\r", "\n");
return text.trim();
}
// 处理文档中的乱码
public String fixEncoding(String text) {
try {
// 尝试多种编码
byte[] bytes = text.getBytes(StandardCharsets.ISO_8859_1);
String utf8Text = new String(bytes, StandardCharsets.UTF_8);
// 检测并修复常见乱码
utf8Text = utf8Text
.replace("â€", "'")
.replace("“", "\"")
.replace("â€", "\"")
.replace("’", "'");
return utf8Text;
} catch (Exception e) {
return text;
}
}
}4.2 结构化数据清洗
@Component
public class StructuredDataCleaner {
// 处理表格数据
public String cleanTableData(String text) {
// 1. 识别表格结构
String[] lines = text.split("\n");
StringBuilder cleaned = new StringBuilder();
for (String line : lines) {
// 2. 移除表格边框字符
line = line.replaceAll("[┌─┬┐├┼┤└┴┘│]", " ");
// 3. 规范化表格内容
line = line.replaceAll("\\s+\\|\\s+", " | ");
line = line.replaceAll("^\\||\\|$", "");
cleaned.append(line.trim()).append("\n");
}
return cleaned.toString();
}
// 处理代码块
public String cleanCodeBlocks(String text) {
// 保留代码结构，但移除多余空白
Pattern codePattern = Pattern.compile("```(.*?)```", Pattern.DOTALL);
Matcher matcher = codePattern.matcher(text);
StringBuffer result = new StringBuffer();
while (matcher.find()) {
String codeBlock = matcher.group(1);
// 清理代码块内的多余空白
String cleanedCode = codeBlock.trim().replaceAll("\\n\\s+", "\n");
matcher.appendReplacement(result, "```" + cleanedCode + "```");
}
matcher.appendTail(result);
return result.toString();
}
}文档分块策略
5.1 基于Token的分块
import org.springframework.ai.transformer.splitter.TokenTextSplitter;
import org.springframework.ai.transformer.splitter.TextSplitter;
@Component
public class TokenBasedSplitter {
public List<Document> splitByTokens(List<Document> documents) {
// 配置文本分割器
TokenTextSplitter splitter = new TokenTextSplitter(
1000, // chunkSize: 每个分块的最大token数
200, // chunkOverlap: 分块之间的重叠token数
true, // keepSeparator: 是否保留分隔符
"gpt-4" // 模型名称，用于计算token数
);
return splitter.apply(documents);
}
// 自定义分隔符的分块
public List<Document> splitWithCustomSeparators(List<Document> documents) {
TokenTextSplitter splitter = new TokenTextSplitter.Builder()
.setChunkSize(1000)
.setChunkOverlap(200)
.setSeparator("\n\n") // 使用空行作为主要分隔符
.setSecondarySeparators(List.of("\n", "。", "！", "？", "；"))
.build();
return splitter.apply(documents);
}
}5.2 基于语义的分块
import org.springframework.ai.embedding.EmbeddingModel;
import org.springframework.ai.transformer.splitter.SemanticTextSplitter;
@Component
public class SemanticSplitter {
private final EmbeddingModel embeddingModel;
public SemanticSplitter(EmbeddingModel embeddingModel) {
this.embeddingModel = embeddingModel;
}
public List<Document> splitSemantically(List<Document> documents) {
SemanticTextSplitter splitter = new SemanticTextSplitter(
embeddingModel,
1000, // chunkSize
200, // chunkOverlap
0.8, // similarityThreshold: 语义相似度阈值
50 // windowSize: 滑动窗口大小
);
return splitter.apply(documents);
}
}5.3 智能分块策略
@Component
public class IntelligentChunkingStrategy {
// 混合分块策略
public List<Document> intelligentChunking(List<Document> documents) {
List<Document> allChunks = new ArrayList<>();
for (Document doc : documents) {
String content = doc.getContent();
// 1. 按章节分块（如果有章节结构）
List<String> chapterChunks = splitByChapters(content);
if (!chapterChunks.isEmpty()) {
allChunks.addAll(createDocuments(chapterChunks, doc.getMetadata()));
continue;
}
// 2. 按段落分块
List<String> paragraphChunks = splitByParagraphs(content);
if (paragraphChunks.size() > 1) {
allChunks.addAll(mergeSmallParagraphs(paragraphChunks, doc.getMetadata()));
continue;
}
// 3. 按句子分块（最后手段）
List<String> sentenceChunks = splitBySentences(content);
allChunks.addAll(mergeSmallSentences(sentenceChunks, doc.getMetadata()));
}
return allChunks;
}
private List<String> splitByChapters(String content) {
List<String> chapters = new ArrayList<>();
// 检测章节标题（支持多种格式）
Pattern chapterPattern = Pattern.compile(
"(?:第[一二三四五六七八九十零百千]+章|[1-9][0-9]*\\.\\s|Chapter\\s+[1-9][0-9]*).*?(?=\\n第|\\n[1-9]|\\nChapter|$)",
Pattern.DOTALL
);
Matcher matcher = chapterPattern.matcher(content);
while (matcher.find()) {
chapters.add(matcher.group());
}
return chapters;
}
private List<String> splitByParagraphs(String content) {
// 按空行分割段落
return Arrays.stream(content.split("\\n\\s*\\n"))
.filter(p -> !p.trim().isEmpty())
.collect(Collectors.toList());
}
private List<String> splitBySentences(String content) {
// 简单的句子分割（支持中英文）
Pattern sentencePattern = Pattern.compile(
"[^。！？.!?]*[。！？.!?]",
Pattern.MULTILINE
);
List<String> sentences = new ArrayList<>();
Matcher matcher = sentencePattern.matcher(content);
while (matcher.find()) {
sentences.add(matcher.group().trim());
}
return sentences;
}
private List<Document> mergeSmallParagraphs(List<String> paragraphs, Map<String, Object> metadata) {
List<Document> documents = new ArrayList<>();
StringBuilder currentChunk = new StringBuilder();
int currentSize = 0;
for (String paragraph : paragraphs) {
int paragraphSize = paragraph.length();
if (currentSize + paragraphSize > 1000 && currentSize > 0) {
// 保存当前分块
documents.add(new Document(currentChunk.toString(), metadata));
currentChunk = new StringBuilder();
currentSize = 0;
}
if (currentSize > 0) {
currentChunk.append("\n\n");
}
currentChunk.append(paragraph);
currentSize += paragraphSize;
}
// 添加最后一个分块
if (currentSize > 0) {
documents.add(new Document(currentChunk.toString(), metadata));
}
return documents;
}
}完整处理流程
6.1 文档处理管道
@Component
public class DocumentProcessingPipeline {
private final PdfDocumentProcessor pdfProcessor;
private final WordDocumentProcessor wordProcessor;
private final TextDocumentProcessor textProcessor;
private final DocumentCleaner cleaner;
private final IntelligentChunkingStrategy chunkingStrategy;
public DocumentProcessingPipeline(
PdfDocumentProcessor pdfProcessor,
WordDocumentProcessor wordProcessor,
TextDocumentProcessor textProcessor,
DocumentCleaner cleaner,
IntelligentChunkingStrategy chunkingStrategy) {
this.pdfProcessor = pdfProcessor;
this.wordProcessor = wordProcessor;
this.textProcessor = textProcessor;
this.cleaner = cleaner;
this.chunkingStrategy = chunkingStrategy;
}
public List<Document> processDocument(Resource documentResource) {
// 1. 根据文件类型选择解析器
List<Document> rawDocuments = parseDocument(documentResource);
// 2. 清洗文档内容
List<Document> cleanedDocuments = cleanDocuments(rawDocuments);
// 3. 智能分块
List<Document> chunks = chunkingStrategy.intelligentChunking(cleanedDocuments);
// 4. 添加元数据
return enhanceMetadata(chunks, documentResource);
}
private List<Document> parseDocument(Resource resource) {
String filename = resource.getFilename();
if (filename == null) {
throw new IllegalArgumentException("无法识别文件类型");
}
if (filename.toLowerCase().endsWith(".pdf")) {
return pdfProcessor.parsePdf(resource);
} else if (filename.toLowerCase().endsWith(".docx") ||
filename.toLowerCase().endsWith(".doc")) {
return wordProcessor.parseWordDocument(resource);
} else if (filename.toLowerCase().endsWith(".txt")) {
return textProcessor.parseTextFile(resource);
} else {
// 尝试使用Tika解析其他格式
TikaDocumentReader reader = new TikaDocumentReader(resource);
return reader.get();
}
}
private List<Document> cleanDocuments(List<Document> documents) {
return documents.stream()
.map(doc -> {
String cleanedContent = cleaner.cleanText(doc.getContent());
cleanedContent = cleaner.fixEncoding(cleanedContent);
return new Document(cleanedContent, doc.getMetadata());
})
.collect(Collectors.toList());
}
private List<Document> enhanceMetadata(List<Document> chunks, Resource resource) {
return chunks.stream()
.map(doc -> {
Map<String, Object> metadata = new HashMap<>(doc.getMetadata());
metadata.put("source", resource.getFilename());
metadata.put("processed_time", LocalDateTime.now().toString());
metadata.put("chunk_size", doc.getContent().length());
metadata.put("chunk_hash", calculateHash(doc.getContent()));
return new Document(doc.getContent(), metadata);
})
.collect(Collectors.toList());
}
private String calculateHash(String content) {
try {
MessageDigest md = MessageDigest.getInstance("SHA-256");
byte[] hash = md.digest(content.getBytes(StandardCharsets.UTF_8));
return Base64.getEncoder().encodeToString(hash);
} catch (NoSuchAlgorithmException e) {
return "";
}
}
}6.2 批量处理服务
@Service
public class BatchDocumentProcessor {
@Autowired
private DocumentProcessingPipeline pipeline;
@Value("${document.process.batch.size:10}")
private int batchSize;
@Async
public CompletableFuture<List<Document>> processDocumentsInBatch(List<Resource> resources) {
List<Document> allProcessed = new ArrayList<>();
// 分批处理，避免内存溢出
for (int i = 0; i < resources.size(); i += batchSize) {
List<Resource> batch = resources.subList(
i, Math.min(i + batchSize, resources.size())
);
List<Document> batchResults = batch.stream()
.parallel()
.flatMap(resource -> {
try {
return pipeline.processDocument(resource).stream();
} catch (Exception e) {
log.error("处理文档失败: {}", resource.getFilename(), e);
return Stream.empty();
}
})
.collect(Collectors.toList());
allProcessed.addAll(batchResults);
}
return CompletableFuture.completedFuture(allProcessed);
}
}配置示例
7.1 应用配置
# application.yml
spring:
ai:
document:
processing:
chunk-size: 1000
chunk-overlap: 200
max-document-size: 10485760 # 10MB
supported-formats:
- pdf
- docx
- doc
- txt
- md
cleaning:
remove-special-chars: true
normalize-whitespace: true
fix-encoding: true
chunking:
strategy: intelligent
semantic-threshold: 0.8
preserve-structure: true7.2 Bean配置
@Configuration
public class DocumentProcessingConfig {
@Bean
public PdfDocumentReaderConfig pdfDocumentReaderConfig() {
return PdfDocumentReaderConfig.builder()
.withPagesPerDocument(1)
.withPageTopMargin(0)
.withPageBottomMargin(0)
.build();
}高级特性与优化
8.1 文档质量评估
@Component
public class DocumentQualityAssessor {
// 评估文档质量分数
public double assessDocumentQuality(Document document) {
String content = document.getContent();
double score = 0.0;
// 1. 内容完整性检查
score += checkCompleteness(content) * 0.3;
// 2. 可读性检查
score += checkReadability(content) * 0.3;
// 3. 信息密度检查
score += checkInformationDensity(content) * 0.2;
// 4. 结构合理性检查
score += checkStructure(content) * 0.2;
return Math.min(score, 1.0);
}
private double checkCompleteness(String content) {
// 检查是否有明显缺失
if (content.length() < 50) return 0.2;
if (content.length() < 200) return 0.5;
// 检查句子完整性
long completeSentences = countCompleteSentences(content);
long totalSentences = countTotalSentences(content);
if (totalSentences == 0) return 0.3;
return (double) completeSentences / totalSentences;
}
private double checkReadability(String content) {
// 计算Flesch Reading Ease（简化版）
int words = countWords(content);
int sentences = countTotalSentences(content);
if (sentences == 0 || words == 0) return 0.3;
double wordsPerSentence = (double) words / sentences;
double syllablesPerWord = estimateSyllablesPerWord(content);
// 简化版Flesch公式
double readability = 206.835 - 1.015 * wordsPerSentence - 84.6 * syllablesPerWord;
// 归一化到0-1
return Math.max(0, Math.min(1, readability / 100));
}
private double checkInformationDensity(String content) {
// 计算信息密度（关键词占比）
Set<String> keywords = extractKeywords(content);
int totalWords = countWords(content);
if (totalWords == 0) return 0.2;
double keywordDensity = (double) keywords.size() * 10 / totalWords;
return Math.min(keywordDensity, 1.0);
}
private double checkStructure(String content) {
// 检查文档结构
double structureScore = 0.0;
// 是否有标题
if (hasHeadings(content)) structureScore += 0.3;
// 是否有段落结构
if (hasParagraphs(content)) structureScore += 0.3;
// 是否有列表
if (hasLists(content)) structureScore += 0.2;
// 是否有链接或引用
if (hasReferences(content)) structureScore += 0.2;
return structureScore;
}
// 辅助方法
private long countCompleteSentences(String text) {
return Arrays.stream(text.split("[。！？.!?]"))
.filter(s -> s.trim().length() > 5)
.count();
}
private long countTotalSentences(String text) {
return Arrays.stream(text.split("[。！？.!?]"))
.filter(s -> !s.trim().isEmpty())
.count();
}
private int countWords(String text) {
return text.split("\\s+").length;
}
private double estimateSyllablesPerWord(String text) {
// 简化的音节估计
String[] words = text.toLowerCase().split("\\s+");
int totalSyllables = 0;
for (String word : words) {
totalSyllables += estimateWordSyllables(word);
}
return words.length > 0 ? (double) totalSyllables / words.length : 0;
}
private int estimateWordSyllables(String word) {
// 简化的音节计数规则
word = word.replaceAll("[^a-z]", "");
if (word.length() <= 3) return 1;
int syllables = 0;
boolean prevVowel = false;
for (char c : word.toCharArray()) {
boolean isVowel = "aeiou".indexOf(c) >= 0;
if (isVowel && !prevVowel) {
syllables++;
}
prevVowel = isVowel;
}
// 特殊情况
if (word.endsWith("e") && syllables > 1) syllables--;
if (word.endsWith("le") && syllables == 1) syllables++;
return Math.max(1, syllables);
}
private Set<String> extractKeywords(String content) {
// 简单的关键词提取（基于词频和停用词）
Set<String> stopWords = Set.of("the", "a", "an", "and", "or", "but", "in", "on", "at", "to", "for");
return Arrays.stream(content.toLowerCase().split("\\s+"))
.filter(word -> word.length() > 3)
.filter(word -> !stopWords.contains(word))
.collect(Collectors.toSet());
}
private boolean hasHeadings(String content) {
return content.matches("(?s).*# .*|.*==+.*|.*[一二三四五六七八九十]、.*");
}
private boolean hasParagraphs(String content) {
return content.split("\\n\\s*\\n").length > 1;
}
private boolean hasLists(String content) {
return content.matches("(?s).*\\* .*|.*- .*|.*\\d+\\. .*");
}
private boolean hasReferences(String content) {
return content.matches("(?s).*\\[.*\\].*|.*http://.*|.*https://.*");
}
}8.2 增量文档处理
@Component
public class IncrementalDocumentProcessor {
@Autowired
private DocumentRepository documentRepository;
@Autowired
private EmbeddingModel embeddingModel;
// 增量处理文档，只处理新内容
public List<Document> processIncrementally(Resource resource, String documentId) {
// 1. 解析文档
List<Document> newChunks = parseDocument(resource);
// 2. 获取已存在的文档块
List<Document> existingChunks = documentRepository.findBySourceDocumentId(documentId);
// 3. 计算差异
DocumentDiffResult diff = calculateDiff(newChunks, existingChunks);
// 4. 处理新增和修改的块
List<Document> chunksToProcess = new ArrayList<>();
chunksToProcess.addAll(diff.getAddedChunks());
chunksToProcess.addAll(diff.getModifiedChunks());
// 5. 清理已删除的块
diff.getRemovedChunks().forEach(chunk ->
documentRepository.deleteById(chunk.getId())
);
return chunksToProcess;
}
private DocumentDiffResult calculateDiff(List<Document> newChunks, List<Document> existingChunks) {
DocumentDiffResult result = new DocumentDiffResult();
// 计算哈希值用于比较
Map<String, Document> existingChunksByHash = existingChunks.stream()
.collect(Collectors.toMap(
this::calculateChunkHash,
Function.identity()
));
Map<String, Document> newChunksByHash = newChunks.stream()
.collect(Collectors.toMap(
this::calculateChunkHash,
Function.identity()
));
// 找出新增的块
newChunksByHash.keySet().stream()
.filter(hash -> !existingChunksByHash.containsKey(hash))
.map(newChunksByHash::get)
.forEach(result::addAddedChunk);
// 找出删除的块
existingChunksByHash.keySet().stream()
.filter(hash -> !newChunksByHash.containsKey(hash))
.map(existingChunksByHash::get)
.forEach(result::addRemovedChunk);
// 找出修改的块（哈希相同但内容不同）
newChunksByHash.keySet().stream()
.filter(existingChunksByHash::containsKey)
.filter(hash -> !newChunksByHash.get(hash).getContent()
.equals(existingChunksByHash.get(hash).getContent()))
.map(newChunksByHash::get)
.forEach(result::addModifiedChunk);
return result;
}
private String calculateChunkHash(Document chunk) {
try {
String content = chunk.getContent();
Map<String, Object> metadata = chunk.getMetadata();
// 组合内容和关键元数据计算哈希
String toHash = content + "|" +
metadata.getOrDefault("chunk_index", "") + "|" +
metadata.getOrDefault("section", "");
MessageDigest md = MessageDigest.getInstance("SHA-256");
byte[] hash = md.digest(toHash.getBytes(StandardCharsets.UTF_8));
return Base64.getEncoder().encodeToString(hash);
} catch (Exception e) {
return "";
}
}
// 差异结果类
@Data
public static class DocumentDiffResult {
private List<Document> addedChunks = new ArrayList<>();
private List<Document> removedChunks = new ArrayList<>();
private List<Document> modifiedChunks = new ArrayList<>();
}
}8.3 多语言支持
@Component
public class MultilingualDocumentProcessor {
private static final Map<String, String> LANGUAGE_PATTERNS = Map.of(
"zh", "[\\u4e00-\\u9fff]",
"en", "[a-zA-Z]",
"ja", "[\\u3040-\\u309F\\u30A0-\\u30FF\\u4E00-\\u9FFF]",
"ko", "[\\uAC00-\\uD7AF]",
"ru", "[\\u0400-\\u04FF]"
);
// 检测文档语言
public String detectLanguage(String text) {
if (text == null || text.trim().isEmpty()) {
return "unknown";
}
Map<String, Integer> languageScores = new HashMap<>();
for (Map.Entry<String, String> entry : LANGUAGE_PATTERNS.entrySet()) {
Pattern pattern = Pattern.compile(entry.getValue());
Matcher matcher = pattern.matcher(text);
int count = 0;
while (matcher.find()) {
count++;
}
languageScores.put(entry.getKey(), count);
}
// 返回得分最高的语言
return languageScores.entrySet().stream()
.max(Map.Entry.comparingByValue())
.map(Map.Entry::getKey)
.orElse("unknown");
}
// 语言特定的清洗规则
public String cleanByLanguage(String text, String language) {
String cleaned = text;
switch (language) {
case "zh":
// 中文特定清洗
cleaned = cleanChineseText(text);
break;
case "ja":
// 日文特定清洗
cleaned = cleanJapaneseText(text);
break;
case "ko":
// 韩文特定清洗
cleaned = cleanKoreanText(text);
break;
default:
// 默认清洗（主要针对英文）
cleaned = cleanEnglishText(text);
}
return cleaned;
}
private String cleanChineseText(String text) {
// 中文文本清洗
String cleaned = text;
// 移除全角空格
cleaned = cleaned.replaceAll(" ", " ");
// 标准化中文标点
cleaned = cleaned.replaceAll("，", ", ")
.replaceAll("。", ". ")
.replaceAll("！", "! ")
.replaceAll("？", "? ")
.replaceAll("；", "; ")
.replaceAll("：", ": ")
.replaceAll("「", "\"")
.replaceAll("」", "\"")
.replaceAll("『", "'")
.replaceAll("』", "'");
// 移除多余空格
cleaned = cleaned.replaceAll("\\s+", " ");
return cleaned.trim();
}
private String cleanJapaneseText(String text) {
// 日文文本清洗
String cleaned = text;
// 处理日文特定字符
cleaned = cleaned.replaceAll("【", "[")
.replaceAll("】", "]")
.replaceAll("｢", "\"")
.replaceAll("｣", "\"");
return cleaned;
}
private String cleanKoreanText(String text) {
// 韩文文本清洗
String cleaned = text;
// 韩文特定处理
cleaned = cleaned.replaceAll("《", "<")
.replaceAll("》", ">");
return cleaned;
}
private String cleanEnglishText(String text) {
// 英文文本清洗
String cleaned = text;
// 移除连字符换行
cleaned = cleaned.replaceAll("-\\n", "");
// 标准化引号
cleaned = cleaned.replaceAll("[“”]", "\"")
.replaceAll("[‘']", "'");
// 处理缩写
cleaned = cleaned.replaceAll("\\bDr\\.", "Dr")
.replaceAll("\\bMr\\.", "Mr")
.replaceAll("\\bMrs\\.", "Mrs")
.replaceAll("\\bMs\\.", "Ms")
.replaceAll("\\bvs\\.", "vs")
.replaceAll("\\betc\\.", "etc");
return cleaned;
}
// 语言特定的分块策略
public List<Document> chunkByLanguage(List<Document> documents, String language) {
switch (language) {
case "zh":
return chunkChineseText(documents);
case "ja":
return chunkJapaneseText(documents);
case "ko":
return chunkKoreanText(documents);
default:
return chunkEnglishText(documents);
}
}
private List<Document> chunkChineseText(List<Document> documents) {
// 中文分块策略：按段落和句子
List<Document> chunks = new ArrayList<>();
for (Document doc : documents) {
String content = doc.getContent();
// 先按段落分割
String[] paragraphs = content.split("\\n\\s*\\n");
for (String paragraph : paragraphs) {
if (paragraph.trim().isEmpty()) continue;
// 如果段落太长，再按句子分割
if (paragraph.length() > 1000) {
String[] sentences = paragraph.split("[。！？.!?]");
StringBuilder currentChunk = new StringBuilder();
for (String sentence : sentences) {
if (sentence.trim().isEmpty()) continue;
if (currentChunk.length() + sentence.length() > 800) {
if (currentChunk.length() > 0) {
chunks.add(new Document(currentChunk.toString(), doc.getMetadata()));
currentChunk = new StringBuilder();
}
}
if (currentChunk.length() > 0) {
currentChunk.append("。");
}
currentChunk.append(sentence);
}
if (currentChunk.length() > 0) {
chunks.add(new Document(currentChunk.toString(), doc.getMetadata()));
}
} else {
chunks.add(new Document(paragraph, doc.getMetadata()));
}
}
}
return chunks;
}
}3. 向量库批量导入、增量更新、过期数据清理
批量数据导入
1.1 使用 VectorStore 批量操作
@Service
public class VectorDataBatchService {
@Autowired
private VectorStore vectorStore;
@Autowired
private DocumentReader documentReader;
/**
* 批量导入文档
*/
public void batchImportDocuments(List<Document> documents, int batchSize) {
List<List<Document>> batches = partitionList(documents, batchSize);
for (List<Document> batch : batches) {
// 添加元数据
List<Document> enrichedBatch = batch.stream()
.map(doc -> {
Map<String, Object> metadata = new HashMap<>(doc.getMetadata());
metadata.put("import_time", System.currentTimeMillis());
metadata.put("batch_id", UUID.randomUUID().toString());
metadata.put("version", "1.0");
return new Document(doc.getId(), doc.getContent(), metadata);
})
.collect(Collectors.toList());
// 批量添加
vectorStore.add(enrichedBatch);
// 避免速率限制
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
/**
* 分批次处理
*/
private <T> List<List<T>> partitionList(List<T> list, int batchSize) {
List<List<T>> batches = new ArrayList<>();
for (int i = 0; i < list.size(); i += batchSize) {
batches.add(list.subList(i, Math.min(i + batchSize, list.size())));
}
return batches;
}
/**
* 从文件批量导入
*/
public void batchImportFromDirectory(String directoryPath) {
List<Document> documents = new ArrayList<>();
// 读取目录下所有文件
try (Stream<Path> paths = Files.walk(Paths.get(directoryPath))) {
paths.filter(Files::isRegularFile)
.forEach(filePath -> {
try {
String content = Files.readString(filePath);
Document doc = new Document(
UUID.randomUUID().toString(),
content,
Map.of(
"source", filePath.toString(),
"file_name", filePath.getFileName().toString(),
"import_time", System.currentTimeMillis()
)
);
documents.add(doc);
} catch (IOException e) {
log.error("Error reading file: {}", filePath, e);
}
});
} catch (IOException e) {
log.error("Error walking directory", e);
}
// 批量导入
batchImportDocuments(documents, 100);
}
}1.2 使用异步批量导入
@Service
@Slf4j
public class AsyncBatchImportService {
@Autowired
private VectorStore vectorStore;
private final ExecutorService executorService =
Executors.newFixedThreadPool(5);
/**
* 异步批量导入
*/
@Async
public CompletableFuture<Void> asyncBatchImport(
List<Document> documents,
String importSessionId) {
return CompletableFuture.runAsync(() -> {
log.info("Starting batch import session: {}", importSessionId);
AtomicInteger successCount = new AtomicInteger(0);
AtomicInteger failCount = new AtomicInteger(0);
List<List<Document>> batches = partitionList(documents, 50);
List<CompletableFuture<Void>> futures = batches.stream()
.map(batch -> CompletableFuture.runAsync(() -> {
try {
vectorStore.add(batch);
successCount.addAndGet(batch.size());
log.debug("Batch processed: {} documents", batch.size());
} catch (Exception e) {
failCount.addAndGet(batch.size());
log.error("Batch processing failed", e);
}
}, executorService))
.collect(Collectors.toList());
// 等待所有批次完成
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
.join();
log.info("Import session {} completed. Success: {}, Failed: {}",
importSessionId, successCount.get(), failCount.get());
});
}
/**
* 带进度监控的批量导入
*/
public void batchImportWithProgress(
List<Document> documents,
Consumer<ImportProgress> progressCallback) {
int total = documents.size();
AtomicInteger processed = new AtomicInteger(0);
List<List<Document>> batches = partitionList(documents, 50);
for (List<Document> batch : batches) {
vectorStore.add(batch);
processed.addAndGet(batch.size());
// 更新进度
progressCallback.accept(new ImportProgress(
processed.get(),
total,
(double) processed.get() / total * 100
));
}
}
@Data
@AllArgsConstructor
public static class ImportProgress {
private int processed;
private int total;
private double percentage;
}
}增量更新策略
2.1 基于时间戳的增量更新
@Service
@Slf4j
public class IncrementalUpdateService {
@Autowired
private VectorStore vectorStore;
@Autowired
private DocumentSourceService documentSource;
/**
* 增量更新 - 基于最后更新时间
*/
public void incrementalUpdateByTimestamp(Duration updateInterval) {
// 获取上次更新时间
Long lastUpdateTime = getLastUpdateTime();
Long currentTime = System.currentTimeMillis();
// 获取新增或修改的文档
List<Document> changedDocuments = documentSource
.getDocumentsModifiedAfter(lastUpdateTime);
if (!changedDocuments.isEmpty()) {
// 处理新增和更新
processChangedDocuments(changedDocuments, lastUpdateTime);
// 更新最后更新时间
updateLastUpdateTime(currentTime);
}
log.info("Incremental update completed. Changed documents: {}",
changedDocuments.size());
}
/**
* 处理变更的文档
*/
private void processChangedDocuments(
List<Document> changedDocuments,
Long lastUpdateTime) {
for (Document doc : changedDocuments) {
String documentId = doc.getId();
// 检查是否已存在
List<Document> existing = vectorStore.similaritySearch(
SearchRequest.defaults()
.withQuery(doc.getContent().substring(0, Math.min(100, doc.getContent().length())))
.withTopK(1)
.withFilterExpression(
"metadata.document_id == '" + documentId + "'"
)
);
if (!existing.isEmpty()) {
// 更新现有文档
updateExistingDocument(existing.get(0), doc);
} else {
// 添加新文档
addNewDocument(doc);
}
}
}
/**
* 基于内容哈希的增量更新
*/
public void incrementalUpdateByContentHash() {
// 获取所有源文档
List<Document> sourceDocuments = documentSource.getAllDocuments();
// 计算内容哈希
Map<String, String> sourceHashes = sourceDocuments.stream()
.collect(Collectors.toMap(
Document::getId,
doc -> calculateContentHash(doc.getContent())
));
// 获取向量库中现有文档的哈希
Map<String, String> existingHashes = getExistingDocumentHashes();
// 找出需要更新的文档
List<Document> documentsToUpdate = sourceDocuments.stream()
.filter(doc -> {
String sourceHash = sourceHashes.get(doc.getId());
String existingHash = existingHashes.get(doc.getId());
return existingHash == null || !existingHash.equals(sourceHash);
})
.collect(Collectors.toList());
// 批量更新
if (!documentsToUpdate.isEmpty()) {
batchUpdateDocuments(documentsToUpdate);
}
}
/**
* 计算内容哈希
*/
private String calculateContentHash(String content) {
try {
MessageDigest digest = MessageDigest.getInstance("SHA-256");
byte[] hash = digest.digest(content.getBytes(StandardCharsets.UTF_8));
return Base64.getEncoder().encodeToString(hash);
} catch (NoSuchAlgorithmException e) {
throw new RuntimeException("Hash algorithm not available", e);
}
}
/**
* 智能合并更新
*/
public void smartMergeUpdate(Document newDocument) {
// 查找相似文档
List<Document> similarDocs = vectorStore.similaritySearch(
SearchRequest.defaults()
.withQuery(newDocument.getContent())
.withTopK(3)
.withFilterExpression(
"metadata.document_type == '" +
newDocument.getMetadata().get("document_type") + "'"
)
);
if (similarDocs.isEmpty()) {
// 新增文档
vectorStore.add(List.of(newDocument));
} else {
// 合并或更新
Document mostSimilar = similarDocs.get(0);
double similarity = calculateSimilarity(
newDocument.getContent(),
mostSimilar.getContent()
);
if (similarity > 0.9) {
// 高度相似，更新现有文档
updateDocumentWithMerge(mostSimilar, newDocument);
} else {
// 新增文档
vectorStore.add(List.of(newDocument));
}
}
}
}2.2 基于变更数据捕获（CDC）的更新
@Service
@Slf4j
public class CDCUpdateService {
@Autowired
private VectorStore vectorStore;
@Autowired
private KafkaTemplate<String, DocumentChangeEvent> kafkaTemplate;
/**
* 监听文档变更事件
*/
@KafkaListener(topics = "document-changes")
public void handleDocumentChange(DocumentChangeEvent event) {
switch (event.getChangeType()) {
case CREATE:
handleCreate(event.getDocument());
break;
case UPDATE:
handleUpdate(event.getDocument());
break;
case DELETE:
handleDelete(event.getDocumentId());
break;
}
}
/**
* 处理文档创建
*/
private void handleCreate(Document document) {
// 添加元数据
Map<String, Object> metadata = new HashMap<>(document.getMetadata());
metadata.put("created_at", System.currentTimeMillis());
metadata.put("updated_at", System.currentTimeMillis());
metadata.put("version", 1);
Document enrichedDoc = new Document(
document.getId(),
document.getContent(),
metadata
);
vectorStore.add(List.of(enrichedDoc));
log.info("Document created: {}", document.getId());
}
/**
* 处理文档更新
*/
private void handleUpdate(Document document) {
// 先删除旧版本
vectorStore.delete(
List.of(document.getId()),
Optional.empty()
);
// 添加新版本
Map<String, Object> metadata = new HashMap<>(document.getMetadata());
metadata.put("updated_at", System.currentTimeMillis());
metadata.put("version",
((Integer) metadata.getOrDefault("version", 0)) + 1);
Document updatedDoc = new Document(
document.getId(),
document.getContent(),
metadata
);
vectorStore.add(List.of(updatedDoc));
log.info("Document updated: {}", document.getId());
}
/**
* 处理文档删除
*/
private void handleDelete(String documentId) {
vectorStore.delete(
List.of(documentId),
Optional.empty()
);
log.info("Document deleted: {}", documentId);
}
@Data
@AllArgsConstructor
@NoArgsConstructor
public static class DocumentChangeEvent {
private String documentId;
private Document document;
private ChangeType changeType;
private Long timestamp;
public enum ChangeType {
CREATE, UPDATE, DELETE
}
}
}过期数据清理
3.1 基于TTL的自动清理
@Service
@Slf4j
public class DataExpirationService {
@Autowired
private VectorStore vectorStore;
@Value("${vectorstore.data.ttl.days:30}")
private int dataTtlDays;
@Scheduled(cron = "0 0 2 * * ?") // 每天凌晨2点执行
public void cleanupExpiredData() {
log.info("Starting expired data cleanup...");
long cutoffTime = System.currentTimeMillis() -
TimeUnit.DAYS.toMillis(dataTtlDays);
// 查找过期数据
List<String> expiredIds = findExpiredDocumentIds(cutoffTime);
if (!expiredIds.isEmpty()) {
// 批量删除
int batchSize = 100;
for (int i = 0; i < expiredIds.size(); i += batchSize) {
List<String> batch = expiredIds.subList(
i, Math.min(i + batchSize, expiredIds.size())
);
vectorStore.delete(batch, Optional.empty());
log.info("Deleted batch of {} expired documents", batch.size());
// 避免过载
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
log.info("Cleanup completed. Total deleted: {}", expiredIds.size());
} else {
log.info("No expired documents found");
}
}
/**
* 查找过期文档ID
*/
private List<String> findExpiredDocumentIds(long cutoffTime) {
// 注意：这里需要根据具体的向量库实现来查询
// 以下为伪代码，实际实现取决于向量库的查询能力
List<String> expiredIds = new ArrayList<>();
// 示例：使用元数据过滤查找过期文档
// 实际实现可能需要分页查询所有文档
int pageSize = 1000;
int offset = 0;
boolean hasMore = true;
while (hasMore) {
// 模拟查询 - 实际需要根据向量库API调整
List<Document> documents = vectorStore.similaritySearch(
SearchRequest.defaults()
.withQuery("") // 空查询获取所有文档
.withTopK(pageSize)
.withFilterExpression("metadata.created_at < " + cutoffTime)
);
documents.stream()
.map(Document::getId)
.forEach(expiredIds::add);
hasMore = documents.size() == pageSize;
offset += pageSize;
}
return expiredIds;
}
/**
* 基于访问频率的智能清理
*/
@Scheduled(cron = "0 0 3 * * SUN") // 每周日凌晨3点执行
public void cleanupByAccessFrequency() {
log.info("Starting access-based cleanup...");
long thirtyDaysAgo = System.currentTimeMillis() -
TimeUnit.DAYS.toMillis(30);
long sevenDaysAgo = System.currentTimeMillis() -
TimeUnit.DAYS.toMillis(7);
// 获取文档访问统计
Map<String, DocumentAccessStats> accessStats =
getDocumentAccessStatistics();
List<String> toDelete = new ArrayList<>();
for (Map.Entry<String, DocumentAccessStats> entry : accessStats.entrySet()) {
DocumentAccessStats stats = entry.getValue();
// 规则1：超过30天未访问
if (stats.getLastAccessTime() < thirtyDaysAgo) {
toDelete.add(entry.getKey());
}
// 规则2：创建超过60天且最近7天访问次数少于3次
else if (stats.getCreatedTime() < thirtyDaysAgo * 2 &&
stats.getAccessCountLast7Days() < 3) {
toDelete.add(entry.getKey());
}
}
// 执行删除
if (!toDelete.isEmpty()) {
batchDeleteDocuments(toDelete);
log.info("Deleted {} documents based on access frequency", toDelete.size());
}
}
/**
* 版本化数据清理
*/
public void cleanupOldVersions(String documentId, int keepVersions) {
// 查找该文档的所有版本
List<Document> versions = findDocumentVersions(documentId);
if (versions.size() > keepVersions) {
// 按版本号排序，保留最新的几个版本
versions.sort((a, b) -> {
int versionA = (int) a.getMetadata().getOrDefault("version", 0);
int versionB = (int) b.getMetadata().getOrDefault("version", 0);
return Integer.compare(versionB, versionA); // 降序
});
// 删除旧版本
List<String> oldVersionIds = versions.stream()
.skip(keepVersions)
.map(Document::getId)
.collect(Collectors.toList());
vectorStore.delete(oldVersionIds, Optional.empty());
log.info("Deleted {} old versions of document {}",
oldVersionIds.size(), documentId);
}
}
@Data
public static class DocumentAccessStats {
private String documentId;
private long createdTime;
private long lastAccessTime;
private int totalAccessCount;
private int accessCountLast7Days;
private int accessCountLast30Days;
}
}4. 检索优化：相似度阈值、TopK筛选、重排序优化
Spring AI 中的检索优化是提升 RAG（检索增强生成）系统效果的关键环节。以下是针对相似度阈值、TopK筛选和重排序优化的详细方案：
一、相似度阈值优化
动态阈值策略
@Component
public class DynamicThresholdRetriever {
@Value("${retrieval.similarity.base-threshold:0.7}")
private double baseThreshold;
@Value("${retrieval.similarity.adaptive-factor:0.1}")
private double adaptiveFactor;
public List<Document> retrieveWithDynamicThreshold(
String query,
VectorStore vectorStore,
int maxResults) {
// 1. 先获取更多结果用于分析
List<Document> candidates = vectorStore.similaritySearch(
SearchRequest.query(query)
.withTopK(maxResults * 2)
.withSimilarityThreshold(0.3) // 低阈值获取更多候选
);
// 2. 计算动态阈值
double dynamicThreshold = calculateDynamicThreshold(candidates, query);
// 3. 应用阈值筛选
return candidates.stream()
.filter(doc -> doc.getMetadata()
.getOrDefault("similarity", 0.0) >= dynamicThreshold)
.limit(maxResults)
.collect(Collectors.toList());
}
private double calculateDynamicThreshold(List<Document> candidates, String query) {
if (candidates.isEmpty()) return baseThreshold;
// 方法1：基于分数分布
double maxScore = candidates.stream()
.mapToDouble(doc -> (Double) doc.getMetadata().get("similarity"))
.max().orElse(0.0);
// 方法2：基于查询复杂度（简单实现）
int queryLength = query.split(" ").length;
double lengthFactor = Math.min(1.0, queryLength / 20.0);
// 方法3：结合平均分
double avgScore = candidates.stream()
.limit(5)
.mapToDouble(doc -> (Double) doc.getMetadata().get("similarity"))
.average().orElse(0.0);
return Math.max(baseThreshold,
avgScore * 0.7 + maxScore * 0.3 - adaptiveFactor);
}
}分档阈值策略
# application.yml
retrieval:
thresholds:
high-precision: 0.85 # 高精度场景
balanced: 0.75 # 平衡场景
high-recall: 0.65 # 高召回场景
query-type-detection:
factual-threshold: 0.8 # 事实性问题
conceptual-threshold: 0.7 # 概念性问题二、TopK 优化策略
自适应 TopK
@Component
public class AdaptiveTopKRetriever {
private final QueryAnalyzer queryAnalyzer;
public List<Document> adaptiveRetrieve(String query, VectorStore vectorStore) {
// 1. 查询复杂度分析
QueryAnalysis analysis = queryAnalyzer.analyze(query);
// 2. 动态确定 TopK
int topK = determineTopK(analysis);
// 3. 分阶段检索
return retrieveWithAdaptiveStrategy(query, vectorStore, topK, analysis);
}
private int determineTopK(QueryAnalysis analysis) {
int baseK = 10;
// 基于查询类型调整
if (analysis.isComplex()) {
return baseK * 2; // 复杂查询需要更多上下文
}
// 基于查询长度调整
int wordCount = analysis.getWordCount();
if (wordCount > 15) {
return baseK + 5;
}
// 基于意图调整
switch (analysis.getIntent()) {
case "factual":
return baseK; // 事实性问题需要精确
case "comparative":
return baseK * 2; // 比较性问题需要更多资料
case "creative":
return baseK + 3; // 创意性问题适中
default:
return baseK;
}
}
private List<Document> retrieveWithAdaptiveStrategy(
String query,
VectorStore vectorStore,
int topK,
QueryAnalysis analysis) {
// 方法1：分步检索
if (analysis.isBroadTopic()) {
// 宽泛主题：先宽后精
List<Document> broadResults = vectorStore.similaritySearch(
SearchRequest.query(query).withTopK(topK * 2)
);
return refineResults(broadResults, query, topK);
}
// 方法2：直接精确检索
return vectorStore.similaritySearch(
SearchRequest.query(query)
.withTopK(topK)
.withSimilarityThreshold(0.7)
);
}
}混合检索策略
@Configuration
public class HybridRetrievalConfig {
@Bean
public Retriever hybridRetriever(
VectorStore vectorStore,
KeywordRetriever keywordRetriever) {
return query -> {
// 1. 向量检索
List<Document> vectorResults = vectorStore.similaritySearch(
SearchRequest.query(query).withTopK(8)
);
// 2. 关键词检索（BM25）
List<Document> keywordResults = keywordRetriever.retrieve(query, 8);
// 3. 结果融合
return fuseResults(vectorResults, keywordResults, query);
};
}
private List<Document> fuseResults(
List<Document> vectorResults,
List<Document> keywordResults,
String query) {
// Reciprocal Rank Fusion (RRF)
Map<String, DocumentScore> scoreMap = new HashMap<>();
// 计算 RRF 分数
addToScoreMap(vectorResults, scoreMap, 1);
addToScoreMap(keywordResults, scoreMap, 2);
// 排序并返回 TopK
return scoreMap.entrySet().stream()
.sorted((a, b) -> Double.compare(b.getValue().getScore(), a.getValue().getScore()))
.limit(10)
.map(entry -> entry.getValue().getDocument())
.collect(Collectors.toList());
}
}三、重排序优化
多阶段重排序
@Component
public class RerankingPipeline {
private final List<Reranker> rerankers;
private final CrossEncoderReranker crossEncoderReranker;
@Autowired
public RerankingPipeline(
@Qualifier("simpleReranker") Reranker simpleReranker,
@Qualifier("metadataReranker") Reranker metadataReranker,
CrossEncoderReranker crossEncoderReranker) {
this.rerankers = Arrays.asList(simpleReranker, metadataReranker);
this.crossEncoderReranker = crossEncoderReranker;
}
public List<Document> rerank(String query, List<Document> candidates) {
List<Document> results = new ArrayList<>(candidates);
// 阶段1：轻量级重排序
for (Reranker reranker : rerankers) {
results = reranker.rerank(query, results);
}
// 阶段2：精排（使用交叉编码器，计算量较大）
if (shouldUseCrossEncoder(query, results)) {
results = crossEncoderReranker.rerank(query, results.subList(0, 20));
}
return results;
}
private boolean shouldUseCrossEncoder(String query, List<Document> documents) {
// 只在必要时使用计算密集型的交叉编码器
if (documents.size() < 5) return false;
// 检查分数是否接近，需要精细排序
double scoreRange = calculateScoreRange(documents);
return scoreRange < 0.3; // 分数接近时需要精排
}
}基于语义和元数据的重排序
@Component
public class SemanticMetadataReranker implements Reranker {
@Override
public List<Document> rerank(String query, List<Document> documents) {
return documents.stream()
.sorted((doc1, doc2) -> {
double score1 = calculateCombinedScore(query, doc1);
double score2 = calculateCombinedScore(query, doc2);
return Double.compare(score2, score1);
})
.collect(Collectors.toList());
}
private double calculateCombinedScore(String query, Document document) {
double semanticScore = (Double) document.getMetadata()
.getOrDefault("similarity", 0.0);
double metadataScore = calculateMetadataScore(document);
double recencyScore = calculateRecencyScore(document);
double authorityScore = calculateAuthorityScore(document);
// 加权组合
return semanticScore * 0.5
+ metadataScore * 0.2
+ recencyScore * 0.2
+ authorityScore * 0.1;
}
private double calculateRecencyScore(Document document) {
Object dateObj = document.getMetadata().get("date");
if (dateObj instanceof LocalDateTime) {
LocalDateTime docDate = (LocalDateTime) dateObj;
long daysOld = ChronoUnit.DAYS.between(docDate, LocalDateTime.now());
return Math.max(0, 1.0 - daysOld / 365.0); // 一年内线性衰减
}
return 0.5; // 默认值
}
}使用交叉编码器精排
@Component
public class CrossEncoderReranker {
private final RestTemplate restTemplate;
private final String crossEncoderUrl;
public List<Document> rerank(String query, List<Document> documents) {
// 准备重排序请求
RerankRequest request = new RerankRequest(query, documents);
// 调用重排序服务（如 Cohere, Jina, 或本地部署的模型）
RerankResponse response = restTemplate.postForObject(
crossEncoderUrl,
request,
RerankResponse.class
);
// 按新分数排序
return response.getRerankedDocuments().stream()
.sorted(Comparator.comparingDouble(DocumentScore::getScore).reversed())
.map(DocumentScore::getDocument)
.collect(Collectors.toList());
}
// 本地轻量级交叉编码器（使用 ONNX 或 SentenceTransformers）
public List<Document> localRerank(String query, List<Document> documents) {
try (OnnxSession session = new OnnxSession("cross-encoder.onnx")) {
return documents.parallelStream()
.map(doc -> {
double score = session.score(query, doc.getContent());
doc.getMetadata().put("rerank_score", score);
return doc;
})
.sorted((d1, d2) -> Double.compare(
(Double) d2.getMetadata().get("rerank_score"),
(Double) d1.getMetadata().get("rerank_score")
))
.collect(Collectors.toList());
}
}
}四、完整检索管道配置
@Configuration
@Slf4j
public class RetrievalPipelineConfig {
@Bean
public RetrievalPipeline retrievalPipeline(
VectorStore vectorStore,
KeywordRetriever keywordRetriever,
RerankingPipeline rerankingPipeline,
@Value("${retrieval.strategy:hybrid}") String strategy) {
return new RetrievalPipeline() {
@Override
public List<Document> retrieve(String query, RetrievalOptions options) {
// 1. 初始检索
List<Document> initialResults = performInitialRetrieval(
query, vectorStore, keywordRetriever, strategy, options
);
// 2. 重排序
List<Document> rerankedResults = rerankingPipeline.rerank(
query, initialResults
);
// 3. 去重和过滤
List<Document> finalResults = deduplicateAndFilter(
rerankedResults, options
);
log.debug("Retrieved {} documents for query: {}",
finalResults.size(), query);
return finalResults;
}
};
}
private List<Document> performInitialRetrieval(
String query,
VectorStore vectorStore,
KeywordRetriever keywordRetriever,
String strategy,
RetrievalOptions options) {
switch (strategy.toLowerCase()) {
case "vector":
return vectorStore.similaritySearch(
SearchRequest.query(query)
.withTopK(options.getTopK())
.withSimilarityThreshold(options.getThreshold())
);
case "hybrid":
// 混合检索
List<Document> vectorResults = vectorStore.similaritySearch(
SearchRequest.query(query).withTopK(options.getTopK())
);
List<Document> keywordResults = keywordRetriever.retrieve(
query, options.getTopK()
);
return fuseResults(vectorResults, keywordResults);
case "multi-vector":
// 多向量检索（针对长文档）
return multiVectorRetrieval(query, vectorStore, options);
default:
throw new IllegalArgumentException("Unknown retrieval strategy: " + strategy);
}
}
}五、性能优化建议
缓存策略
@Component
@CacheConfig(cacheNames = "retrievalCache")
public class CachedRetriever {
@Cacheable(key = "#query.hashCode() + '|' + #options.hashCode()")
public List<Document> retrieveWithCache(String query, RetrievalOptions options) {
// 实际检索逻辑
return retrievalPipeline.retrieve(query, options);
}
@CacheEvict(allEntries = true)
public void clearCache() {
// 清理缓存
}
}异步处理
@Service
public class AsyncRetrievalService {
@Async("retrievalTaskExecutor")
public CompletableFuture<List<Document>> retrieveAsync(String query) {
return CompletableFuture.completedFuture(
retrievalPipeline.retrieve(query, defaultOptions())
);
}
@Bean("retrievalTaskExecutor")
public TaskExecutor retrievalTaskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(5);
executor.setMaxPoolSize(10);
executor.setQueueCapacity(100);
executor.setThreadNamePrefix("retrieval-");
return executor;
}
}六、监控和评估
@Component
@Slf4j
public class RetrievalMonitor {
private final MeterRegistry meterRegistry;
public void monitorRetrieval(String query, List<Document> results, long duration) {
// 记录指标
meterRegistry.timer("retrieval.duration").record(duration, TimeUnit.MILLISECONDS);
meterRegistry.gauge("retrieval.result.count", results.size());
// 计算质量指标
double avgScore = results.stream()
.mapToDouble(doc -> (Double) doc.getMetadata().get("similarity"))
.average()
.orElse(0.0);
meterRegistry.gauge("retrieval.avg_score", avgScore);
// 记录查询特征
log.info("Retrieval completed - Query: {}, Results: {}, Avg Score: {:.3f}, Duration: {}ms",
query, results.size(), avgScore, duration);
}
public void logRetrievalQuality(String query, List<Document> results, boolean wasHelpful) {
// 用于后续优化阈值和策略
RetrievalQualityMetric metric = new RetrievalQualityMetric(
query,
results.size(),
calculateScoreDistribution(results),
wasHelpful
);
// 存储到数据库或发送到监控系统
qualityMetricRepository.save(metric);
}
}配置示例
# application.yml
spring:
ai:
retrieval:
strategy: hybrid
top-k:
base: 10
adaptive: true
max: 50
similarity:
threshold: 0.75
adaptive: true
min-threshold: 0.5
reranking:
enabled: true
stages:
- type: metadata
weight: 0.3
- type: cross-encoder
enabled: true
model: cross-encoder/ms-marco-MiniLM-L-6-v2
top-n: 20
cache:
enabled: true
ttl: 300s
monitoring:
enabled: true
log-queries: true最佳实践建议
- 渐进式优化 ：从简单策略开始，逐步增加复杂度
- A/B测试 ：对比不同策略的效果
- 监控反馈 ：收集用户反馈调整参数
- 领域适配 ：根据具体领域调整权重和阈值
- 性能平衡 ：在效果和延迟之间找到平衡点
这些优化策略可以根据具体业务需求进行组合和调整，建议通过实验确定最适合您场景的参数配置。
5. 问答链路开发：检索知识库+大模型整合回答
我来详细讲解如何使用 Spring AI 实现知识库检索与大模型整合的问答系统。
一、整体架构
用户问题 → Spring AI → 向量检索 → 相关文档 → 大模型整合 → 最终答案二、核心组件配置
依赖配置 (pom.xml)
<dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-openai-spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-vector-store-postgresql</artifactId>
</dependency>
<!-- 或使用其他向量数据库 -->
<dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-pgvector-store</artifactId>
</dependency>
<dependency>
<artifactId>spring-ai-transformers-spring-boot-starter</artifactId>
<groupId>org.springframework.ai</groupId>
</dependency>应用配置 (application.yml)
spring:
ai:
openai:
api-key: ${OPENAI_API_KEY}
chat:
options:
model: gpt-4
temperature: 0.7
vectorstore:
postgresql:
enabled: true
initialize-schema: true
embedding:
openai:
enabled: true
options:
model: text-embedding-3-small三、核心实现代码
向量存储与文档处理
@Service
public class VectorStoreService {
@Autowired
private VectorStore vectorStore;
@Autowired
private EmbeddingModel embeddingModel;
/**
* 文档分块与向量化存储
*/
public void storeDocuments(List<Document> documents) {
// 文档分块
TextSplitter textSplitter = new TokenTextSplitter(
1000, // chunkSize
200, // chunkOverlap
true // keepSeparator
);
List<Document> chunks = textSplitter.split(documents);
// 创建文档向量
List<Document> embeddedDocs = chunks.stream()
.map(doc -> {
List<Double> embedding = embeddingModel.embed(doc.getContent());
doc.setEmbedding(embedding);
return doc;
})
.collect(Collectors.toList());
// 存储到向量数据库
vectorStore.add(embeddedDocs);
}
/**
* 相似度检索
*/
public List<Document> retrieveSimilarDocuments(String query, int topK) {
// 查询向量化
List<Double> queryEmbedding = embeddingModel.embed(query);
// 相似度检索
SearchRequest searchRequest = SearchRequest.query(query)
.withTopK(topK)
.withSimilarityThreshold(0.7); // 相似度阈值
return vectorStore.similaritySearch(searchRequest);
}
}检索增强生成 (RAG) 服务
@Service
public class RagService {
@Autowired
private ChatClient chatClient;
@Autowired
private VectorStoreService vectorStoreService;
/**
* RAG问答主流程
*/
public String answerWithRAG(String question) {
// 1. 检索相关文档
List<Document> relevantDocs = vectorStoreService
.retrieveSimilarDocuments(question, 5);
// 2. 构建上下文提示词
String context = buildContextFromDocuments(relevantDocs);
String prompt = buildPrompt(question, context);
// 3. 调用大模型生成答案
ChatResponse response = chatClient.call(
new Prompt(prompt)
);
return response.getResult().getOutput().getContent();
}
/**
* 构建上下文
*/
private String buildContextFromDocuments(List<Document> documents) {
StringBuilder context = new StringBuilder();
context.append("基于以下参考信息回答问题：\n\n");
for (int i = 0; i < documents.size(); i++) {
Document doc = documents.get(i);
context.append(String.format("[参考文档 %d]\n", i + 1));
context.append(doc.getContent());
context.append("\n\n");
}
return context.toString();
}
/**
* 构建提示词模板
*/
private String buildPrompt(String question, String context) {
return String.format("""
你是一个专业的问答助手，请根据提供的参考信息回答问题。
如果参考信息不足以回答问题，请如实告知。
参考信息：
%s
问题：%s
要求：
1. 基于参考信息回答问题
2. 保持回答简洁准确
3. 如果参考信息中没有相关内容，请说"根据现有信息无法回答"
4. 在回答末尾标注引用来源的文档编号
回答：
""", context, question);
}
/**
* 流式响应版本
*/
public Flux<String> answerWithRAGStream(String question) {
List<Document> relevantDocs = vectorStoreService
.retrieveSimilarDocuments(question, 5);
String context = buildContextFromDocuments(relevantDocs);
String prompt = buildPrompt(question, context);
return chatClient.stream(new Prompt(prompt))
.map(response -> response.getResult().getOutput().getContent());
}
}控制器层
@RestController
@RequestMapping("/api/rag")
public class RagController {
@Autowired
private RagService ragService;
@PostMapping("/answer")
public ResponseEntity<AnswerResponse> answer(@RequestBody QuestionRequest request) {
String answer = ragService.answerWithRAG(request.getQuestion());
return ResponseEntity.ok(new AnswerResponse(answer));
}
@GetMapping(value = "/answer/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> answerStream(@RequestParam String question) {
return ragService.answerWithRAGStream(question);
}
// DTO类
@Data
public static class QuestionRequest {
private String question;
}
@Data
public static class AnswerResponse {
private String answer;
private LocalDateTime timestamp;
public AnswerResponse(String answer) {
this.answer = answer;
this.timestamp = LocalDateTime.now();
}
}
}四、高级特性实现
多路检索与重排序
@Service
public class AdvancedRagService {
@Autowired
private List<RetrievalStrategy> retrievalStrategies;
@Autowired
private RerankerService rerankerService;
public String advancedRAG(String question) {
// 多路检索
List<Document> allDocuments = new ArrayList<>();
for (RetrievalStrategy strategy : retrievalStrategies) {
allDocuments.addAll(strategy.retrieve(question));
}
// 重排序
List<Document> rerankedDocs = rerankerService.rerank(question, allDocuments);
// 去重
List<Document> finalDocs = deduplicateDocuments(rerankedDocs);
// 上下文窗口管理
String context = manageContextWindow(finalDocs, 4000);
// 生成答案
return generateAnswer(question, context);
}
}对话历史管理
@Component
public class ConversationManager {
private final Map<String, List<Message>> conversationHistory = new ConcurrentHashMap<>();
public void addToHistory(String sessionId, String question, String answer) {
List<Message> history = conversationHistory
.computeIfAbsent(sessionId, k -> new ArrayList<>());
history.add(new Message("user", question));
history.add(new Message("assistant", answer));
// 限制历史长度
if (history.size() > 20) {
history = history.subList(history.size() - 10, history.size());
conversationHistory.put(sessionId, history);
}
}
public String buildPromptWithHistory(String sessionId, String question, String context) {
List<Message> history = conversationHistory.getOrDefault(sessionId, new ArrayList<>());
StringBuilder prompt = new StringBuilder();
prompt.append("对话历史：\n");
for (Message msg : history) {
prompt.append(msg.getRole()).append(": ").append(msg.getContent()).append("\n");
}
prompt.append("\n参考信息：\n").append(context);
prompt.append("\n\n当前问题：").append(question);
return prompt.toString();
}
}混合检索策略
@Service
public class HybridRetrievalService {
@Autowired
private VectorStore vectorStore;
@Autowired
private KeywordSearchService keywordSearchService;
public List<Document> hybridRetrieve(String query, int topK) {
// 向量检索
List<Document> vectorResults = vectorStore
.similaritySearch(SearchRequest.query(query).withTopK(topK));
// 关键词检索
List<Document> keywordResults = keywordSearchService.search(query, topK);
// 结果融合（RRF算法）
Map<String, Document> combined = new HashMap<>();
// 合并结果，计算综合分数
combineResults(vectorResults, keywordResults, combined);
return combined.values().stream()
.sorted(Comparator.comparing(Document::getScore).reversed())
.limit(topK)
.collect(Collectors.toList());
}
}五、性能优化建议
缓存策略
@Service
@Cacheable
public class CachedRagService {
@Cacheable(value = "ragAnswers", key = "#question.hashCode()")
public String getCachedAnswer(String question) {
return ragService.answerWithRAG(question);
}
@Cacheable(value = "documentEmbeddings", key = "#content.hashCode()")
public List<Double> getCachedEmbedding(String content) {
return embeddingModel.embed(content);
}
}异步处理
@Service
public class AsyncRagService {
@Async
public CompletableFuture<String> answerAsync(String question) {
return CompletableFuture.completedFuture(ragService.answerWithRAG(question));
}
public Flux<String> processBatch(List<String> questions) {
return Flux.fromIterable(questions)
.parallel()
.runOn(Schedulers.boundedElastic())
.flatMap(q -> Mono.fromCallable(() -> ragService.answerWithRAG(q)))
.sequential();
}
}六、监控与评估
@Component
public class RagMonitor {
private final MeterRegistry meterRegistry;
public void recordMetrics(String question, List<Document> retrievedDocs,
String answer, long latency) {
// 记录检索相关度
double relevanceScore = calculateRelevance(question, retrievedDocs);
meterRegistry.summary("rag.relevance").record(relevanceScore);
// 记录响应时间
meterRegistry.timer("rag.latency").record(latency, TimeUnit.MILLISECONDS);
// 记录答案长度
meterRegistry.summary("rag.answer.length").record(answer.length());
}
private double calculateRelevance(String question, List<Document> docs) {
// 实现相关度计算逻辑
return 0.0;
}
}七、部署配置
Docker Compose 配置
version: '3.8'
services:
postgres:
image: ankane/pgvector
environment:
POSTGRES_DB: ragdb
POSTGRES_USER: raguser
POSTGRES_PASSWORD: ragpass
ports:
- "5432:5432"
volumes:
- pgdata:/var/lib/postgresql/data
app:
build: .
depends_on:
- postgres
environment:
SPRING_DATASOURCE_URL: jdbc:postgresql://postgres:5432/ragdb
SPRING_AI_OPENAI_API_KEY: ${OPENAI_API_KEY}
ports:
- "8080:8080"八、最佳实践建议
- 文档预处理
- 清理HTML/特殊字符
- 统一编码格式
- 分段合理，保持语义完整性
- 检索优化
- 调整chunk大小（通常500-1500字符）
- 设置合理的重叠窗口
- 使用混合检索策略
- 提示工程
- 明确指令，避免歧义
- 提供格式要求
- 设置回答约束
- 错误处理
- 网络超时重试
- 限流控制
- 降级策略
这个完整的 Spring AI RAG 实现方案提供了从基础到高级的功能，可以根据具体需求进行调整和扩展。
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
13
0- 0
已为社区贡献21条内容
所有评论(0)