fix: align graph keyword search with Python (#19778)

This commit is contained in:
buua436 authored and GitHub committed 2026-09-17 17:24:26 +08:00
1 parent 37d88a65c2
commit df016eb5ea
9 files changed
+431 -130

No files matched your search

+13 -1
View File
@@ -255,13 +255,25 @@ func (h *DatasetArtifactHandler) GetArtifactAlteration(c *gin.Context) {
}
// GetArtifactGraph handles GET /artifacts/graph — wiki entity/relation graph.
// keywords seeds the graph with BM25 matches; top_n/topN controls the entity cap.
func (h *DatasetArtifactHandler) GetArtifactGraph(c *gin.Context) {
_, tenantID, _ := h.datasetOwner(c, c.Param("dataset_id"))
if tenantID == "" {
return
}
datasetID := c.Param("dataset_id")
graph, err := h.svc.GetWikiGraph(c.Request.Context(), tenantID, datasetID)
keywords := strings.TrimSpace(c.Query("keywords"))
topNValue := c.Query("top_n")
if topNValue == "" {
topNValue = c.Query("topN")
}
var topN *int
if topNValue != "" {
if value, parseErr := strconv.Atoi(topNValue); parseErr == nil {
topN = &value
}
}
graph, err := h.svc.GetWikiGraph(c.Request.Context(), tenantID, datasetID, keywords, topN)
if err != nil {
common.ErrorWithCode(c, common.CodeDataError, err.Error())
return
@@ -714,12 +714,108 @@ func (p *wikiPipeline) mapBatch(batch []common.Chunk) (wikiExtract, error) {
}
func (p *wikiPipeline) runPlan() (wikiPlan, error) {
if p.wikiMode() == "entity" {
return p.runEntityPlan()
}
if p.wikiMode() == "topic" && p.deps.Embed != nil {
return p.runTopicPlan()
}
return p.runLegacyPlan()
}
// runEntityPlan builds one page for every reduced entity or concept. Entity
// mode already has a canonical identity from MAP/REDUCE, so a PLAN LLM call
// would only repeat fields that the compiler can derive deterministically.
func (p *wikiPipeline) runEntityPlan() (wikiPlan, error) {
if err := p.ctx.Err(); err != nil {
return wikiPlan{}, err
}
plan := wikiPlan{Pages: buildWikiEntityPlanPages(p.reduced)}
plan.Pages = assembleWikiPlanRelatedPages(plan.Pages, p.reduced.Relations)
plan.Pages = normalizeWikiPlanPages(plan.Pages, p.reduced)
plan.Pages = normalizeWikiPlanPageLinks(plan.Pages)
reconciled, err := p.reconcilePlan(plan)
if err != nil {
return wikiPlan{}, err
}
reconciled.Pages = normalizeWikiPlanPageLinks(reconciled.Pages)
return reconciled, nil
}
// buildWikiEntityPlanPages projects MAP/REDUCE identities into deterministic
// page metadata. Topic assignment uses source-chunk overlap with MAP topics;
// the page itself remains one-to-one with the extracted entity or concept.
func buildWikiEntityPlanPages(reduced wikiExtract) []wikiPlanPage {
pages := make([]wikiPlanPage, 0, len(reduced.Entities)+len(reduced.Concepts))
for _, entity := range reduced.Entities {
name := strings.TrimSpace(entity.Name)
if name == "" {
continue
}
pages = append(pages, wikiPlanPage{
Action: "CREATE",
Slug: entityPageSlug(name, entity.Type),
Title: name,
PageType: "entity",
Topic: selectWikiIdentityTopic(name, "entity", sourceChunkIDsForWikiIdentity(name, entity.SourceChunkIDs, reduced.Claims), reduced.Topics),
EntityNames: uniqueStrings(append([]string{name}, entity.Aliases...)),
Priority: len(pages) + 1,
})
}
for _, concept := range reduced.Concepts {
term := strings.TrimSpace(concept.Term)
if term == "" {
continue
}
pages = append(pages, wikiPlanPage{
Action: "CREATE",
Slug: "concept/" + slugify(term),
Title: term,
PageType: "concept",
Topic: selectWikiIdentityTopic(term, "concept", sourceChunkIDsForWikiIdentity(term, concept.SourceChunkIDs, reduced.Claims), reduced.Topics),
EntityNames: []string{term},
Priority: len(pages) + 1,
})
}
return pages
}
func sourceChunkIDsForWikiIdentity(name string, sourceChunkIDs []string, claims []wikiClaim) []string {
ids := append([]string(nil), sourceChunkIDs...)
for _, claim := range claims {
if normKey(claim.Subject) == normKey(name) {
ids = append(ids, claim.SourceChunkIDs...)
}
}
return uniqueStrings(ids)
}
func selectWikiIdentityTopic(name, pageType string, sourceChunkIDs []string, topics []wikiTopic) string {
identityChunks := normalizedStringSet(sourceChunkIDs)
bestPath := ""
bestOverlap := 0
for _, topic := range topics {
path := common.NormalizeWikiTopicPath(topic.Path)
if path == "" || (pageType != "topic" && normKey(common.WikiTopicLeaf(path)) == normKey(name)) {
continue
}
overlap := normalizedOverlapCount(identityChunks, topic.SourceChunkIDs)
if overlap == 0 || overlap < bestOverlap {
continue
}
if overlap == bestOverlap && bestPath != "" && normKey(path) >= normKey(bestPath) {
continue
}
bestPath = path
bestOverlap = overlap
}
if bestPath == "" {
return common.GeneralWikiTopic
}
return bestPath
}
func (p *wikiPipeline) runLegacyPlan() (wikiPlan, error) {
batches := packWikiPlanBatches(p.reduced, wikiPlanTokenBudget)
if len(batches) == 0 {
@@ -235,7 +235,8 @@ func TestRunPlan_PromptMaxPagesNeverExceedsCap(t *testing.T) {
entities = append(entities, wikiEntity{Name: "Ent " + itoa(i) + big})
}
p := &wikiPipeline{
ctx: context.Background(),
ctx: context.Background(),
param: topicModeParam(),
deps: common.Deps{
ModelContextLen: 1024,
Chat: chatFunc(func(_ context.Context, req common.ChatRequest) (*common.ChatResponse, error) {
@@ -352,7 +353,8 @@ func TestRunPlan_TruncatesToGlobalHardCap(t *testing.T) {
}
payload := map[string]any{"pages": pages}
p := &wikiPipeline{
ctx: context.Background(),
ctx: context.Background(),
param: topicModeParam(),
deps: common.Deps{
Chat: reconcileChatStub{resp: mustJSON(payload)},
},
@@ -440,7 +442,8 @@ func TestRunPlan_ParallelBatchesMergeInOrder(t *testing.T) {
// batch2 (token budget 3500).
big := strings.Repeat("x", 7000)
p := &wikiPipeline{
ctx: context.Background(),
ctx: context.Background(),
param: topicModeParam(),
deps: common.Deps{
Chat: batchPlanChatStub{},
},
@@ -499,7 +502,8 @@ func TestRunPlan_ParallelBatchesFirstError(t *testing.T) {
big := strings.Repeat("x", 7000)
boom := errors.New("planning failed")
p := &wikiPipeline{
ctx: context.Background(),
ctx: context.Background(),
param: topicModeParam(),
deps: common.Deps{
Chat: failPlanChatStub{err: boom},
},
@@ -548,7 +552,8 @@ func TestRunPlan_CancelledCtxAborts(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
cancel()
p := &wikiPipeline{
ctx: ctx,
ctx: ctx,
param: topicModeParam(),
deps: common.Deps{
Chat: batchPlanChatStub{},
},
@@ -565,3 +570,8 @@ func TestRunPlan_CancelledCtxAborts(t *testing.T) {
t.Fatalf("runPlan err = nil, want context cancelled")
}
}
func topicModeParam() common.Param {
enabled := true
return common.Param{Plan: &enabled}
}
@@ -34,6 +34,69 @@ func TestReduceExtracts_MergesProvenance(t *testing.T) {
}
}
func TestRunPlan_EntityModeBuildsPagesFromReducedExtract(t *testing.T) {
p := &wikiPipeline{
ctx: context.Background(),
reduced: wikiExtract{
Entities: []wikiEntity{{
Name: "刘备",
Type: "person",
Aliases: []string{"玄德"},
SourceChunkIDs: []string{"c1"},
}},
Concepts: []wikiConcept{{
Term: "三国演义",
Definition: "历史小说",
SourceChunkIDs: []string{"c2"},
}},
Claims: []wikiClaim{
{Subject: "刘备", SourceChunkIDs: []string{"c1"}},
{Subject: "三国演义", SourceChunkIDs: []string{"c2"}},
},
Relations: []wikiRelation{{From: "刘备", To: "三国演义", Type: "appears_in"}},
Topics: []wikiTopic{
{Path: "历史/人物", SourceChunkIDs: []string{"c1"}},
{Path: "历史/作品", SourceChunkIDs: []string{"c2"}},
},
},
docID: "doc-1",
}
plan, err := p.runPlan()
if err != nil {
t.Fatalf("runPlan err = %v", err)
}
if len(plan.Pages) != 2 {
t.Fatalf("plan pages = %d, want 2", len(plan.Pages))
}
bySlug := make(map[string]wikiPlanPage, len(plan.Pages))
for _, page := range plan.Pages {
bySlug[page.Slug] = page
}
entityPage, ok := bySlug["entity/person/刘备"]
if !ok {
t.Fatalf("entity page missing: %#v", plan.Pages)
}
if entityPage.Title != "刘备" || entityPage.PageType != "entity" || entityPage.Topic != "历史/人物" {
t.Fatalf("entity page metadata = %#v", entityPage)
}
if !slices.Equal(entityPage.EntityNames, []string{"刘备", "玄德"}) {
t.Fatalf("entity page names = %#v", entityPage.EntityNames)
}
if !slices.Equal(entityPage.RelatedKB, []string{"concept/三国演义"}) {
t.Fatalf("entity related pages = %#v", entityPage.RelatedKB)
}
conceptPage, ok := bySlug["concept/三国演义"]
if !ok {
t.Fatalf("concept page missing: %#v", plan.Pages)
}
if conceptPage.Topic != "历史/作品" || !slices.Equal(conceptPage.RelatedKB, []string{"entity/person/刘备"}) {
t.Fatalf("concept page metadata = %#v", conceptPage)
}
}
func TestParseWikiExtractNormalizesTopicPathAndProvenance(t *testing.T) {
extract := parseWikiExtract(map[string]any{
"topics": []any{map[string]any{
@@ -18,6 +18,7 @@ package knowledge_compile
import (
"context"
"encoding/json"
"strings"
"testing"
)
@@ -51,6 +52,27 @@ func findRelation(rows []map[string]interface{}, from, to string) map[string]int
const kbForTest = "kb1"
func TestProjectWikiGraphRowsWritesSearchContent(t *testing.T) {
w := engineWriter{}
p := page("entity", "刘备")
p.Summary = "三国演义中的人物"
rows, err := w.projectWikiGraphRows(context.Background(), "t1", kbForTest, []wikiPageProjection{p})
if err != nil {
t.Fatalf("project: %v", err)
}
entity := findEntity(rows, "entity/刘备")
if entity == nil {
t.Fatal("missing entity")
}
for _, field := range []string{"title_tks", "title_sm_tks", "content_ltks", "content_sm_ltks"} {
value, ok := entity[field].(string)
if !ok || strings.TrimSpace(value) == "" {
t.Fatalf("entity[%q] = %#v, want non-empty search content", field, entity[field])
}
}
}
// TestProjectWikiGraphRowsCrossBatchEdgeSurvives verifies a relation whose two
// endpoints land in different "batches" (here modeled as two pages) is still
// materialized: the projection is global over the full page set, so edges never
@@ -34,6 +34,7 @@ import (
"ragflow/internal/entity"
kccommon "ragflow/internal/ingestion/component/knowledge_compiler/common"
"ragflow/internal/service/file"
"ragflow/internal/tokenizer"
"ragflow/internal/utility"
)
@@ -1385,6 +1386,35 @@ func wikiGraphBareKey(slug string) string {
return strings.TrimSpace(s)
}
// tokenizeWikiGraphContent prepares the lexical fields used by wiki graph
// keyword search. Keep the raw text as a fallback so graph rows remain
// searchable when the tokenizer pool is unavailable during startup or tests.
func tokenizeWikiGraphContent(title, slug, description string) (string, string, string, string) {
title = strings.TrimSpace(title)
content := strings.TrimSpace(strings.Join([]string{slug, description}, " "))
if title == "" && content == "" {
return "", "", "", ""
}
titleLTKS, err := tokenizer.Tokenize(title)
if err != nil || titleLTKS == "" {
titleLTKS = title
}
titleSMLTKS, err := tokenizer.FineGrainedTokenize(titleLTKS)
if err != nil || titleSMLTKS == "" {
titleSMLTKS = titleLTKS
}
contentLTKS, err := tokenizer.Tokenize(content)
if err != nil || contentLTKS == "" {
contentLTKS = content
}
contentSMLTKS, err := tokenizer.FineGrainedTokenize(contentLTKS)
if err != nil || contentSMLTKS == "" {
contentSMLTKS = contentLTKS
}
return titleLTKS, titleSMLTKS, contentLTKS, contentSMLTKS
}
// wikiPageProjection is the subset of a merged wiki_page row that the graph
// projection needs. It is reconstructed from the stored display columns (the
// same fields GetWikiGraph reads back), not from the JSON payload.
@@ -1799,6 +1829,7 @@ func (w engineWriter) projectWikiGraphRows(_ context.Context, tenant, kb string,
if err != nil {
return nil, err
}
titleLTKS, titleSMLTKS, contentLTKS, contentSMLTKS := tokenizeWikiGraphContent(p.Title, p.Slug, p.Summary)
rows = append(rows, map[string]interface{}{
"id": entityID,
"doc_id": kb,
@@ -1815,6 +1846,10 @@ func (w engineWriter) projectWikiGraphRows(_ context.Context, tenant, kb string,
"weight_int": weight,
"source_chunk_ids": p.SourceChunkIDs,
"source_doc_ids": capSourceDocs(p.SourceDocIDs),
"title_tks": titleLTKS,
"title_sm_tks": titleSMLTKS,
"content_ltks": contentLTKS,
"content_sm_ltks": contentSMLTKS,
"content_with_weight": string(content),
})
+64 -10
View File
@@ -42,6 +42,8 @@ const (
CompileKwdSkill = "skill"
CompileKwdSkillAll = "skill_all"
CompileKwdDatasetNav = "dataset_nav"
defaultWikiGraphTopN = 512
maxWikiGraphTopN = 1024
)
// DatasetArtifactService reads knowledge-compilation artifacts (wiki pages,
@@ -61,6 +63,10 @@ func wikiIndexName(tenantID string) string {
// searchCompiled runs a filtered search over the tenant's document index for
// the given dataset, returning the matching chunks.
func (s *DatasetArtifactService) searchCompiled(ctx context.Context, tenantID, datasetID string, filter map[string]interface{}, selectFields []string, offset, limit int, orderBy *types.OrderByExpr) ([]map[string]interface{}, int64, error) {
return s.searchCompiledWithMatch(ctx, tenantID, datasetID, filter, selectFields, offset, limit, orderBy, nil)
}
func (s *DatasetArtifactService) searchCompiledWithMatch(ctx context.Context, tenantID, datasetID string, filter map[string]interface{}, selectFields []string, offset, limit int, orderBy *types.OrderByExpr, matchExprs []interface{}) ([]map[string]interface{}, int64, error) {
docEngine := engine.Get()
if docEngine == nil {
return nil, 0, fmt.Errorf("document engine is not initialized")
@@ -80,6 +86,7 @@ func (s *DatasetArtifactService) searchCompiled(ctx context.Context, tenantID, d
SelectFields: selectFields,
Filter: merged,
OrderBy: orderBy,
MatchExprs: matchExprs,
})
if err != nil {
return nil, 0, err
@@ -544,8 +551,12 @@ var wikiTopicCollator = collate.New(language.Chinese)
// WikiGraph is the entity/relation graph for a dataset's wiki artifacts.
type WikiGraph struct {
Entities []WikiGraphEntity `json:"entities"`
Relations []WikiGraphRelation `json:"relations"`
Entities []WikiGraphEntity `json:"entities"`
Relations []WikiGraphRelation `json:"relations"`
TotalEntities int `json:"total_entities"`
TotalRelations int `json:"total_relations"`
ReturnedEntities int `json:"returned_entities"`
ReturnedRelations int `json:"returned_relations"`
}
// WikiGraphEntity is a single wiki graph entity.
@@ -565,12 +576,47 @@ type WikiGraphRelation struct {
To string `json:"to"`
}
// GetWikiGraph returns the wiki entity/relation graph for a dataset.
func (s *DatasetArtifactService) GetWikiGraph(ctx context.Context, tenantID, datasetID string) (*WikiGraph, error) {
entityChunks, _, err := s.searchCompiled(ctx, tenantID, datasetID,
map[string]interface{}{"compile_kwd": []string{CompileKwdWikiEntity}, "available_int": 1},
[]string{"slug_kwd", "title_kwd", "aliases_kwd", "description_with_weight", "entity_type_kwd", "weight_int", "source_chunk_ids"},
0, 1000, (&types.OrderByExpr{}).Desc("weight_int"))
// GetWikiGraph returns an incremental wiki graph slice for a dataset. keywords
// seeds the overview with BM25 entity matches; topN overrides the entity cap.
func (s *DatasetArtifactService) GetWikiGraph(ctx context.Context, tenantID, datasetID, keywords string, topN *int) (*WikiGraph, error) {
entityFilter := map[string]interface{}{"compile_kwd": []string{CompileKwdWikiEntity}, "available_int": 1}
relationFilter := map[string]interface{}{"compile_kwd": []string{CompileKwdWikiRelation}, "available_int": 1}
_, totalEntities, err := s.searchCompiled(ctx, tenantID, datasetID, entityFilter, []string{"id"}, 0, 1, nil)
if err != nil {
return nil, err
}
_, totalRelations, err := s.searchCompiled(ctx, tenantID, datasetID, relationFilter, []string{"id"}, 0, 1, nil)
if err != nil {
return nil, err
}
limit := defaultWikiGraphTopN
if topN != nil {
limit = *topN
if limit < 1 {
limit = 1
}
if limit > maxWikiGraphTopN {
limit = maxWikiGraphTopN
}
}
keywords = strings.TrimSpace(keywords)
var matchExprs []interface{}
var orderBy *types.OrderByExpr
if keywords != "" {
matchExprs = []interface{}{&types.MatchTextExpr{
Fields: []string{"title_tks^10", "title_sm_tks^5", "content_ltks^2", "content_sm_ltks"},
MatchingText: keywords,
TopN: limit,
ExtraOptions: map[string]interface{}{"original_query": keywords},
}}
} else {
orderBy = (&types.OrderByExpr{}).Desc("weight_int")
}
entityChunks, _, err := s.searchCompiledWithMatch(ctx, tenantID, datasetID, entityFilter,
[]string{"id", "slug_kwd", "title_kwd", "aliases_kwd", "description_with_weight", "entity_type_kwd", "weight_int", "source_chunk_ids"},
0, limit, orderBy, matchExprs)
if err != nil {
return nil, err
}
@@ -578,7 +624,12 @@ func (s *DatasetArtifactService) GetWikiGraph(ctx context.Context, tenantID, dat
for _, c := range entityChunks {
fromKwds = append(fromKwds, firstStringValue(c["slug_kwd"]))
}
graph := &WikiGraph{Entities: []WikiGraphEntity{}, Relations: []WikiGraphRelation{}}
graph := &WikiGraph{
Entities: []WikiGraphEntity{},
Relations: []WikiGraphRelation{},
TotalEntities: int(totalEntities),
TotalRelations: int(totalRelations),
}
for _, c := range entityChunks {
w := intValue(c["weight_int"])
// Match the GetWikiPage / ListWikiPages contract: expose the bare slug
@@ -605,8 +656,9 @@ func (s *DatasetArtifactService) GetWikiGraph(ctx context.Context, tenantID, dat
})
}
if len(fromKwds) > 0 {
relationFilter["from_kwd"] = fromKwds
relChunks, _, err := s.searchCompiled(ctx, tenantID, datasetID,
map[string]interface{}{"compile_kwd": []string{CompileKwdWikiRelation}, "available_int": 1, "from_kwd": fromKwds},
relationFilter,
[]string{"from_kwd", "to_kwd"}, 0, 10000, nil)
if err != nil {
return nil, err
@@ -621,6 +673,8 @@ func (s *DatasetArtifactService) GetWikiGraph(ctx context.Context, tenantID, dat
})
}
}
graph.ReturnedEntities = len(graph.Entities)
graph.ReturnedRelations = len(graph.Relations)
return graph, nil
}
+108 -114
View File
@@ -25,7 +25,7 @@ import (
const (
graphFullThreshold = 1024 // below this combined count, return all rows
graphTopEntities = 256 // seed set A size for large buckets
graphKeywordCandidates = 16 // keyword/KNN candidate rows
graphKeywordCandidates = 16 // keyword candidate rows
graphExpansionCap = 4096 // hub-node expansion cap
)
@@ -684,65 +684,6 @@ func structureKindForBucket(kind string) string {
return strings.ReplaceAll(k, "-", "_")
}
// loadDocumentTemplateMeta resolves the doc's configured template group into
// configured_ids + template_meta maps, mirroring the Python resolution.
func (s *DatasetArtifactService) loadDocumentTemplateMeta(ctx context.Context, tenantID, documentID string) ([]string, map[string]map[string]interface{}, error) {
empty := map[string]map[string]interface{}{}
doc, err := dao.NewDocumentDAO().GetByID(ctx, dao.DB, documentID)
if err != nil || doc == nil {
return nil, empty, err
}
groupID := ""
if doc.ParserConfig != nil {
if gid, ok := doc.ParserConfig["compilation_template_group_id"].(string); ok && gid != "" {
groupID = gid
} else if ext, ok := doc.ParserConfig["ext"].(map[string]interface{}); ok {
if gid, ok := ext["compilation_template_group_id"].(string); ok && gid != "" {
groupID = gid
}
}
}
if groupID == "" {
return nil, empty, nil
}
group, err := NewCompilationTemplateGroupService().GetSaved(ctx, tenantID, groupID)
if err != nil {
return nil, empty, err
}
configuredIDs := []string{}
meta := map[string]map[string]interface{}{}
seen := map[string]bool{}
if group != nil {
for _, t := range group.Templates {
tid := t.ID
if tid == "" || seen[tid] {
continue
}
rawKind := ""
if t.Config != nil {
if k, ok := t.Config["kind"].(string); ok {
rawKind = k
}
}
if rawKind == "" {
rawKind = t.Kind
}
kindNorm := compilationTemplateKind(rawKind)
if kindNorm == "wiki" {
continue
}
seen[tid] = true
configuredIDs = append(configuredIDs, tid)
meta[tid] = map[string]interface{}{
"template_id": tid,
"template_name": t.Name,
"kind": rawKind,
}
}
}
return configuredIDs, meta, nil
}
// DocumentStructureGraphInput is the parsed request for GetDocumentGraph.
type DocumentStructureGraphInput struct {
TenantID string
@@ -755,10 +696,8 @@ type DocumentStructureGraphInput struct {
// GetDocumentGraph returns the per-template structure graph of a document,
// mirroring Python get_document_structure_graph (normal + keywords modes).
func (s *DatasetArtifactService) GetDocumentGraph(ctx context.Context, in DocumentStructureGraphInput) (*DocumentStructureGraphResponse, error) {
configuredIDs, templateMeta, err := s.loadDocumentTemplateMeta(ctx, in.TenantID, in.DocumentID)
if err != nil {
return nil, err
}
templateIDs := []string{}
templateMeta := map[string]map[string]interface{}{}
resp := &DocumentStructureGraphResponse{Templates: []DocumentStructureGraphTemplate{}}
entityCountFilter := map[string]interface{}{
@@ -771,13 +710,17 @@ func (s *DatasetArtifactService) GetDocumentGraph(ctx context.Context, in Docume
}
resp.TotalEntities = int(entityTotal)
// keywords mode: name matching/KNN → matched entities' subgraph.
// keywords mode: name/BM25 matching → matched entities' subgraph.
if in.Keywords != "" {
bucketMeta, entities, relations, err := s.keywordSubgraph(ctx, in.TenantID, in.DatasetID, in.DocumentID, in.Keywords, templateMeta, nil)
if err != nil {
return nil, err
}
if bucketMeta == nil || (len(entities) == 0 && len(relations) == 0) {
if err := s.discoverDocumentGraphTemplateMeta(ctx, in.TenantID, in.DatasetID, in.DocumentID, &templateIDs, templateMeta); err != nil {
return nil, err
}
resp.Templates = buildDocumentGraphTemplateShells(templateIDs, templateMeta)
return resp, nil
}
resp.Templates = append(resp.Templates, DocumentStructureGraphTemplate{
@@ -817,10 +760,9 @@ func (s *DatasetArtifactService) GetDocumentGraph(ctx context.Context, in Docume
bucketScopes := map[string]map[string]interface{}{}
templateDAO := dao.NewCompilationTemplateDAO()
for _, row := range metaRows {
// Pipeline-produced rows may contain a template id even when the
// document parser config does not contain the corresponding template
// group. Resolve that id directly so template_name is the user-visible
// name instead of the opaque id.
// Resolve template metadata from the persisted graph row. The template
// name is loaded only as display metadata; the row remains authoritative
// for whether the template exists in this document.
templateID := rowTemplateID(row)
if templateID != "" {
if _, known := templateMeta[templateID]; !known {
@@ -836,6 +778,9 @@ func (s *DatasetArtifactService) GetDocumentGraph(ctx context.Context, in Docume
}
meta, scope := resolveGraphBucket(row, templateMeta, in.DocumentID)
bid := graphStr(meta["template_id"])
if bid != "" && !containsStr(templateIDs, bid) {
templateIDs = append(templateIDs, bid)
}
if _, ok := bucketMetas[bid]; !ok {
bucketMetas[bid] = meta
bucketScopes[bid] = scope
@@ -859,9 +804,9 @@ func (s *DatasetArtifactService) GetDocumentGraph(ctx context.Context, in Docume
}
}
// Order: configured templates first, then discovered.
// Order: discovered templates first, then any buckets not seen in the scan.
orderedIDs := []string{}
for _, tid := range configuredIDs {
for _, tid := range templateIDs {
if _, ok := grouped[tid]; ok && !containsStr(orderedIDs, tid) {
orderedIDs = append(orderedIDs, tid)
}
@@ -889,6 +834,93 @@ func containsStr(list []string, v string) bool {
return false
}
func buildDocumentGraphTemplateShells(templateIDs []string, templateMeta map[string]map[string]interface{}) []DocumentStructureGraphTemplate {
templates := make([]DocumentStructureGraphTemplate, 0, len(templateIDs))
for _, templateID := range templateIDs {
meta := templateMeta[templateID]
if meta == nil {
continue
}
templateName := graphStr(meta["template_name"])
if templateName == "" {
templateName = templateID
}
templates = append(templates, DocumentStructureGraphTemplate{
TemplateID: templateID,
TemplateName: templateName,
Kind: graphStr(meta["kind"]),
Entities: []StructureGraphNode{},
Relations: []StructureGraphRelation{},
})
}
return templates
}
func (s *DatasetArtifactService) discoverDocumentGraphTemplateMeta(ctx context.Context, tenantID, datasetID, documentID string, templateIDs *[]string, templateMeta map[string]map[string]interface{}) error {
if templateIDs == nil || templateMeta == nil {
return nil
}
seen := make(map[string]bool, len(*templateIDs))
for _, templateID := range *templateIDs {
seen[templateID] = true
}
templateDAO := dao.NewCompilationTemplateDAO()
fields := []string{"id", "compile_kwd", "compilation_template_ids", "compilation_template_kind_kwd"}
filter := map[string]interface{}{
"doc_id": []string{documentID},
"knowledge_graph_kwd": []string{"entity", "relation"},
}
for offset := 0; ; offset += 1000 {
rows, total, err := graphRowSearch(ctx, tenantID, datasetID, fields, filter, nil, offset, 1000, nil)
if err != nil {
return err
}
for _, row := range rows {
templateID := rowTemplateID(row)
kind := firstStringValue(row["compilation_template_kind_kwd"])
if kind == "" {
kind = firstStringValue(row["compile_kwd"])
}
if templateID == "" {
compileKind := firstStringValue(row["compile_kwd"])
if compileKind == "" {
continue
}
templateID = "legacy:" + compileKind
}
if seen[templateID] {
continue
}
templateName := templateID
if !strings.HasPrefix(templateID, "legacy:") {
if template, loadErr := templateDAO.GetTemplate(ctx, dao.DB, tenantID, templateID); loadErr == nil && template != nil {
templateName = template.Name
if kind == "" {
kind = template.Kind
}
}
} else {
templateName = "Legacy (" + strings.TrimPrefix(templateID, "legacy:") + ")"
}
if templateName == "" {
templateName = templateID
}
templateMeta[templateID] = map[string]interface{}{
"template_id": templateID,
"template_name": templateName,
"kind": kind,
}
*templateIDs = append(*templateIDs, templateID)
seen[templateID] = true
}
if int64(offset+1000) >= total || len(rows) == 0 {
break
}
}
return nil
}
// DatasetStructureGraphInput is the parsed request for the dataset-scope
// structure graph endpoint (GET /datasets/:id/artifacts/structure).
// Kind is REQUIRED (mirrors Python dataset_api.py:715): missing or invalid →
@@ -922,8 +954,7 @@ type DatasetStructureGraphResponse struct {
// way). It collects distinct template ids, then reads each template's dataset
// entity/relation rows via buildBucket. It does NOT read kg_build_meta (write/
// delete-side only). When Keywords is set, it returns the keyword-matched
// entity subgraph using the same BM25/KNN and relation expansion path as the
// document-level endpoint.
// entity subgraph using the name/BM25 and relation expansion path.
func (s *DatasetArtifactService) GetDatasetStructure(ctx context.Context, in DatasetStructureGraphInput) (*DatasetStructureGraphResponse, error) {
resolved := resolveDatasetStructureKind(in.Kind)
if resolved == "" {
@@ -1189,7 +1220,8 @@ func nameMatchesQuery(node StructureGraphNode, query string) bool {
return true
}
// keywordSubgraph mirrors sgc.keyword_subgraph (BM25 + KNN fallback + ancestor walk).
// keywordSubgraph mirrors sgc.keyword_subgraph (lexical lookup and ancestor
// walk).
func (s *DatasetArtifactService) keywordSubgraph(ctx context.Context, tenantID, datasetID, documentID, keywords string, templateMeta map[string]map[string]interface{}, baseEntityCond map[string]interface{}) (map[string]interface{}, []StructureGraphNode, []StructureGraphRelation, error) {
if baseEntityCond == nil {
baseEntityCond = map[string]interface{}{
@@ -1208,10 +1240,8 @@ func (s *DatasetArtifactService) keywordSubgraph(ctx context.Context, tenantID,
resolveBucket := func(row map[string]interface{}) (map[string]interface{}, map[string]interface{}) {
templateID := rowTemplateID(row)
if templateID != "" && !templateMetaLoaded[templateID] {
// Pipeline-produced rows can carry a template id even when the
// document parser config does not contain the corresponding group.
// Resolve it here as well as in the normal path so keyword searches
// expose the user-visible template name instead of the opaque id.
// Resolve the display metadata from the persisted template id so
// keyword searches expose the user-visible name instead of the id.
templateMetaLoaded[templateID] = true
if template, err := templateDAO.GetTemplate(ctx, dao.DB, tenantID, templateID); err == nil && template != nil {
templateMeta[templateID] = map[string]interface{}{
@@ -1323,26 +1353,6 @@ func (s *DatasetArtifactService) keywordSubgraph(ctx context.Context, tenantID,
candidates = detailCandidates
}
// Semantic fallback via KNN.
if len(candidates) == 0 {
vec, err := embedQuery(ctx, tenantID, datasetID, keywords)
if err != nil || len(vec) == 0 {
return nil, nil, nil, nil
}
denseExpr := &types.MatchDenseExpr{
VectorColumnName: fmt.Sprintf("q_%d_vec", len(vec)),
EmbeddingData: vec,
EmbeddingDataType: "float",
DistanceType: "cosine",
TopN: graphKeywordCandidates,
ExtraOptions: map[string]interface{}{"similarity": 0.3},
}
topMap, _, err := graphRowSearch(ctx, tenantID, datasetID, topFields, baseEntityCond, nil, 0, graphKeywordCandidates, []interface{}{denseExpr})
if err != nil {
return nil, nil, nil, err
}
candidates = validTop(topMap)
}
if len(candidates) == 0 {
return nil, nil, nil, nil
}
@@ -1498,19 +1508,3 @@ func (s *DatasetArtifactService) keywordSubgraph(ctx context.Context, tenantID,
}
return bucketMeta, entities, normalizeRelationEndpoints(entities, relations), nil
}
// embedQuery embeds a query via the nav embedder (best-effort). The embedder
// resolves the tenant's embedding model on demand; an empty result (or error) is
// treated as "no semantic candidates", matching Python's fallback behavior.
func embedQuery(ctx context.Context, tenantID, datasetID, query string) ([]float64, error) {
embedder := NewNavEmbedder(NewModelProviderService(), "")
vecs, err := embedder.Encode(ctx, tenantID, []string{query})
if err != nil || len(vecs) == 0 || len(vecs[0]) == 0 {
return nil, fmt.Errorf("embed query failed")
}
f := make([]float64, len(vecs[0]))
for i, v := range vecs[0] {
f[i] = float64(v)
}
return f, nil
}
@@ -126,3 +126,18 @@ func TestResolveGraphBucket_UsesRawStructureMetadata(t *testing.T) {
t.Fatalf("scope = %#v, want template-scoped raw-row filter", scope)
}
}
func TestBuildDocumentGraphTemplateShells_PreservesTemplateMetadata(t *testing.T) {
templates := buildDocumentGraphTemplateShells([]string{"tree-1"}, map[string]map[string]interface{}{
"tree-1": {"template_name": "Tree", "kind": "tree"},
})
if len(templates) != 1 {
t.Fatalf("got %d templates, want 1", len(templates))
}
if templates[0].TemplateID != "tree-1" || templates[0].TemplateName != "Tree" || templates[0].Kind != "tree" {
t.Fatalf("template = %#v, want metadata for tree-1", templates[0])
}
if templates[0].Entities == nil || templates[0].Relations == nil {
t.Fatal("empty template collections must be non-nil")
}
}