mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 19:23:42 +08:00
feat(wiki): semantic hybrid search + chunk persistence + deep research pipeline
This commit is contained in:
matevip
parent
a6e9a17208
commit
1fdf87b31e
@ -91,4 +91,15 @@ public class WikiProperties {
|
||||
* 默认 true（M2 上线）；遇问题可在 application.yml 配 mate.wiki.use-two-phase-digest=false 回退到旧行为。
|
||||
*/
|
||||
private boolean useTwoPhaseDigest = true;
|
||||
|
||||
// ==================== RFC-011: Embedding ====================
|
||||
|
||||
/** 嵌入模型名称（DashScope） */
|
||||
private String embeddingModel = "text-embedding-v3";
|
||||
|
||||
/** 嵌入批量大小（一次 API 调用处理多少 chunk） */
|
||||
private int embeddingBatchSize = 16;
|
||||
|
||||
/** 混合搜索默认模式：keyword / semantic / hybrid */
|
||||
private String searchDefaultMode = "hybrid";
|
||||
}
|
||||
|
||||
@ -0,0 +1,118 @@
|
||||
package vip.mate.wiki.controller;
|
||||
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import io.swagger.v3.oas.annotations.tags.Tag;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.web.bind.annotation.*;
|
||||
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
|
||||
import vip.mate.channel.web.ChatStreamTracker;
|
||||
import vip.mate.common.result.R;
|
||||
import vip.mate.wiki.service.WikiKnowledgeBaseService;
|
||||
import vip.mate.wiki.service.WikiResearchService;
|
||||
import vip.mate.workspace.core.annotation.RequireWorkspaceRole;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
|
||||
/**
|
||||
* RFC-011 Phase 3: Wiki Deep Research REST + SSE 接口
|
||||
*
|
||||
* @author MateClaw Team
|
||||
*/
|
||||
@Slf4j
|
||||
@Tag(name = "Wiki Deep Research")
|
||||
@RestController
|
||||
@RequestMapping("/api/v1/wiki/research")
|
||||
@RequiredArgsConstructor
|
||||
public class WikiResearchController {
|
||||
|
||||
private final WikiResearchService researchService;
|
||||
private final WikiKnowledgeBaseService kbService;
|
||||
private final ChatStreamTracker streamTracker;
|
||||
|
||||
private static final ExecutorService RESEARCH_EXEC = Executors.newVirtualThreadPerTaskExecutor();
|
||||
|
||||
/**
|
||||
* 启动 research。返回 sessionId，前端用它订阅 SSE 流。
|
||||
*/
|
||||
@RequireWorkspaceRole("member")
|
||||
@Operation(summary = "启动 Deep Research，返回 SSE sessionId")
|
||||
@PostMapping("/start")
|
||||
public R<Map<String, Object>> startResearch(
|
||||
@RequestBody Map<String, Object> body,
|
||||
@RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) {
|
||||
|
||||
Long kbId = body.get("kbId") != null ? Long.valueOf(body.get("kbId").toString()) : null;
|
||||
String topic = (String) body.get("topic");
|
||||
Integer topK = body.get("topKPerQuestion") != null
|
||||
? Integer.valueOf(body.get("topKPerQuestion").toString()) : null;
|
||||
|
||||
if (kbId == null || topic == null || topic.isBlank()) {
|
||||
return R.fail("kbId and topic are required");
|
||||
}
|
||||
if (kbService.getById(kbId) == null) {
|
||||
return R.fail("Knowledge base not found");
|
||||
}
|
||||
|
||||
// 生成 SSE 会话 ID
|
||||
String sessionId = "research-" + UUID.randomUUID();
|
||||
streamTracker.register(sessionId);
|
||||
|
||||
// 异步跑 research，事件通过 streamTracker 推送
|
||||
// 【Review Bug 4】register 后需要 incrementFlux 配平，否则 complete 永远不会清理 RunState
|
||||
streamTracker.incrementFlux(sessionId);
|
||||
RESEARCH_EXEC.submit(() -> {
|
||||
try {
|
||||
researchService.research(kbId, topic, sessionId, topK);
|
||||
} catch (Exception e) {
|
||||
log.error("[ResearchController] Execution failed for sessionId={}: {}", sessionId, e.getMessage(), e);
|
||||
} finally {
|
||||
// 先发结束标记，让前端关闭 EventSource
|
||||
try {
|
||||
streamTracker.broadcast(sessionId, "done", "{}");
|
||||
} catch (Exception ignored) {}
|
||||
// 然后清理 RunState（递减 flux count，所有 flux 完成时自动 remove）
|
||||
try {
|
||||
streamTracker.complete(sessionId);
|
||||
} catch (Exception ignored) {}
|
||||
}
|
||||
});
|
||||
|
||||
return R.ok(Map.of(
|
||||
"sessionId", sessionId,
|
||||
"kbId", kbId,
|
||||
"topic", topic,
|
||||
"streamUrl", "/api/v1/wiki/research/stream/" + sessionId
|
||||
));
|
||||
}
|
||||
|
||||
/**
|
||||
* SSE 端点：订阅指定 sessionId 的 research 事件流
|
||||
*/
|
||||
@RequireWorkspaceRole("viewer")
|
||||
@Operation(summary = "订阅 Deep Research SSE 事件流")
|
||||
@GetMapping(value = "/stream/{sessionId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
|
||||
public SseEmitter stream(@PathVariable String sessionId) {
|
||||
// 10 分钟超时（research 典型 < 1 分钟，10 分钟给重连留余地）
|
||||
SseEmitter emitter = new SseEmitter(10 * 60 * 1000L);
|
||||
|
||||
boolean attached = streamTracker.attach(sessionId, emitter);
|
||||
if (!attached) {
|
||||
try {
|
||||
emitter.send(SseEmitter.event().name("error")
|
||||
.data("{\"message\":\"session not found or already ended\"}"));
|
||||
emitter.complete();
|
||||
} catch (Exception ignored) {}
|
||||
}
|
||||
|
||||
emitter.onCompletion(() -> streamTracker.detach(sessionId, emitter));
|
||||
emitter.onTimeout(() -> streamTracker.detach(sessionId, emitter));
|
||||
emitter.onError(err -> streamTracker.detach(sessionId, emitter));
|
||||
|
||||
return emitter;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,62 @@
|
||||
package vip.mate.wiki.model;
|
||||
|
||||
import com.baomidou.mybatisplus.annotation.*;
|
||||
import lombok.Data;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* Wiki chunk 实体
|
||||
* <p>
|
||||
* RFC-013 最小切片：持久化 splitIntoChunks() 的产物，为后续 embedding (RFC-011)、
|
||||
* citation、chunk 级增量处理提供基础。
|
||||
*
|
||||
* @author MateClaw Team
|
||||
*/
|
||||
@Data
|
||||
@TableName("mate_wiki_chunk")
|
||||
public class WikiChunkEntity {
|
||||
|
||||
@TableId(type = IdType.ASSIGN_ID)
|
||||
private Long id;
|
||||
|
||||
/** 所属知识库 ID */
|
||||
private Long kbId;
|
||||
|
||||
/** 来源原始材料 ID */
|
||||
private Long rawId;
|
||||
|
||||
/** chunk 在材料内的序号（0-based） */
|
||||
private Integer ordinal;
|
||||
|
||||
/** chunk 文本内容 */
|
||||
@TableField(updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String content;
|
||||
|
||||
/** 字符数 */
|
||||
private Integer charCount;
|
||||
|
||||
/** 在原始文本中的起始偏移 */
|
||||
private Integer startOffset;
|
||||
|
||||
/** 在原始文本中的结束偏移 */
|
||||
private Integer endOffset;
|
||||
|
||||
/** 内容 SHA-256 哈希（增量处理依据） */
|
||||
private String contentHash;
|
||||
|
||||
/** RFC-011：向量 embedding（float32[] little-endian 序列化） */
|
||||
private byte[] embedding;
|
||||
|
||||
/** RFC-011：生成该 embedding 的模型名称（切模型时需全量重嵌） */
|
||||
private String embeddingModel;
|
||||
|
||||
@TableField(fill = FieldFill.INSERT)
|
||||
private LocalDateTime createTime;
|
||||
|
||||
@TableField(fill = FieldFill.INSERT_UPDATE)
|
||||
private LocalDateTime updateTime;
|
||||
|
||||
@TableLogic
|
||||
private Integer deleted;
|
||||
}
|
||||
@ -0,0 +1,14 @@
|
||||
package vip.mate.wiki.repository;
|
||||
|
||||
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import vip.mate.wiki.model.WikiChunkEntity;
|
||||
|
||||
/**
|
||||
* Wiki chunk 数据访问层
|
||||
*
|
||||
* @author MateClaw Team
|
||||
*/
|
||||
@Mapper
|
||||
public interface WikiChunkMapper extends BaseMapper<WikiChunkEntity> {
|
||||
}
|
||||
@ -0,0 +1,203 @@
|
||||
package vip.mate.wiki.service;
|
||||
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
import vip.mate.wiki.WikiProperties;
|
||||
import vip.mate.wiki.model.WikiChunkEntity;
|
||||
import vip.mate.wiki.model.WikiPageEntity;
|
||||
|
||||
import java.util.*;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
* RFC-011: 混合检索服务
|
||||
* <p>
|
||||
* 支持三种模式：
|
||||
* <ul>
|
||||
* <li>{@code keyword} — DB LIKE 搜索（现有 WikiPageService.searchPages）</li>
|
||||
* <li>{@code semantic} — chunk 向量 cosine 相似度 → 回溯到 page</li>
|
||||
* <li>{@code hybrid} — 两者融合，RRF (Reciprocal Rank Fusion) 排名</li>
|
||||
* </ul>
|
||||
*
|
||||
* @author MateClaw Team
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class HybridRetriever {
|
||||
|
||||
private final WikiPageService pageService;
|
||||
private final WikiChunkService chunkService;
|
||||
private final WikiEmbeddingService embeddingService;
|
||||
private final WikiProperties properties;
|
||||
|
||||
public enum Mode { KEYWORD, SEMANTIC, HYBRID }
|
||||
|
||||
/**
|
||||
* 搜索结果（页面级）
|
||||
*/
|
||||
public record PageHit(Long pageId, String slug, String title, String summary, double score) {}
|
||||
|
||||
/**
|
||||
* 搜索结果（chunk 级，语义搜索专用）
|
||||
*/
|
||||
public record ChunkHit(Long chunkId, Long rawId, String snippet, float score) {}
|
||||
|
||||
/**
|
||||
* 执行混合搜索，返回页面级结果
|
||||
*/
|
||||
public List<PageHit> searchPages(Long kbId, String query, String modeStr, int topK) {
|
||||
Mode mode = parseMode(modeStr);
|
||||
|
||||
List<RankedItem> semantic = List.of();
|
||||
List<RankedItem> keyword = List.of();
|
||||
|
||||
if (mode != Mode.KEYWORD && embeddingService.isAvailable()) {
|
||||
semantic = semanticSearch(kbId, query, topK * 3);
|
||||
}
|
||||
if (mode != Mode.SEMANTIC) {
|
||||
keyword = keywordSearch(kbId, query, topK * 3);
|
||||
}
|
||||
|
||||
// 如果 semantic 不可用（no embedding model），回退到 keyword
|
||||
if (mode == Mode.SEMANTIC && semantic.isEmpty()) {
|
||||
log.debug("[HybridRetriever] Semantic unavailable, falling back to keyword");
|
||||
keyword = keywordSearch(kbId, query, topK * 3);
|
||||
}
|
||||
|
||||
List<RankedItem> fused;
|
||||
if (mode == Mode.KEYWORD || semantic.isEmpty()) {
|
||||
fused = keyword;
|
||||
} else if (mode == Mode.SEMANTIC) {
|
||||
fused = semantic;
|
||||
} else {
|
||||
fused = rrfFuse(semantic, keyword, 60);
|
||||
}
|
||||
|
||||
// 取 topK，装配 PageHit
|
||||
return fused.stream()
|
||||
.limit(topK)
|
||||
.map(ri -> {
|
||||
WikiPageEntity page = pageService.getById(ri.pageId);
|
||||
if (page == null) return null;
|
||||
return new PageHit(ri.pageId, page.getSlug(), page.getTitle(),
|
||||
page.getSummary(), ri.score);
|
||||
})
|
||||
.filter(Objects::nonNull)
|
||||
.toList();
|
||||
}
|
||||
|
||||
/**
|
||||
* chunk 级语义搜索（Agent 直接拿 chunk 片段作为证据）
|
||||
*/
|
||||
public List<ChunkHit> searchChunks(Long kbId, String query, int topK) {
|
||||
if (!embeddingService.isAvailable()) return List.of();
|
||||
|
||||
float[] queryVec = embeddingService.embedQuery(query);
|
||||
if (queryVec == null) return List.of();
|
||||
|
||||
List<WikiChunkEntity> allChunks = chunkService.listByKbId(kbId);
|
||||
|
||||
return allChunks.stream()
|
||||
.filter(c -> c.getEmbedding() != null)
|
||||
.map(c -> {
|
||||
float[] chunkVec = WikiEmbeddingService.bytesToFloats(c.getEmbedding());
|
||||
float score = WikiEmbeddingService.cosine(queryVec, chunkVec);
|
||||
String snippet = c.getContent().length() > 300
|
||||
? c.getContent().substring(0, 300) + "..."
|
||||
: c.getContent();
|
||||
return new ChunkHit(c.getId(), c.getRawId(), snippet, score);
|
||||
})
|
||||
.sorted(Comparator.comparingDouble(ChunkHit::score).reversed())
|
||||
.limit(topK)
|
||||
.toList();
|
||||
}
|
||||
|
||||
// ==================== 内部方法 ====================
|
||||
|
||||
/** 语义搜索：chunk cosine → 聚合到 page（同页多 chunk 取最高分） */
|
||||
private List<RankedItem> semanticSearch(Long kbId, String query, int limit) {
|
||||
float[] queryVec = embeddingService.embedQuery(query);
|
||||
if (queryVec == null) return List.of();
|
||||
|
||||
List<WikiChunkEntity> allChunks = chunkService.listByKbId(kbId);
|
||||
if (allChunks.isEmpty()) return List.of();
|
||||
|
||||
// chunk → score, 然后 需要映射到 page。
|
||||
// 当前没有 chunk → page 的直接关联（chunk 只有 rawId）。
|
||||
// 走 rawId → 找该 rawId 对应的所有 page（source_raw_ids 含该 rawId）
|
||||
// 这是个近似：一个 rawId 可能产出多个 page，都算命中。
|
||||
Map<Long, Float> chunkScores = new HashMap<>();
|
||||
for (WikiChunkEntity chunk : allChunks) {
|
||||
if (chunk.getEmbedding() == null) continue;
|
||||
float[] vec = WikiEmbeddingService.bytesToFloats(chunk.getEmbedding());
|
||||
float score = WikiEmbeddingService.cosine(queryVec, vec);
|
||||
chunkScores.merge(chunk.getRawId(), score, Math::max); // rawId 级聚合
|
||||
}
|
||||
|
||||
// rawId → page IDs
|
||||
List<WikiPageEntity> allPages = pageService.listByKbId(kbId);
|
||||
Map<Long, Double> pageScores = new HashMap<>();
|
||||
for (WikiPageEntity page : allPages) {
|
||||
String rawIds = page.getSourceRawIds();
|
||||
if (rawIds == null) continue;
|
||||
// 解析 "[1,2,3]" 格式
|
||||
for (String rawIdStr : rawIds.replaceAll("[\\[\\]\\s]", "").split(",")) {
|
||||
try {
|
||||
long rawId = Long.parseLong(rawIdStr.trim());
|
||||
Float score = chunkScores.get(rawId);
|
||||
if (score != null) {
|
||||
pageScores.merge(page.getId(), (double) score, Math::max);
|
||||
}
|
||||
} catch (NumberFormatException ignored) {}
|
||||
}
|
||||
}
|
||||
|
||||
return pageScores.entrySet().stream()
|
||||
.sorted(Map.Entry.<Long, Double>comparingByValue().reversed())
|
||||
.limit(limit)
|
||||
.map(e -> new RankedItem(e.getKey(), e.getValue()))
|
||||
.toList();
|
||||
}
|
||||
|
||||
/** 关键词搜索：走现有 DB LIKE */
|
||||
private List<RankedItem> keywordSearch(Long kbId, String query, int limit) {
|
||||
List<WikiPageEntity> results = pageService.searchPages(kbId, query);
|
||||
List<RankedItem> ranked = new ArrayList<>();
|
||||
for (int i = 0; i < Math.min(results.size(), limit); i++) {
|
||||
// LIKE 无分数，用倒序排名作为伪分数
|
||||
ranked.add(new RankedItem(results.get(i).getId(), 1.0 / (i + 1)));
|
||||
}
|
||||
return ranked;
|
||||
}
|
||||
|
||||
/** RRF 融合：score = Σ 1/(k + rank_i) */
|
||||
private List<RankedItem> rrfFuse(List<RankedItem> a, List<RankedItem> b, int k) {
|
||||
Map<Long, Double> fused = new HashMap<>();
|
||||
for (int i = 0; i < a.size(); i++) fused.merge(a.get(i).pageId, 1.0 / (k + i + 1), Double::sum);
|
||||
for (int i = 0; i < b.size(); i++) fused.merge(b.get(i).pageId, 1.0 / (k + i + 1), Double::sum);
|
||||
return fused.entrySet().stream()
|
||||
.sorted(Map.Entry.<Long, Double>comparingByValue().reversed())
|
||||
.map(e -> new RankedItem(e.getKey(), e.getValue()))
|
||||
.toList();
|
||||
}
|
||||
|
||||
private Mode parseMode(String mode) {
|
||||
if (mode == null || mode.isBlank()) {
|
||||
String defaultMode = properties.getSearchDefaultMode();
|
||||
return switch (defaultMode) {
|
||||
case "keyword" -> Mode.KEYWORD;
|
||||
case "semantic" -> Mode.SEMANTIC;
|
||||
default -> Mode.HYBRID;
|
||||
};
|
||||
}
|
||||
return switch (mode.toLowerCase()) {
|
||||
case "keyword" -> Mode.KEYWORD;
|
||||
case "semantic" -> Mode.SEMANTIC;
|
||||
default -> Mode.HYBRID;
|
||||
};
|
||||
}
|
||||
|
||||
private record RankedItem(Long pageId, double score) {}
|
||||
}
|
||||
@ -0,0 +1,192 @@
|
||||
package vip.mate.wiki.service;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import vip.mate.wiki.model.WikiChunkEntity;
|
||||
import vip.mate.wiki.repository.WikiChunkMapper;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.security.MessageDigest;
|
||||
import java.util.*;
|
||||
|
||||
/**
|
||||
* Wiki chunk 服务
|
||||
* <p>
|
||||
* RFC-013 最小切片：chunk 持久化 + hash 级增量对账。
|
||||
* <ul>
|
||||
* <li>{@link #persistChunks} — 切分后一次性入库，返回 chunk ID 列表供后续流程使用</li>
|
||||
* <li>{@link #reconcile} — 增量对账：hash 相同的 chunk 保留（含 embedding），hash 不同的重建</li>
|
||||
* <li>{@link #deleteByRawId} — 材料删除时级联清理</li>
|
||||
* </ul>
|
||||
*
|
||||
* @author MateClaw Team
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class WikiChunkService {
|
||||
|
||||
private final WikiChunkMapper chunkMapper;
|
||||
|
||||
/**
|
||||
* 将文本切片列表持久化为 chunk 记录。
|
||||
* <p>
|
||||
* 如果该 rawId 已有 chunk 记录，走 {@link #reconcile} 增量对账；否则全量插入。
|
||||
*
|
||||
* @param kbId 知识库 ID
|
||||
* @param rawId 原始材料 ID
|
||||
* @param chunks 切分后的文本列表（有序）
|
||||
* @param offsets 每个 chunk 对应的 [startOffset, endOffset] 数组
|
||||
* @return 持久化后的 chunk ID 列表（与 chunks 同序）
|
||||
*/
|
||||
@Transactional
|
||||
public List<Long> persistChunks(Long kbId, Long rawId, List<String> chunks, List<int[]> offsets) {
|
||||
List<WikiChunkEntity> existing = listByRawId(rawId);
|
||||
|
||||
if (existing.isEmpty()) {
|
||||
// 全量插入
|
||||
return insertAll(kbId, rawId, chunks, offsets);
|
||||
}
|
||||
|
||||
// 增量对账
|
||||
return reconcile(kbId, rawId, chunks, offsets, existing);
|
||||
}
|
||||
|
||||
/**
|
||||
* 增量对账：比对 hash，保留不变的 chunk（保护未来的 embedding），重建变化的。
|
||||
*
|
||||
* @return 对账后的 chunk ID 列表（与 chunks 同序）
|
||||
*/
|
||||
private List<Long> reconcile(Long kbId, Long rawId, List<String> chunks, List<int[]> offsets,
|
||||
List<WikiChunkEntity> existing) {
|
||||
// 旧 chunk 按 ordinal 索引
|
||||
Map<Integer, WikiChunkEntity> oldByOrdinal = new HashMap<>();
|
||||
for (WikiChunkEntity e : existing) {
|
||||
oldByOrdinal.put(e.getOrdinal(), e);
|
||||
}
|
||||
|
||||
List<Long> resultIds = new ArrayList<>(chunks.size());
|
||||
Set<Long> retainedIds = new HashSet<>();
|
||||
int retained = 0, rebuilt = 0;
|
||||
|
||||
for (int i = 0; i < chunks.size(); i++) {
|
||||
String text = chunks.get(i);
|
||||
String hash = computeHash(text);
|
||||
int[] offset = offsets.get(i);
|
||||
|
||||
WikiChunkEntity old = oldByOrdinal.get(i);
|
||||
if (old != null && hash.equals(old.getContentHash())) {
|
||||
// hash 相同 → 保留（embedding 等附加数据不丢）
|
||||
// 但更新 offset（材料可能在其他位置变了导致偏移变化）
|
||||
if (!old.getStartOffset().equals(offset[0]) || !old.getEndOffset().equals(offset[1])) {
|
||||
old.setStartOffset(offset[0]);
|
||||
old.setEndOffset(offset[1]);
|
||||
chunkMapper.updateById(old);
|
||||
}
|
||||
resultIds.add(old.getId());
|
||||
retainedIds.add(old.getId());
|
||||
retained++;
|
||||
} else {
|
||||
// hash 不同或 ordinal 超出旧范围 → 新建
|
||||
WikiChunkEntity entity = buildEntity(kbId, rawId, i, text, hash, offset);
|
||||
chunkMapper.insert(entity);
|
||||
resultIds.add(entity.getId());
|
||||
rebuilt++;
|
||||
}
|
||||
}
|
||||
|
||||
// 删除多余的旧 chunk（数量缩减的情况）
|
||||
int deleted = 0;
|
||||
for (WikiChunkEntity old : existing) {
|
||||
if (!retainedIds.contains(old.getId()) && !resultIds.contains(old.getId())) {
|
||||
chunkMapper.deleteById(old.getId());
|
||||
deleted++;
|
||||
}
|
||||
}
|
||||
|
||||
log.info("[WikiChunk] Reconciled raw={}: retained={}, rebuilt={}, deleted={}",
|
||||
rawId, retained, rebuilt, deleted);
|
||||
return resultIds;
|
||||
}
|
||||
|
||||
/**
|
||||
* 全量插入
|
||||
*/
|
||||
private List<Long> insertAll(Long kbId, Long rawId, List<String> chunks, List<int[]> offsets) {
|
||||
List<Long> ids = new ArrayList<>(chunks.size());
|
||||
for (int i = 0; i < chunks.size(); i++) {
|
||||
String text = chunks.get(i);
|
||||
String hash = computeHash(text);
|
||||
int[] offset = offsets.get(i);
|
||||
WikiChunkEntity entity = buildEntity(kbId, rawId, i, text, hash, offset);
|
||||
chunkMapper.insert(entity);
|
||||
ids.add(entity.getId());
|
||||
}
|
||||
log.info("[WikiChunk] Inserted {} chunks for raw={}", chunks.size(), rawId);
|
||||
return ids;
|
||||
}
|
||||
|
||||
public List<WikiChunkEntity> listByRawId(Long rawId) {
|
||||
return chunkMapper.selectList(
|
||||
new LambdaQueryWrapper<WikiChunkEntity>()
|
||||
.eq(WikiChunkEntity::getRawId, rawId)
|
||||
.orderByAsc(WikiChunkEntity::getOrdinal));
|
||||
}
|
||||
|
||||
public List<WikiChunkEntity> listByKbId(Long kbId) {
|
||||
return chunkMapper.selectList(
|
||||
new LambdaQueryWrapper<WikiChunkEntity>()
|
||||
.eq(WikiChunkEntity::getKbId, kbId)
|
||||
.orderByAsc(WikiChunkEntity::getRawId)
|
||||
.orderByAsc(WikiChunkEntity::getOrdinal));
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public void deleteByRawId(Long rawId) {
|
||||
int deleted = chunkMapper.delete(
|
||||
new LambdaQueryWrapper<WikiChunkEntity>()
|
||||
.eq(WikiChunkEntity::getRawId, rawId));
|
||||
if (deleted > 0) {
|
||||
log.info("[WikiChunk] Deleted {} chunks for raw={}", deleted, rawId);
|
||||
}
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public void deleteByKbId(Long kbId) {
|
||||
int deleted = chunkMapper.delete(
|
||||
new LambdaQueryWrapper<WikiChunkEntity>()
|
||||
.eq(WikiChunkEntity::getKbId, kbId));
|
||||
if (deleted > 0) {
|
||||
log.info("[WikiChunk] Deleted {} chunks for kbId={}", deleted, kbId);
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Helpers ====================
|
||||
|
||||
private WikiChunkEntity buildEntity(Long kbId, Long rawId, int ordinal, String text, String hash, int[] offset) {
|
||||
WikiChunkEntity entity = new WikiChunkEntity();
|
||||
entity.setKbId(kbId);
|
||||
entity.setRawId(rawId);
|
||||
entity.setOrdinal(ordinal);
|
||||
entity.setContent(text);
|
||||
entity.setCharCount(text.length());
|
||||
entity.setStartOffset(offset[0]);
|
||||
entity.setEndOffset(offset[1]);
|
||||
entity.setContentHash(hash);
|
||||
return entity;
|
||||
}
|
||||
|
||||
private String computeHash(String content) {
|
||||
try {
|
||||
MessageDigest digest = MessageDigest.getInstance("SHA-256");
|
||||
byte[] hash = digest.digest(content.getBytes(StandardCharsets.UTF_8));
|
||||
return HexFormat.of().formatHex(hash);
|
||||
} catch (Exception e) {
|
||||
log.warn("[WikiChunk] Hash computation failed: {}", e.getMessage());
|
||||
return "HASH_ERROR_" + content.length();
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,166 @@
|
||||
package vip.mate.wiki.service;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.ai.embedding.EmbeddingModel;
|
||||
import org.springframework.ai.embedding.EmbeddingRequest;
|
||||
import org.springframework.ai.embedding.EmbeddingResponse;
|
||||
import org.springframework.beans.factory.ObjectProvider;
|
||||
import org.springframework.stereotype.Service;
|
||||
import vip.mate.wiki.WikiProperties;
|
||||
import vip.mate.wiki.model.WikiChunkEntity;
|
||||
import vip.mate.wiki.repository.WikiChunkMapper;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.ByteOrder;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* RFC-011: Wiki 嵌入服务
|
||||
* <p>
|
||||
* 使用 Spring AI 的 {@link EmbeddingModel}（DashScope auto-config）对 chunk 做向量化。
|
||||
* <ul>
|
||||
* <li>{@link #embedMissingChunks} — 批量嵌入缺失 embedding 的 chunk（材料处理后异步调用）</li>
|
||||
* <li>{@link #embedQuery} — 查询向量化（混合搜索时调用）</li>
|
||||
* </ul>
|
||||
*
|
||||
* @author MateClaw Team
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
public class WikiEmbeddingService {
|
||||
|
||||
private final EmbeddingModel embeddingModel;
|
||||
private final WikiChunkMapper chunkMapper;
|
||||
private final WikiProperties properties;
|
||||
private final boolean available;
|
||||
|
||||
public WikiEmbeddingService(ObjectProvider<EmbeddingModel> embeddingModelProvider,
|
||||
WikiChunkMapper chunkMapper, WikiProperties properties) {
|
||||
this.chunkMapper = chunkMapper;
|
||||
this.properties = properties;
|
||||
EmbeddingModel model = embeddingModelProvider.getIfAvailable();
|
||||
this.embeddingModel = model;
|
||||
this.available = model != null;
|
||||
if (!available) {
|
||||
log.warn("[WikiEmbedding] No EmbeddingModel bean found — semantic search disabled. "
|
||||
+ "Ensure spring-ai-alibaba-starter-dashscope is on classpath and DASHSCOPE_API_KEY is set.");
|
||||
} else {
|
||||
log.info("[WikiEmbedding] EmbeddingModel available: {}", model.getClass().getSimpleName());
|
||||
}
|
||||
}
|
||||
|
||||
public boolean isAvailable() { return available; }
|
||||
|
||||
/**
|
||||
* 批量嵌入指定 KB 中缺失 embedding 的 chunk。
|
||||
* <p>
|
||||
* 只嵌入 embedding 为 NULL 或 embeddingModel 与当前配置不匹配的 chunk。
|
||||
* 模型切换时自动触发全量重嵌（通过 embeddingModel 字段比对）。
|
||||
*/
|
||||
public int embedMissingChunks(Long kbId) {
|
||||
if (!available) {
|
||||
log.debug("[WikiEmbedding] Skipping — no EmbeddingModel available");
|
||||
return 0;
|
||||
}
|
||||
|
||||
String modelName = properties.getEmbeddingModel();
|
||||
List<WikiChunkEntity> pending = chunkMapper.selectList(
|
||||
new LambdaQueryWrapper<WikiChunkEntity>()
|
||||
.eq(WikiChunkEntity::getKbId, kbId)
|
||||
.and(w -> w.isNull(WikiChunkEntity::getEmbedding)
|
||||
.or().ne(WikiChunkEntity::getEmbeddingModel, modelName)));
|
||||
|
||||
if (pending.isEmpty()) {
|
||||
log.debug("[WikiEmbedding] No chunks need embedding for kbId={}", kbId);
|
||||
return 0;
|
||||
}
|
||||
|
||||
int batchSize = Math.max(1, properties.getEmbeddingBatchSize());
|
||||
int total = 0;
|
||||
|
||||
for (int offset = 0; offset < pending.size(); offset += batchSize) {
|
||||
List<WikiChunkEntity> batch = pending.subList(offset, Math.min(offset + batchSize, pending.size()));
|
||||
try {
|
||||
List<String> inputs = batch.stream()
|
||||
.map(WikiChunkEntity::getContent)
|
||||
.toList();
|
||||
|
||||
EmbeddingResponse resp = embeddingModel.call(
|
||||
new EmbeddingRequest(inputs, null));
|
||||
|
||||
for (int i = 0; i < batch.size(); i++) {
|
||||
float[] vec = resp.getResults().get(i).getOutput();
|
||||
WikiChunkEntity chunk = batch.get(i);
|
||||
chunk.setEmbedding(floatsToBytes(vec));
|
||||
chunk.setEmbeddingModel(modelName);
|
||||
chunkMapper.updateById(chunk);
|
||||
}
|
||||
total += batch.size();
|
||||
} catch (Exception e) {
|
||||
log.error("[WikiEmbedding] Batch embedding failed (kbId={}, batchSize={}): {}",
|
||||
kbId, batch.size(), e.getMessage());
|
||||
// 继续下一批，不中断
|
||||
}
|
||||
}
|
||||
|
||||
log.info("[WikiEmbedding] Embedded {}/{} chunks for kbId={}, model={}",
|
||||
total, pending.size(), kbId, modelName);
|
||||
return total;
|
||||
}
|
||||
|
||||
/**
|
||||
* 查询向量化（混合搜索时调用）
|
||||
*/
|
||||
public float[] embedQuery(String query) {
|
||||
if (!available) return null;
|
||||
try {
|
||||
EmbeddingResponse resp = embeddingModel.call(
|
||||
new EmbeddingRequest(List.of(query), null));
|
||||
return resp.getResults().get(0).getOutput();
|
||||
} catch (Exception e) {
|
||||
log.error("[WikiEmbedding] Query embedding failed: {}", e.getMessage());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 清空指定 KB 的所有 embedding（模型切换时调用）
|
||||
*/
|
||||
public void clearEmbeddings(Long kbId) {
|
||||
chunkMapper.update(null, new LambdaUpdateWrapper<WikiChunkEntity>()
|
||||
.eq(WikiChunkEntity::getKbId, kbId)
|
||||
.set(WikiChunkEntity::getEmbedding, null)
|
||||
.set(WikiChunkEntity::getEmbeddingModel, null));
|
||||
log.info("[WikiEmbedding] Cleared all embeddings for kbId={}", kbId);
|
||||
}
|
||||
|
||||
// ==================== 向量序列化 ====================
|
||||
|
||||
public static byte[] floatsToBytes(float[] vec) {
|
||||
ByteBuffer buf = ByteBuffer.allocate(vec.length * 4).order(ByteOrder.LITTLE_ENDIAN);
|
||||
for (float v : vec) buf.putFloat(v);
|
||||
return buf.array();
|
||||
}
|
||||
|
||||
public static float[] bytesToFloats(byte[] bytes) {
|
||||
ByteBuffer buf = ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN);
|
||||
float[] vec = new float[bytes.length / 4];
|
||||
for (int i = 0; i < vec.length; i++) vec[i] = buf.getFloat();
|
||||
return vec;
|
||||
}
|
||||
|
||||
/** 余弦相似度 */
|
||||
public static float cosine(float[] a, float[] b) {
|
||||
if (a.length != b.length) return 0f;
|
||||
float dot = 0, normA = 0, normB = 0;
|
||||
for (int i = 0; i < a.length; i++) {
|
||||
dot += a[i] * b[i];
|
||||
normA += a[i] * a[i];
|
||||
normB += b[i] * b[i];
|
||||
}
|
||||
float denom = (float) (Math.sqrt(normA) * Math.sqrt(normB));
|
||||
return denom == 0 ? 0f : dot / denom;
|
||||
}
|
||||
}
|
||||
@ -46,6 +46,8 @@ public class WikiProcessingService {
|
||||
private final WikiKnowledgeBaseService kbService;
|
||||
private final WikiRawMaterialService rawService;
|
||||
private final WikiPageService pageService;
|
||||
private final WikiChunkService chunkService;
|
||||
private final WikiEmbeddingService embeddingService;
|
||||
private final WikiProperties properties;
|
||||
private final ModelConfigService modelConfigService;
|
||||
private final AgentGraphBuilder agentGraphBuilder;
|
||||
@ -172,6 +174,13 @@ public class WikiProcessingService {
|
||||
if (textContent.length() > properties.getMaxChunkSize()) {
|
||||
result = processInChunks(kb, raw, textContent, existingPagesIndex);
|
||||
} else {
|
||||
// 单 chunk 也持久化（RFC-013：保证所有 chunk 都入库）
|
||||
try {
|
||||
chunkService.persistChunks(kb.getId(), rawId,
|
||||
List.of(textContent), List.of(new int[]{0, textContent.length()}));
|
||||
} catch (Exception e) {
|
||||
log.warn("[Wiki] Single chunk persistence failed for raw={}: {}", rawId, e.getMessage());
|
||||
}
|
||||
int pages = processChunk(kb, raw, textContent, existingPagesIndex);
|
||||
result = new int[]{pages, pages == 0 ? 1 : 0, 1};
|
||||
}
|
||||
@ -236,6 +245,25 @@ public class WikiProcessingService {
|
||||
log.info("[Wiki] Processing completed for raw={}, kbId={}, generatedPages={}, totalPages={}",
|
||||
rawId, kb.getId(), totalPages, pageCount);
|
||||
|
||||
// RFC-011：异步嵌入新 chunk（不阻塞处理管线）
|
||||
// 注意：此方法目前未加 @Transactional，每个 DB 操作短事务独立提交。
|
||||
// 如果未来加了事务包裹 processRawMaterial，这里的异步任务需要改用
|
||||
// TransactionSynchronizationManager.registerSynchronization(afterCommit)
|
||||
// 否则新线程会查不到 chunk（事务未提交）导致 embedding 静默跳过。
|
||||
if (totalPages > 0) {
|
||||
final Long fKbId = kb.getId();
|
||||
WIKI_EXECUTOR.submit(() -> {
|
||||
try {
|
||||
int embedded = embeddingService.embedMissingChunks(fKbId);
|
||||
if (embedded > 0) {
|
||||
log.info("[Wiki] Async embedding completed: kbId={}, embedded={}", fKbId, embedded);
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
log.warn("[Wiki] Async embedding failed for kbId={}: {}", fKbId, ex.getMessage());
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("[Wiki] Processing failed for raw={}: {}", rawId, e.getMessage(), e);
|
||||
rawService.updateProcessingStatus(rawId, "failed", e.getMessage());
|
||||
@ -295,11 +323,21 @@ public class WikiProcessingService {
|
||||
*/
|
||||
private int[] processInChunks(WikiKnowledgeBaseEntity kb, WikiRawMaterialEntity raw, String text,
|
||||
String existingPagesIndex) {
|
||||
// Phase 1: 切分文本为 chunks
|
||||
List<String> chunks = splitIntoChunks(text);
|
||||
// Phase 1: 切分文本为 chunks（带偏移，供持久化）
|
||||
List<ChunkWithOffset> chunksWithOffset = splitIntoChunksWithOffsets(text);
|
||||
List<String> chunks = chunksWithOffset.stream().map(ChunkWithOffset::text).toList();
|
||||
int totalChunks = chunks.size();
|
||||
log.info("[Wiki] Split into {} chunks for raw={}, kbId={}", totalChunks, raw.getId(), kb.getId());
|
||||
|
||||
// RFC-013：持久化 chunk 到 mate_wiki_chunk（增量对账：hash 不变的保留）
|
||||
try {
|
||||
List<int[]> offsets = chunksWithOffset.stream()
|
||||
.map(c -> new int[]{c.startOffset(), c.endOffset()}).toList();
|
||||
chunkService.persistChunks(kb.getId(), raw.getId(), chunks, offsets);
|
||||
} catch (Exception e) {
|
||||
log.warn("[Wiki] Chunk persistence failed for raw={}, continuing without: {}", raw.getId(), e.getMessage());
|
||||
}
|
||||
|
||||
if (totalChunks == 1) {
|
||||
// 单 chunk 不走并行
|
||||
try {
|
||||
@ -356,9 +394,19 @@ public class WikiProcessingService {
|
||||
* 将文本切分为多个 chunks（智能句子边界，支持中英文）
|
||||
*/
|
||||
private List<String> splitIntoChunks(String text) {
|
||||
return splitIntoChunksWithOffsets(text).stream().map(ChunkWithOffset::text).toList();
|
||||
}
|
||||
|
||||
/** chunk 文本 + 在原始文本中的偏移 */
|
||||
record ChunkWithOffset(String text, int startOffset, int endOffset) {}
|
||||
|
||||
/**
|
||||
* 切分并记录每个 chunk 的原始偏移（RFC-013：供 WikiChunkService 持久化）
|
||||
*/
|
||||
private List<ChunkWithOffset> splitIntoChunksWithOffsets(String text) {
|
||||
int chunkSize = properties.getMaxChunkSize();
|
||||
int overlap = Math.min(500, chunkSize / 10);
|
||||
List<String> chunks = new ArrayList<>();
|
||||
List<ChunkWithOffset> chunks = new ArrayList<>();
|
||||
int start = 0;
|
||||
|
||||
while (start < text.length()) {
|
||||
@ -372,7 +420,7 @@ public class WikiProcessingService {
|
||||
}
|
||||
}
|
||||
|
||||
chunks.add(text.substring(start, end));
|
||||
chunks.add(new ChunkWithOffset(text.substring(start, end), start, end));
|
||||
|
||||
// 前进（带 overlap 防止边界上下文丢失）
|
||||
int nextStart = end - overlap;
|
||||
|
||||
@ -36,6 +36,8 @@ public class WikiRawMaterialService {
|
||||
private final WikiProperties properties;
|
||||
private final ApplicationEventPublisher eventPublisher;
|
||||
private final DocumentExtractTool documentExtractTool;
|
||||
/** RFC-013：删除时级联清理 chunk */
|
||||
private final WikiChunkService chunkService;
|
||||
|
||||
/**
|
||||
* RFC-012 follow-up #3：从 partial 状态触发的 reprocess 会在此 set 中打标，
|
||||
@ -278,6 +280,14 @@ public class WikiRawMaterialService {
|
||||
@Transactional
|
||||
public void delete(Long id) {
|
||||
rawMapper.deleteById(id);
|
||||
// RFC-013：级联清理 chunk，避免语义搜索命中孤儿 chunk
|
||||
try {
|
||||
if (chunkService != null) {
|
||||
chunkService.deleteByRawId(id);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.warn("[Wiki] Failed to cascade-delete chunks for raw={}: {}", id, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@ -0,0 +1,297 @@
|
||||
package vip.mate.wiki.service;
|
||||
|
||||
import cn.hutool.json.JSONArray;
|
||||
import cn.hutool.json.JSONObject;
|
||||
import cn.hutool.json.JSONUtil;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.ai.chat.messages.SystemMessage;
|
||||
import org.springframework.ai.chat.messages.UserMessage;
|
||||
import org.springframework.ai.chat.model.ChatModel;
|
||||
import org.springframework.ai.chat.model.ChatResponse;
|
||||
import org.springframework.ai.chat.prompt.Prompt;
|
||||
import org.springframework.retry.support.RetryTemplate;
|
||||
import org.springframework.stereotype.Service;
|
||||
import vip.mate.agent.AgentGraphBuilder;
|
||||
import vip.mate.agent.prompt.PromptLoader;
|
||||
import vip.mate.channel.web.ChatStreamTracker;
|
||||
import vip.mate.llm.model.ModelConfigEntity;
|
||||
import vip.mate.llm.service.ModelConfigService;
|
||||
import vip.mate.wiki.model.WikiRawMaterialEntity;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
* RFC-011 Phase 3: Wiki Deep Research 服务
|
||||
* <p>
|
||||
* 三阶段管线（不走 StateGraph 框架，保持轻量）：
|
||||
* <ol>
|
||||
* <li><b>Plan</b>：LLM 把 topic 拆为 3-5 个子问题</li>
|
||||
* <li><b>Retrieve + Draft</b>：并行对每个子问题调 {@link HybridRetriever} + LLM 起草段落</li>
|
||||
* <li><b>Compose</b>：LLM 把段落组装为最终报告</li>
|
||||
* </ol>
|
||||
* 事件通过 {@link ChatStreamTracker#broadcast} 推送，前端用 SSE 订阅。
|
||||
*
|
||||
* @author MateClaw Team
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class WikiResearchService {
|
||||
|
||||
private final HybridRetriever hybridRetriever;
|
||||
private final WikiRawMaterialService rawService;
|
||||
private final ModelConfigService modelConfigService;
|
||||
private final AgentGraphBuilder agentGraphBuilder;
|
||||
private final ChatStreamTracker streamTracker;
|
||||
|
||||
private static final RetryTemplate NO_RETRY = RetryTemplate.builder().maxAttempts(1).build();
|
||||
private static final ExecutorService EXECUTOR = Executors.newVirtualThreadPerTaskExecutor();
|
||||
private static final int DEFAULT_TOP_K_PER_QUESTION = 5;
|
||||
private static final int MAX_PARALLEL_QUESTIONS = 3;
|
||||
|
||||
/**
|
||||
* 执行 Deep Research，通过 SSE 流式推送进度
|
||||
*
|
||||
* @param kbId 知识库 ID
|
||||
* @param topic 研究主题
|
||||
* @param sessionId SSE 会话 ID（前端订阅用）
|
||||
* @param topKPerQuestion 每个子问题召回的材料数（默认 5）
|
||||
* @return 最终报告
|
||||
*/
|
||||
public ResearchResult research(Long kbId, String topic, String sessionId, Integer topKPerQuestion) {
|
||||
int topK = topKPerQuestion != null && topKPerQuestion > 0 ? topKPerQuestion : DEFAULT_TOP_K_PER_QUESTION;
|
||||
log.info("[Research] Start: kbId={}, topic={}, sessionId={}", kbId, topic, sessionId);
|
||||
|
||||
try {
|
||||
// Stage 1: Plan
|
||||
List<SubQuestion> questions = planStage(topic);
|
||||
if (questions.isEmpty()) {
|
||||
broadcast(sessionId, "research.error", Map.of("message", "主题无法分解为可研究的子问题"));
|
||||
return new ResearchResult(topic, List.of(), "无法为该主题生成研究计划。");
|
||||
}
|
||||
broadcast(sessionId, "research.plan", Map.of(
|
||||
"questions", questions.stream().map(q -> Map.of("question", q.question, "intent", q.intent)).toList()
|
||||
));
|
||||
|
||||
// Stage 2: Retrieve + Draft (并行)
|
||||
List<Section> sections = draftStage(kbId, questions, topK, sessionId);
|
||||
if (sections.stream().allMatch(s -> s.content == null || s.content.isBlank())) {
|
||||
broadcast(sessionId, "research.error", Map.of("message", "所有子问题都未能起草出内容"));
|
||||
return new ResearchResult(topic, sections, "没有足够的材料回答该主题。");
|
||||
}
|
||||
|
||||
// Stage 3: Compose
|
||||
String report = composeStage(topic, sections);
|
||||
broadcast(sessionId, "research.done", Map.of(
|
||||
"report", report,
|
||||
"sections", sections.size(),
|
||||
"materialsUsed", sections.stream().flatMap(s -> s.materialRefs.stream()).distinct().count()
|
||||
));
|
||||
return new ResearchResult(topic, sections, report);
|
||||
} catch (Exception e) {
|
||||
log.error("[Research] Failed: kbId={}, topic={}: {}", kbId, topic, e.getMessage(), e);
|
||||
broadcast(sessionId, "research.error", Map.of("message", e.getMessage() != null ? e.getMessage() : "研究失败"));
|
||||
return new ResearchResult(topic, List.of(), "研究过程失败: " + e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Stage 1: Plan ====================
|
||||
|
||||
private List<SubQuestion> planStage(String topic) {
|
||||
String systemPrompt = PromptLoader.loadPrompt("research/plan-system");
|
||||
String userPrompt = PromptLoader.loadPrompt("research/plan-user").replace("{topic}", topic);
|
||||
|
||||
String response = callLlm(systemPrompt, userPrompt, "plan");
|
||||
if (response == null || response.isBlank()) return List.of();
|
||||
|
||||
try {
|
||||
String cleaned = stripCodeFences(response);
|
||||
JSONObject obj = JSONUtil.parseObj(cleaned);
|
||||
JSONArray arr = obj.getJSONArray("questions");
|
||||
if (arr == null) return List.of();
|
||||
List<SubQuestion> questions = new ArrayList<>();
|
||||
for (Object item : arr) {
|
||||
JSONObject q = (JSONObject) item;
|
||||
String question = q.getStr("question");
|
||||
String intent = q.getStr("intent", "");
|
||||
if (question != null && !question.isBlank()) {
|
||||
questions.add(new SubQuestion(question, intent));
|
||||
}
|
||||
}
|
||||
log.info("[Research] Plan produced {} sub-questions", questions.size());
|
||||
return questions;
|
||||
} catch (Exception e) {
|
||||
log.warn("[Research] Plan JSON parse failed: {}", e.getMessage());
|
||||
return List.of();
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Stage 2: Retrieve + Draft ====================
|
||||
|
||||
private List<Section> draftStage(Long kbId, List<SubQuestion> questions, int topK, String sessionId) {
|
||||
Semaphore semaphore = new Semaphore(MAX_PARALLEL_QUESTIONS);
|
||||
List<CompletableFuture<Section>> futures = new ArrayList<>(questions.size());
|
||||
|
||||
for (int i = 0; i < questions.size(); i++) {
|
||||
final int idx = i;
|
||||
final SubQuestion q = questions.get(i);
|
||||
futures.add(CompletableFuture.supplyAsync(() -> {
|
||||
try {
|
||||
semaphore.acquire();
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
return new Section(q.question, "", List.of());
|
||||
}
|
||||
try {
|
||||
Section section = draftOneSection(kbId, q, topK);
|
||||
broadcast(sessionId, "research.draft", Map.of(
|
||||
"index", idx,
|
||||
"question", q.question,
|
||||
"content", section.content,
|
||||
"materialRefs", section.materialRefs
|
||||
));
|
||||
return section;
|
||||
} finally {
|
||||
semaphore.release();
|
||||
}
|
||||
}, EXECUTOR));
|
||||
}
|
||||
|
||||
return futures.stream()
|
||||
.map(CompletableFuture::join)
|
||||
.toList();
|
||||
}
|
||||
|
||||
private Section draftOneSection(Long kbId, SubQuestion q, int topK) {
|
||||
// 检索 chunk 级材料片段
|
||||
List<HybridRetriever.ChunkHit> hits = hybridRetriever.searchChunks(kbId, q.question, topK);
|
||||
|
||||
if (hits.isEmpty()) {
|
||||
return new Section(q.question, "现有材料中未找到与该问题相关的内容。", List.of());
|
||||
}
|
||||
|
||||
// 装配材料文本（带编号）
|
||||
StringBuilder materials = new StringBuilder();
|
||||
List<MaterialRef> refs = new ArrayList<>();
|
||||
Map<Long, String> rawTitleCache = new HashMap<>();
|
||||
|
||||
for (int i = 0; i < hits.size(); i++) {
|
||||
HybridRetriever.ChunkHit hit = hits.get(i);
|
||||
String rawTitle = rawTitleCache.computeIfAbsent(hit.rawId(), id -> {
|
||||
WikiRawMaterialEntity raw = rawService.getById(id);
|
||||
return raw != null ? raw.getTitle() : "unknown";
|
||||
});
|
||||
materials.append("### 材料 ").append(i + 1)
|
||||
.append("（来自《").append(rawTitle).append("》）\n")
|
||||
.append(hit.snippet())
|
||||
.append("\n\n");
|
||||
refs.add(new MaterialRef(i + 1, hit.chunkId(), hit.rawId(), rawTitle));
|
||||
}
|
||||
|
||||
String systemPrompt = PromptLoader.loadPrompt("research/draft-system");
|
||||
String userPrompt = PromptLoader.loadPrompt("research/draft-user")
|
||||
.replace("{question}", q.question)
|
||||
.replace("{intent}", q.intent != null ? q.intent : "")
|
||||
.replace("{materials}", materials.toString());
|
||||
|
||||
String content = callLlm(systemPrompt, userPrompt, "draft: " + q.question);
|
||||
if (content == null || content.isBlank()) {
|
||||
content = "现有材料不足以回答该子问题。";
|
||||
}
|
||||
|
||||
return new Section(q.question, content, refs);
|
||||
}
|
||||
|
||||
// ==================== Stage 3: Compose ====================
|
||||
|
||||
private String composeStage(String topic, List<Section> sections) {
|
||||
StringBuilder sectionsText = new StringBuilder();
|
||||
LinkedHashMap<Integer, String> usedMaterials = new LinkedHashMap<>();
|
||||
|
||||
for (int i = 0; i < sections.size(); i++) {
|
||||
Section s = sections.get(i);
|
||||
sectionsText.append("### 子问题 ").append(i + 1).append("：").append(s.question).append("\n");
|
||||
sectionsText.append(s.content).append("\n\n");
|
||||
for (MaterialRef ref : s.materialRefs) {
|
||||
usedMaterials.putIfAbsent(ref.index, ref.rawTitle);
|
||||
}
|
||||
}
|
||||
|
||||
StringBuilder materialsRef = new StringBuilder();
|
||||
usedMaterials.forEach((idx, title) ->
|
||||
materialsRef.append("- 材料 ").append(idx).append("：").append(title).append("\n"));
|
||||
|
||||
String systemPrompt = PromptLoader.loadPrompt("research/compose-system");
|
||||
String userPrompt = PromptLoader.loadPrompt("research/compose-user")
|
||||
.replace("{topic}", topic)
|
||||
.replace("{sections}", sectionsText.toString())
|
||||
.replace("{materials_ref}", materialsRef.toString());
|
||||
|
||||
String report = callLlm(systemPrompt, userPrompt, "compose");
|
||||
if (report == null || report.isBlank()) {
|
||||
// 降级：直接拼接段落
|
||||
return "# " + topic + "\n\n" + sectionsText;
|
||||
}
|
||||
return report;
|
||||
}
|
||||
|
||||
// ==================== Helpers ====================
|
||||
|
||||
private String callLlm(String systemPrompt, String userPrompt, String ctx) {
|
||||
try {
|
||||
ChatModel chatModel = buildChatModel();
|
||||
Prompt prompt = new Prompt(List.of(
|
||||
new SystemMessage(systemPrompt),
|
||||
new UserMessage(userPrompt)));
|
||||
ChatResponse response = chatModel.call(prompt);
|
||||
if (response == null || response.getResult() == null
|
||||
|| response.getResult().getOutput() == null) {
|
||||
return null;
|
||||
}
|
||||
return response.getResult().getOutput().getText();
|
||||
} catch (Exception e) {
|
||||
log.error("[Research] LLM call failed ({}): {}", ctx, e.getMessage());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private ChatModel buildChatModel() {
|
||||
ModelConfigEntity model = modelConfigService.getDefaultModel();
|
||||
return agentGraphBuilder.buildRuntimeChatModel(model, NO_RETRY);
|
||||
}
|
||||
|
||||
private void broadcast(String sessionId, String eventName, Map<String, Object> payload) {
|
||||
if (sessionId == null || sessionId.isBlank()) return;
|
||||
try {
|
||||
streamTracker.broadcast(sessionId, eventName, JSONUtil.toJsonStr(payload));
|
||||
} catch (Exception e) {
|
||||
log.debug("[Research] SSE broadcast failed for {}: {}", sessionId, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private String stripCodeFences(String text) {
|
||||
if (text == null) return "";
|
||||
String t = text.strip();
|
||||
if (t.startsWith("```json")) t = t.substring(7);
|
||||
else if (t.startsWith("```")) t = t.substring(3);
|
||||
if (t.endsWith("```")) t = t.substring(0, t.length() - 3);
|
||||
return t.strip();
|
||||
}
|
||||
|
||||
// ==================== DTO ====================
|
||||
|
||||
public record SubQuestion(String question, String intent) {}
|
||||
|
||||
public record MaterialRef(int index, Long chunkId, Long rawId, String rawTitle) {}
|
||||
|
||||
public record Section(String question, String content, List<MaterialRef> materialRefs) {}
|
||||
|
||||
public record ResearchResult(String topic, List<Section> sections, String report) {}
|
||||
}
|
||||
@ -13,6 +13,7 @@ import org.springframework.stereotype.Component;
|
||||
import vip.mate.wiki.model.WikiKnowledgeBaseEntity;
|
||||
import vip.mate.wiki.model.WikiPageEntity;
|
||||
import vip.mate.wiki.model.WikiRawMaterialEntity;
|
||||
import vip.mate.wiki.service.HybridRetriever;
|
||||
import vip.mate.wiki.service.WikiKnowledgeBaseService;
|
||||
import vip.mate.wiki.service.WikiPageService;
|
||||
import vip.mate.wiki.service.WikiRawMaterialService;
|
||||
@ -35,6 +36,7 @@ public class WikiTool {
|
||||
private final WikiPageService pageService;
|
||||
private final WikiKnowledgeBaseService kbService;
|
||||
private final WikiRawMaterialService rawService;
|
||||
private final HybridRetriever hybridRetriever;
|
||||
|
||||
@Tool(description = """
|
||||
读取 Wiki 知识库中指定页面的完整内容。
|
||||
@ -102,11 +104,13 @@ public class WikiTool {
|
||||
|
||||
@Tool(description = """
|
||||
在 Wiki 知识库中搜索页面。
|
||||
按关键词搜索页面标题、摘要和正文内容，返回匹配的页面列表及其来源文件。
|
||||
支持三种模式：keyword（关键词匹配）、semantic（语义向量相似度）、hybrid（两者融合，默认）。
|
||||
返回匹配的页面列表及其来源文件。
|
||||
""")
|
||||
public String wiki_search_pages(
|
||||
@ToolParam(description = "当前 Agent 的 ID") Long agentId,
|
||||
@ToolParam(description = "搜索关键词") String query) {
|
||||
@ToolParam(description = "搜索关键词或自然语言问题") String query,
|
||||
@ToolParam(description = "搜索模式：keyword | semantic | hybrid（默认 hybrid）", required = false) String mode) {
|
||||
|
||||
if (query == null || query.isBlank()) {
|
||||
return error("query is required");
|
||||
@ -117,32 +121,82 @@ public class WikiTool {
|
||||
return error("No wiki knowledge base found for this agent");
|
||||
}
|
||||
|
||||
// DB 级别搜索（不加载 content CLOB 到 Java 内存）
|
||||
List<WikiPageEntity> matched = pageService.searchPages(kbId, query);
|
||||
// RFC-011：走混合检索
|
||||
List<HybridRetriever.PageHit> hits = hybridRetriever.searchPages(kbId, query, mode, 20);
|
||||
|
||||
// Agent 引用追踪（搜索结果中的页面都算被引用）
|
||||
for (WikiPageEntity p : matched) {
|
||||
pageService.trackReference(kbId, p.getSlug());
|
||||
// Agent 引用追踪
|
||||
for (HybridRetriever.PageHit h : hits) {
|
||||
pageService.trackReference(kbId, h.slug());
|
||||
}
|
||||
|
||||
JSONArray arr = new JSONArray();
|
||||
for (WikiPageEntity page : matched) {
|
||||
JSONObject obj = JSONUtil.createObj()
|
||||
.set("title", page.getTitle())
|
||||
.set("slug", page.getSlug())
|
||||
.set("summary", page.getSummary())
|
||||
.set("sourceFiles", resolveSourceFiles(page.getSourceRawIds()));
|
||||
arr.add(obj);
|
||||
for (HybridRetriever.PageHit hit : hits) {
|
||||
arr.add(JSONUtil.createObj()
|
||||
.set("title", hit.title())
|
||||
.set("slug", hit.slug())
|
||||
.set("summary", hit.summary())
|
||||
.set("score", String.format("%.4f", hit.score())));
|
||||
}
|
||||
|
||||
return JSONUtil.createObj()
|
||||
.set("kbId", kbId)
|
||||
.set("query", query)
|
||||
.set("matchCount", matched.size())
|
||||
.set("mode", mode != null ? mode : "hybrid")
|
||||
.set("matchCount", hits.size())
|
||||
.set("pages", arr)
|
||||
.toString();
|
||||
}
|
||||
|
||||
@Tool(description = """
|
||||
在 Wiki 知识库中进行 chunk 级语义搜索。
|
||||
返回与查询语义最接近的原始文本片段（chunk），包含相似度分数。
|
||||
当 wiki_search_pages 返回的页面摘要不够具体时，使用此工具获取精确的源文本证据。
|
||||
""")
|
||||
public String wiki_semantic_search(
|
||||
@ToolParam(description = "当前 Agent 的 ID") Long agentId,
|
||||
@ToolParam(description = "自然语言查询") String query,
|
||||
@ToolParam(description = "返回条数（默认 5）", required = false) Integer topK) {
|
||||
|
||||
if (query == null || query.isBlank()) {
|
||||
return error("query is required");
|
||||
}
|
||||
|
||||
Long kbId = resolveKbId(agentId);
|
||||
if (kbId == null) {
|
||||
return error("No wiki knowledge base found for this agent");
|
||||
}
|
||||
|
||||
int k = (topK != null && topK > 0) ? Math.min(topK, 20) : 5;
|
||||
List<HybridRetriever.ChunkHit> hits = hybridRetriever.searchChunks(kbId, query, k);
|
||||
|
||||
if (hits.isEmpty()) {
|
||||
return JSONUtil.createObj()
|
||||
.set("kbId", kbId)
|
||||
.set("query", query)
|
||||
.set("matchCount", 0)
|
||||
.set("message", "No semantic matches found. Try wiki_search_pages with mode=keyword.")
|
||||
.toString();
|
||||
}
|
||||
|
||||
JSONArray arr = new JSONArray();
|
||||
for (HybridRetriever.ChunkHit hit : hits) {
|
||||
// 解析 raw material 标题
|
||||
WikiRawMaterialEntity raw = rawService.getById(hit.rawId());
|
||||
arr.add(JSONUtil.createObj()
|
||||
.set("chunkId", hit.chunkId())
|
||||
.set("rawTitle", raw != null ? raw.getTitle() : "unknown")
|
||||
.set("snippet", hit.snippet())
|
||||
.set("score", String.format("%.4f", hit.score())));
|
||||
}
|
||||
|
||||
return JSONUtil.createObj()
|
||||
.set("kbId", kbId)
|
||||
.set("query", query)
|
||||
.set("matchCount", hits.size())
|
||||
.set("chunks", arr)
|
||||
.toString();
|
||||
}
|
||||
|
||||
@Tool(description = """
|
||||
追溯 Wiki 页面的来源原始文件。
|
||||
查询指定页面是由哪些原始文档生成的，返回文件名、类型、路径等信息。
|
||||
|
||||
@ -0,0 +1,23 @@
|
||||
-- V12: Wiki chunk persistence (RFC-013 minimal slice → enables RFC-011 embedding)
|
||||
-- Chunks are persisted after splitIntoChunks(), enabling:
|
||||
-- 1. chunk-level incremental reprocessing (hash compare)
|
||||
-- 2. future embedding storage (ALTER ADD embedding BLOB in Phase 2)
|
||||
-- 3. fine-grained FTS indexing
|
||||
|
||||
CREATE TABLE IF NOT EXISTS mate_wiki_chunk (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
kb_id BIGINT NOT NULL,
|
||||
raw_id BIGINT NOT NULL,
|
||||
ordinal INT NOT NULL,
|
||||
content TEXT NOT NULL,
|
||||
char_count INT NOT NULL,
|
||||
start_offset INT NOT NULL,
|
||||
end_offset INT NOT NULL,
|
||||
content_hash VARCHAR(64) NOT NULL,
|
||||
create_time DATETIME NOT NULL,
|
||||
update_time DATETIME NOT NULL,
|
||||
deleted INT NOT NULL DEFAULT 0
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_wiki_chunk_kb ON mate_wiki_chunk(kb_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_wiki_chunk_raw ON mate_wiki_chunk(raw_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_wiki_chunk_hash ON mate_wiki_chunk(content_hash);
|
||||
@ -0,0 +1,4 @@
|
||||
-- V13: Add embedding column to mate_wiki_chunk (RFC-011 Phase 2)
|
||||
-- float32[] serialized as little-endian byte[], stored in BLOB
|
||||
ALTER TABLE mate_wiki_chunk ADD COLUMN IF NOT EXISTS embedding BLOB DEFAULT NULL;
|
||||
ALTER TABLE mate_wiki_chunk ADD COLUMN IF NOT EXISTS embedding_model VARCHAR(64) DEFAULT NULL;
|
||||
@ -0,0 +1,18 @@
|
||||
-- V12: Wiki chunk persistence (RFC-013 minimal slice → enables RFC-011 embedding)
|
||||
CREATE TABLE IF NOT EXISTS mate_wiki_chunk (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
kb_id BIGINT NOT NULL,
|
||||
raw_id BIGINT NOT NULL,
|
||||
ordinal INT NOT NULL,
|
||||
content TEXT NOT NULL,
|
||||
char_count INT NOT NULL,
|
||||
start_offset INT NOT NULL,
|
||||
end_offset INT NOT NULL,
|
||||
content_hash VARCHAR(64) NOT NULL,
|
||||
create_time DATETIME NOT NULL,
|
||||
update_time DATETIME NOT NULL,
|
||||
deleted INT NOT NULL DEFAULT 0
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_wiki_chunk_kb ON mate_wiki_chunk(kb_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_wiki_chunk_raw ON mate_wiki_chunk(raw_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_wiki_chunk_hash ON mate_wiki_chunk(content_hash);
|
||||
@ -0,0 +1,3 @@
|
||||
-- V13: Add embedding column to mate_wiki_chunk (RFC-011 Phase 2)
|
||||
ALTER TABLE mate_wiki_chunk ADD COLUMN IF NOT EXISTS embedding BLOB DEFAULT NULL;
|
||||
ALTER TABLE mate_wiki_chunk ADD COLUMN IF NOT EXISTS embedding_model VARCHAR(64) DEFAULT NULL;
|
||||
@ -0,0 +1,11 @@
|
||||
你是一个研究助手。基于**已写好的若干段落**（每段对应一个子问题的回答），组装成一份结构清晰的综合研究报告。
|
||||
|
||||
规则：
|
||||
1. **不要重写段落内容**——只做：组织顺序、加标题、去重
|
||||
2. 开头加一句话概述（基于主题）
|
||||
3. 每个子问题作为一个二级标题（## 标题）
|
||||
4. 段落之间做必要的衔接，但不要引入新信息
|
||||
5. 末尾加一段"### 参考材料"，列出所有段落中引用过的材料序号和对应的材料标题
|
||||
6. 输出 Markdown 格式
|
||||
|
||||
如果某些段落说"材料不足"，在报告中保留这个声明，不要掩盖。
|
||||
@ -0,0 +1,14 @@
|
||||
## 研究主题
|
||||
{topic}
|
||||
|
||||
## 已写好的段落
|
||||
|
||||
{sections}
|
||||
|
||||
## 使用过的材料
|
||||
|
||||
{materials_ref}
|
||||
|
||||
---
|
||||
|
||||
请组装为一份 Markdown 格式的综合研究报告。
|
||||
@ -0,0 +1,10 @@
|
||||
你是一个研究助手。基于提供的**材料片段**，针对**子问题**写一段 150-300 字的中文回答。
|
||||
|
||||
规则：
|
||||
1. **只基于提供的材料片段**——不要引入外部知识、不要虚构
|
||||
2. **如果材料不足以回答，明确说"现有材料不足以回答"**，不要强行编造
|
||||
3. 语言简洁准确，适合作为综合报告的一节
|
||||
4. 不要使用 markdown 标题（##），输出纯段落文本
|
||||
5. 尽量在段末注明使用了哪个材料片段的序号，格式 `[材料 1]`、`[材料 2, 3]`
|
||||
|
||||
输出只包含正文段落，不要任何额外说明。
|
||||
@ -0,0 +1,13 @@
|
||||
## 子问题
|
||||
{question}
|
||||
|
||||
## 意图
|
||||
{intent}
|
||||
|
||||
## 材料片段（按相关度排序）
|
||||
|
||||
{materials}
|
||||
|
||||
---
|
||||
|
||||
请基于上述材料片段写出 150-300 字的回答段落。
|
||||
@ -0,0 +1,16 @@
|
||||
你是一个研究助手。针对用户提出的主题，把它分解为 3-5 个子问题，每个子问题应该：
|
||||
|
||||
1. 独立可检索——能在知识库中找到直接相关的材料
|
||||
2. 互相覆盖主题的不同侧面——避免重复，保证广度
|
||||
3. 简短明确——一个子问题一句话表达
|
||||
|
||||
严格输出 JSON（不要 markdown 代码块包裹）：
|
||||
|
||||
{
|
||||
"questions": [
|
||||
{"question": "子问题 1", "intent": "这个问题意在了解什么"},
|
||||
{"question": "子问题 2", "intent": "..."}
|
||||
]
|
||||
}
|
||||
|
||||
如果主题太宽泛或无法分解，questions 可以是空数组。
|
||||
@ -0,0 +1,3 @@
|
||||
研究主题：{topic}
|
||||
|
||||
请拆解为 3-5 个可独立检索的子问题。
|
||||
Loading…
Reference in New Issue
Block a user