diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java b/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java index 312da9fa..939a01a6 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java @@ -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"; } diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiResearchController.java b/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiResearchController.java new file mode 100644 index 00000000..79b55ed8 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiResearchController.java @@ -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> startResearch( + @RequestBody Map 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; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiChunkEntity.java b/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiChunkEntity.java new file mode 100644 index 00000000..1c0f28c2 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiChunkEntity.java @@ -0,0 +1,62 @@ +package vip.mate.wiki.model; + +import com.baomidou.mybatisplus.annotation.*; +import lombok.Data; + +import java.time.LocalDateTime; + +/** + * Wiki chunk 实体 + *

+ * 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; +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiChunkMapper.java b/mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiChunkMapper.java new file mode 100644 index 00000000..0b07c55b --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiChunkMapper.java @@ -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 { +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/HybridRetriever.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/HybridRetriever.java new file mode 100644 index 00000000..1d590b2e --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/HybridRetriever.java @@ -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: 混合检索服务 + *

+ * 支持三种模式： + *

+ * + * @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 searchPages(Long kbId, String query, String modeStr, int topK) { + Mode mode = parseMode(modeStr); + + List semantic = List.of(); + List 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 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 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 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 semanticSearch(Long kbId, String query, int limit) { + float[] queryVec = embeddingService.embedQuery(query); + if (queryVec == null) return List.of(); + + List 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 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 allPages = pageService.listByKbId(kbId); + Map 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.comparingByValue().reversed()) + .limit(limit) + .map(e -> new RankedItem(e.getKey(), e.getValue())) + .toList(); + } + + /** 关键词搜索：走现有 DB LIKE */ + private List keywordSearch(Long kbId, String query, int limit) { + List results = pageService.searchPages(kbId, query); + List 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 rrfFuse(List a, List b, int k) { + Map 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.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) {} +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiChunkService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiChunkService.java new file mode 100644 index 00000000..a29af317 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiChunkService.java @@ -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 服务 + *

+ * RFC-013 最小切片：chunk 持久化 + hash 级增量对账。 + *

    + *
  • {@link #persistChunks} — 切分后一次性入库，返回 chunk ID 列表供后续流程使用
    • + *
  • {@link #reconcile} — 增量对账：hash 相同的 chunk 保留（含 embedding），hash 不同的重建
    • + *
  • {@link #deleteByRawId} — 材料删除时级联清理
    • + *
+ * + * @author MateClaw Team + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class WikiChunkService { + + private final WikiChunkMapper chunkMapper; + + /** + * 将文本切片列表持久化为 chunk 记录。 + *

+ * 如果该 rawId 已有 chunk 记录，走 {@link #reconcile} 增量对账；否则全量插入。 + * + * @param kbId 知识库 ID + * @param rawId 原始材料 ID + * @param chunks 切分后的文本列表（有序） + * @param offsets 每个 chunk 对应的 [startOffset, endOffset] 数组 + * @return 持久化后的 chunk ID 列表（与 chunks 同序） + */ + @Transactional + public List persistChunks(Long kbId, Long rawId, List chunks, List offsets) { + List 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 reconcile(Long kbId, Long rawId, List chunks, List offsets, + List existing) { + // 旧 chunk 按 ordinal 索引 + Map oldByOrdinal = new HashMap<>(); + for (WikiChunkEntity e : existing) { + oldByOrdinal.put(e.getOrdinal(), e); + } + + List resultIds = new ArrayList<>(chunks.size()); + Set 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 insertAll(Long kbId, Long rawId, List chunks, List offsets) { + List 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 listByRawId(Long rawId) { + return chunkMapper.selectList( + new LambdaQueryWrapper() + .eq(WikiChunkEntity::getRawId, rawId) + .orderByAsc(WikiChunkEntity::getOrdinal)); + } + + public List listByKbId(Long kbId) { + return chunkMapper.selectList( + new LambdaQueryWrapper() + .eq(WikiChunkEntity::getKbId, kbId) + .orderByAsc(WikiChunkEntity::getRawId) + .orderByAsc(WikiChunkEntity::getOrdinal)); + } + + @Transactional + public void deleteByRawId(Long rawId) { + int deleted = chunkMapper.delete( + new LambdaQueryWrapper() + .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() + .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(); + } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiEmbeddingService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiEmbeddingService.java new file mode 100644 index 00000000..dd94540a --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiEmbeddingService.java @@ -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 嵌入服务 + *

+ * 使用 Spring AI 的 {@link EmbeddingModel}（DashScope auto-config）对 chunk 做向量化。 + *

    + *
  • {@link #embedMissingChunks} — 批量嵌入缺失 embedding 的 chunk（材料处理后异步调用）
    • + *
  • {@link #embedQuery} — 查询向量化（混合搜索时调用）
    • + *
+ * + * @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 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。 + *

+ * 只嵌入 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 pending = chunkMapper.selectList( + new LambdaQueryWrapper() + .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 batch = pending.subList(offset, Math.min(offset + batchSize, pending.size())); + try { + List 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() + .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; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java index b2b36adc..a5c5af2e 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java @@ -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 chunks = splitIntoChunks(text); + // Phase 1: 切分文本为 chunks（带偏移，供持久化） + List chunksWithOffset = splitIntoChunksWithOffsets(text); + List 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 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 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 splitIntoChunksWithOffsets(String text) { int chunkSize = properties.getMaxChunkSize(); int overlap = Math.min(500, chunkSize / 10); - List chunks = new ArrayList<>(); + List 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; diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java index d2bb4781..010a53e5 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java @@ -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()); + } } /** diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiResearchService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiResearchService.java new file mode 100644 index 00000000..38b0aa10 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiResearchService.java @@ -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 服务 + *

+ * 三阶段管线（不走 StateGraph 框架，保持轻量）： + *

    + *
  1. Plan：LLM 把 topic 拆为 3-5 个子问题
    2. + *
  2. Retrieve + Draft：并行对每个子问题调 {@link HybridRetriever} + LLM 起草段落
    3. + *
  3. Compose：LLM 把段落组装为最终报告
    4. + *
+ * 事件通过 {@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 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
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 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 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
draftStage(Long kbId, List questions, int topK, String sessionId) { + Semaphore semaphore = new Semaphore(MAX_PARALLEL_QUESTIONS); + List> 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 hits = hybridRetriever.searchChunks(kbId, q.question, topK); + + if (hits.isEmpty()) { + return new Section(q.question, "现有材料中未找到与该问题相关的内容。", List.of()); + } + + // 装配材料文本（带编号） + StringBuilder materials = new StringBuilder(); + List refs = new ArrayList<>(); + Map 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
sections) { + StringBuilder sectionsText = new StringBuilder(); + LinkedHashMap 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 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 materialRefs) {} + + public record ResearchResult(String topic, List
sections, String report) {} +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/tool/WikiTool.java b/mateclaw-server/src/main/java/vip/mate/wiki/tool/WikiTool.java index dbdcb921..bf680716 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/tool/WikiTool.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/tool/WikiTool.java @@ -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 matched = pageService.searchPages(kbId, query); + // RFC-011：走混合检索 + List 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 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 页面的来源原始文件。 查询指定页面是由哪些原始文档生成的，返回文件名、类型、路径等信息。 diff --git a/mateclaw-server/src/main/resources/db/migration/h2/V12__wiki_chunk_table.sql b/mateclaw-server/src/main/resources/db/migration/h2/V12__wiki_chunk_table.sql new file mode 100644 index 00000000..981975b4 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/h2/V12__wiki_chunk_table.sql @@ -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); diff --git a/mateclaw-server/src/main/resources/db/migration/h2/V13__wiki_chunk_embedding.sql b/mateclaw-server/src/main/resources/db/migration/h2/V13__wiki_chunk_embedding.sql new file mode 100644 index 00000000..15ef8f19 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/h2/V13__wiki_chunk_embedding.sql @@ -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; diff --git a/mateclaw-server/src/main/resources/db/migration/mysql/V12__wiki_chunk_table.sql b/mateclaw-server/src/main/resources/db/migration/mysql/V12__wiki_chunk_table.sql new file mode 100644 index 00000000..0dd23f97 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/mysql/V12__wiki_chunk_table.sql @@ -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); diff --git a/mateclaw-server/src/main/resources/db/migration/mysql/V13__wiki_chunk_embedding.sql b/mateclaw-server/src/main/resources/db/migration/mysql/V13__wiki_chunk_embedding.sql new file mode 100644 index 00000000..7ea10c8b --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/mysql/V13__wiki_chunk_embedding.sql @@ -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; diff --git a/mateclaw-server/src/main/resources/prompts/research/compose-system.txt b/mateclaw-server/src/main/resources/prompts/research/compose-system.txt new file mode 100644 index 00000000..427a9854 --- /dev/null +++ b/mateclaw-server/src/main/resources/prompts/research/compose-system.txt @@ -0,0 +1,11 @@ +你是一个研究助手。基于**已写好的若干段落**（每段对应一个子问题的回答），组装成一份结构清晰的综合研究报告。 + +规则： +1. **不要重写段落内容**——只做：组织顺序、加标题、去重 +2. 开头加一句话概述（基于主题） +3. 每个子问题作为一个二级标题（## 标题） +4. 段落之间做必要的衔接，但不要引入新信息 +5. 末尾加一段"### 参考材料"，列出所有段落中引用过的材料序号和对应的材料标题 +6. 输出 Markdown 格式 + +如果某些段落说"材料不足"，在报告中保留这个声明，不要掩盖。 diff --git a/mateclaw-server/src/main/resources/prompts/research/compose-user.txt b/mateclaw-server/src/main/resources/prompts/research/compose-user.txt new file mode 100644 index 00000000..0bd43e3e --- /dev/null +++ b/mateclaw-server/src/main/resources/prompts/research/compose-user.txt @@ -0,0 +1,14 @@ +## 研究主题 +{topic} + +## 已写好的段落 + +{sections} + +## 使用过的材料 + +{materials_ref} + +--- + +请组装为一份 Markdown 格式的综合研究报告。 diff --git a/mateclaw-server/src/main/resources/prompts/research/draft-system.txt b/mateclaw-server/src/main/resources/prompts/research/draft-system.txt new file mode 100644 index 00000000..33ec1cd9 --- /dev/null +++ b/mateclaw-server/src/main/resources/prompts/research/draft-system.txt @@ -0,0 +1,10 @@ +你是一个研究助手。基于提供的**材料片段**，针对**子问题**写一段 150-300 字的中文回答。 + +规则： +1. **只基于提供的材料片段**——不要引入外部知识、不要虚构 +2. **如果材料不足以回答，明确说"现有材料不足以回答"**，不要强行编造 +3. 语言简洁准确，适合作为综合报告的一节 +4. 不要使用 markdown 标题（##），输出纯段落文本 +5. 尽量在段末注明使用了哪个材料片段的序号，格式 `[材料 1]`、`[材料 2, 3]` + +输出只包含正文段落，不要任何额外说明。 diff --git a/mateclaw-server/src/main/resources/prompts/research/draft-user.txt b/mateclaw-server/src/main/resources/prompts/research/draft-user.txt new file mode 100644 index 00000000..78856afc --- /dev/null +++ b/mateclaw-server/src/main/resources/prompts/research/draft-user.txt @@ -0,0 +1,13 @@ +## 子问题 +{question} + +## 意图 +{intent} + +## 材料片段（按相关度排序） + +{materials} + +--- + +请基于上述材料片段写出 150-300 字的回答段落。 diff --git a/mateclaw-server/src/main/resources/prompts/research/plan-system.txt b/mateclaw-server/src/main/resources/prompts/research/plan-system.txt new file mode 100644 index 00000000..76765f3f --- /dev/null +++ b/mateclaw-server/src/main/resources/prompts/research/plan-system.txt @@ -0,0 +1,16 @@ +你是一个研究助手。针对用户提出的主题，把它分解为 3-5 个子问题，每个子问题应该： + +1. 独立可检索——能在知识库中找到直接相关的材料 +2. 互相覆盖主题的不同侧面——避免重复，保证广度 +3. 简短明确——一个子问题一句话表达 + +严格输出 JSON（不要 markdown 代码块包裹）： + +{ + "questions": [ + {"question": "子问题 1", "intent": "这个问题意在了解什么"}, + {"question": "子问题 2", "intent": "..."} + ] +} + +如果主题太宽泛或无法分解，questions 可以是空数组。 diff --git a/mateclaw-server/src/main/resources/prompts/research/plan-user.txt b/mateclaw-server/src/main/resources/prompts/research/plan-user.txt new file mode 100644 index 00000000..6593169f --- /dev/null +++ b/mateclaw-server/src/main/resources/prompts/research/plan-user.txt @@ -0,0 +1,3 @@ +研究主题：{topic} + +请拆解为 3-5 个可独立检索的子问题。