mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(wiki): pipeline definition CRUD/YAML API, run query API, page-created trigger
This commit is contained in:
matevip
parent
c627f898ec
commit
7363ee8668
@ -62,6 +62,9 @@ public class WikiController {
|
||||
private final WikiPageTypeProfileService pageTypeProfileService;
|
||||
private final WikiSourcePathValidator pathValidator;
|
||||
private final vip.mate.wiki.service.WikiSourceWatcherService sourceWatcherService;
|
||||
private final vip.mate.wiki.pipeline.WikiPipelineDefinitionService pipelineDefinitionService;
|
||||
private final vip.mate.wiki.repository.WikiPipelineRunMapper pipelineRunMapper;
|
||||
private final vip.mate.wiki.repository.WikiPipelineStepRunMapper pipelineStepRunMapper;
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
// ==================== Knowledge Base ====================
|
||||
@ -344,6 +347,86 @@ public class WikiController {
|
||||
return R.ok(out);
|
||||
}
|
||||
|
||||
// ==================== Pipeline ====================
|
||||
|
||||
@RequireWorkspaceRole("viewer")
|
||||
@Operation(summary = "列出知识库的 pipeline 定义")
|
||||
@GetMapping("/knowledge-bases/{kbId}/pipelines")
|
||||
public R<List<vip.mate.wiki.model.WikiPipelineDefinitionEntity>> listPipelines(
|
||||
@PathVariable Long kbId, @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) {
|
||||
verifyKBWorkspace(kbId, workspaceId);
|
||||
return R.ok(pipelineDefinitionService.list(kbId));
|
||||
}
|
||||
|
||||
@RequireWorkspaceRole("admin")
|
||||
@Operation(summary = "保存(创建/更新)pipeline 定义(YAML/JSON)")
|
||||
@PostMapping("/knowledge-bases/{kbId}/pipelines")
|
||||
public R<vip.mate.wiki.model.WikiPipelineDefinitionEntity> savePipeline(
|
||||
@PathVariable Long kbId, @RequestBody Map<String, String> body,
|
||||
@RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) {
|
||||
verifyKBWorkspace(kbId, workspaceId);
|
||||
boolean yaml = !"json".equalsIgnoreCase(body.getOrDefault("format", "yaml"));
|
||||
try {
|
||||
return R.ok(pipelineDefinitionService.saveFromConfig(kbId, body.get("config"), yaml));
|
||||
} catch (IllegalArgumentException e) {
|
||||
return R.fail(400, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
@RequireWorkspaceRole("member")
|
||||
@Operation(summary = "校验 pipeline 配置(不保存)")
|
||||
@PostMapping("/knowledge-bases/{kbId}/pipelines/validate")
|
||||
public R<Map<String, Object>> validatePipeline(
|
||||
@PathVariable Long kbId, @RequestBody Map<String, String> body,
|
||||
@RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) {
|
||||
verifyKBWorkspace(kbId, workspaceId);
|
||||
boolean yaml = !"json".equalsIgnoreCase(body.getOrDefault("format", "yaml"));
|
||||
List<String> issues = pipelineDefinitionService.validateConfig(body.getOrDefault("config", ""), yaml);
|
||||
Map<String, Object> out = new LinkedHashMap<>();
|
||||
out.put("valid", issues.isEmpty());
|
||||
out.put("issues", issues);
|
||||
return R.ok(out);
|
||||
}
|
||||
|
||||
@RequireWorkspaceRole("admin")
|
||||
@Operation(summary = "删除 pipeline 定义")
|
||||
@DeleteMapping("/knowledge-bases/{kbId}/pipelines/{id}")
|
||||
public R<Void> deletePipeline(@PathVariable Long kbId, @PathVariable Long id,
|
||||
@RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) {
|
||||
verifyKBWorkspace(kbId, workspaceId);
|
||||
pipelineDefinitionService.delete(id);
|
||||
return R.ok();
|
||||
}
|
||||
|
||||
@RequireWorkspaceRole("viewer")
|
||||
@Operation(summary = "查询 pipeline 运行记录")
|
||||
@GetMapping("/knowledge-bases/{kbId}/pipelines/{id}/runs")
|
||||
public R<List<vip.mate.wiki.model.WikiPipelineRunEntity>> listPipelineRuns(
|
||||
@PathVariable Long kbId, @PathVariable Long id,
|
||||
@RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) {
|
||||
verifyKBWorkspace(kbId, workspaceId);
|
||||
return R.ok(pipelineRunMapper.selectList(
|
||||
com.baomidou.mybatisplus.core.toolkit.Wrappers.<vip.mate.wiki.model.WikiPipelineRunEntity>lambdaQuery()
|
||||
.eq(vip.mate.wiki.model.WikiPipelineRunEntity::getDefinitionId, id)
|
||||
.orderByDesc(vip.mate.wiki.model.WikiPipelineRunEntity::getCreateTime)));
|
||||
}
|
||||
|
||||
@RequireWorkspaceRole("viewer")
|
||||
@Operation(summary = "查询单次 run 的步骤明细")
|
||||
@GetMapping("/knowledge-bases/{kbId}/pipeline-runs/{runId}")
|
||||
public R<Map<String, Object>> getPipelineRun(
|
||||
@PathVariable Long kbId, @PathVariable Long runId,
|
||||
@RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) {
|
||||
verifyKBWorkspace(kbId, workspaceId);
|
||||
Map<String, Object> out = new LinkedHashMap<>();
|
||||
out.put("run", pipelineRunMapper.selectById(runId));
|
||||
out.put("steps", pipelineStepRunMapper.selectList(
|
||||
com.baomidou.mybatisplus.core.toolkit.Wrappers.<vip.mate.wiki.model.WikiPipelineStepRunEntity>lambdaQuery()
|
||||
.eq(vip.mate.wiki.model.WikiPipelineStepRunEntity::getRunId, runId)
|
||||
.orderByAsc(vip.mate.wiki.model.WikiPipelineStepRunEntity::getCreateTime)));
|
||||
return R.ok(out);
|
||||
}
|
||||
|
||||
// ==================== Raw Materials ====================
|
||||
|
||||
@RequireWorkspaceRole("viewer")
|
||||
|
||||
@ -7,5 +7,5 @@ package vip.mate.wiki.event;
|
||||
*
|
||||
* @author MateClaw Team
|
||||
*/
|
||||
public record WikiPageCreatedEvent(Long kbId, String pageType) {
|
||||
public record WikiPageCreatedEvent(Long kbId, String pageType, Long pageId) {
|
||||
}
|
||||
|
||||
@ -0,0 +1,164 @@
|
||||
package vip.mate.wiki.pipeline;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
import vip.mate.wiki.model.WikiPipelineDefinitionEntity;
|
||||
import vip.mate.wiki.repository.WikiPipelineDefinitionMapper;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* CRUD + YAML/JSON parsing for user-defined pipelines. A definition's config
|
||||
* (YAML or JSON) carries {@code name}, {@code owner_agent}, a {@code trigger}
|
||||
* object and a {@code steps} array; this service parses it into the persisted
|
||||
* entity (trigger / steps stored as JSON).
|
||||
*
|
||||
* @author MateClaw Team
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
public class WikiPipelineDefinitionService {
|
||||
|
||||
private static final java.util.Set<String> KNOWN_EXECUTORS = java.util.Set.of("llm", "skill", "python");
|
||||
private static final java.util.Set<String> KNOWN_TRIGGERS =
|
||||
java.util.Set.of("page_type_count", "page_created", "stale_marked");
|
||||
|
||||
private final WikiPipelineDefinitionMapper definitionMapper;
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
public WikiPipelineDefinitionService(WikiPipelineDefinitionMapper definitionMapper, ObjectMapper objectMapper) {
|
||||
this.definitionMapper = definitionMapper;
|
||||
this.objectMapper = objectMapper;
|
||||
}
|
||||
|
||||
public List<WikiPipelineDefinitionEntity> list(Long kbId) {
|
||||
return definitionMapper.selectList(new LambdaQueryWrapper<WikiPipelineDefinitionEntity>()
|
||||
.eq(WikiPipelineDefinitionEntity::getKbId, kbId)
|
||||
.orderByDesc(WikiPipelineDefinitionEntity::getCreateTime));
|
||||
}
|
||||
|
||||
public WikiPipelineDefinitionEntity get(Long id) {
|
||||
return definitionMapper.selectById(id);
|
||||
}
|
||||
|
||||
public void delete(Long id) {
|
||||
definitionMapper.deleteById(id);
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse a YAML/JSON pipeline config and upsert it (by kb + name). Throws
|
||||
* {@link IllegalArgumentException} on a structural problem.
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public WikiPipelineDefinitionEntity saveFromConfig(Long kbId, String config, boolean yaml) {
|
||||
Map<String, Object> root = parse(config, yaml);
|
||||
List<String> issues = validateParsed(root);
|
||||
if (!issues.isEmpty()) {
|
||||
throw new IllegalArgumentException("Invalid pipeline config: " + String.join("; ", issues));
|
||||
}
|
||||
String name = String.valueOf(root.get("name"));
|
||||
Object owner = root.get("owner_agent");
|
||||
Map<String, Object> trigger = (Map<String, Object>) root.get("trigger");
|
||||
Object steps = root.get("steps");
|
||||
|
||||
WikiPipelineDefinitionEntity entity = definitionMapper.selectOne(
|
||||
new LambdaQueryWrapper<WikiPipelineDefinitionEntity>()
|
||||
.eq(WikiPipelineDefinitionEntity::getKbId, kbId)
|
||||
.eq(WikiPipelineDefinitionEntity::getName, name)
|
||||
.last("LIMIT 1"));
|
||||
boolean isNew = entity == null;
|
||||
if (isNew) {
|
||||
entity = new WikiPipelineDefinitionEntity();
|
||||
entity.setKbId(kbId);
|
||||
entity.setName(name);
|
||||
entity.setEnabled(1);
|
||||
}
|
||||
entity.setOwnerAgentId(Long.valueOf(String.valueOf(owner)));
|
||||
entity.setTriggerType(String.valueOf(trigger.get("type")));
|
||||
try {
|
||||
entity.setTriggerConfigJson(objectMapper.writeValueAsString(trigger));
|
||||
entity.setStepsJson(objectMapper.writeValueAsString(steps));
|
||||
} catch (Exception e) {
|
||||
throw new IllegalArgumentException("Failed to serialize pipeline config: " + e.getMessage());
|
||||
}
|
||||
Object dedup = trigger.get("dedup_window_seconds");
|
||||
entity.setDedupWindowSeconds(dedup instanceof Number ? ((Number) dedup).intValue() : 0);
|
||||
|
||||
if (isNew) {
|
||||
definitionMapper.insert(entity);
|
||||
} else {
|
||||
definitionMapper.updateById(entity);
|
||||
}
|
||||
return entity;
|
||||
}
|
||||
|
||||
/** Validate a config without saving; returns human-readable issues (empty = valid). */
|
||||
public List<String> validateConfig(String config, boolean yaml) {
|
||||
try {
|
||||
return validateParsed(parse(config, yaml));
|
||||
} catch (Exception e) {
|
||||
return List.of("Unparseable config: " + e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private Map<String, Object> parse(String config, boolean yaml) {
|
||||
if (config == null || config.isBlank()) {
|
||||
throw new IllegalArgumentException("config is empty");
|
||||
}
|
||||
try {
|
||||
if (yaml) {
|
||||
Object loaded = new org.yaml.snakeyaml.Yaml().load(config);
|
||||
if (!(loaded instanceof Map)) {
|
||||
throw new IllegalArgumentException("YAML root must be a mapping");
|
||||
}
|
||||
return (Map<String, Object>) loaded;
|
||||
}
|
||||
return objectMapper.readValue(config, Map.class);
|
||||
} catch (IllegalArgumentException e) {
|
||||
throw e;
|
||||
} catch (Exception e) {
|
||||
throw new IllegalArgumentException(e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private List<String> validateParsed(Map<String, Object> root) {
|
||||
List<String> issues = new ArrayList<>();
|
||||
if (root.get("name") == null || String.valueOf(root.get("name")).isBlank()) {
|
||||
issues.add("missing 'name'");
|
||||
}
|
||||
if (root.get("owner_agent") == null) {
|
||||
issues.add("missing 'owner_agent' (steps run under this agent)");
|
||||
}
|
||||
Object trig = root.get("trigger");
|
||||
if (!(trig instanceof Map)) {
|
||||
issues.add("missing 'trigger' object");
|
||||
} else {
|
||||
String type = String.valueOf(((Map<String, Object>) trig).get("type"));
|
||||
if (!KNOWN_TRIGGERS.contains(type)) {
|
||||
issues.add("unknown trigger type '" + type + "' (expected one of " + KNOWN_TRIGGERS + ")");
|
||||
}
|
||||
}
|
||||
Object steps = root.get("steps");
|
||||
if (!(steps instanceof List) || ((List<?>) steps).isEmpty()) {
|
||||
issues.add("'steps' must be a non-empty array");
|
||||
} else {
|
||||
for (Object s : (List<Object>) steps) {
|
||||
if (!(s instanceof Map)) { issues.add("each step must be an object"); continue; }
|
||||
String ex = String.valueOf(((Map<String, Object>) s).get("executor"));
|
||||
if (!KNOWN_EXECUTORS.contains(ex)) {
|
||||
issues.add("step executor '" + ex + "' unknown (expected " + KNOWN_EXECUTORS + ")");
|
||||
}
|
||||
if ("python".equals(ex)) {
|
||||
issues.add("python executor needs a sandbox and is not enabled in this build");
|
||||
}
|
||||
}
|
||||
}
|
||||
return issues;
|
||||
}
|
||||
}
|
||||
@ -30,6 +30,7 @@ public class WikiPipelineTriggerListener {
|
||||
public void onPageCreated(WikiPageCreatedEvent event) {
|
||||
try {
|
||||
triggerService.onPageTypeCount(event.kbId(), event.pageType());
|
||||
triggerService.onPageCreated(event.kbId(), event.pageType(), event.pageId());
|
||||
} catch (Exception e) {
|
||||
log.warn("[WikiPipeline] trigger evaluation failed for kb={} pageType={}: {}",
|
||||
event.kbId(), event.pageType(), e.getMessage());
|
||||
|
||||
@ -49,6 +49,36 @@ public class WikiPipelineTriggerService {
|
||||
* number of runs actually started (0 when no threshold bucket was newly
|
||||
* crossed). Safe to call after every page create.
|
||||
*/
|
||||
/**
|
||||
* Fire {@code page_created} definitions once per matching page creation
|
||||
* (deduped by page id). Optional {@code page_type} in the trigger config
|
||||
* narrows which page types fire. Returns the number of runs started.
|
||||
*/
|
||||
public int onPageCreated(Long kbId, String pageType, Long pageId) {
|
||||
if (kbId == null || pageType == null || pageId == null) {
|
||||
return 0;
|
||||
}
|
||||
List<WikiPipelineDefinitionEntity> defs = definitionMapper.selectList(
|
||||
new LambdaQueryWrapper<WikiPipelineDefinitionEntity>()
|
||||
.eq(WikiPipelineDefinitionEntity::getKbId, kbId)
|
||||
.eq(WikiPipelineDefinitionEntity::getTriggerType, "page_created")
|
||||
.eq(WikiPipelineDefinitionEntity::getEnabled, 1));
|
||||
int started = 0;
|
||||
for (WikiPipelineDefinitionEntity def : defs) {
|
||||
TriggerConfig cfg = parseConfig(def.getTriggerConfigJson());
|
||||
if (cfg != null && cfg.pageType != null && !pageType.equalsIgnoreCase(cfg.pageType)) {
|
||||
continue; // type filter set and doesn't match
|
||||
}
|
||||
String input = "{\"pageType\":\"" + pageType + "\",\"pageId\":\"" + pageId + "\"}";
|
||||
WikiPipelineService.RunOutcome outcome =
|
||||
pipelineService.execute(def, pageType, "page:" + pageId, input);
|
||||
if (!outcome.duplicate() && outcome.run() != null) {
|
||||
started++;
|
||||
}
|
||||
}
|
||||
return started;
|
||||
}
|
||||
|
||||
public int onPageTypeCount(Long kbId, String pageType) {
|
||||
if (kbId == null || pageType == null || pageType.isBlank()) {
|
||||
return 0;
|
||||
|
||||
@ -1508,7 +1508,7 @@ public class WikiProcessingService {
|
||||
// transactions), so the count is accurate. Idempotent + dedup-guarded
|
||||
// downstream, so firing on update paths is safe.
|
||||
if (eventPublisher != null && pageType != null && !pageType.isBlank()) {
|
||||
eventPublisher.publishEvent(new vip.mate.wiki.event.WikiPageCreatedEvent(kbId, pageType));
|
||||
eventPublisher.publishEvent(new vip.mate.wiki.event.WikiPageCreatedEvent(kbId, pageType, pageId));
|
||||
}
|
||||
// When an existing fact page is updated, propagate staleness to the
|
||||
// experience pages depending on it (async, off the ingest thread).
|
||||
|
||||
@ -0,0 +1,103 @@
|
||||
package vip.mate.wiki.pipeline;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import vip.mate.wiki.model.WikiPipelineDefinitionEntity;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* E2E for {@link WikiPipelineDefinitionService}: YAML/JSON parsing, structural
|
||||
* validation, and upsert/list/delete against H2.
|
||||
*/
|
||||
@SpringBootTest(
|
||||
webEnvironment = SpringBootTest.WebEnvironment.NONE,
|
||||
properties = {
|
||||
"spring.flyway.enabled=true",
|
||||
"spring.flyway.locations=classpath:db/migration/h2",
|
||||
"mateclaw.feature-flag.refresh-ms=999999"
|
||||
}
|
||||
)
|
||||
class WikiPipelineDefinitionServiceE2ETest {
|
||||
|
||||
@Autowired
|
||||
private WikiPipelineDefinitionService service;
|
||||
|
||||
private static final java.util.concurrent.atomic.AtomicLong SEQ =
|
||||
new java.util.concurrent.atomic.AtomicLong(System.nanoTime());
|
||||
|
||||
private static final String YAML = """
|
||||
name: episode-to-pattern
|
||||
owner_agent: 2055137662148763649
|
||||
trigger:
|
||||
type: page_type_count
|
||||
page_type: episode
|
||||
threshold: 20
|
||||
dedup_window_seconds: 3600
|
||||
steps:
|
||||
- id: summarize
|
||||
executor: llm
|
||||
prompt: pattern-analysis
|
||||
- id: enrich
|
||||
executor: skill
|
||||
skill: wiki-link-enrich
|
||||
""";
|
||||
|
||||
@Test
|
||||
void parsesYamlAndUpserts() {
|
||||
long kb = SEQ.incrementAndGet();
|
||||
WikiPipelineDefinitionEntity def = service.saveFromConfig(kb, YAML, true);
|
||||
assertNotNull(def.getId());
|
||||
assertEquals("episode-to-pattern", def.getName());
|
||||
assertEquals(2055137662148763649L, def.getOwnerAgentId());
|
||||
assertEquals("page_type_count", def.getTriggerType());
|
||||
assertEquals(3600, def.getDedupWindowSeconds());
|
||||
assertTrue(def.getTriggerConfigJson().contains("episode"));
|
||||
assertTrue(def.getStepsJson().contains("wiki-link-enrich"));
|
||||
|
||||
// upsert: same name → update in place, not a second row
|
||||
service.saveFromConfig(kb, YAML.replace("threshold: 20", "threshold: 40"), true);
|
||||
List<WikiPipelineDefinitionEntity> all = service.list(kb);
|
||||
assertEquals(1, all.size());
|
||||
assertTrue(all.get(0).getTriggerConfigJson().contains("40"));
|
||||
|
||||
service.delete(def.getId());
|
||||
assertNull(service.get(def.getId()));
|
||||
}
|
||||
|
||||
@Test
|
||||
void parsesJson() {
|
||||
long kb = SEQ.incrementAndGet();
|
||||
String json = "{\"name\":\"p\",\"owner_agent\":42,\"trigger\":{\"type\":\"page_created\"},"
|
||||
+ "\"steps\":[{\"id\":\"s\",\"executor\":\"llm\",\"prompt\":\"x\"}]}";
|
||||
WikiPipelineDefinitionEntity def = service.saveFromConfig(kb, json, false);
|
||||
assertEquals("page_created", def.getTriggerType());
|
||||
}
|
||||
|
||||
@Test
|
||||
void validation_reportsIssues() {
|
||||
assertTrue(service.validateConfig(YAML, true).isEmpty());
|
||||
|
||||
// missing name + unknown trigger + python step
|
||||
String bad = """
|
||||
owner_agent: 1
|
||||
trigger:
|
||||
type: nope
|
||||
steps:
|
||||
- id: x
|
||||
executor: python
|
||||
""";
|
||||
List<String> issues = service.validateConfig(bad, true);
|
||||
assertFalse(issues.isEmpty());
|
||||
assertTrue(issues.stream().anyMatch(s -> s.contains("name")));
|
||||
assertTrue(issues.stream().anyMatch(s -> s.contains("trigger type")));
|
||||
assertTrue(issues.stream().anyMatch(s -> s.contains("python")));
|
||||
}
|
||||
}
|
||||
@ -20,7 +20,7 @@ class WikiPipelineTriggerListenerTest {
|
||||
WikiPipelineTriggerService trigger = mock(WikiPipelineTriggerService.class);
|
||||
when(trigger.onPageTypeCount(7L, "episode")).thenReturn(1);
|
||||
|
||||
new WikiPipelineTriggerListener(trigger).onPageCreated(new WikiPageCreatedEvent(7L, "episode"));
|
||||
new WikiPipelineTriggerListener(trigger).onPageCreated(new WikiPageCreatedEvent(7L, "episode", 100L));
|
||||
|
||||
verify(trigger).onPageTypeCount(7L, "episode");
|
||||
}
|
||||
@ -31,6 +31,6 @@ class WikiPipelineTriggerListenerTest {
|
||||
doThrow(new RuntimeException("boom")).when(trigger).onPageTypeCount(7L, "episode");
|
||||
|
||||
// Must not throw — ingest must be unaffected by a pipeline failure.
|
||||
new WikiPipelineTriggerListener(trigger).onPageCreated(new WikiPageCreatedEvent(7L, "episode"));
|
||||
new WikiPipelineTriggerListener(trigger).onPageCreated(new WikiPageCreatedEvent(7L, "episode", 100L));
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user