mirror of
https://github.com/Tencent/WeKnora.git
synced 2026-10-28 03:29:30 +08:00
fix(graph): feed OCR chunks to graph extraction instead of image-link placeholders (#3092)
* fix(graph): feed OCR chunks to graph extraction instead of image-link placeholders KnowledgePostProcessService.Handle gathered "text-like" chunks through chunkService.ListChunksByKnowledgeID, which is text-only by design, so the image_ocr / image_caption chunks written by multimodal parsing were never returned. For a scanned PDF the text chunks contain nothing but markdown image links, so graph extraction ran on empty input: most calls failed with "unexpected end of JSON input" and the rest echoed the few-shot example from the extract config into the graph. - Fetch the exact chunk types the post-process enriches via chunkRepo.ListChunksByKnowledgeIDAndTypes (text, image_ocr, image_caption); the service now receives interfaces.ChunkRepository through dig like knowledgeService already does - Build a dedicated graphChunks list for graph extraction: skip image_caption chunks (they duplicate the OCR text) and chunks that contain only image links (chunkHasExtractableText) - graphChunkCount, the enqueue loop and the log line use graphChunks; summary, wiki, auto-tag and question generation keep using textChunks - Tests: fake ChunkRepository in the wiki-enqueue test, table test for chunkHasExtractableText * fix(graph): skip figure OCR and harden empty-chunk detection Reuse the CommonMark image scanner plus extractRealText so titles with parentheses and <image> wrappers are not treated as prose. Only enqueue OCR for graph extraction when the parent text chunk has no extractable text, and skip link-only chunks for question generation as well. * style(graph): wrap long post-process extract log line --------- Co-authored-by: wizardchen <wizardchen@tencent.com>
This commit is contained in:
co-authored by
wizardchen
parent
4d95cd4581
commit
8d7298fb5d
@@ -9,6 +9,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/application/access"
|
||||
"github.com/Tencent/WeKnora/internal/infrastructure/docparser"
|
||||
"github.com/Tencent/WeKnora/internal/logger"
|
||||
"github.com/Tencent/WeKnora/internal/tracing/langfuse"
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
@@ -23,6 +24,7 @@ type KnowledgePostProcessService struct {
|
||||
knowledgeRepo interfaces.KnowledgeRepository
|
||||
kbService interfaces.KnowledgeBaseService
|
||||
chunkService interfaces.ChunkService
|
||||
chunkRepo interfaces.ChunkRepository
|
||||
taskEnqueuer interfaces.TaskEnqueuer
|
||||
pendingRepo interfaces.TaskPendingOpsRepository
|
||||
redisClient *redis.Client
|
||||
@@ -33,6 +35,7 @@ func NewKnowledgePostProcessService(
|
||||
knowledgeRepo interfaces.KnowledgeRepository,
|
||||
kbService interfaces.KnowledgeBaseService,
|
||||
chunkService interfaces.ChunkService,
|
||||
chunkRepo interfaces.ChunkRepository,
|
||||
taskEnqueuer interfaces.TaskEnqueuer,
|
||||
pendingRepo interfaces.TaskPendingOpsRepository,
|
||||
redisClient *redis.Client,
|
||||
@@ -42,6 +45,7 @@ func NewKnowledgePostProcessService(
|
||||
knowledgeRepo: knowledgeRepo,
|
||||
kbService: kbService,
|
||||
chunkService: chunkService,
|
||||
chunkRepo: chunkRepo,
|
||||
taskEnqueuer: taskEnqueuer,
|
||||
pendingRepo: pendingRepo,
|
||||
redisClient: redisClient,
|
||||
@@ -158,8 +162,13 @@ func (s *KnowledgePostProcessService) Handle(ctx context.Context, task *asynq.Ta
|
||||
processOverrides, _ := knowledge.ProcessOverrides()
|
||||
eff := ResolveProcessConfig(kb, processOverrides)
|
||||
|
||||
// 2. Fetch all chunks
|
||||
chunks, err := s.chunkService.ListChunksByKnowledgeID(ctx, payload.KnowledgeID)
|
||||
// 2. Fetch all text-like chunks. ListChunksByKnowledgeID is text-only by
|
||||
// design, which silently dropped the OCR / Caption chunks that
|
||||
// multimodal parsing adds for scanned PDFs — leaving graph extraction
|
||||
// with nothing but image-link placeholders. Ask the repository for the
|
||||
// exact chunk types we enrich instead.
|
||||
chunks, err := s.chunkRepo.ListChunksByKnowledgeIDAndTypes(ctx, payload.TenantID, payload.KnowledgeID,
|
||||
[]types.ChunkType{types.ChunkTypeText, types.ChunkTypeImageOCR, types.ChunkTypeImageCaption})
|
||||
if err != nil {
|
||||
return fmt.Errorf("list chunks for knowledge %s: %w", payload.KnowledgeID, err)
|
||||
}
|
||||
@@ -179,6 +188,8 @@ func (s *KnowledgePostProcessService) Handle(ctx context.Context, task *asynq.Ta
|
||||
}
|
||||
}
|
||||
|
||||
graphChunks := selectGraphChunks(textChunks)
|
||||
|
||||
// 3. Compute the enrichment subtask count up front so we can flip to
|
||||
// "finalizing" with the right counter BEFORE spawning any subtasks.
|
||||
// Each subtask handler atomically decrements pending_subtasks_count
|
||||
@@ -207,12 +218,14 @@ func (s *KnowledgePostProcessService) Handle(ctx context.Context, task *asynq.Ta
|
||||
// call retries / cancels / traces independently. We only target
|
||||
// ChunkTypeText here — OCR / Caption chunks were never fed to question
|
||||
// generation in the legacy whole-knowledge loop, so excluding them
|
||||
// keeps behavior identical. Sorted by StartAt so the per-chunk
|
||||
// context (prev / next) matches the legacy ordering.
|
||||
// keeps behavior identical. Link-only text chunks (scanned PDF pages)
|
||||
// are skipped: the LLM has nothing to ask about and tends to echo the
|
||||
// few-shot example. Sorted by StartAt so the per-chunk context
|
||||
// (prev / next) matches the legacy ordering.
|
||||
var questionChunks []*types.Chunk
|
||||
if willSpawnQuestion {
|
||||
for _, c := range textChunks {
|
||||
if c.ChunkType == types.ChunkTypeText {
|
||||
if c.ChunkType == types.ChunkTypeText && chunkHasExtractableText(c.Content) {
|
||||
questionChunks = append(questionChunks, c)
|
||||
}
|
||||
}
|
||||
@@ -229,7 +242,7 @@ func (s *KnowledgePostProcessService) Handle(ctx context.Context, task *asynq.Ta
|
||||
|
||||
graphChunkCount := 0
|
||||
if eff.GraphEnabled {
|
||||
graphChunkCount = len(textChunks)
|
||||
graphChunkCount = len(graphChunks)
|
||||
}
|
||||
expectedSubtasks := 0
|
||||
if willSpawnSummary {
|
||||
@@ -401,8 +414,10 @@ func (s *KnowledgePostProcessService) Handle(ctx context.Context, task *asynq.Ta
|
||||
// 5. Spawn Graph RAG Tasks — only when graph indexing is enabled in IndexingStrategy
|
||||
enqueuedGraphCount := 0
|
||||
if graphChunkCount > 0 {
|
||||
logger.Infof(ctx, "[KnowledgePostProcess] Spawning Graph RAG extract tasks for %d text-like chunks", len(textChunks))
|
||||
for i, chunk := range textChunks {
|
||||
logger.Infof(ctx,
|
||||
"[KnowledgePostProcess] Spawning Graph RAG extract tasks for %d text/OCR chunks (of %d text-like)",
|
||||
len(graphChunks), len(textChunks))
|
||||
for i, chunk := range graphChunks {
|
||||
ok, err := NewChunkExtractTask(ctx, s.taskEnqueuer, payload.TenantID, chunk.ID, kb.SummaryModelID,
|
||||
payload.KnowledgeID, attempt, i)
|
||||
if err != nil {
|
||||
@@ -671,3 +686,48 @@ func (s *KnowledgePostProcessService) enqueueQuestionGenerationTasks(
|
||||
enqueued, total, questionGenChunkBatchSize, payload.KnowledgeID, questionCount)
|
||||
return enqueued
|
||||
}
|
||||
|
||||
// selectGraphChunks picks the chunks that should be sent to graph extraction.
|
||||
// Captions are dropped (they describe the page visually and duplicate OCR).
|
||||
// Text chunks that are only image placeholders are skipped so the extractor
|
||||
// cannot echo the few-shot example. OCR is kept when the parent text chunk
|
||||
// has no extractable prose (scanned PDF pages); OCR of figures next to real
|
||||
// text is skipped to avoid doubling LLM cost on illustrated documents.
|
||||
func selectGraphChunks(chunks []*types.Chunk) []*types.Chunk {
|
||||
textByID := make(map[string]*types.Chunk, len(chunks))
|
||||
for _, c := range chunks {
|
||||
if c != nil && c.ChunkType == types.ChunkTypeText {
|
||||
textByID[c.ID] = c
|
||||
}
|
||||
}
|
||||
var out []*types.Chunk
|
||||
for _, c := range chunks {
|
||||
if c == nil {
|
||||
continue
|
||||
}
|
||||
switch c.ChunkType {
|
||||
case types.ChunkTypeImageCaption:
|
||||
continue
|
||||
case types.ChunkTypeImageOCR:
|
||||
if !chunkHasExtractableText(c.Content) {
|
||||
continue
|
||||
}
|
||||
if parent := textByID[c.ParentChunkID]; parent != nil && chunkHasExtractableText(parent.Content) {
|
||||
continue
|
||||
}
|
||||
out = append(out, c)
|
||||
case types.ChunkTypeText:
|
||||
if chunkHasExtractableText(c.Content) {
|
||||
out = append(out, c)
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// chunkHasExtractableText reports whether a chunk still contains prose once
|
||||
// markdown images and <image> wrappers are removed, i.e. whether graph
|
||||
// extraction (or question generation) can find entities in it.
|
||||
func chunkHasExtractableText(content string) bool {
|
||||
return extractRealText(docparser.StripMarkdownImages(content)) != ""
|
||||
}
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"testing"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
"github.com/hibiken/asynq"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestChunkHasExtractableText(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
content string
|
||||
want bool
|
||||
}{
|
||||
{"empty", "", false},
|
||||
{"whitespace only", " \n\t ", false},
|
||||
{"single image link", "", false},
|
||||
{"several image links with newlines", "\n\n\n", false},
|
||||
{"prose", "第1章 総則", true},
|
||||
{"prose around an image link", "See figure:  below.", true},
|
||||
{"image link followed by OCR text", "\nSection 2: Definitions", true},
|
||||
{"title containing right paren", ` 结果")`, false},
|
||||
{"path with balanced parens", `.png "title")`, false},
|
||||
{"empty image wrapper", `<image url="x"><image_original></image_original></image>`, false},
|
||||
{"wrapper with OCR body", `<image url="images/p1.png">
|
||||
<image_original></image_original>
|
||||
<image_caption>scanned letter</image_caption>
|
||||
<image_ocr>Sehr geehrter Herr Mustermann</image_ocr>
|
||||
</image>`, true},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := chunkHasExtractableText(tt.content); got != tt.want {
|
||||
t.Fatalf("chunkHasExtractableText(%q) = %v, want %v", tt.content, got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSelectGraphChunks(t *testing.T) {
|
||||
chunks := []*types.Chunk{
|
||||
{ID: "text-link", ChunkType: types.ChunkTypeText, Content: ""},
|
||||
{ID: "ocr-scanned", ChunkType: types.ChunkTypeImageOCR, ParentChunkID: "text-link", Content: "第1章 総則"},
|
||||
{
|
||||
ID: "cap-scanned", ChunkType: types.ChunkTypeImageCaption,
|
||||
ParentChunkID: "text-link", Content: "a scanned page of printed text",
|
||||
},
|
||||
{ID: "text-prose", ChunkType: types.ChunkTypeText, Content: "Contract between Alice and Bob."},
|
||||
{ID: "ocr-figure", ChunkType: types.ChunkTypeImageOCR, ParentChunkID: "text-prose", Content: "ACME logo"},
|
||||
{
|
||||
ID: "text-title-paren", ChunkType: types.ChunkTypeText,
|
||||
Content: ` 结果")`,
|
||||
},
|
||||
{ID: "ocr-orphan", ChunkType: types.ChunkTypeImageOCR, Content: "standalone OCR"},
|
||||
}
|
||||
|
||||
got := selectGraphChunks(chunks)
|
||||
gotIDs := make([]string, len(got))
|
||||
for i, c := range got {
|
||||
gotIDs[i] = c.ID
|
||||
}
|
||||
assert.Equal(t, []string{"ocr-scanned", "text-prose", "ocr-orphan"}, gotIDs)
|
||||
}
|
||||
|
||||
func TestKnowledgePostProcessGraphEnqueueSelection(t *testing.T) {
|
||||
t.Setenv("NEO4J_ENABLE", "true")
|
||||
|
||||
const knowledgeID = "knowledge-graph-ocr"
|
||||
const kbID = "kb-graph"
|
||||
repo := &wikiEnqueueFailureKnowledgeRepo{
|
||||
knowledge: &types.Knowledge{
|
||||
ID: knowledgeID,
|
||||
TenantID: 7,
|
||||
KnowledgeBaseID: kbID,
|
||||
ParseStatus: types.ParseStatusProcessing,
|
||||
},
|
||||
}
|
||||
queue := &wikiEnqueueFailureTaskQueue{}
|
||||
bind := func(id, parent, content string, chunkType types.ChunkType) *types.Chunk {
|
||||
return &types.Chunk{
|
||||
ID: id,
|
||||
TenantID: 7,
|
||||
KnowledgeID: knowledgeID,
|
||||
KnowledgeBaseID: kbID,
|
||||
ParentChunkID: parent,
|
||||
ChunkType: chunkType,
|
||||
Content: content,
|
||||
}
|
||||
}
|
||||
service := &KnowledgePostProcessService{
|
||||
knowledgeRepo: repo,
|
||||
kbService: &wikiEnqueueFailureKBService{kb: &types.KnowledgeBase{
|
||||
ID: kbID,
|
||||
TenantID: 7,
|
||||
IndexingStrategy: types.IndexingStrategy{
|
||||
GraphEnabled: true,
|
||||
},
|
||||
ExtractConfig: &types.ExtractConfig{Enabled: true},
|
||||
}},
|
||||
chunkRepo: &wikiEnqueueFailureChunkRepo{chunks: []*types.Chunk{
|
||||
bind("text-link", "", "", types.ChunkTypeText),
|
||||
bind("ocr-scanned", "text-link", "第1章 総則", types.ChunkTypeImageOCR),
|
||||
bind("cap-scanned", "text-link", "a scanned page", types.ChunkTypeImageCaption),
|
||||
bind("text-prose", "", "Contract between Alice and Bob.", types.ChunkTypeText),
|
||||
bind("ocr-figure", "text-prose", "ACME logo", types.ChunkTypeImageOCR),
|
||||
}},
|
||||
taskEnqueuer: queue,
|
||||
}
|
||||
|
||||
payload, err := json.Marshal(types.KnowledgePostProcessPayload{
|
||||
TenantID: 7,
|
||||
KnowledgeID: knowledgeID,
|
||||
KnowledgeBaseID: kbID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
err = service.Handle(context.Background(), asynq.NewTask(types.TypeKnowledgePostProcess, payload))
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Equal(t, []string{"ocr-scanned", "text-prose"}, queue.extractChunkIDs)
|
||||
assert.Equal(t, []string{
|
||||
types.TypeSummaryGeneration,
|
||||
types.TypeChunkExtract,
|
||||
types.TypeChunkExtract,
|
||||
}, queue.taskTypes)
|
||||
assert.Equal(t, 3, repo.expectedSubtasks, "summary plus two graph extract slots")
|
||||
}
|
||||
@@ -36,6 +36,13 @@ func (r *wikiEnqueueFailureKnowledgeRepo) SetFinalizing(
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (r *wikiEnqueueFailureKnowledgeRepo) FinalizeSubtask(
|
||||
context.Context,
|
||||
string,
|
||||
) (int, bool, error) {
|
||||
return 0, false, nil
|
||||
}
|
||||
|
||||
func (r *wikiEnqueueFailureKnowledgeRepo) UpdateKnowledgeColumn(
|
||||
context.Context,
|
||||
string,
|
||||
@@ -71,8 +78,9 @@ func (s *wikiEnqueueFailureChunkService) ListChunksByKnowledgeID(
|
||||
|
||||
type wikiEnqueueFailureTaskQueue struct {
|
||||
interfaces.TaskEnqueuer
|
||||
taskTypes []string
|
||||
wikiErr error
|
||||
taskTypes []string
|
||||
extractChunkIDs []string
|
||||
wikiErr error
|
||||
}
|
||||
|
||||
func (q *wikiEnqueueFailureTaskQueue) Enqueue(
|
||||
@@ -83,6 +91,12 @@ func (q *wikiEnqueueFailureTaskQueue) Enqueue(
|
||||
if task.Type() == types.TypeWikiIngest {
|
||||
return nil, q.wikiErr
|
||||
}
|
||||
if task.Type() == types.TypeChunkExtract {
|
||||
var payload types.ExtractChunkPayload
|
||||
if err := json.Unmarshal(task.Payload(), &payload); err == nil {
|
||||
q.extractChunkIDs = append(q.extractChunkIDs, payload.ChunkID)
|
||||
}
|
||||
}
|
||||
return &asynq.TaskInfo{ID: "queued", Type: task.Type()}, nil
|
||||
}
|
||||
|
||||
@@ -110,6 +124,30 @@ func (r *wikiEnqueueFailurePendingRepo) SeedKnowledgeFinalizingWithPendingOp(
|
||||
return true, nil
|
||||
}
|
||||
|
||||
type wikiEnqueueFailureChunkRepo struct {
|
||||
interfaces.ChunkRepository
|
||||
chunks []*types.Chunk
|
||||
}
|
||||
|
||||
func (r *wikiEnqueueFailureChunkRepo) ListChunksByKnowledgeIDAndTypes(
|
||||
_ context.Context,
|
||||
_ uint64,
|
||||
_ string,
|
||||
chunkTypes []types.ChunkType,
|
||||
) ([]*types.Chunk, error) {
|
||||
allowed := make(map[types.ChunkType]struct{}, len(chunkTypes))
|
||||
for _, ct := range chunkTypes {
|
||||
allowed[ct] = struct{}{}
|
||||
}
|
||||
var out []*types.Chunk
|
||||
for _, c := range r.chunks {
|
||||
if _, ok := allowed[c.ChunkType]; ok {
|
||||
out = append(out, c)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func newWikiEnqueueTestService(
|
||||
knowledgeID string,
|
||||
pendingRepo *wikiEnqueueFailurePendingRepo,
|
||||
@@ -142,6 +180,16 @@ func newWikiEnqueueTestService(
|
||||
ChunkType: types.ChunkTypeText,
|
||||
},
|
||||
}},
|
||||
chunkRepo: &wikiEnqueueFailureChunkRepo{chunks: []*types.Chunk{
|
||||
{
|
||||
ID: "chunk-1",
|
||||
TenantID: 7,
|
||||
KnowledgeID: knowledgeID,
|
||||
KnowledgeBaseID: "kb-wiki",
|
||||
ChunkType: types.ChunkTypeText,
|
||||
Content: "plain text",
|
||||
},
|
||||
}},
|
||||
taskEnqueuer: queue,
|
||||
pendingRepo: pendingRepo,
|
||||
}, repo
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
)
|
||||
|
||||
type markdownImageTargetSpan struct {
|
||||
ImageStart int
|
||||
TargetStart int
|
||||
TargetEnd int
|
||||
}
|
||||
@@ -30,6 +31,7 @@ func scanMarkdownImageTargets(markdown string) []markdownImageTargetSpan {
|
||||
continue
|
||||
}
|
||||
spans = append(spans, markdownImageTargetSpan{
|
||||
ImageStart: i,
|
||||
TargetStart: targetStart,
|
||||
TargetEnd: targetEnd,
|
||||
})
|
||||
@@ -38,6 +40,32 @@ func scanMarkdownImageTargets(markdown string) []markdownImageTargetSpan {
|
||||
return spans
|
||||
}
|
||||
|
||||
// StripMarkdownImages removes complete  constructs,
|
||||
// including destinations and titles that contain parentheses. Leftover
|
||||
// prose is preserved so callers can tell whether a chunk still has
|
||||
// extractable text after image placeholders are gone.
|
||||
func StripMarkdownImages(markdown string) string {
|
||||
spans := scanMarkdownImageTargets(markdown)
|
||||
if len(spans) == 0 {
|
||||
return markdown
|
||||
}
|
||||
var b strings.Builder
|
||||
b.Grow(len(markdown))
|
||||
last := 0
|
||||
for _, span := range spans {
|
||||
if span.ImageStart < last {
|
||||
continue
|
||||
}
|
||||
b.WriteString(markdown[last:span.ImageStart])
|
||||
last = span.TargetEnd + 1
|
||||
if last > len(markdown) {
|
||||
last = len(markdown)
|
||||
}
|
||||
}
|
||||
b.WriteString(markdown[last:])
|
||||
return b.String()
|
||||
}
|
||||
|
||||
func findMarkdownImageAltEnd(markdown string, start int) int {
|
||||
for i := start; i+1 < len(markdown); i++ {
|
||||
if markdown[i] == ']' && markdown[i+1] == '(' && !isEscaped(markdown, i) {
|
||||
|
||||
@@ -201,3 +201,39 @@ func TestSplitMarkdownImageTarget(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStripMarkdownImages(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
input string
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "simple image",
|
||||
input: "before  after",
|
||||
want: "before after",
|
||||
},
|
||||
{
|
||||
name: "title containing right paren",
|
||||
input: ` 结果")`,
|
||||
want: "",
|
||||
},
|
||||
{
|
||||
name: "prose around image",
|
||||
input: "See figure:  below.",
|
||||
want: "See figure: below.",
|
||||
},
|
||||
{
|
||||
name: "escaped image marker is kept",
|
||||
input: `\ kept`,
|
||||
want: `\ kept`,
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := StripMarkdownImages(tt.input); got != tt.want {
|
||||
t.Fatalf("StripMarkdownImages(%q) = %q, want %q", tt.input, got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user