diff --git a/internal/ingestion/component/knowledge_compiler/common/relations.go b/internal/ingestion/component/knowledge_compiler/common/relations.go new file mode 100644 index 0000000000..f0e81e8ad1 --- /dev/null +++ b/internal/ingestion/component/knowledge_compiler/common/relations.go @@ -0,0 +1,58 @@ +package common + +import "strings" + +// FilterOrphanRelations removes relation products whose endpoints are not +// present in the same compilation result. Entity and relation rows are scoped +// by variant, template, and compile kind so unrelated templates cannot satisfy +// each other's endpoints. +func FilterOrphanRelations(products []Product) []Product { + entities := make(map[string]struct{}, len(products)) + for _, product := range products { + if product.Meta == nil || strings.TrimSpace(metaString(product.Meta, "kind")) != "entity" { + continue + } + name := normalizeRelationEndpoint(metaString(product.Meta, "name")) + if name != "" { + entities[relationScope(product)+"\x00"+name] = struct{}{} + } + } + + filtered := make([]Product, 0, len(products)) + for _, product := range products { + from := normalizeRelationEndpoint(metaString(product.Meta, "from")) + to := normalizeRelationEndpoint(metaString(product.Meta, "to")) + isRelation := strings.TrimSpace(metaString(product.Meta, "kind")) == "relation" || (from != "" || to != "") + if isRelation && (from == "" || to == "" || !hasRelationEntity(entities, product, from) || !hasRelationEntity(entities, product, to)) { + continue + } + filtered = append(filtered, product) + } + return filtered +} + +func hasRelationEntity(entities map[string]struct{}, product Product, name string) bool { + _, ok := entities[relationScope(product)+"\x00"+name] + return ok +} + +func relationScope(product Product) string { + compileKind := metaString(product.Meta, "compile_kwd") + if compileKind == "" { + compileKind = string(product.Variant) + } + template := strings.TrimSpace(product.TemplateID) + if template == "" { + template = strings.TrimSpace(product.Kind) + } + return string(product.Variant) + "\x00" + template + "\x00" + compileKind +} + +func normalizeRelationEndpoint(value string) string { + return strings.ToLower(strings.Join(strings.Fields(value), " ")) +} + +func metaString(meta map[string]any, key string) string { + value, _ := meta[key].(string) + return value +} diff --git a/internal/ingestion/component/knowledge_compiler/common/relations_test.go b/internal/ingestion/component/knowledge_compiler/common/relations_test.go new file mode 100644 index 0000000000..e9e6f76e42 --- /dev/null +++ b/internal/ingestion/component/knowledge_compiler/common/relations_test.go @@ -0,0 +1,30 @@ +package common + +import "testing" + +func TestFilterOrphanRelations(t *testing.T) { + products := []Product{ + {Variant: VariantStructure, TemplateID: "graph", Meta: map[string]any{"kind": "entity", "name": "A"}}, + {Variant: VariantStructure, TemplateID: "graph", Meta: map[string]any{"kind": "entity", "name": "B"}}, + {Variant: VariantStructure, TemplateID: "graph", Meta: map[string]any{"kind": "relation", "from": " A ", "to": "B"}}, + {Variant: VariantStructure, TemplateID: "graph", Meta: map[string]any{"kind": "relation", "from": "A", "to": "Missing"}}, + } + + filtered := FilterOrphanRelations(products) + if len(filtered) != 3 { + t.Fatalf("got %d products, want 3", len(filtered)) + } + if filtered[2].Meta["to"] != "B" { + t.Fatalf("valid relation was not preserved: %+v", filtered[2]) + } +} + +func TestFilterOrphanRelationsScopesTemplates(t *testing.T) { + products := []Product{ + {Variant: VariantStructure, TemplateID: "graph", Meta: map[string]any{"kind": "entity", "name": "A"}}, + {Variant: VariantStructure, TemplateID: "timeline", Meta: map[string]any{"kind": "relation", "from": "A", "to": "B"}}, + } + if got := FilterOrphanRelations(products); len(got) != 1 { + t.Fatalf("relation satisfied by another template: got %d products, want 1", len(got)) + } +} diff --git a/internal/ingestion/component/knowledge_compiler/structure/compile.go b/internal/ingestion/component/knowledge_compiler/structure/compile.go index b13b0dde3f..e29bbe1ca7 100644 --- a/internal/ingestion/component/knowledge_compiler/structure/compile.go +++ b/internal/ingestion/component/knowledge_compiler/structure/compile.go @@ -71,6 +71,12 @@ func extractHypergraph(ctx context.Context, deps common.Deps, cfg CompileConfig, known = append(known, v) } + // Relations can only reference entities from this batch. If entity extraction + // returned nothing, skip the second LLM call and avoid creating orphan edges. + if len(known) == 0 { + return nodes, nil, nil + } + if strings.TrimSpace(edgePromptTmpl) == "" { return nodes, nil, nil } diff --git a/internal/ingestion/component/knowledge_compiler/structure/structure_test.go b/internal/ingestion/component/knowledge_compiler/structure/structure_test.go index 23ceb0ce9a..237e1d34ca 100644 --- a/internal/ingestion/component/knowledge_compiler/structure/structure_test.go +++ b/internal/ingestion/component/knowledge_compiler/structure/structure_test.go @@ -388,6 +388,20 @@ func TestFillKnownNodes(t *testing.T) { } } +func TestExtractHypergraphSkipsEdgesWithoutEntities(t *testing.T) { + chat := &graphChat{} + nodes, edges, err := extractHypergraph(context.Background(), common.Deps{Chat: chat}, CompileConfig{LLMID: "llm", ParserConfig: map[string]any{"entity": map[string]any{"id": "name"}}}, "node", "## Known Entities: {known_nodes}", "unrelated text") + if err != nil { + t.Fatalf("extractHypergraph: %v", err) + } + if len(nodes) != 0 || len(edges) != 0 { + t.Fatalf("empty entity batch returned nodes=%v edges=%v", nodes, edges) + } + if chat.nodeCalls != 1 || chat.edgeCalls != 0 { + t.Fatalf("LLM calls = node:%d edge:%d, want node:1 edge:0", chat.nodeCalls, chat.edgeCalls) + } +} + // ---- payload helpers ---- func TestPayloadChunkIDs(t *testing.T) { diff --git a/internal/ingestion/knowledge_compile/consumer.go b/internal/ingestion/knowledge_compile/consumer.go index f8598b783b..7fd9444541 100644 --- a/internal/ingestion/knowledge_compile/consumer.go +++ b/internal/ingestion/knowledge_compile/consumer.go @@ -1563,6 +1563,7 @@ func structureEntityNavLine(payloadJSON string) string { // dataset_structure_merger._merge_bucket but is a self-contained from-scratch // path (B3: the legacy unified merge never aggregated structure by name). func (c *Consumer) mergeStructureDataset(ctx context.Context, tenant, kb string, products []kccommon.Product) error { + products = kccommon.FilterOrphanRelations(products) common.Info("knowledge_compile: mergeStructureDataset entry", zap.String("kb_id", kb), zap.Int("products", len(products))) diff --git a/internal/ingestion/knowledge_compile/structure_merge_test.go b/internal/ingestion/knowledge_compile/structure_merge_test.go index 8da59fe663..eb30b2d8e2 100644 --- a/internal/ingestion/knowledge_compile/structure_merge_test.go +++ b/internal/ingestion/knowledge_compile/structure_merge_test.go @@ -119,6 +119,10 @@ func TestMergeStructureDataset_IsolatesTemplates(t *testing.T) { func TestMergeStructureDataset_RelationsNotDropped(t *testing.T) { c := &Consumer{writer: &fakeWriter{}} products := []kccommon.Product{ + {Variant: kccommon.VariantStructure, DocID: "d1", Content: "entity A", + Meta: map[string]any{"kind": "entity", "name": "A"}}, + {Variant: kccommon.VariantStructure, DocID: "d1", Content: "entity B", + Meta: map[string]any{"kind": "entity", "name": "B"}}, {Variant: kccommon.VariantStructure, DocID: "d1", Content: "rel desc", Meta: map[string]any{"kind": "relation", "from": "A", "to": "B", "source_chunk_ids": []string{"c1"}}}, {Variant: kccommon.VariantStructure, DocID: "d2", Content: "rel desc 2", @@ -128,10 +132,19 @@ func TestMergeStructureDataset_RelationsNotDropped(t *testing.T) { t.Fatalf("mergeStructureDataset: %v", err) } fw := c.writer.(*fakeWriter) - if len(fw.buckets) != 1 { - t.Fatalf("want 1 relation bucket, got %d", len(fw.buckets)) + if len(fw.buckets) != 3 { + t.Fatalf("want 2 entity buckets and 1 relation bucket, got %d", len(fw.buckets)) + } + var b *StructureBucket + for i := range fw.buckets { + if fw.buckets[i].Type == "relation" { + b = &fw.buckets[i] + break + } + } + if b == nil { + t.Fatalf("missing relation bucket: %+v", fw.buckets) } - b := fw.buckets[0] if b.Type != "relation" || b.FromEntity != "A" || b.ToEntity != "B" { t.Errorf("relation bucket = %+v, want Type=relation From=A To=B", b) } @@ -143,6 +156,10 @@ func TestMergeStructureDataset_RelationsNotDropped(t *testing.T) { func TestMergeStructureDataset_NormalizesRelationIdentity(t *testing.T) { c := &Consumer{writer: &fakeWriter{}} products := []kccommon.Product{ + {Variant: kccommon.VariantStructure, DocID: "d1", Content: "entity A", + Meta: map[string]any{"kind": "entity", "name": "A"}}, + {Variant: kccommon.VariantStructure, DocID: "d1", Content: "entity B", + Meta: map[string]any{"kind": "entity", "name": "B"}}, {Variant: kccommon.VariantStructure, DocID: "d1", Content: "first", Meta: map[string]any{"kind": "relation", "from": " A ", "to": " B ", "relation_type": " Related ", "source_chunk_ids": []string{"c1"}}}, {Variant: kccommon.VariantStructure, DocID: "d2", Content: "second", @@ -152,10 +169,20 @@ func TestMergeStructureDataset_NormalizesRelationIdentity(t *testing.T) { t.Fatalf("mergeStructureDataset: %v", err) } buckets := c.writer.(*fakeWriter).buckets - if len(buckets) != 1 { - t.Fatalf("normalized relation identity should produce one bucket, got %d: %+v", len(buckets), buckets) + if len(buckets) != 3 { + t.Fatalf("normalized relation identity should produce two entity buckets and one relation bucket, got %d: %v", len(buckets), buckets) } - if got := len(buckets[0].SourceDocIDs); got != 2 { + var relationBucket *StructureBucket + for i := range buckets { + if buckets[i].Type == "relation" { + relationBucket = &buckets[i] + break + } + } + if relationBucket == nil { + t.Fatalf("missing relation bucket: %+v", buckets) + } + if got := len(relationBucket.SourceDocIDs); got != 2 { t.Fatalf("relation source docs = %v, want two docs", buckets[0].SourceDocIDs) } } @@ -193,6 +220,9 @@ func TestMergeStructureDataset_DescriptionFallbacks(t *testing.T) { {Variant: kccommon.VariantMindmap, DocID: "d1", Content: `{"name":"Root","type":"mindmap"}`, Meta: map[string]any{"kind": "entity", "name": "Root", "entity_type": "mindmap"}}, + {Variant: kccommon.VariantMindmap, DocID: "d1", + Content: `{"name":"Child","type":"mindmap"}`, + Meta: map[string]any{"kind": "entity", "name": "Child", "entity_type": "mindmap"}}, {Variant: kccommon.VariantMindmap, DocID: "d1", Content: `{"from":"Root","to":"Child","type":"related"}`, Meta: map[string]any{"kind": "relation", "from": "Root", "to": "Child", "relation_type": "related"}}, @@ -201,14 +231,15 @@ func TestMergeStructureDataset_DescriptionFallbacks(t *testing.T) { t.Fatalf("mergeStructureDataset: %v", err) } fw := c.writer.(*fakeWriter) - if len(fw.buckets) != 2 { - t.Fatalf("want entity and relation buckets, got %d", len(fw.buckets)) + if len(fw.buckets) != 3 { + t.Fatalf("want two entity and one relation bucket, got %d", len(fw.buckets)) } for _, b := range fw.buckets { switch b.Type { case "mindmap": - if b.Description != "Root" { - t.Errorf("entity fallback description = %q, want %q", b.Description, "Root") + want := b.Name + if b.Description != want { + t.Errorf("entity %q fallback description = %q, want %q", b.Name, b.Description, want) } case "relation": if b.Description != "Root related Child" { diff --git a/internal/ingestion/knowledge_compile/wiki_merge.go b/internal/ingestion/knowledge_compile/wiki_merge.go index df1d622236..d6ab34ff48 100644 --- a/internal/ingestion/knowledge_compile/wiki_merge.go +++ b/internal/ingestion/knowledge_compile/wiki_merge.go @@ -175,7 +175,7 @@ func unionWikiProvenance(a, b map[string]any) map[string]any { } // The incoming page replaces current page metadata. Identity and creation // time remain from the existing row. - for _, key := range []string{"summary", "kind"} { + for _, key := range []string{"summary", "kind", "content_md_raw"} { if v, ok := b[key]; ok { out[key] = v }