EasyFlow-Tech-Dev/Easy-Agents
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 修复知识库索引并发与向量化边界

Browse Source 
- 复用 Lucene 和 Elasticsearch 客户端并支持有界批量写入与删除

- 记录脱敏后的 Embedding 失败请求与完整响应

- 为 BGE-M3 分块统一增加上下文硬上限
This commit is contained in:
陈子默
2026-08-07 12:38:09 +08:00
parent bdb69a2250
commit f13e24751a
12 changed files with 1472 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files

246
easy-agents-rag/easy-agents-rag-ingestion/src/main/java/com/easyagents/rag/ingestion/chunk/RagSplitStrategyRegistry.java
View File

@@ -21,19 +21,19 @@ public class RagSplitStrategyRegistry {
strategyCode = analysisResult.getRecommendedStrategyCode();
}
String normalizedContent = analysisResult.getNormalizedContent();
List<RagChunk> chunks;
if (RagStrategyCodes.MARKDOWN_SECTION.equals(strategyCode)) {
return buildMarkdownChunks(normalizedContent, strategyConfig);
chunks = buildMarkdownChunks(normalizedContent, strategyConfig);
} else if (RagStrategyCodes.OUTLINE_SECTION.equals(strategyCode)) {
chunks = buildOutlineChunks(normalizedContent, strategyConfig);
} else if (RagStrategyCodes.QA_PAIR.equals(strategyCode)) {
chunks = buildQaChunks(normalizedContent, strategyConfig);
} else if (RagStrategyCodes.CUSTOM_REGEX.equals(strategyCode)) {
chunks = buildRegexChunks(normalizedContent, strategyConfig);
} else {
chunks = buildParagraphChunks(normalizedContent, strategyConfig);
}
if (RagStrategyCodes.OUTLINE_SECTION.equals(strategyCode)) {
return buildOutlineChunks(normalizedContent, strategyConfig);
}
if (RagStrategyCodes.QA_PAIR.equals(strategyCode)) {
return buildQaChunks(normalizedContent, strategyConfig);
}
if (RagStrategyCodes.CUSTOM_REGEX.equals(strategyCode)) {
return buildRegexChunks(normalizedContent, strategyConfig);
}
return buildParagraphChunks(normalizedContent, strategyConfig);
return postProcess(enforceBgeM3HardLimit(chunks));
}
private List<RagChunk> buildMarkdownChunks(String content, StrategyConfig strategyConfig) {
@@ -143,7 +143,7 @@ public class RagSplitStrategyRegistry {
));
}
}
return postProcess(result);
return result;
}
private List<RagChunk> buildQaChunks(String content, StrategyConfig strategyConfig) {
@@ -162,7 +162,14 @@ public class RagSplitStrategyRegistry {
Matcher questionMatcher = QUESTION_PREFIX.matcher(line);
Matcher answerMatcher = ANSWER_PREFIX.matcher(line);
if (questionMatcher.matches()) {
qaIndex = flushQaChunk(result, currentQuestion, questionSlices, answerSlices, qaIndex, strategyConfig);
qaIndex = flushQaChunk(
result,
content,
currentQuestion,
questionSlices,
answerSlices,
qaIndex,
strategyConfig);
currentQuestion = questionMatcher.group(2).trim();
questionSlices = new ArrayList<LineSlice>();
answerSlices = new ArrayList<LineSlice>();
@@ -187,11 +194,19 @@ public class RagSplitStrategyRegistry {
questionSlices.add(lineSlice);
}
}
flushQaChunk(result, currentQuestion, questionSlices, answerSlices, qaIndex, strategyConfig);
return postProcess(result);
flushQaChunk(
result,
content,
currentQuestion,
questionSlices,
answerSlices,
qaIndex,
strategyConfig);
return result;
}
private int flushQaChunk(List<RagChunk> result,
String content,
String currentQuestion,
List<LineSlice> questionSlices,
List<LineSlice> answerSlices,
@@ -203,27 +218,59 @@ public class RagSplitStrategyRegistry {
if (answerSlices == null || answerSlices.isEmpty()) {
return qaIndex;
}
String question = joinLineSlices(questionSlices);
String answer = joinLineSlices(answerSlices);
String baseContent = "问题：" + question + "\n答案：" + answer;
List<String> subContents = baseContent.length() > safeChunkSize(strategyConfig)
? splitLongContent(baseContent, strategyConfig.getChunkSize())
: Collections.singletonList(baseContent);
int total = subContents.size();
List<TextRange> sourceRanges = buildQaSourceRanges(questionSlices, answerSlices);
for (int i = 0; i < subContents.size(); i++) {
TextRange questionRange = mergeLineSlices(questionSlices);
TextRange rawAnswerRange = mergeLineSlices(answerSlices);
TextRange answerRange = rawAnswerRange == null
? null
: trimRange(content, rawAnswerRange.start, rawAnswerRange.end);
if (questionRange == null || answerRange == null) {
return qaIndex;
}
String question = currentQuestion.trim();
String answer = content.substring(answerRange.start, answerRange.end);
String qaPrefix = "问题：" + question + "\n答案：";
int requestedTokens = Math.min(
safeChunkSize(strategyConfig),
RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT);
int prefixTokens = BgeM3ChunkSafety.estimateContentTokens(qaPrefix);
boolean includePrefix = prefixTokens
< RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT;
int answerTokenLimit = includePrefix
? Math.max(1, Math.max(requestedTokens, prefixTokens + 1) - prefixTokens)
: RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT;
List<TextRange> answerPartRanges = new ArrayList<TextRange>();
for (BgeM3ChunkSafety.ChunkRange answerPart
: BgeM3ChunkSafety.splitToTokenLimit(answer, answerTokenLimit)) {
TextRange absoluteAnswerRange = trimRange(
content,
answerRange.start + answerPart.getStart(),
answerRange.start + answerPart.getEnd());
if (absoluteAnswerRange != null) {
answerPartRanges.add(absoluteAnswerRange);
}
}
int total = answerPartRanges.size();
for (int i = 0; i < answerPartRanges.size(); i++) {
TextRange absoluteAnswerRange = answerPartRanges.get(i);
String answerFragment = content.substring(
absoluteAnswerRange.start,
absoluteAnswerRange.end);
List<TextRange> sourceRanges = new ArrayList<TextRange>();
sourceRanges.add(questionRange);
sourceRanges.add(absoluteAnswerRange);
RagChunk chunk = createChunk(
RagChunkTypes.QA_PAIR,
"Q" + qaIndex + " " + question,
Collections.<String>emptyList(),
subContents.get(i),
includePrefix ? qaPrefix + answerFragment : answerFragment,
result.size() + 1,
i + 1,
total,
sourceRanges
);
chunk.setQuestion(question);
chunk.setAnswer(answer);
chunk.setAnswer(answerFragment);
chunk.getOptions().put(RagMetadataKeys.QA_GROUP_ID, "qa-" + qaIndex);
result.add(chunk);
}
@@ -254,7 +301,7 @@ public class RagSplitStrategyRegistry {
));
index++;
}
return postProcess(result);
return result;
}
private List<RagChunk> buildRegexChunks(String content, StrategyConfig strategyConfig) {
@@ -264,42 +311,129 @@ public class RagSplitStrategyRegistry {
Pattern pattern = Pattern.compile(regex);
Matcher matcher = pattern.matcher(content);
int segmentStart = 0;
boolean retainRegexMatch = Boolean.TRUE.equals(
strategyConfig.getRetainRegexMatch());
while (matcher.find()) {
index = addRegexChunk(result, content, segmentStart, matcher.start(), index);
segmentStart = matcher.end();
// 保留时将匹配内容归入下一分块；默认继续从匹配结束位置开始。
segmentStart = retainRegexMatch ? matcher.start() : matcher.end();
}
addRegexChunk(result, content, segmentStart, content.length(), index);
return postProcess(result);
return result;
}
private List<String> splitLongContent(String content, Integer chunkSize) {
int size = chunkSize == null || chunkSize.intValue() <= 0 ? RagDefaults.CHUNK_SIZE : chunkSize.intValue();
String[] paragraphs = content.split("\\n\\s*\\n");
int size = chunkSize == null || chunkSize.intValue() <= 0
? RagDefaults.CHUNK_SIZE
: Math.min(chunkSize.intValue(), RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT);
List<String> parts = new ArrayList<String>();
StringBuilder current = new StringBuilder();
for (String paragraph : paragraphs) {
String text = paragraph.trim();
if (!StringUtil.hasText(text)) {
continue;
for (BgeM3ChunkSafety.ChunkRange range
: BgeM3ChunkSafety.splitToTokenLimit(content, size)) {
String part = content.substring(range.getStart(), range.getEnd()).trim();
if (StringUtil.hasText(part)) {
parts.add(part);
}
if (current.length() > 0 && current.length() + text.length() + 2 > size) {
parts.add(current.toString().trim());
current = new StringBuilder();
}
if (current.length() > 0) {
current.append("\n\n");
}
current.append(text);
}
if (current.length() > 0) {
parts.add(current.toString().trim());
}
if (parts.isEmpty()) {
parts.add(content);
}
return parts;
}
/**
* Applies the shared BGE-M3 context safety guard after strategy-specific splitting.
*
* @param chunks chunks produced by a semantic or length strategy
* @return chunks whose conservative token estimates stay within the hard limit
*/
private List<RagChunk> enforceBgeM3HardLimit(List<RagChunk> chunks) {
List<RagChunk> result = new ArrayList<RagChunk>();
for (RagChunk chunk : chunks) {
String content = chunk.getContent();
if (BgeM3ChunkSafety.isWithinHardLimit(content)) {
result.add(chunk);
continue;
}
List<BgeM3ChunkSafety.ChunkRange> parts =
BgeM3ChunkSafety.splitToHardLimit(content);
for (int i = 0; i < parts.size(); i++) {
BgeM3ChunkSafety.ChunkRange range = parts.get(i);
result.add(copyHardSplitChunk(
chunk,
content,
new TextRange(range.getStart(), range.getEnd()),
i + 1,
parts.size()));
}
}
return result;
}
/**
* Copies a strategy chunk while replacing only content and hard-split metadata.
*
* @param source original strategy chunk
* @param sourceContent original chunk content
* @param range selected content range
* @param partNo forced part number
* @param partTotal forced part count
* @return copied hard-split chunk
*/
private RagChunk copyHardSplitChunk(RagChunk source,
String sourceContent,
TextRange range,
int partNo,
int partTotal) {
RagChunk copy = new RagChunk();
copy.setChunkType(source.getChunkType());
copy.setSourceLabel(source.getSourceLabel());
copy.setHeadingPath(new ArrayList<String>(source.getHeadingPath()));
copy.setContent(sourceContent.substring(range.start, range.end));
copy.setQuestion(source.getQuestion());
copy.setAnswer(RagChunkTypes.QA_PAIR.equals(source.getChunkType())
? copy.getContent()
: source.getAnswer());
copy.setPartNo(Integer.valueOf(partNo));
copy.setPartTotal(Integer.valueOf(partTotal));
copy.setWarnings(new ArrayList<String>(source.getWarnings()));
copy.setOptions(new LinkedHashMap<String, Object>(source.getOptions()));
copy.getOptions().put("hardSplit", Boolean.TRUE);
copy.getOptions().put("hardSplitTokenLimit",
Integer.valueOf(RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT));
adjustSingleSourceRange(copy, sourceContent.length(), range);
return copy;
}
/**
* Narrows an exact single source range to the forced subrange.
*
* @param chunk copied hard-split chunk
* @param sourceLength original chunk content length
* @param relativeRange forced range relative to the original chunk
*/
@SuppressWarnings("unchecked")
private void adjustSingleSourceRange(RagChunk chunk, int sourceLength, TextRange relativeRange) {
Object rawRanges = chunk.getOptions().get(RagMetadataKeys.SOURCE_RANGES);
if (!(rawRanges instanceof List) || ((List<?>) rawRanges).size() != 1) {
return;
}
Object rawRange = ((List<?>) rawRanges).get(0);
if (!(rawRange instanceof Map)) {
return;
}
Map<String, Object> sourceRange = (Map<String, Object>) rawRange;
Object rawStart = sourceRange.get("start");
Object rawEnd = sourceRange.get("end");
if (!(rawStart instanceof Number) || !(rawEnd instanceof Number)) {
return;
}
int sourceStart = ((Number) rawStart).intValue();
int sourceEnd = ((Number) rawEnd).intValue();
if (sourceEnd - sourceStart != sourceLength) {
return;
}
chunk.getOptions().put(RagMetadataKeys.SOURCE_RANGES,
toSourceRangeMaps(Collections.singletonList(
new TextRange(sourceStart + relativeRange.start, sourceStart + relativeRange.end))));
}
private List<RagChunk> postProcess(List<RagChunk> chunks) {
List<RagChunk> result = new ArrayList<RagChunk>();
Set<String> dedup = new HashSet<String>();
@@ -318,7 +452,8 @@ public class RagSplitStrategyRegistry {
}
chunk.setChunkId("chunk-" + index);
chunk.setCharCount(Integer.valueOf(content.length()));
chunk.setTokenEstimate(Integer.valueOf(Math.max(1, content.length() / 4)));
chunk.setTokenEstimate(Integer.valueOf(
Math.max(1, BgeM3ChunkSafety.estimateContentTokens(content))));
result.add(chunk);
index++;
}
@@ -460,7 +595,14 @@ public class RagSplitStrategyRegistry {
private int safeOverlap(StrategyConfig strategyConfig) {
Integer overlapSize = strategyConfig.getOverlapSize();
return overlapSize == null || overlapSize.intValue() < 0 ? RagDefaults.OVERLAP_SIZE : overlapSize.intValue();
int overlap = overlapSize == null || overlapSize.intValue() < 0
? RagDefaults.OVERLAP_SIZE
: overlapSize.intValue();
int chunkSize = safeChunkSize(strategyConfig);
if (overlap >= chunkSize) {
throw new IllegalArgumentException("overlapSize must be smaller than chunkSize");
}
return overlap;
}
private String joinAndTrim(List<String> lines) {

24
easy-agents-rag/easy-agents-rag-ingestion/src/main/java/com/easyagents/rag/ingestion/model/StrategyConfig.java
View File

@@ -11,6 +11,10 @@ public class StrategyConfig implements Serializable {
private Integer chunkSize = RagDefaults.CHUNK_SIZE;
private Integer overlapSize = RagDefaults.OVERLAP_SIZE;
private String regex;
/**
* Whether a custom-regex match is retained at the beginning of the next chunk.
*/
private Boolean retainRegexMatch;
private Integer rowsPerChunk = RagDefaults.ROWS_PER_CHUNK;
private Integer mdSplitterLevel = RagDefaults.MD_SPLITTER_LEVEL;
@@ -24,6 +28,7 @@ public class StrategyConfig implements Serializable {
copy.setChunkSize(this.chunkSize);
copy.setOverlapSize(this.overlapSize);
copy.setRegex(this.regex);
copy.setRetainRegexMatch(this.retainRegexMatch);
copy.setRowsPerChunk(this.rowsPerChunk);
copy.setMdSplitterLevel(this.mdSplitterLevel);
return copy;
@@ -61,6 +66,25 @@ public class StrategyConfig implements Serializable {
this.regex = regex;
}
/**
* Returns whether custom-regex matches are retained in the next chunk.
*
* @return {@code true} to retain each match at the beginning of the next chunk;
* otherwise {@code false} or {@code null}
*/
public Boolean getRetainRegexMatch() {
return retainRegexMatch;
}
/**
* Sets whether custom-regex matches are retained in the next chunk.
*
* @param retainRegexMatch {@code true} to retain each match; {@code false} to discard it
*/
public void setRetainRegexMatch(Boolean retainRegexMatch) {
this.retainRegexMatch = retainRegexMatch;
}
public Integer getRowsPerChunk() {
return rowsPerChunk;
}

152
easy-agents-rag/easy-agents-rag-ingestion/src/test/java/com/easyagents/rag/ingestion/RagIngestionPipelineTest.java
View File

@@ -2,6 +2,7 @@ package com.easyagents.rag.ingestion;
import com.easyagents.rag.core.RagChunk;
import com.easyagents.rag.core.RagChunkTypes;
import com.easyagents.rag.core.RagDefaults;
import com.easyagents.rag.core.RagStrategyCodes;
import com.easyagents.rag.ingestion.analysis.DocumentStructureAnalyzer;
import com.easyagents.rag.ingestion.chunk.RagSplitStrategyRegistry;
@@ -97,6 +98,157 @@ public class RagIngestionPipelineTest {
assertHasValidSourceRanges(analysis, chunks.get(0));
}
@Test
public void shouldEnforceBgeM3HardLimitForAutoQaTxt() {
StringBuilder qa = new StringBuilder("问：自动导入为什么失败？\n答：");
int i = 0;
while (qa.length() < 77946) {
qa.append((char) ('一' + i % 20));
if (i > 0 && i % 700 == 0) {
qa.append('。');
}
i++;
}
qa.setLength(77946);
AnalysisResult analysis = recommender.recommend(analyzer.analyze(qa.toString(), "txt"));
StrategyConfig config = StrategyConfig.defaults();
config.setStrategyCode(RagStrategyCodes.AUTO);
List<RagChunk> chunks = registry.split(analysis, config);
Assert.assertEquals(RagStrategyCodes.QA_PAIR, analysis.getRecommendedStrategyCode());
Assert.assertEquals(77946, analysis.getNormalizedContent().length());
Assert.assertTrue(chunks.size() > 1);
assertWithinBgeM3HardLimit(chunks);
Assert.assertTrue(chunks.stream().allMatch(
chunk -> chunk.getTokenEstimate().intValue() <= RagDefaults.CHUNK_SIZE));
Assert.assertTrue(chunks.stream().allMatch(
chunk -> chunk.getAnswer() != null
&& chunk.getAnswer().length() < analysis.getNormalizedContent().length()));
assertHasValidSourceRanges(analysis, chunks.get(0));
}
@Test
public void shouldEnforceBgeM3HardLimitForCustomRegexChunk() {
StringBuilder content = new StringBuilder();
for (int i = 0; i < 20000; i++) {
content.append((char) ('甲' + i % 16));
if (i > 0 && i % 997 == 0) {
content.append('；');
}
}
AnalysisResult analysis = recommender.recommend(analyzer.analyze(content.toString(), "txt"));
StrategyConfig config = StrategyConfig.defaults();
config.setStrategyCode(RagStrategyCodes.CUSTOM_REGEX);
config.setRegex("\\|NEVER_MATCH\\|");
List<RagChunk> chunks = registry.split(analysis, config);
Assert.assertTrue(chunks.size() > 1);
assertWithinBgeM3HardLimit(chunks);
assertHasValidSourceRanges(analysis, chunks.get(0));
}
/**
* Verifies that custom-regex matches remain discarded when the new option is omitted.
*/
@Test
public void shouldDiscardCustomRegexMatchByDefault() {
AnalysisResult analysis = new AnalysisResult();
analysis.setNormalizedContent(
"<QUESTION>第一个问题及其答案内容。<QUESTION>第二个问题及其答案内容。");
StrategyConfig config = StrategyConfig.defaults();
config.setStrategyCode(RagStrategyCodes.CUSTOM_REGEX);
config.setRegex("<QUESTION>");
List<RagChunk> chunks = registry.split(analysis, config);
Assert.assertEquals(2, chunks.size());
Assert.assertEquals("第一个问题及其答案内容。", chunks.get(0).getContent());
Assert.assertEquals("第二个问题及其答案内容。", chunks.get(1).getContent());
}
/**
* Verifies that an enabled option retains each match at the beginning of the next chunk.
*/
@Test
public void shouldRetainCustomRegexMatchInNextChunk() {
AnalysisResult analysis = new AnalysisResult();
analysis.setNormalizedContent(
"<QUESTION>第一个问题及其答案内容。<QUESTION>第二个问题及其答案内容。");
StrategyConfig config = StrategyConfig.defaults();
config.setStrategyCode(RagStrategyCodes.CUSTOM_REGEX);
config.setRegex("<QUESTION>");
config.setRetainRegexMatch(Boolean.TRUE);
List<RagChunk> chunks = registry.split(analysis, config);
Assert.assertEquals(2, chunks.size());
Assert.assertEquals(
"<QUESTION>第一个问题及其答案内容。",
chunks.get(0).getContent());
Assert.assertEquals(
"<QUESTION>第二个问题及其答案内容。",
chunks.get(1).getContent());
assertHasValidSourceRanges(analysis, chunks.get(1));
}
@Test
public void shouldUseChineseAwareTokenEstimate() {
String content = "知识库自动导入向量化失败，需要检查分块大小和模型上下文窗口。";
AnalysisResult analysis = recommender.recommend(analyzer.analyze(content, "txt"));
StrategyConfig config = StrategyConfig.defaults();
config.setStrategyCode(RagStrategyCodes.PARAGRAPH_LENGTH);
List<RagChunk> chunks = registry.split(analysis, config);
Assert.assertEquals(Integer.valueOf(content.codePointCount(0, content.length())),
chunks.get(0).getTokenEstimate());
}
@Test
public void shouldRejectOverlapEqualToChunkSize() {
AnalysisResult analysis = recommender.recommend(
analyzer.analyze("知识库分块参数需要保证游标持续向前推进。", "txt"));
StrategyConfig config = StrategyConfig.defaults();
config.setStrategyCode(RagStrategyCodes.PARAGRAPH_LENGTH);
config.setChunkSize(128);
config.setOverlapSize(128);
try {
registry.split(analysis, config);
Assert.fail("overlapSize 等于 chunkSize 时应拒绝分块");
} catch (IllegalArgumentException expected) {
Assert.assertTrue(expected.getMessage().contains("overlapSize"));
}
}
@Test
public void shouldRejectOverlapGreaterThanChunkSize() {
AnalysisResult analysis = recommender.recommend(
analyzer.analyze("知识库分块参数需要保证游标持续向前推进。", "txt"));
StrategyConfig config = StrategyConfig.defaults();
config.setStrategyCode(RagStrategyCodes.PARAGRAPH_LENGTH);
config.setChunkSize(128);
config.setOverlapSize(512);
try {
registry.split(analysis, config);
Assert.fail("overlapSize 大于 chunkSize 时应拒绝分块");
} catch (IllegalArgumentException expected) {
Assert.assertTrue(expected.getMessage().contains("overlapSize"));
}
}
private void assertWithinBgeM3HardLimit(List<RagChunk> chunks) {
for (RagChunk chunk : chunks) {
Assert.assertNotNull(chunk.getTokenEstimate());
Assert.assertTrue(
"chunk token estimate exceeds BGE-M3 hard limit: " + chunk.getTokenEstimate(),
chunk.getTokenEstimate().intValue() <= RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT);
}
}
@SuppressWarnings("unchecked")
private void assertHasValidSourceRanges(AnalysisResult analysis, RagChunk chunk) {
Object rawRanges = chunk.getOptions().get("sourceRanges");