mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 11:13:43 +08:00
feat(wiki): cross-material transformation aggregator — map-reduce all runs of a template into one KB page
This commit is contained in:
matevip
parent
a5b952050c
commit
90936dba4f
@ -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 '<template-name>-aggregate'.")
|
||||
@PostMapping("/{id}/aggregate")
|
||||
public R<Map<String, Object>> 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")
|
||||
|
||||
@ -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 <template-name>-aggregate}.
|
||||
*
|
||||
* <p>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<WikiTransformationRunEntity> 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<Long, WikiTransformationRunEntity> latestByRaw = new java.util.LinkedHashMap<>();
|
||||
for (WikiTransformationRunEntity r : runs) {
|
||||
latestByRaw.putIfAbsent(r.getRawId(), r); // listRunsByTransformation orders DESC by createTime
|
||||
}
|
||||
List<WikiTransformationRunEntity> 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<Long> 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<Long> ids) {
|
||||
try { return objectMapper.writeValueAsString(ids); }
|
||||
catch (Exception e) {
|
||||
return ids.toString().replace(" ", "");
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -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) {
|
||||
|
||||
@ -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.
|
||||
@ -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}
|
||||
@ -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) =>
|
||||
|
||||
@ -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…',
|
||||
|
||||
@ -1850,6 +1850,11 @@ export default {
|
||||
promptHelp: '可使用 {input_text} 和 {title} 占位符',
|
||||
saveBtn: '保存',
|
||||
cancelBtn: '取消',
|
||||
aggregateBtn: '聚合所有运行',
|
||||
aggregating: '聚合中…',
|
||||
aggregateDone: '聚合页已生成',
|
||||
aggregateFailed: '聚合失败',
|
||||
aggregateNoRuns: '当前模板还没有任何完成的运行可供聚合',
|
||||
runs: '运行历史',
|
||||
rerunBtn: '重新运行',
|
||||
rerunning: '重新运行中…',
|
||||
|
||||
@ -72,6 +72,13 @@
|
||||
<span v-if="runningTemplateId === tpl.id">{{ t('wiki.transformations.running') }}</span>
|
||||
<span v-else>{{ t('wiki.transformations.runApply') }}</span>
|
||||
</button>
|
||||
<button class="btn-secondary"
|
||||
:disabled="aggregatingTemplateId === tpl.id"
|
||||
@click="onAggregate(tpl)">
|
||||
{{ aggregatingTemplateId === tpl.id
|
||||
? t('wiki.transformations.aggregating')
|
||||
: t('wiki.transformations.aggregateBtn') }}
|
||||
</button>
|
||||
<button class="btn-secondary" @click="openEdit(tpl)">{{ t('wiki.transformations.editBtn') }}</button>
|
||||
<button class="btn-secondary btn-danger" @click="onDelete(tpl)">{{ t('wiki.transformations.deleteBtn') }}</button>
|
||||
</div>
|
||||
@ -295,6 +302,7 @@ const saving = ref(false)
|
||||
const savingRunId = ref<number | null>(null)
|
||||
const cancellingRunId = ref<number | null>(null)
|
||||
const rerunningRunId = ref<number | null>(null)
|
||||
const aggregatingTemplateId = ref<number | null>(null)
|
||||
interface ModelOption { id: number; name: string; provider: string; modelName: string }
|
||||
const availableModels = ref<ModelOption[]>([])
|
||||
|
||||
@ -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 {
|
||||
|
||||
Loading…
Reference in New Issue
Block a user