From 90936dba4f8dbf8e9b0f5479bcd00fbf7eacad7c Mon Sep 17 00:00:00 2001 From: matevip Date: Tue, 12 May 2026 14:10:54 +0800 Subject: [PATCH] =?UTF-8?q?feat(wiki):=20cross-material=20transformation?= =?UTF-8?q?=20aggregator=20=E2=80=94=20map-reduce=20all=20runs=20of=20a=20?= =?UTF-8?q?template=20into=20one=20KB=20page?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../WikiTransformationController.java | 32 +++ .../service/WikiTransformationAggregator.java | 193 ++++++++++++++++++ .../java/vip/mate/wiki/tool/WikiTool.java | 52 +++++ .../wiki/transformation-aggregate-system.txt | 16 ++ .../wiki/transformation-aggregate-user.txt | 9 + mateclaw-ui/src/api/index.ts | 4 + mateclaw-ui/src/i18n/locales/en-US.ts | 5 + mateclaw-ui/src/i18n/locales/zh-CN.ts | 5 + .../Wiki/components/TransformationsPanel.vue | 29 +++ 9 files changed, 345 insertions(+) create mode 100644 mateclaw-server/src/main/java/vip/mate/wiki/service/WikiTransformationAggregator.java create mode 100644 mateclaw-server/src/main/resources/prompts/wiki/transformation-aggregate-system.txt create mode 100644 mateclaw-server/src/main/resources/prompts/wiki/transformation-aggregate-user.txt diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiTransformationController.java b/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiTransformationController.java index 3f7084e2..be3cb422 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiTransformationController.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiTransformationController.java @@ -11,6 +11,7 @@ import vip.mate.wiki.model.WikiKnowledgeBaseEntity; import vip.mate.wiki.model.WikiTransformationEntity; import vip.mate.wiki.model.WikiTransformationRunEntity; import vip.mate.wiki.service.WikiKnowledgeBaseService; +import vip.mate.wiki.service.WikiTransformationAggregator; import vip.mate.wiki.service.WikiTransformationExecutor; import vip.mate.wiki.service.WikiTransformationService; import vip.mate.workspace.core.annotation.RequireWorkspaceRole; @@ -34,6 +35,7 @@ public class WikiTransformationController { private final WikiTransformationService transformationService; private final WikiTransformationExecutor executor; + private final WikiTransformationAggregator aggregator; private final WikiKnowledgeBaseService kbService; // ==================== Templates ==================== @@ -136,6 +138,36 @@ public class WikiTransformationController { return R.ok(); } + @RequireWorkspaceRole("member") + @Operation(summary = "Aggregate all completed runs of a template into one KB-level synthesis page", + description = "Map-reduces across every completed run of the template within the given KB. " + + "Upserts the merged document at slug '-aggregate'.") + @PostMapping("/{id}/aggregate") + public R> aggregate(@PathVariable Long id, + @RequestParam Long kbId, + @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) { + WikiTransformationEntity t = transformationService.getById(id); + if (t == null) return R.fail("Transformation not found"); + verifyTemplateWorkspace(t, workspaceId); + verifyKBWorkspace(kbId, workspaceId != null ? workspaceId : 1L); + + try { + WikiTransformationAggregator.Result res = aggregator.aggregate(t, kbId, "manual"); + if (res.pageId() == null) { + return R.fail(res.title()); // when sources are empty we put the reason in title field + } + return R.ok(Map.of( + "pageId", res.pageId(), + "slug", res.slug(), + "title", res.title(), + "sourcesUsed", res.sourcesUsed(), + "charsFed", res.charsFed(), + "created", res.created())); + } catch (IllegalStateException | IllegalArgumentException e) { + return R.fail(e.getMessage()); + } + } + // ==================== Runs ==================== @RequireWorkspaceRole("viewer") diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiTransformationAggregator.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiTransformationAggregator.java new file mode 100644 index 00000000..5e21a5f4 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiTransformationAggregator.java @@ -0,0 +1,193 @@ +package vip.mate.wiki.service; + +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.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; +import vip.mate.agent.prompt.PromptLoader; +import vip.mate.wiki.job.WikiJobStep; +import vip.mate.wiki.job.WikiModelRoutingService; +import vip.mate.wiki.model.WikiPageEntity; +import vip.mate.wiki.model.WikiRawMaterialEntity; +import vip.mate.wiki.model.WikiTransformationEntity; +import vip.mate.wiki.model.WikiTransformationRunEntity; + +import java.util.ArrayList; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * Map-reduce a single transformation template across all completed runs in + * a KB: produces one synthesis wiki page that unifies the per-source + * outputs. The map step is already done by the executor — each run carries + * its per-source output. This service is the reduce step: load the runs, + * stack them with source labels, ask an LLM to merge + dedupe, and persist + * the merged document as a synthesis page slugged + * {@code -aggregate}. + * + *

Idempotent: re-running upserts the same slug, so the aggregate page + * stays current as new runs land. Page-level embedding + reverse-citation + * extraction run the same way they do for single-source synthesis pages. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class WikiTransformationAggregator { + + /** Hard cap on combined input chars fed to the LLM merge call. */ + private static final int MAX_AGG_INPUT_CHARS = 80_000; + + /** Per-source output is truncated to keep the merge prompt within the cap when many sources exist. */ + private static final int PER_SOURCE_SOFT_CAP = 12_000; + + private final WikiTransformationService transformationService; + private final WikiRawMaterialService rawService; + private final WikiPageService pageService; + + @Autowired(required = false) + private WikiModelRoutingService modelRoutingService; + + @Autowired(required = false) + private WikiEmbeddingService embeddingService; + + private final com.fasterxml.jackson.databind.ObjectMapper objectMapper = + new com.fasterxml.jackson.databind.ObjectMapper(); + + public record Result(Long pageId, String slug, String title, + int sourcesUsed, int charsFed, boolean created) { + public static Result empty(String reason) { + return new Result(null, null, reason, 0, 0, false); + } + } + + public Result aggregate(WikiTransformationEntity template, Long kbId, String triggeredBy) { + if (template == null) throw new IllegalArgumentException("template is required"); + if (kbId == null) throw new IllegalArgumentException("kbId is required"); + if (modelRoutingService == null) throw new IllegalStateException("ModelRoutingService unavailable"); + + // Load every completed run for this template against this KB. Cap at + // 100 sources so a degenerate KB doesn't push past the input window. + List runs = transformationService + .listRunsByTransformation(template.getId(), 200) + .stream() + .filter(r -> "completed".equalsIgnoreCase(r.getStatus()) + && kbId.equals(r.getKbId()) + && r.getOutput() != null && !r.getOutput().isBlank() + && r.getRawId() != null) + .limit(100) + .toList(); + if (runs.isEmpty()) { + return Result.empty("no completed runs to aggregate"); + } + + // Deduplicate by rawId — only the most recent completed run per raw + // contributes, so a template that's been re-run several times against + // the same source doesn't get its old outputs included. + java.util.Map latestByRaw = new java.util.LinkedHashMap<>(); + for (WikiTransformationRunEntity r : runs) { + latestByRaw.putIfAbsent(r.getRawId(), r); // listRunsByTransformation orders DESC by createTime + } + List distinct = new ArrayList<>(latestByRaw.values()); + + // Build the per-source block, truncating each section to keep the + // merged prompt within the model's context window. + StringBuilder outputs = new StringBuilder(); + Set sourceRawIds = new LinkedHashSet<>(); + int totalChars = 0; + int sourcesIncluded = 0; + for (WikiTransformationRunEntity run : distinct) { + WikiRawMaterialEntity raw = rawService.getById(run.getRawId()); + String sourceTitle = raw != null && raw.getTitle() != null && !raw.getTitle().isBlank() + ? raw.getTitle() : ("raw#" + run.getRawId()); + String body = run.getOutput().length() > PER_SOURCE_SOFT_CAP + ? run.getOutput().substring(0, PER_SOURCE_SOFT_CAP) + "\n…(truncated for merge)" + : run.getOutput(); + String block = "### From: " + sourceTitle + "\n\n" + body + "\n\n---\n\n"; + if (totalChars + block.length() > MAX_AGG_INPUT_CHARS) { + log.warn("[WikiAggregator] template={} kb={} stopping merge at {} sources to stay under {} chars", + template.getName(), kbId, sourcesIncluded, MAX_AGG_INPUT_CHARS); + break; + } + outputs.append(block); + sourceRawIds.add(run.getRawId()); + totalChars += block.length(); + sourcesIncluded++; + } + if (sourcesIncluded == 0) return Result.empty("all source outputs were empty after dedup"); + + // LLM call + String systemPrompt = PromptLoader.loadPrompt("wiki/transformation-aggregate-system"); + String userPrompt = PromptLoader.loadPrompt("wiki/transformation-aggregate-user") + .replace("{template_title}", template.getTitle() == null ? template.getName() : template.getTitle()) + .replace("{template_description}", template.getDescription() == null ? "" : template.getDescription()) + .replace("{outputs}", outputs.toString()); + + Long modelId = template.getModelId() != null + ? template.getModelId() + : modelRoutingService.selectModelId(kbId, "heavy_ingest", WikiJobStep.CREATE_PAGE); + ChatModel chatModel = modelRoutingService.buildChatModel(modelId); + + ChatResponse resp = chatModel.call(new Prompt(List.of( + new SystemMessage(systemPrompt), new UserMessage(userPrompt)))); + String mergedOutput = (resp == null || resp.getResult() == null + || resp.getResult().getOutput() == null) ? null : resp.getResult().getOutput().getText(); + if (mergedOutput == null || mergedOutput.isBlank()) { + throw new IllegalStateException("Aggregator LLM returned empty output"); + } + mergedOutput = WikiTransformationExecutor.cleanLlmOutput(mergedOutput); + + // Upsert the aggregate page on a deterministic slug so re-aggregation + // refreshes it in place instead of spawning duplicates. + String slug = template.getName() + "-aggregate"; + String title = (template.getTitle() == null ? template.getName() : template.getTitle()) + + "（KB 聚合）"; + String summary = sourcesIncluded + " 个原始材料合并 · " + + (triggeredBy == null ? "manual" : triggeredBy); + String sourceRawIdsJson = toJsonArray(new ArrayList<>(sourceRawIds)); + + WikiPageEntity existing = pageService.getBySlug(kbId, slug); + WikiPageEntity persisted; + boolean created; + if (existing == null) { + persisted = pageService.createPage(kbId, slug, title, mergedOutput, summary, + sourceRawIdsJson, "synthesis"); + created = true; + } else { + persisted = pageService.updatePageByAi(kbId, slug, mergedOutput, summary, + sourceRawIds.iterator().next()); + if (persisted == null) persisted = existing; + created = false; + } + log.info("[WikiAggregator] {} aggregate page slug={} for template={} kb={} ({} sources, {} chars in)", + created ? "created" : "updated", slug, template.getName(), kbId, + sourcesIncluded, totalChars); + + // Fire-and-forget page embed so the aggregate joins semantic search. + if (embeddingService != null) { + final Long pid = persisted.getId(); + Thread.startVirtualThread(() -> { + try { embeddingService.embedPage(pid); } + catch (Exception ee) { + log.warn("[WikiAggregator] post-aggregate embed failed pageId={}: {}", + pid, ee.getMessage()); + } + }); + } + + return new Result(persisted.getId(), slug, title, sourcesIncluded, totalChars, created); + } + + private String toJsonArray(List ids) { + try { return objectMapper.writeValueAsString(ids); } + catch (Exception e) { + return ids.toString().replace(" ", ""); + } + } +} 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 4078cc4b..b1502472 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 @@ -68,6 +68,9 @@ public class WikiTool { @Autowired(required = false) private WikiTransformationExecutor transformationExecutor; + @Autowired(required = false) + private WikiTransformationAggregator transformationAggregator; + public WikiTool(WikiPageService pageService, WikiKnowledgeBaseService kbService, WikiRawMaterialService rawService, @@ -762,6 +765,55 @@ public class WikiTool { } } + @Tool(description = """ + Aggregate all completed runs of a transformation template across every + raw material in this KB into a single synthesis wiki page. Use this + after running a template against multiple sources to get a KB-level + unified document (e.g. one consolidated 题型库 across 5 different + mock exam PDFs, one customer-account brief across all sources for an + account). Idempotent — re-running upserts the same slug. + """) + public String wiki_aggregate_transformation( + @ToolParam(description = "Agent ID") Long agentId, + @ToolParam(description = "Transformation name (from wiki_list_transformations)") String name) { + if (name == null || name.isBlank()) return error("name is required"); + Long kbId = resolveKbId(agentId); + if (kbId == null) return error("No wiki knowledge base found for this agent"); + if (transformationService == null || transformationAggregator == null) { + return error("Transformations not available"); + } + + WikiKnowledgeBaseEntity kb = kbService.getById(kbId); + Long wsId = (kb == null || kb.getWorkspaceId() == null) ? 1L : kb.getWorkspaceId(); + + WikiTransformationEntity template = transformationService.findByName(kbId, wsId, name).orElse(null); + if (template == null) return error("Transformation not found: " + name); + + try { + var res = transformationAggregator.aggregate(template, kbId, "agent_tool"); + if (res.pageId() == null) { + return JSONUtil.createObj() + .set("ok", true) + .set("aggregated", false) + .set("reason", res.title()) + .toString(); + } + return JSONUtil.createObj() + .set("ok", true) + .set("aggregated", true) + .set("pageSlug", res.slug()) + .set("pageTitle", res.title()) + .set("sourcesUsed", res.sourcesUsed()) + .set("created", res.created()) + .toString(); + } catch (IllegalStateException | IllegalArgumentException e) { + return error(e.getMessage()); + } catch (Exception e) { + log.warn("[WikiTool] wiki_aggregate_transformation failed: {}", e.getMessage()); + return error("Aggregate failed: " + e.getMessage()); + } + } + // ==================== Helpers ==================== private Long resolveKbId(Long agentId) { diff --git a/mateclaw-server/src/main/resources/prompts/wiki/transformation-aggregate-system.txt b/mateclaw-server/src/main/resources/prompts/wiki/transformation-aggregate-system.txt new file mode 100644 index 00000000..605cb321 --- /dev/null +++ b/mateclaw-server/src/main/resources/prompts/wiki/transformation-aggregate-system.txt @@ -0,0 +1,16 @@ +You are a senior synthesis editor merging several AI-generated extracts. +Each extract was produced by running the same template against a different +source material; your job is to produce one unified KB-level document. + +Rules: +- Merge entries that describe the same concept / theorem / clause / person / + signal. Keep the entry once but list every source that contributed it. +- Preserve the per-source output structure (headings, tables, bullets) but + compress all sources into one cohesive document, not a concatenation. +- Add a "Sources" section at the very top listing every source you merged, + with a one-line note on each. +- Within each merged entry, when a fact came from more than one source, + cite the source titles in parentheses. +- Never invent content. If sources disagree, surface the disagreement + rather than smoothing it over. +- Output only Markdown. No preamble. No closing remarks. diff --git a/mateclaw-server/src/main/resources/prompts/wiki/transformation-aggregate-user.txt b/mateclaw-server/src/main/resources/prompts/wiki/transformation-aggregate-user.txt new file mode 100644 index 00000000..84683e8d --- /dev/null +++ b/mateclaw-server/src/main/resources/prompts/wiki/transformation-aggregate-user.txt @@ -0,0 +1,9 @@ +## Template + +**{template_title}** — {template_description} + +The per-source extracts below were all produced by running this template. + +## Per-source outputs + +{outputs} diff --git a/mateclaw-ui/src/api/index.ts b/mateclaw-ui/src/api/index.ts index c33e01bd..4438969f 100644 --- a/mateclaw-ui/src/api/index.ts +++ b/mateclaw-ui/src/api/index.ts @@ -707,6 +707,10 @@ export const wikiApi = { http.delete(`/wiki/transformations/${id}`), applyTransformation: (id: number, rawId: number, sync = true) => http.post(`/wiki/transformations/${id}/apply`, { rawId }, { params: { sync } }), + applyTransformationToPage: (id: number, pageId: number, sync = true) => + http.post(`/wiki/transformations/${id}/apply`, { pageId }, { params: { sync } }), + aggregateTransformation: (id: number, kbId: number) => + http.post(`/wiki/transformations/${id}/aggregate`, undefined, { params: { kbId } }), listTransformationRuns: (params: { rawId?: number; kbId?: number; transformationId?: number; limit?: number }) => http.get('/wiki/transformations/runs', { params }), getTransformationRun: (runId: number) => diff --git a/mateclaw-ui/src/i18n/locales/en-US.ts b/mateclaw-ui/src/i18n/locales/en-US.ts index 7aaf7dba..a4b39f63 100644 --- a/mateclaw-ui/src/i18n/locales/en-US.ts +++ b/mateclaw-ui/src/i18n/locales/en-US.ts @@ -1838,6 +1838,11 @@ export default { promptHelp: 'Supports {input_text} and {title} placeholders', saveBtn: 'Save', cancelBtn: 'Cancel', + aggregateBtn: 'Aggregate all runs', + aggregating: 'Aggregating…', + aggregateDone: 'Aggregate page produced', + aggregateFailed: 'Aggregate failed', + aggregateNoRuns: 'This template has no completed runs to aggregate yet', runs: 'Run history', rerunBtn: 'Re-run', rerunning: 'Re-running…', diff --git a/mateclaw-ui/src/i18n/locales/zh-CN.ts b/mateclaw-ui/src/i18n/locales/zh-CN.ts index 4c72bd31..6dda20e9 100644 --- a/mateclaw-ui/src/i18n/locales/zh-CN.ts +++ b/mateclaw-ui/src/i18n/locales/zh-CN.ts @@ -1850,6 +1850,11 @@ export default { promptHelp: '可使用 {input_text} 和 {title} 占位符', saveBtn: '保存', cancelBtn: '取消', + aggregateBtn: '聚合所有运行', + aggregating: '聚合中…', + aggregateDone: '聚合页已生成', + aggregateFailed: '聚合失败', + aggregateNoRuns: '当前模板还没有任何完成的运行可供聚合', runs: '运行历史', rerunBtn: '重新运行', rerunning: '重新运行中…', diff --git a/mateclaw-ui/src/views/Wiki/components/TransformationsPanel.vue b/mateclaw-ui/src/views/Wiki/components/TransformationsPanel.vue index 2b17a47c..8739416d 100644 --- a/mateclaw-ui/src/views/Wiki/components/TransformationsPanel.vue +++ b/mateclaw-ui/src/views/Wiki/components/TransformationsPanel.vue @@ -72,6 +72,13 @@ {{ t('wiki.transformations.running') }} {{ t('wiki.transformations.runApply') }} + @@ -295,6 +302,7 @@ const saving = ref(false) const savingRunId = ref(null) const cancellingRunId = ref(null) const rerunningRunId = ref(null) +const aggregatingTemplateId = ref(null) interface ModelOption { id: number; name: string; provider: string; modelName: string } const availableModels = ref([]) @@ -514,6 +522,27 @@ async function onSaveRunAsPage(tpl: WikiTransformation, run: WikiTransformationR } } +async function onAggregate(tpl: WikiTransformation) { + if (!store.currentKB) return + aggregatingTemplateId.value = tpl.id + try { + const resp: any = await wikiApi.aggregateTransformation(tpl.id, store.currentKB.id) + const payload = resp?.data ?? {} + if (payload && payload.pageId) { + ElMessage.success(`${t('wiki.transformations.aggregateDone')} · ${payload.sourcesUsed} sources`) + // Refresh the page list in the wiki store so the new aggregate page + // shows up in the sidebar without a manual reload. + try { await store.fetchPages(store.currentKB.id) } catch {} + } else { + ElMessage.info(t('wiki.transformations.aggregateNoRuns')) + } + } catch (e: any) { + ElMessage.error(e?.message ?? t('wiki.transformations.aggregateFailed')) + } finally { + aggregatingTemplateId.value = null + } +} + async function onCancelRun(tpl: WikiTransformation, run: WikiTransformationRun) { cancellingRunId.value = run.id try {