Files
ragflow/internal/ingestion/component/docx_vision_dispatch.go
Jack 554925b583 Fix(go): align ingestion pipeline with Python (parser/media dispatch + PDF coordinate chain + Chunker) (#17349)
## Summary
Aligns the Go ingestion pipeline with the Python implementation, closing
several behavioral gaps found during the Python→Go migration (tracked in
`docs/migration_python_go_diff.md`). Covers parser/media dispatch
alignment, the PDF coordinate-chain (preview images, outline→title,
chunk coordinate finalization), and the Chunker Token/QA batches below.

Commits are grouped as follows.

### 1. Fix parser params (c524f450e)
Fixes parser/media wiring and several dispatch gaps:
- **docx/pdf vision dispatch**: correct parameter handling and VLM
invocation.
- **markdown vision (diff 2.5)**: also enhance items whose
`doc_type_kwd` is `table`, not only `image` (parser/utils.py:181).
- **media audio (diff 2.11)**: when `output_format` is `json`, carry the
ASR transcription as a JSON item instead of only the `Text` field (the
Invoke switch had no `json` branch and dropped it).
- **email (diff 2.2)**: default `output_format` is `json`
(parser.py:212), not `text`.
- **tokenizer**: handle empty/whitespace-only names; trim before
embedding.
- **extractor**: tag-matching parameter wiring.
- **split**: keyword-split regex now covers CJK/English separators.
- **parser.go**: parser-param plumbing.

### 2. fix parser gap (373537da1)
Image dispatch now mirrors `rag/app/picture.py:chunk()`:
- Always OCR the image (PaddleOCR or local ONNX).
- When OCR text is short, also call VLM (`describe`) and combine `OCR +
VLM` text.
- Emits a **structured JSON item** carrying the image data-URI and
`doc_type_kwd:"image"`, instead of a bare `Text` string. This fixes the
payload being rejected downstream by OneChunker/TokenChunker (JSON=nil).

### 3. PDF coordinate-chain fixes (55367a820, 727f8167c)
Closes three items from the migration tracker in the
chunker/tokenizer/task layer:

- **(Chunker-1.3) `restore_pdf_text_previews`** — `needsCrop` now also
returns true for `text` chunks that carry PDF positions
(`pdfcrop_cgo.go`), so text blocks get a rendered preview image uploaded
to storage via `imageUploadDecorator`/`ChunkImageUploader`, matching
Python `restore_pdf_text_previews` + `image2id`.

- **(Chunker-1.5) PDF outline → title levels** — `title.go` adds
`outlineSimilarity` (rune-bigram Jaccard, mirroring
`common.py:_outline_similarity`), `resolveOutlineLevels` (matches text
lines to outline entries at similarity > 0.8, with a sparse guard
`len(outline)/len(records) <= 0.03`), and `outlineFromInputs` (reads
`file.outline`). Wired into `newLevelContext` in both `group.go` and
`hierarchy.go`; falls back to the title-shape heuristic when no outline
is present.

- **(Tokenizer-(T)1) `finalize_pdf_chunk`** — the coordinate →
`position_int`/`page_num_int`/`top_int` conversion is owned by the task
layer (`processChunkPositions`→`AddPositions`), which runs *after* the
tokenizer and consumes the tokenizer-owned fields. The tokenizer only
preserves the raw `positions`/`_pdf_positions` (no duplicate
conversion), pinned by `TestChunkDocsToMaps_PreservesPDFPositions`.

### 4. Integration test made environment-free
(`internal/ingestion/task/pipeline_real_integration_test.go`)
- Removed the `//go:build integration` tag so the contract tests run
under the default `build.sh --test` (which does not pass `-tags
integration`).
- External dependencies replaced with in-memory substitutes so no
MySQL/MinIO/ES is required:
- MySQL → on-disk sqlite (`glebarez/sqlite`) with the needed tables
auto-migrated.
  - MinIO → `storage.NewMemoryStorage()`.
- Elasticsearch → chunks captured via `WithInsertFunc` instead of
`engine.InsertChunks`/`Search`.
- `requireTokenizerPool` still skips gracefully when the native
tokenizer pool is unavailable; `WithLogCreateFunc(noop)` avoids
depending on the operation-log table.
- Added `taskChunkFieldEqualsStr` to tolerate `kb_id` being a
`[]string`/`[]any` in the raw chunk payload (the search engine flattens
it to a string on read).

### 5. TokenChunker alignment — Batch 1
(`internal/ingestion/component/chunker/token.go`)
Closes four Chunker items from the migration tracker:
- **(Chunker-2.1) sentence delimiter** — the boundary regex now also
breaks on ASCII `!`/`?`. Extracted to a package-level `var
sentenceDelimiter` and used in `mergeByTokenSize`, matching Python's
full delimiter set.
- **(Chunker-2.2) overlap tag leakage** — when a new chunk starts, its
overlap prefix is taken from the previous chunk *after* `removeTag`, in
both the text path (`mergeByTokenSize`) and the JSON path
(`mergeByTokenSizeFromJSON`). Parser tags (`@@…##`) no longer leak into
the overlap region (mirrors `nlp/__init__.py:1181`).
- **(Chunker-2.11) empty-text merge** — merging a non-empty chunk into
an empty previous chunk now assigns the text directly instead of being
skipped (`mergeByTokenSizeFromJSON`), mirroring
`token_chunker.py:236-239`.
- **(Chunker-2.4) overlap token counting** —
`takeFromEnd`/`takeFromStart` now count tokens exactly via `tokenizeStr`
instead of the 4-bytes/token heuristic, fixing over-counting for CJK
text.

### 6. QA Chunker alignment — Batch 2
(`internal/ingestion/component/chunker/qa.go` + `schema`)
Closes three Chunker items from the migration tracker:
- **(Chunker-2.13) default language** — an empty `lang` now defaults to
Chinese prefixes (`问题:`/`回答:`) instead of English, matching `qa.py:299`.
- **(Chunker-2.12) `rmQAPrefix` regex** — the separator is changed to
`[\t:: ]+` (one-or-more), matching `qa.py:241`, so multiple separators
(e.g. `Q:: answer`) are fully stripped.
- **(Chunker-1.8 QA) missing chunk fields** — QA chunks now preserve:
- `top_int` — the source row/record index, threaded through the
tab/csv/markdown extractors (mirrors `qa.py` `beAdoc(..., row_num=i)`);
  - `image` + `doc_type_kwd:"image"`;
  - `_pdf_positions` / `positions` carried from the upstream JSON item.
`schema.ChunkDoc` gains a `TopInt []int` field (serialized as `top_int`,
registered in `UnmarshalJSON`). Note: the Tag/Table/Presentation/One
chunker field gaps under 1.8 remain pending.

## Test plan
- Added/updated unit tests: `pdfcrop_cgo_test.go` (`TestNeedsCrop`,
`TestRestorePDFTextPreview`), `title_test.go`
(`TestResolveOutlineLevels`, `TestResolveOutlineLevels_SparseGuard`,
`TestNewLevelContext_OutlineBranch`, `TestOutlineFromInputs`),
`tokenizer_unit_test.go` (`TestChunkDocsToMaps_PreservesPDFPositions`),
`token_pdfpos_test.go`.
- **Batch 1** — `token_batch1_test.go`:
`TestSentenceDelimiterMatchesBangAndQuestion`,
`TestMergeByTokenSizeFromJSON_OverlapStripsTags`,
`TestMergeByTokenSizeFromJSON_EmptyPrevKeepsChunk`,
`TestTakeFromEndRespectsTokenCount`,
`TestTakeFromStartRespectsTokenCount`.
- **Batch 2** — `qa_batch2_test.go`:
`TestQAChunker_DefaultLangIsChinese`,
`TestRmQAPrefixStripsMultipleSeparators`, `TestQAChunker_SetsTopInt`,
`TestQAChunker_CarriesImageAndPositions`. Existing `qa_test.go`
expectations were updated to the corrected language default / separator
behavior.
- `pipeline_real_integration_test.go`
(`TestPipelineExecutor_Run_RealCanvasDSL_UsesGeneralPipeline`,
`TestPipelineExecutor_Run_RealPDF_ProducesIndexedChunks`,
`TestRunPipeline_RealPipelineOutput_ProducesIndexFields`) now runs
without any external service.
- `bash build.sh --test ./internal/ingestion/...` passes.
- No files deleted.
2026-07-24 21:06:38 +08:00

273 lines
8.2 KiB
Go

//
// Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// DOCX vision figure dispatch: enriches the parse result with
// LLM-generated descriptions of embedded images, mirroring
// Python's enhance_media_sections_with_vision in
// rag/flow/parser/utils.py (invoked from parser.py:_doc's JSON branch).
//
// Unlike the PDF vision path (which replaces dispatchParse entirely),
// DOCX vision is a post-processing step. It mirrors Python exactly:
// vision enrichment happens ONLY on the JSON output path, where each
// item carries a doc_type_kwd and an optional image. The markdown path
// performs no vision enrichment in Python, so it must not here either.
package component
import (
"context"
"fmt"
"os"
"path/filepath"
"strings"
"sync"
"ragflow/internal/entity"
modelModule "ragflow/internal/entity/models"
"ragflow/internal/ingestion/component/schema"
"ragflow/internal/utility"
)
var (
docxVisionPromptBuilder = buildDOCXVisionPrompt
visionChatInvoker = defaultVisionChatInvoker
docxVisionConcurrency uint = 10
)
const (
docxVisionPromptFile = "vision_llm_figure_describe_prompt.md"
docxVisionPromptWithContextFile = "vision_llm_figure_describe_prompt_with_context.md"
)
var (
docxVisionPromptsBase string
docxVisionPromptsOnce sync.Once
docxVisionPromptCache = make(map[string]string)
docxVisionPromptMu sync.RWMutex
)
// maybeDispatchDOCXVision enriches a DOCX parse result with vision-model
// descriptions of embedded images. It mirrors Python's
// enhance_media_sections_with_vision (rag/flow/parser/utils.py:162), which
// runs only in the JSON output branch of parser.py:_doc.
//
// For each JSON item whose doc_type_kwd is "image" or "table" AND that
// carries a non-empty "image" field, the vision model describes the image
// and the description is appended to the item's text (Python:
// item["text"] = f"{text}\n{parsed_text}" if text else parsed_text). Items
// without an image (e.g. DOCX tables) are left untouched, exactly as Python
// skips them via `if item.get("image") is None: continue`.
//
// The markdown output path receives no vision enrichment — Python's DOCX
// markdown branch only concatenates text and never calls the vision model.
func maybeDispatchDOCXVision(
ctx context.Context,
fileType utility.FileType,
dispatched parserDispatchResult,
inputs map[string]any,
setups map[string]schema.ParserSetup,
) (parserDispatchResult, bool, error) {
if fileType != utility.FileTypeDOCX {
return dispatched, false, nil
}
// Python triggers vision enrichment only on the JSON path
// (parser.py:_doc → enhance_media_sections_with_vision).
if dispatched.Err != nil || dispatched.OutputFormat != "json" || len(dispatched.JSON) == 0 {
return dispatched, false, nil
}
tenantID := getStringOr(inputs, "tenant_id", "")
if tenantID == "" {
return dispatched, false, nil
}
// Resolve the tenant's IMAGE2TEXT model.
driver, modelName, apiConfig, _, err := resolveTenantModelByType(tenantID, entity.ModelTypeImage2Text)
if err != nil {
// Model not available — skip vision enhancement silently,
// matching Python's try/except pass behaviour.
return dispatched, false, nil
}
// Collect the indices of JSON items that carry an embeddable image.
type target struct {
idx int
}
var targets []target
for i, item := range dispatched.JSON {
kd, _ := item["doc_type_kwd"].(string)
if kd != "image" && kd != "table" {
continue
}
img, _ := item["image"].(string)
if img == "" {
continue
}
targets = append(targets, target{idx: i})
}
if len(targets) == 0 {
return dispatched, false, nil
}
descriptions := make([]string, len(targets))
var wg sync.WaitGroup
sem := make(chan struct{}, docxVisionConcurrency)
for slot, tg := range targets {
wg.Add(1)
go func(slot int, itemIdx int) {
defer wg.Done()
sem <- struct{}{}
defer func() { <-sem }()
img, _ := dispatched.JSON[itemIdx]["image"].(string)
if img == "" {
return
}
// DOCX JSON items have no surrounding context (unlike the
// former markdown path), so use the bare figure prompt —
// matching Python's VisionFigureParser(context_size=0).
prompt, perr := docxVisionPromptBuilder("", "")
if perr != nil {
return
}
messages := buildVisionMessages(prompt, img)
resp, ierr := visionChatInvoker(ctx, driver, modelName, messages, apiConfig)
if ierr != nil {
return
}
descriptions[slot] = extractDOCXVisionAnswer(resp)
}(slot, tg.idx)
}
wg.Wait()
modified := false
for slot, tg := range targets {
desc := strings.TrimSpace(descriptions[slot])
if desc == "" {
continue
}
existing, _ := dispatched.JSON[tg.idx]["text"].(string)
if existing != "" {
dispatched.JSON[tg.idx]["text"] = existing + "\n" + desc
} else {
dispatched.JSON[tg.idx]["text"] = desc
}
modified = true
}
return dispatched, modified, nil
}
// buildDOCXVisionPrompt loads the figure-describe prompt template
// and, when context text is available, renders it with the
// with-context variant. Mirrors Python:
//
// if context_above or context_below:
// prompt = vision_llm_figure_describe_prompt_with_context(context_above, context_below)
// else:
// prompt = vision_llm_figure_describe_prompt()
func buildDOCXVisionPrompt(contextAbove, contextBelow string) (string, error) {
hasContext := strings.TrimSpace(contextAbove) != "" || strings.TrimSpace(contextBelow) != ""
var templateName string
if hasContext {
templateName = docxVisionPromptWithContextFile
} else {
templateName = docxVisionPromptFile
}
template, err := loadDOCXVisionPromptFile(templateName)
if err != nil {
return "", err
}
if hasContext {
template = strings.ReplaceAll(template, "{{ context_above }}", contextAbove)
template = strings.ReplaceAll(template, "{{ context_below }}", contextBelow)
}
return template, nil
}
func loadDOCXVisionPromptFile(filename string) (string, error) {
docxVisionPromptMu.RLock()
if cached, ok := docxVisionPromptCache[filename]; ok {
docxVisionPromptMu.RUnlock()
return cached, nil
}
docxVisionPromptMu.RUnlock()
baseDir, err := docxVisionPromptsBaseDir()
if err != nil {
return "", err
}
promptPath := filepath.Join(baseDir, "rag", "prompts", filename)
content, err := os.ReadFile(promptPath)
if err != nil {
return "", fmt.Errorf("docx vision prompt %q: %w", filename, err)
}
cached := strings.TrimSpace(string(content))
docxVisionPromptMu.Lock()
docxVisionPromptCache[filename] = cached
docxVisionPromptMu.Unlock()
return cached, nil
}
func docxVisionPromptsBaseDir() (string, error) {
var initErr error
docxVisionPromptsOnce.Do(func() {
root := utility.GetProjectRoot()
if _, statErr := os.Stat(filepath.Join(root, "rag", "prompts")); statErr == nil {
docxVisionPromptsBase = root
return
}
initErr = fmt.Errorf("rag/prompts not found under project root %q", root)
})
if initErr != nil {
return "", initErr
}
return docxVisionPromptsBase, nil
}
func buildVisionMessages(prompt, imageBase64 string) []modelModule.Message {
dataURI := "data:image/png;base64," + imageBase64
return []modelModule.Message{{
Role: "user",
Content: []interface{}{
map[string]any{"type": "text", "text": prompt},
map[string]any{"type": "image_url", "image_url": map[string]any{"url": dataURI}},
},
}}
}
func extractDOCXVisionAnswer(resp *modelModule.ChatResponse) string {
if resp == nil || resp.Answer == nil {
return ""
}
return strings.TrimSpace(*resp.Answer)
}
func defaultVisionChatInvoker(
ctx context.Context,
driver modelModule.ModelDriver,
modelName string,
messages []modelModule.Message,
apiConfig *modelModule.APIConfig,
) (*modelModule.ChatResponse, error) {
vision := true
return driver.ChatWithMessages(ctx, modelName, messages, apiConfig, &modelModule.ChatConfig{Vision: &vision}, nil)
}