Implement Tagger in Extractor component in GO (#17128)

### Summary

Implement Tagger in Extractor component in GO

internal/engine/elasticsearch/chunk.go

@@ -96,6 +96,17 @@ func (e *elasticsearchEngine) CreateChunkStore(ctx context.Context, baseName, da
 				"number_of_shards":   1,
 				"number_of_replicas": 0,
 			},
+			"mappings": map[string]interface{}{
+				"properties": map[string]interface{}{
+					// tag_feas must be an object of numeric values (the
+					// "rank_features" ES type) so tag-based ranking works and
+					// inserts succeed. Without this, dynamic mapping could
+					// infer an incompatible type when a JSON string is sent.
+					"tag_feas": map[string]interface{}{
+						"type": "rank_features",
+					},
+				},
+			},
 		}
 	}
 
@@ -222,10 +233,38 @@ func (e *elasticsearchEngine) InsertChunks(ctx context.Context, chunks []map[str
 		return nil, fmt.Errorf("failed to parse bulk response: %w", err)
 	}
 
-	// Check for errors in bulk response
+	// Check for errors in bulk response. ES reports a 200 at the
+	// top level even when individual bulk items fail, so we must
+	// inspect the items and surface the reasons instead of returning
+	// success while the chunks were silently dropped.
 	if errors, ok := bulkResponse["errors"].(bool); ok && errors {
-		common.Warn("Bulk request had some errors")
-		// Could iterate through items to find specific errors if needed
+		var reasons []string
+		if items, ok := bulkResponse["items"].([]interface{}); ok {
+			for _, it := range items {
+				im, ok := it.(map[string]interface{})
+				if !ok {
+					continue
+				}
+				for _, op := range im {
+					om, ok := op.(map[string]interface{})
+					if !ok {
+						continue
+					}
+					if errObj, ok := om["error"].(map[string]interface{}); ok {
+						if reason, ok := errObj["reason"].(string); ok && reason != "" {
+							reasons = append(reasons, reason)
+						}
+					}
+				}
+			}
+		}
+		if len(reasons) > 5 {
+			reasons = reasons[:5]
+		}
+		common.Error("Elasticsearch bulk request had item errors",
+			fmt.Errorf("index %s: %d failed items; first reasons: %v", baseName, len(reasons), reasons))
+		return nil, fmt.Errorf("elasticsearch bulk request had %d item errors (index %s); first reasons: %v",
+			len(reasons), baseName, reasons)
 	}
 
 	common.Info("ElasticsearchConnection.InsertChunks result", zap.String("index_name", baseName), zap.Int("count", len(chunks)))

internal/ingestion/component/extractor.go

@@ -175,6 +175,12 @@ func NewExtractorComponent(params map[string]any) (runtime.Component, error) {
 		if v, ok := params["auto_questions"]; ok {
 			p.AutoQuestions = mapInt(v)
 		}
+		if v, ok := params["auto_tags"]; ok {
+			p.AutoTags = mapInt(v)
+		}
+		if v, ok := params["tag_file_id"].(string); ok {
+			p.TagFileID = v
+		}
 	}
 	if err := p.Validate(); err != nil {
 		return nil, fmt.Errorf("extractor: param check: %w", err)
@@ -512,6 +518,15 @@ func (c *ExtractorComponent) Invoke(ctx context.Context, inputs map[string]any)
 	}
 
 	if err := runtime.WithTimeout(ctx, extractorTimeout, func(timeoutCtx context.Context) error {
+		// Tag phase: run when auto_tags > 0 and we have chunks.
+		if c.Param.AutoTags > 0 && len(in.chunks) > 0 {
+			tagged, tagErr := c.runAutoTags(timeoutCtx, in)
+			if tagErr != nil {
+				return tagErr
+			}
+			in.chunks = tagged
+		}
+
 		if len(in.chunks) == 0 {
 			ans, callErr := c.call(timeoutCtx, in, "")
 			if callErr != nil {
@@ -521,9 +536,9 @@ func (c *ExtractorComponent) Invoke(ctx context.Context, inputs map[string]any)
 			return nil
 		}
 		for i, ck := range in.chunks {
-			text, _ := ck["text"].(string)
+			text, _ := ck["content_with_weight"].(string)
 			if strings.TrimSpace(text) == "" {
-				text, _ = ck["content_with_weight"].(string)
+				text, _ = ck["text"].(string)
 			}
 
 			if c.Param.AutoKeywords > 0 {
@@ -544,8 +559,6 @@ func (c *ExtractorComponent) Invoke(ctx context.Context, inputs map[string]any)
 				}
 				ck[in.fieldName] = ans
 			}
-
-			in.chunks[i] = ck
 		}
 		return nil
 	}); err != nil {

internal/ingestion/component/extractor_tag.go

@@ -0,0 +1,853 @@
+package component
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/csv"
+	"encoding/json"
+	"fmt"
+	"io"
+	"math/rand/v2"
+	"path/filepath"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/xuri/excelize/v2"
+
+	"github.com/cespare/xxhash/v2"
+	eschema "github.com/cloudwego/eino/schema"
+	"go.uber.org/zap"
+
+	"ragflow/internal/agent/runtime"
+	"ragflow/internal/common"
+	"ragflow/internal/dao"
+	"ragflow/internal/engine/redis"
+	"ragflow/internal/entity"
+	"ragflow/internal/ingestion/component/globals"
+	"ragflow/internal/ingestion/component/schema"
+	"ragflow/internal/tokenizer"
+)
+
+const matchOverlapThreshold = 0.5
+
+const taggerLLMConcurrency = 8
+
+var taggerTimeout = 60 * time.Second
+
+const taggerPromptTmpl = `## Role
+You are a text analyzer.
+
+## Task
+Add tags (labels) to a given piece of text content based on the examples and the entire tag set.
+
+## Steps
+- Review the tag/label set.
+- Review examples which all consist of both text content and assigned tags with relevance score in JSON format.
+- Summarize the text content, and tag it with the top %d most relevant tags from the set of tags/labels and the corresponding relevance score.
+
+## Requirements
+- The tags MUST be from the tag set.
+- The output MUST be in JSON format only, the key is tag and the value is its relevance score.
+- The relevance score must range from 1 to 10.
+- Output keywords ONLY.
+
+# TAG SET
+%s
+
+%s
+# Real Data
+### Text Content
+%s
+
+Output:`
+
+const taggerExampleBlock = `# Examples %d
+### Text Content
+%s
+
+Output:
+%s
+
+`
+
+type indexedTagSource struct {
+	examples  []schema.TagLabel
+	allTags   map[string]float64
+	tagTokens [][]string
+}
+
+const tagSourceCacheMax = 128
+
+type boundedTagCache struct {
+	mu     sync.Mutex
+	cap    int
+	items  map[string]*indexedTagSource
+	recent []string
+}
+
+func newBoundedTagCache(cap int) *boundedTagCache {
+	return &boundedTagCache{
+		cap:   cap,
+		items: make(map[string]*indexedTagSource, cap),
+	}
+}
+
+func (c *boundedTagCache) load(key string) (*indexedTagSource, bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	v, ok := c.items[key]
+	if !ok {
+		return nil, false
+	}
+	c.markRecentLocked(key)
+	return v, true
+}
+
+func (c *boundedTagCache) store(key string, val *indexedTagSource) *indexedTagSource {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if existing, ok := c.items[key]; ok {
+		return existing
+	}
+	c.items[key] = val
+	c.recent = append(c.recent, key)
+	for len(c.items) > c.cap {
+		oldest := c.recent[0]
+		c.recent = c.recent[1:]
+		delete(c.items, oldest)
+	}
+	return val
+}
+
+func (c *boundedTagCache) markRecentLocked(key string) {
+	for i, k := range c.recent {
+		if k == key {
+			c.recent = append(c.recent[:i], c.recent[i+1:]...)
+			c.recent = append(c.recent, k)
+			break
+		}
+	}
+}
+
+var tagSourceFileIndexCache = newBoundedTagCache(tagSourceCacheMax)
+
+func (c *ExtractorComponent) runAutoTags(ctx context.Context, in extractorInputs) ([]map[string]any, error) {
+	indexed, ok := c.resolveTagSource(ctx)
+	if !ok || len(in.chunks) == 0 {
+		common.Info("extractor tags: skipped",
+			zap.Int("chunk_count", len(in.chunks)),
+			zap.Bool("has_tag_source", ok),
+			zap.String("llm_id", in.llmID),
+		)
+		return in.chunks, nil
+	}
+	if len(indexed.examples) == 0 || len(indexed.allTags) == 0 {
+		common.Info("extractor tags: empty indexed source",
+			zap.Int("chunk_count", len(in.chunks)),
+			zap.Int("example_count", len(indexed.examples)),
+			zap.Int("all_tag_count", len(indexed.allTags)),
+			zap.String("llm_id", in.llmID),
+		)
+		return in.chunks, nil
+	}
+
+	topN := c.Param.AutoTags
+
+	var examples []schema.TaggedChunk
+	var docsToTag []map[string]any
+	for _, d := range in.chunks {
+		if ctx.Err() != nil {
+			break
+		}
+		matched := matchAndTagChunk(d, indexed.examples, indexed.tagTokens, indexed.allTags, topN)
+		if matched != nil {
+			examples = append(examples, *matched)
+		} else if in.llmID != "" {
+			docsToTag = append(docsToTag, d)
+		}
+	}
+
+	if len(docsToTag) > 0 && in.llmID != "" {
+		driver, model, apiKey, baseURL := resolveExtractorChatTarget(ctx, in.llmID)
+		if driver != "" && model != "" {
+			inv := getExtractorChatInvoker()
+			sem := make(chan struct{}, taggerLLMConcurrency)
+			var wg sync.WaitGroup
+
+			for i := range docsToTag {
+				wg.Add(1)
+				go func(idx int) {
+					defer wg.Done()
+					select {
+					case sem <- struct{}{}:
+						defer func() { <-sem }()
+					case <-ctx.Done():
+						return
+					}
+					llmTagChunk(ctx, inv, docsToTag[idx], indexed.allTags, examples, in.llmID, driver, model, apiKey, baseURL, topN)
+				}(i)
+			}
+			wg.Wait()
+		}
+	}
+
+	taggedCount := 0
+	for _, chunk := range in.chunks {
+		if chunk[common.TAG_FLD] != nil {
+			taggedCount++
+		}
+	}
+	common.Info("extractor tags: completed",
+		zap.Int("chunk_count", len(in.chunks)),
+		zap.Int("example_count", len(indexed.examples)),
+		zap.Int("all_tag_count", len(indexed.allTags)),
+		zap.Int("phase1_match_count", len(examples)),
+		zap.Int("phase2_candidate_count", len(docsToTag)),
+		zap.Int("tagged_chunk_count", taggedCount),
+		zap.Bool("has_llm", in.llmID != ""),
+	)
+
+	return in.chunks, nil
+}
+
+func (c *ExtractorComponent) resolveTagSource(ctx context.Context) (*indexedTagSource, bool) {
+	if c.Param.TagFileID == "" {
+		return nil, false
+	}
+	return c.loadTagFileIndexed(ctx)
+}
+
+func (c *ExtractorComponent) loadTagFileIndexed(ctx context.Context) (*indexedTagSource, bool) {
+	f, err := dao.NewFileDAO().GetByID(c.Param.TagFileID)
+	if err != nil || f == nil || f.Location == nil || *f.Location == "" {
+		common.Warn(fmt.Sprintf("extractor tags: resolve tag_file_id %q: %v", c.Param.TagFileID, err))
+		return nil, false
+	}
+	cacheKey := tagSourceFileCacheKey(f)
+	if cached, ok := tagSourceFileIndexCache.load(cacheKey); ok {
+		common.Info("extractor tags: reused tag source file index",
+			zap.String("file_id", c.Param.TagFileID),
+			zap.String("bucket", f.ParentID),
+			zap.String("key", *f.Location),
+		)
+		return cached, true
+	}
+	stg := resolveStorage()
+	if stg == nil {
+		common.Warn("extractor tags: no storage backend registered")
+		return nil, false
+	}
+	tenantID := globals.GlobalOrInput(ctx, nil, "tenant_id", "")
+	data, err := stg.Get(f.ParentID, *f.Location, tenantID)
+	if err != nil {
+		common.Warn(fmt.Sprintf("extractor tags: load tag source %q/%q: %v", f.ParentID, *f.Location, err))
+		return nil, false
+	}
+	indexed, ok := buildIndexedTagSourceFromBytes(data, f.Name)
+	if !ok {
+		return nil, false
+	}
+	indexed = tagSourceFileIndexCache.store(cacheKey, indexed)
+	common.Info("extractor tags: loaded tag source file",
+		zap.String("file_id", c.Param.TagFileID),
+		zap.String("bucket", f.ParentID),
+		zap.String("key", *f.Location),
+		zap.Int64("size", f.Size),
+		zap.Int("bytes", len(data)),
+	)
+	return indexed, true
+}
+
+func buildIndexedTagSourceFromBytes(data []byte, filename string) (*indexedTagSource, bool) {
+	examples, err := parseTagSourceByFilename(data, filename)
+	if err != nil {
+		common.Warn(fmt.Sprintf("extractor tags: %v", err))
+		return nil, false
+	}
+	return buildIndexedTagSourceFromExamples(examples), true
+}
+
+// parseTagSourceByFilename mirrors rag/app/tag.py chunk(): the format is chosen
+// by the file extension, and only .xlsx/.xls, .txt and .csv are supported. Any
+// other extension (including no extension) is rejected, matching Python's
+// NotImplementedError for unsupported formats. xlsx is parsed per-sheet (2
+// columns, no header, multiple sheets); .csv uses a quote-aware reader; .txt
+// uses the delimiter-detecting reader.
+func parseTagSourceByFilename(data []byte, filename string) ([]schema.TagLabel, error) {
+	switch strings.ToLower(filepath.Ext(filename)) {
+	case ".xlsx", ".xls":
+		return parseXLSXTagSource(data), nil
+	case ".csv":
+		return parseCSVQuoteAwareReader(bytes.NewReader(data)), nil
+	case ".txt":
+		delimiter := detectCSVDelimiterBytes(data)
+		return parseCSVTagSourceReader(bytes.NewReader(data), delimiter), nil
+	default:
+		return nil, fmt.Errorf("unsupported tag source extension %q: only .xlsx, .txt and .csv are supported", filepath.Ext(filename))
+	}
+}
+
+func buildIndexedTagSourceFromExamples(examples []schema.TagLabel) *indexedTagSource {
+	return &indexedTagSource{
+		examples:  examples,
+		allTags:   buildAllTagsProportions(examples),
+		tagTokens: preTokenizeExamples(examples),
+	}
+}
+
+func tagSourceFileCacheKey(f *entity.File) string {
+	location := ""
+	if f.Location != nil {
+		location = *f.Location
+	}
+	updateTime := int64(0)
+	if f.UpdateTime != nil {
+		updateTime = *f.UpdateTime
+	}
+	return fmt.Sprintf("tag-file:%s:%s:%s:%d:%d", f.ID, f.ParentID, location, f.Size, updateTime)
+}
+
+func parseCSVTagSource(text string) []schema.TagLabel {
+	return parseCSVTagSourceBytes([]byte(text))
+}
+
+func parseCSVTagSourceBytes(data []byte) []schema.TagLabel {
+	return parseCSVTagSourceReader(bytes.NewReader(data), detectCSVDelimiterBytes(data))
+}
+
+// parseCSVTagSourceReader mirrors rag/app/tag.py's txt parsing: lines that do
+// not split into exactly two columns are accumulated as body text and prepended
+// to the next tagged line. The second column holds comma-separated tags.
+func parseCSVTagSourceReader(r io.Reader, delimiter string) []schema.TagLabel {
+	scanner := newTagSourceScannerFromReader(r, scanBufferMax)
+	result := make([]schema.TagLabel, 0)
+	content := ""
+	appendLine := func(s string) {
+		if content == "" {
+			content = s
+		} else {
+			content += "\n" + s
+		}
+	}
+	for scanner.Scan() {
+		line := scanner.Text()
+		arr := strings.Split(line, delimiter)
+		if len(arr) != 2 {
+			appendLine(line)
+			continue
+		}
+		appendLine(arr[0])
+		tags := splitAndTrim(arr[1], ",")
+		result = append(result, schema.TagLabel{Content: content, Tags: tags})
+		content = ""
+	}
+	if scanner.Err() != nil {
+		common.Warn(fmt.Sprintf("extractor tags: parse tag source: %v", scanner.Err()))
+	}
+	return result
+}
+
+// parseCSVQuoteAwareReader mirrors rag/app/tag.py's .csv path: each line is
+// parsed with encoding/csv (so quoted fields containing the delimiter are
+// handled), lines that do not yield exactly two non-empty columns are
+// accumulated as body text and prepended to the next tagged line, and the
+// second column holds comma-separated tags.
+func parseCSVQuoteAwareReader(r io.Reader) []schema.TagLabel {
+	scanner := newTagSourceScannerFromReader(r, scanBufferMax)
+	result := make([]schema.TagLabel, 0)
+	content := ""
+	appendLine := func(s string) {
+		if content == "" {
+			content = s
+		} else {
+			content += "\n" + s
+		}
+	}
+	for scanner.Scan() {
+		line := scanner.Text()
+		rec, err := csv.NewReader(strings.NewReader(line)).Read()
+		if err != nil {
+			appendLine(line)
+			continue
+		}
+		row := stripEmptyFields(rec)
+		if len(row) != 2 {
+			appendLine(line)
+			continue
+		}
+		appendLine(row[0])
+		tags := splitAndTrim(row[1], ",")
+		result = append(result, schema.TagLabel{Content: content, Tags: tags})
+		content = ""
+	}
+	if scanner.Err() != nil {
+		common.Warn(fmt.Sprintf("extractor tags: parse csv tag source: %v", scanner.Err()))
+	}
+	return result
+}
+
+// parseXLSXTagSource mirrors rag/app/tag.py's .xlsx path: every sheet is read
+// with no header, and each row contributes a (content, tags) pair from its
+// first and second non-empty cells. The second cell holds comma-separated tags.
+func parseXLSXTagSource(data []byte) []schema.TagLabel {
+	f, err := excelize.OpenReader(bytes.NewReader(data))
+	if err != nil {
+		common.Warn(fmt.Sprintf("extractor tags: open xlsx tag source: %v", err))
+		return nil
+	}
+	defer f.Close()
+
+	result := make([]schema.TagLabel, 0)
+	for _, sheet := range f.GetSheetList() {
+		rows, err := f.GetRows(sheet)
+		if err != nil {
+			common.Warn(fmt.Sprintf("extractor tags: read xlsx sheet %q: %v", sheet, err))
+			continue
+		}
+		for _, row := range rows {
+			var cells []string
+			for _, c := range row {
+				if c = strings.TrimSpace(c); c != "" {
+					cells = append(cells, c)
+				}
+			}
+			if len(cells) < 2 {
+				continue
+			}
+			tags := splitAndTrim(cells[1], ",")
+			result = append(result, schema.TagLabel{Content: cells[0], Tags: tags})
+		}
+	}
+	return result
+}
+
+func stripEmptyFields(fields []string) []string {
+	out := make([]string, 0, len(fields))
+	for _, f := range fields {
+		if f = strings.TrimSpace(f); f != "" {
+			out = append(out, f)
+		}
+	}
+	return out
+}
+
+const scanBufferMax = 1 << 20
+
+func detectCSVDelimiterBytes(data []byte) string {
+	comma, tab := 0, 0
+	scanner := newTagSourceScanner(bytes.NewReader(data), len(data))
+	for scanner.Scan() {
+		line := scanner.Text()
+		if len(strings.Split(line, ",")) == 2 {
+			comma++
+		}
+		if len(strings.Split(line, "	")) == 2 {
+			tab++
+		}
+	}
+	if scanner.Err() != nil {
+		common.Warn(fmt.Sprintf("extractor tags: delimiter scan: %v", scanner.Err()))
+	}
+	if tab >= comma {
+		return "	"
+	}
+	return ","
+}
+
+func newTagSourceScanner(r io.Reader, dataLen int) *bufio.Scanner {
+	maxToken := dataLen + 1
+	if maxToken < 64*1024 {
+		maxToken = 64 * 1024
+	}
+	if maxToken > scanBufferMax {
+		maxToken = scanBufferMax
+	}
+	return newTagSourceScannerFromReader(r, maxToken)
+}
+
+func newTagSourceScannerFromReader(r io.Reader, maxTokens ...int) *bufio.Scanner {
+	maxToken := 64 * 1024
+	if len(maxTokens) > 0 && maxTokens[0] > maxToken {
+		maxToken = maxTokens[0]
+	}
+	initBuf := min(64*1024, maxToken)
+	scanner := bufio.NewScanner(r)
+	scanner.Buffer(make([]byte, 0, initBuf), maxToken)
+	return scanner
+}
+
+func splitAndTrim(s, sep string) []string {
+	parts := strings.Split(s, sep)
+	out := make([]string, 0, len(parts))
+	for _, p := range parts {
+		if p = strings.TrimSpace(p); p != "" {
+			out = append(out, p)
+		}
+	}
+	return out
+}
+
+func buildAllTagsProportions(tagSource []schema.TagLabel) map[string]float64 {
+	tagCount := make(map[string]int)
+	total := 0
+	for _, ex := range tagSource {
+		for _, t := range ex.Tags {
+			t = strings.TrimSpace(t)
+			if t == "" {
+				continue
+			}
+			tagCount[t]++
+			total++
+		}
+	}
+	S := 1000.0
+	proportions := make(map[string]float64, len(tagCount))
+	for tag, c := range tagCount {
+		proportions[tag] = float64(c+1) / (float64(total) + S)
+	}
+	return proportions
+}
+
+func preTokenizeExamples(tagSource []schema.TagLabel) [][]string {
+	out := make([][]string, len(tagSource))
+	for i := range tagSource {
+		tokens, err := tokenizer.Tokenize(tagSource[i].Content)
+		if err == nil && tokens != "" {
+			out[i] = strings.Fields(tokens)
+		}
+	}
+	return out
+}
+
+func matchAndTagChunk(
+	chunk map[string]any,
+	tagSource []schema.TagLabel,
+	tagTokens [][]string,
+	allTags map[string]float64,
+	topN int,
+) *schema.TaggedChunk {
+	text := getChunkText(chunk)
+	if text == "" {
+		return nil
+	}
+	tokens, err := tokenizer.Tokenize(text)
+	if err != nil || tokens == "" {
+		return nil
+	}
+	chunkWords := strings.Fields(tokens)
+
+	var best *schema.TagLabel
+	var bestScore float64
+	for i := range tagSource {
+		exWords := tagTokens[i]
+		if exWords == nil {
+			continue
+		}
+		score := jaccardOverlap(chunkWords, exWords)
+		if score > bestScore && score >= matchOverlapThreshold {
+			bestScore = score
+			ex := tagSource[i]
+			best = &ex
+		}
+	}
+	if best == nil {
+		return nil
+	}
+
+	S := 1000.0
+	cnt := float64(len(best.Tags))
+	type tagScore struct {
+		name  string
+		score int
+	}
+	var scored []tagScore
+	for _, t := range best.Tags {
+		t = strings.TrimSpace(t)
+		if t == "" {
+			continue
+		}
+		bg := allTags[t]
+		if bg <= 0 {
+			bg = 0.0001
+		}
+		s := roundInt(0.1 * 2.0 / (cnt + S) / max(1e-6, bg))
+		if s > 0 {
+			scored = append(scored, tagScore{name: t, score: s})
+		}
+	}
+
+	if len(scored) == 0 {
+		return nil
+	}
+
+	sort.Slice(scored, func(i, j int) bool { return scored[i].score > scored[j].score })
+	if len(scored) > topN {
+		scored = scored[:topN]
+	}
+
+	tagWeights := make(map[string]int, len(scored))
+	for _, ts := range scored {
+		tagWeights[strings.ReplaceAll(ts.name, ".", "_")] = ts.score
+	}
+	// Store as an object, not a JSON string: the ES index maps tag_feas as
+	// type "rank_features", which requires an object of numeric values. A
+	// string would make every bulk insert fail and drop the whole chunk.
+	chunk[common.TAG_FLD] = tagWeights
+
+	return &schema.TaggedChunk{
+		Content:    text,
+		Tags:       best.Tags,
+		TagWeights: tagWeights,
+	}
+}
+
+func jaccardOverlap(a, b []string) float64 {
+	if len(a) == 0 || len(b) == 0 {
+		return 0
+	}
+	setA := make(map[string]struct{}, len(a))
+	for _, w := range a {
+		setA[w] = struct{}{}
+	}
+	setB := make(map[string]struct{}, len(b))
+	for _, w := range b {
+		setB[w] = struct{}{}
+	}
+	intersection := 0
+	for w := range setA {
+		if _, ok := setB[w]; ok {
+			intersection++
+		}
+	}
+	union := len(setA) + len(setB) - intersection
+	if union == 0 {
+		return 0
+	}
+	return float64(intersection) / float64(union)
+}
+
+func getChunkText(chunk map[string]any) string {
+	if v, ok := chunk["content_with_weight"].(string); ok && v != "" {
+		return v
+	}
+	if v, ok := chunk["text"].(string); ok && v != "" {
+		return v
+	}
+	return ""
+}
+
+func roundInt(f float64) int {
+	if f < 0 {
+		return int(f - 0.5)
+	}
+	return int(f + 0.5)
+}
+
+func llmTagChunk(
+	ctx context.Context,
+	inv extractorChatInvoker,
+	chunk map[string]any,
+	allTags map[string]float64,
+	examples []schema.TaggedChunk,
+	llmID, driver, model, apiKey, baseURL string,
+	topN int,
+) {
+	text := getChunkText(chunk)
+	if text == "" {
+		return
+	}
+
+	if cached := getTaggerLLMCache(llmID, text, allTags, topN); cached != nil {
+		chunk[common.TAG_FLD] = cached
+		return
+	}
+
+	var picked []schema.TaggedChunk
+	if len(examples) > 2 {
+		picked = randomChoices(examples, 2)
+	} else if len(examples) > 0 {
+		picked = examples
+	}
+	if len(picked) == 0 {
+		picked = []schema.TaggedChunk{{Content: "This is an example", TagWeights: map[string]int{"example": 1}}}
+	}
+
+	tagNames := sortedTagNames(allTags)
+	tagSetStr := strings.Join(tagNames, ", ")
+	prompt := buildTaggerPrompt(topN, tagSetStr, picked, text)
+
+	msgs := []eschema.Message{
+		{Role: eschema.System, Content: prompt},
+		{Role: eschema.User, Content: "Output:"},
+	}
+
+	temperature := 0.5
+	var result map[string]int
+	timeoutErr := runtime.WithTimeout(ctx, taggerTimeout, func(timeoutCtx context.Context) error {
+		resp, err := inv.Chat(timeoutCtx, extractorChatRequest{
+			Driver:      driver,
+			ModelName:   model,
+			APIKey:      apiKey,
+			BaseURL:     baseURL,
+			Messages:    msgs,
+			Temperature: &temperature,
+		})
+		if err != nil {
+			common.Error("extractor tags: LLM call failed", err)
+			return nil
+		}
+		result = parseTaggerResponse(resp.Content, topN)
+		return nil
+	})
+	if timeoutErr != nil {
+		common.Error("extractor tags: LLM timeout", timeoutErr)
+
+	}
+
+	if len(result) > 0 {
+		chunk[common.TAG_FLD] = result
+		setTaggerLLMCache(llmID, text, allTags, topN, result)
+	}
+}
+
+func buildTaggerPrompt(topN int, tagSetStr string, examples []schema.TaggedChunk, text string) string {
+	var examplesBlock strings.Builder
+	for i, ex := range examples {
+		tagsJSON, _ := json.Marshal(ex.TagWeights)
+		examplesBlock.WriteString(fmt.Sprintf(taggerExampleBlock, i, ex.Content, string(tagsJSON)))
+	}
+	return fmt.Sprintf(taggerPromptTmpl, topN, tagSetStr, examplesBlock.String(), text)
+}
+
+func parseTaggerResponse(raw string, topN int) map[string]int {
+	raw = strings.TrimSpace(raw)
+	if idx := strings.Index(raw, "</think>"); idx >= 0 {
+		raw = strings.TrimSpace(raw[idx+len("</think>"):])
+	}
+	if strings.Contains(raw, "**ERROR**") {
+		common.Warn("extractor tags: LLM returned **ERROR**")
+		return nil
+	}
+
+	obj, ok := tryParseJSONObject(raw)
+	if !ok {
+		obj = jsonRepairExtract(raw)
+		if obj == nil {
+			return nil
+		}
+	}
+
+	result := make(map[string]int, len(obj))
+	for k, v := range obj {
+		score := 0
+		switch n := v.(type) {
+		case float64:
+			score = int(n)
+		case int:
+			score = n
+		}
+		if score > 0 {
+			result[k] = score
+		}
+	}
+
+	if len(result) > topN {
+		type kv struct {
+			k string
+			v int
+		}
+		sorted := make([]kv, 0, len(result))
+		for k, v := range result {
+			sorted = append(sorted, kv{k, v})
+		}
+		sort.Slice(sorted, func(i, j int) bool { return sorted[i].v > sorted[j].v })
+		result = make(map[string]int, topN)
+		for i := 0; i < topN && i < len(sorted); i++ {
+			result[sorted[i].k] = sorted[i].v
+		}
+	}
+	return result
+}
+
+func jsonRepairExtract(raw string) map[string]any {
+	start := strings.Index(raw, "{")
+	end := strings.LastIndex(raw, "}")
+	if start < 0 || end <= start {
+		return nil
+	}
+	candidate := raw[start : end+1]
+	obj, ok := tryParseJSONObject(candidate)
+	if !ok {
+		return nil
+	}
+	return obj
+}
+
+func taggerCacheKey(llmID, text string, allTags map[string]float64, topN int) string {
+	hasher := xxhash.New()
+	hasher.Write([]byte(llmID))
+	hasher.Write([]byte("\x00"))
+	hasher.Write([]byte(text))
+	hasher.Write([]byte("\x00"))
+	tagNames := sortedTagNames(allTags)
+	hasher.Write([]byte(strings.Join(tagNames, ",")))
+	hasher.Write([]byte("\x00"))
+	hasher.Write([]byte(fmt.Sprintf("%d", topN)))
+	return fmt.Sprintf("tagger:%x", hasher.Sum64())
+}
+
+func getTaggerLLMCache(llmID, text string, allTags map[string]float64, topN int) map[string]int {
+	client := redis.Get()
+	if client == nil {
+		return nil
+	}
+	key := taggerCacheKey(llmID, text, allTags, topN)
+	data, err := client.Get(key)
+	if err != nil || data == "" {
+		return nil
+	}
+	var result map[string]int
+	if err := json.Unmarshal([]byte(data), &result); err != nil {
+		return nil
+	}
+	return result
+}
+
+func setTaggerLLMCache(llmID, text string, allTags map[string]float64, topN int, result map[string]int) {
+	if result == nil {
+		return
+	}
+	client := redis.Get()
+	if client == nil {
+		return
+	}
+	key := taggerCacheKey(llmID, text, allTags, topN)
+	data, err := json.Marshal(result)
+	if err != nil {
+		return
+	}
+	client.Set(key, string(data), 24*time.Hour)
+}
+
+func sortedTagNames(allTags map[string]float64) []string {
+	out := make([]string, 0, len(allTags))
+	for t := range allTags {
+		out = append(out, t)
+	}
+	sort.Strings(out)
+	return out
+}
+
+func randomChoices(slice []schema.TaggedChunk, k int) []schema.TaggedChunk {
+	if len(slice) == 0 {
+		return nil
+	}
+	out := make([]schema.TaggedChunk, k)
+	for i := range k {
+		out[i] = slice[rand.IntN(len(slice))]
+	}
+	return out
+}

internal/ingestion/component/extractor_tag_test.go

@@ -0,0 +1,403 @@
+package component
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"github.com/xuri/excelize/v2"
+
+	"ragflow/internal/agent/runtime"
+	"ragflow/internal/common"
+	"ragflow/internal/ingestion/component/schema"
+)
+
+type stubExtractorTagChat struct {
+	responses map[string]string
+}
+
+func (s *stubExtractorTagChat) Chat(_ context.Context, req extractorChatRequest) (*extractorChatResponse, error) {
+	if s.responses != nil {
+		if msg, ok := s.responses["__static__"]; ok {
+			return &extractorChatResponse{Content: msg}, nil
+		}
+	}
+	return &extractorChatResponse{Content: `{"RAG": 8, "vector database": 6}`}, nil
+}
+
+func pushExtractorTagChatStub(t *testing.T, responses map[string]string) {
+	t.Helper()
+	stub := &stubExtractorTagChat{responses: responses}
+	SetExtractorChatInvoker(stub)
+	t.Cleanup(func() { SetExtractorChatInvoker(nil) })
+}
+
+func pushExtractorTagTargetResolverStub(t *testing.T) {
+	t.Helper()
+	SetExtractorChatTargetResolverOverride(func(llmID string) (string, string, string, string, bool) {
+		return "test_driver", "test_model", "test_key", "", true
+	})
+	t.Cleanup(func() { SetExtractorChatTargetResolverOverride(nil) })
+}
+
+func TestExtractorTags_NoTagFileID(t *testing.T) {
+	pushExtractorTagChatStub(t, nil)
+	comp, _ := NewExtractorComponent(map[string]any{"auto_tags": 3})
+	out, err := comp.Invoke(context.Background(), map[string]any{
+		"chunks": []map[string]any{
+			{"content_with_weight": "test"},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	chunks, _ := out["chunks"].([]map[string]any)
+	if len(chunks) != 1 {
+		t.Fatalf("expected 1 chunk, got %d", len(chunks))
+	}
+	if _, ok := chunks[0][common.TAG_FLD]; ok {
+		t.Fatal("tag_feas should not be set when tag_file_id is absent")
+	}
+}
+
+func TestExtractorTags_NoLLMID(t *testing.T) {
+	pushExtractorTagChatStub(t, nil)
+	comp, _ := NewExtractorComponent(map[string]any{"auto_tags": 3})
+	out, err := comp.Invoke(context.Background(), map[string]any{
+		"chunks": []map[string]any{
+			{"content_with_weight": "some unrelated text"},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	chunks, _ := out["chunks"].([]map[string]any)
+	if _, ok := chunks[0][common.TAG_FLD]; ok {
+		t.Fatal("tag_feas should not be set without tag_file_id")
+	}
+}
+
+func TestExtractorTags_WithKeywords(t *testing.T) {
+	pushExtractorTagChatStub(t, nil)
+	pushExtractorTagTargetResolverStub(t)
+
+	comp, _ := NewExtractorComponent(map[string]any{
+		"llm_id":        "test@test",
+		"auto_tags":     3,
+		"auto_keywords": 3,
+	})
+	out, err := comp.Invoke(context.Background(), map[string]any{
+		"chunks": []map[string]any{
+			{"content_with_weight": "some unrelated textxyz"},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	chunks, _ := out["chunks"].([]map[string]any)
+	if chunks[0][common.TAG_FLD] != nil {
+		t.Fatal("tag_feas should not be set without tag_file_id")
+	}
+}
+
+func TestExtractorTags_ComponentRegistration(t *testing.T) {
+	factory, cat, md, ok := runtime.DefaultRegistry.Lookup(componentNameExtractor)
+	if !ok {
+		t.Fatal("Extractor not registered in runtime.DefaultRegistry")
+	}
+	if cat != runtime.CategoryIngestion {
+		t.Fatalf("expected ingestion category, got %v", cat)
+	}
+	if factory == nil {
+		t.Fatal("factory is nil")
+	}
+	data, _ := json.Marshal(md)
+	t.Logf("Extractor metadata: %s", data)
+}
+
+func TestParseTaggerResponse(t *testing.T) {
+	raw := `</think>{"RAG": 8, "LLM": 3, "open": 1}`
+	result := parseTaggerResponse(raw, 2)
+	if len(result) != 2 {
+		t.Fatalf("expected 2 (top-2), got %d: %v", len(result), result)
+	}
+	if result["RAG"] != 8 {
+		t.Fatalf("expected RAG=8, got %d", result["RAG"])
+	}
+	if _, ok := result["open"]; ok {
+		t.Fatal("open should be trimmed (not top-2)")
+	}
+}
+
+func TestParseTaggerResponse_JSONRepair(t *testing.T) {
+	raw := `some prefix garbage {"RAG": 8, "LLM": 5} trailing stuff`
+	result := parseTaggerResponse(raw, 2)
+	if result["RAG"] != 8 || result["LLM"] != 5 {
+		t.Fatalf("json-repair fallback failed, got %v", result)
+	}
+}
+
+func TestParseCSVTagSource_Comma(t *testing.T) {
+	// A comma-delimited file maps to exactly two columns, so each line can
+	// carry only a single tag (tags are comma-separated within the 2nd col).
+	text := "RAGFlow tutorial,RAG\nsome text,LLM"
+	result := parseCSVTagSource(text)
+	if len(result) != 2 {
+		t.Fatalf("expected 2 examples, got %d", len(result))
+	}
+	if result[0].Content != "RAGFlow tutorial" {
+		t.Errorf("content[0] = %q", result[0].Content)
+	}
+	if len(result[0].Tags) != 1 || result[0].Tags[0] != "RAG" {
+		t.Errorf("tags[0] = %v", result[0].Tags)
+	}
+}
+
+func TestParseCSVTagSourceBytes_Comma(t *testing.T) {
+	data := []byte("RAGFlow tutorial,RAG\nsome text,LLM")
+	result := parseCSVTagSourceBytes(data)
+	if len(result) != 2 {
+		t.Fatalf("expected 2 examples, got %d", len(result))
+	}
+	if result[1].Tags[0] != "LLM" {
+		t.Fatalf("unexpected tags: %v", result[1].Tags)
+	}
+}
+
+func TestParseCSVTagSourceReader_Comma(t *testing.T) {
+	r := bytes.NewBufferString("RAGFlow tutorial,RAG\nsome text,LLM")
+	result := parseCSVTagSourceReader(r, ",")
+	if len(result) != 2 {
+		t.Fatalf("expected 2 examples, got %d", len(result))
+	}
+	if result[0].Content != "RAGFlow tutorial" {
+		t.Fatalf("unexpected content: %q", result[0].Content)
+	}
+	if len(result[1].Tags) != 1 || result[1].Tags[0] != "LLM" {
+		t.Fatalf("unexpected tags: %v", result[1].Tags)
+	}
+}
+
+func TestParseCSVTagSource_Tab(t *testing.T) {
+	text := "RAGFlow tutorial\tRAG,tutorial\nvector database guide\tvector database,config"
+	result := parseCSVTagSource(text)
+	if len(result) != 2 {
+		t.Fatalf("expected 2 examples, got %d", len(result))
+	}
+	if result[0].Content != "RAGFlow tutorial" {
+		t.Errorf("content[0] = %q", result[0].Content)
+	}
+	if len(result[1].Tags) != 2 || result[1].Tags[1] != "config" {
+		t.Errorf("tags[1] = %v", result[1].Tags)
+	}
+}
+
+func TestParseCSVTagSource_Accumulation(t *testing.T) {
+	// Mirrors rag/app/tag.py txt semantics: lines that do not split into two
+	// columns are accumulated as body text and prepended to the next tagged
+	// line's content.
+	text := "intro paragraph\nmore body\nRAGFlow tutorial\tRAG,tutorial"
+	result := parseCSVTagSource(text)
+	if len(result) != 1 {
+		t.Fatalf("expected 1 example, got %d", len(result))
+	}
+	want := "intro paragraph\nmore body\nRAGFlow tutorial"
+	if result[0].Content != want {
+		t.Errorf("content = %q, want %q", result[0].Content, want)
+	}
+	if len(result[0].Tags) != 2 || result[0].Tags[0] != "RAG" || result[0].Tags[1] != "tutorial" {
+		t.Errorf("tags = %v", result[0].Tags)
+	}
+}
+
+func stubXLSXBytes(t *testing.T) []byte {
+	t.Helper()
+	f := excelize.NewFile()
+	defer f.Close()
+	must := func(err error) {
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	must(f.SetCellValue("Sheet1", "A1", "RAGFlow guide"))
+	must(f.SetCellValue("Sheet1", "B1", "RAG, tutorial"))
+	must(f.SetCellValue("Sheet1", "A2", "vector db"))
+	must(f.SetCellValue("Sheet1", "B2", "vector database, config"))
+	_, err := f.NewSheet("Sheet2")
+	must(err)
+	must(f.SetCellValue("Sheet2", "A1", "LLM intro"))
+	must(f.SetCellValue("Sheet2", "B1", "LLM"))
+	buf, err := f.WriteToBuffer()
+	if err != nil {
+		t.Fatal(err)
+	}
+	return buf.Bytes()
+}
+
+func TestParseCSVQuoteAwareReader(t *testing.T) {
+	// A quoted content field containing a comma must not be split into extra
+	// columns: "RAGFlow, the guide",RAG is two fields, not three.
+	text := "\"RAGFlow, the guide\",RAG\nplain line,LLM"
+	result := parseCSVQuoteAwareReader(strings.NewReader(text))
+	if len(result) != 2 {
+		t.Fatalf("expected 2 examples, got %d", len(result))
+	}
+	// First line: quoted content keeps the comma -> content "RAGFlow, the guide", tags [RAG]
+	if result[0].Content != "RAGFlow, the guide" {
+		t.Errorf("content[0] = %q", result[0].Content)
+	}
+	if len(result[0].Tags) != 1 || result[0].Tags[0] != "RAG" {
+		t.Errorf("tags[0] = %v", result[0].Tags)
+	}
+	// Second line: plain comma split -> content "plain line", tags [LLM]
+	if result[1].Content != "plain line" {
+		t.Errorf("content[1] = %q", result[1].Content)
+	}
+	if len(result[1].Tags) != 1 || result[1].Tags[0] != "LLM" {
+		t.Errorf("tags[1] = %v", result[1].Tags)
+	}
+}
+
+func TestParseXLSXTagSource(t *testing.T) {
+	data := stubXLSXBytes(t)
+	result := parseXLSXTagSource(data)
+	if len(result) != 3 {
+		t.Fatalf("expected 3 examples (multi-sheet), got %d", len(result))
+	}
+	// Each row's first cell is content, second is comma-separated tags.
+	if result[0].Content != "RAGFlow guide" {
+		t.Errorf("content[0] = %q", result[0].Content)
+	}
+	if len(result[0].Tags) != 2 || result[0].Tags[1] != "tutorial" {
+		t.Errorf("tags[0] = %v", result[0].Tags)
+	}
+	if result[2].Content != "LLM intro" {
+		t.Errorf("content[2] = %q", result[2].Content)
+	}
+	if len(result[2].Tags) != 1 || result[2].Tags[0] != "LLM" {
+		t.Errorf("tags[2] = %v", result[2].Tags)
+	}
+}
+
+func TestParseCSVTagSource_EmptyTags(t *testing.T) {
+	// Mirrors rag/app/tag.py: a row with exactly two columns is appended even
+	// when the tag column is empty (beAdoc always appends for a 2-column row).
+	text := "content one\tRAG,tutorial\nsolo line\t"
+	result := parseCSVTagSource(text)
+	if len(result) != 2 {
+		t.Fatalf("expected 2 examples, got %d", len(result))
+	}
+	if len(result[0].Tags) != 2 {
+		t.Errorf("tags[0] = %v", result[0].Tags)
+	}
+	// Second row has an empty tag column -> still an example, with no tags.
+	if len(result[1].Tags) != 0 {
+		t.Errorf("tags[1] = %v, want empty (matches Python)", result[1].Tags)
+	}
+	if result[1].Content != "solo line" {
+		t.Errorf("content[1] = %q", result[1].Content)
+	}
+}
+
+func TestParseTagSourceByFilename(t *testing.T) {
+	// .xlsx -> multi-sheet, 2 columns per row.
+	if got, err := parseTagSourceByFilename(stubXLSXBytes(t), "tags.xlsx"); err != nil || len(got) != 3 {
+		t.Errorf("xlsx: expected 3 examples, got %d (err=%v)", len(got), err)
+	}
+	// .csv -> quote-aware comma parsing.
+	csvData := []byte("\"a, b\",RAG\nsimple,LLM")
+	if got, err := parseTagSourceByFilename(csvData, "tags.csv"); err != nil || len(got) != 2 {
+		t.Errorf("csv: expected 2 examples, got %d (err=%v)", len(got), err)
+	}
+	if got, err := parseTagSourceByFilename(csvData, "tags.CSV"); err != nil || len(got) != 2 {
+		t.Errorf("csv (uppercase ext): expected 2 examples, got %d (err=%v)", len(got), err)
+	}
+	// .txt -> delimiter-detecting txt reader.
+	txtData := []byte("content one\tRAG,tutorial")
+	if got, err := parseTagSourceByFilename(txtData, "tags.txt"); err != nil || len(got) != 1 {
+		t.Errorf("txt: expected 1 example, got %d (err=%v)", len(got), err)
+	}
+	// Unsupported / no extension mirrors Python's NotImplementedError: the tag
+	// source is not parsed and an error is returned.
+	if _, err := parseTagSourceByFilename(txtData, "noextension"); err == nil {
+		t.Error("no extension: expected an error (unsupported extension)")
+	}
+	if _, err := parseTagSourceByFilename(txtData, "tags.json"); err == nil {
+		t.Error("unsupported extension: expected an error")
+	}
+}
+
+func TestJaccardOverlap(t *testing.T) {
+	tests := []struct {
+		name     string
+		a, b     []string
+		expected float64
+	}{
+		{"identical", []string{"a", "b"}, []string{"a", "b"}, 1.0},
+		{"half overlap", []string{"a", "b"}, []string{"b", "c"}, 1.0 / 3.0},
+		{"no overlap", []string{"a", "b"}, []string{"c", "d"}, 0.0},
+		{"empty a", []string{}, []string{"a"}, 0.0},
+		{"empty b", []string{"a"}, []string{}, 0.0},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := jaccardOverlap(tt.a, tt.b)
+			if got != tt.expected {
+				t.Fatalf("jaccard(%v, %v) = %.4f, want %.4f", tt.a, tt.b, got, tt.expected)
+			}
+		})
+	}
+}
+
+func TestBuildAllTagsProportions(t *testing.T) {
+	ts := []schema.TagLabel{
+		{Content: "a", Tags: []string{"RAG", "LLM"}},
+		{Content: "b", Tags: []string{"RAG", "open"}},
+	}
+	result := buildAllTagsProportions(ts)
+	if len(result) != 3 {
+		t.Fatalf("expected 3 tags, got %d", len(result))
+	}
+	if result["RAG"] <= result["LLM"] {
+		t.Fatalf("expected RAG proportion > LLM, got RAG=%.6f LLM=%.6f", result["RAG"], result["LLM"])
+	}
+	if result["LLM"] <= 0 {
+		t.Fatal("expected LLM proportion > 0")
+	}
+}
+
+func TestJsonRepairExtract(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    string
+		hasField string
+		want     int
+	}{
+		{"valid json", `{"RAG": 8}`, "RAG", 8},
+		{"prefix garbage", `xxx {"RAG": 8, "LLM": 5} yyy`, "LLM", 5},
+		{"nested braces", `{"outer": "x", "inner": {"k": 1}}`, "outer", 0},
+		{"no json", `no json here`, "", 0},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := jsonRepairExtract(tt.input)
+			if tt.hasField == "" {
+				if result != nil {
+					t.Fatalf("expected nil, got %v", result)
+				}
+				return
+			}
+			if result == nil {
+				t.Fatal("expected non-nil result")
+			}
+			switch v := result[tt.hasField].(type) {
+			case float64:
+				if int(v) != tt.want && tt.want != 0 {
+					t.Fatalf("expected %s=%d, got %d", tt.hasField, tt.want, int(v))
+				}
+			}
+		})
+	}
+}

internal/ingestion/component/schema/extractor.go

@@ -16,6 +16,21 @@
 
 package schema
 
+// TagLabel is a single labeled record from the tag definition file:
+// a piece of content and the tags associated with it.
+type TagLabel struct {
+	Content string   `json:"content"`
+	Tags    []string `json:"tags"`
+}
+
+// TaggedChunk is the result of tagging a chunk: the chunk content, the
+// matched tags, and their computed relevance weights.
+type TaggedChunk struct {
+	Content    string         `json:"content"`
+	Tags       []string       `json:"tags"`
+	TagWeights map[string]int `json:"tag_weights,omitempty"`
+}
+
 // ExtractorFromUpstream is the upstream payload consumed by the
 // Extractor component.
 //
@@ -88,6 +103,17 @@ type ExtractorParam struct {
 	// AutoQuestions enables automatic question generation with a fixed
 	// prompt. The value determines the top-N count.
 	AutoQuestions int `json:"auto_questions,omitempty"`
+
+	// AutoTags enables tag assignment on chunks. When > 0, the
+	// component runs a two-phase tagger: Phase 1 uses Jaccard
+	// matching against tag source examples; Phase 2 uses the LLM
+	// for unmatched chunks. The value determines the top-N tags.
+	AutoTags int `json:"auto_tags,omitempty"`
+
+	// TagFileID references a tag-definition file stored in object
+	// storage. Used only when AutoTags > 0 and no inline tag
+	// source text is wired in.
+	TagFileID string `json:"tag_file_id"`
 }
 
 // Defaults returns the default ExtractorParam.
@@ -99,6 +125,8 @@ func (ExtractorParam) Defaults() ExtractorParam {
 		Prompt:        "",
 		AutoKeywords:  0,
 		AutoQuestions: 0,
+		AutoTags:      0,
+		TagFileID:     "",
 	}
 }
 

internal/ingestion/pipeline/template/ingestion_pipeline_audio.json

@@ -134,7 +134,8 @@
                         },
                         "auto_keywords": 0,
                         "auto_questions": 0,
-                        "llm_id": ""
+                        "llm_id": "",
+                        "auto_tags": 0
                     }
                 }
             }

internal/ingestion/pipeline/template/ingestion_pipeline_book.json

@@ -195,7 +195,8 @@
                         },
                         "auto_keywords": 0,
                         "auto_questions": 0,
-                        "llm_id": ""
+                        "llm_id": "",
+                        "auto_tags": 0
                     }
                 }
             }

internal/ingestion/pipeline/template/ingestion_pipeline_email.json

@@ -139,7 +139,8 @@
                         },
                         "auto_keywords": 0,
                         "auto_questions": 0,
-                        "llm_id": ""
+                        "llm_id": "",
+                        "auto_tags": 0
                     }
                 }
             }

internal/ingestion/pipeline/template/ingestion_pipeline_general.json

@@ -217,7 +217,8 @@
                         },
                         "auto_keywords": 0,
                         "auto_questions": 0,
-                        "llm_id": ""
+                        "llm_id": "",
+                        "auto_tags": 0
                     }
                 }
             }

internal/ingestion/pipeline/template/ingestion_pipeline_laws.json

@@ -207,7 +207,8 @@
                         },
                         "auto_keywords": 0,
                         "auto_questions": 0,
-                        "llm_id": ""
+                        "llm_id": "",
+                        "auto_tags": 0
                     }
                 }
             }

internal/ingestion/pipeline/template/ingestion_pipeline_manual.json

@@ -175,7 +175,8 @@
                         },
                         "auto_keywords": 0,
                         "auto_questions": 0,
-                        "llm_id": ""
+                        "llm_id": "",
+                        "auto_tags": 0
                     }
                 }
             }

internal/ingestion/pipeline/template/ingestion_pipeline_one.json

@@ -188,7 +188,8 @@
                         },
                         "auto_keywords": 0,
                         "auto_questions": 0,
-                        "llm_id": ""
+                        "llm_id": "",
+                        "auto_tags": 0
                     }
                 }
             }

internal/ingestion/pipeline/template/ingestion_pipeline_paper.json

@@ -156,7 +156,8 @@
                         },
                         "auto_keywords": 0,
                         "auto_questions": 0,
-                        "llm_id": ""
+                        "llm_id": "",
+                        "auto_tags": 0
                     }
                 }
             }

internal/ingestion/pipeline/template/ingestion_pipeline_picture.json

@@ -148,7 +148,8 @@
                         },
                         "auto_keywords": 0,
                         "auto_questions": 0,
-                        "llm_id": ""
+                        "llm_id": "",
+                        "auto_tags": 0
                     }
                 }
             }

internal/ingestion/pipeline/template/ingestion_pipeline_presentation.json

@@ -134,7 +134,8 @@
                         },
                         "auto_keywords": 0,
                         "auto_questions": 0,
-                        "llm_id": ""
+                        "llm_id": "",
+                        "auto_tags": 0
                     }
                 }
             }

internal/ingestion/pipeline/template/ingestion_pipeline_resume.json

@@ -48,7 +48,8 @@
                         "topPEnabled": true,
                         "top_p": 0.3,
                         "auto_keywords": 0,
-                        "auto_questions": 0
+                        "auto_questions": 0,
+                        "auto_tags": 0
                     }
                 },
                 "upstream": [

internal/service/nlp/reranker.go

@@ -701,14 +701,13 @@ func applyRankFeatureScores(chunks []map[string]interface{}, sim []float64, rank
 	// Compute tag score for each chunk
 	tagScores := make([]float64, len(chunks))
 	for i, chunk := range chunks {
-		tagFeaStr, ok := chunk[common.TAG_FLD].(string)
-		if !ok || tagFeaStr == "" {
+		// tag_feas may be a JSON string (legacy) or an object as stored by the
+		// "rank_features" ES mapping; normalize to map[string]float64 either way.
+		tagFeaMap := extractTagFeasMap(chunk[common.TAG_FLD])
+		if len(tagFeaMap) == 0 {
 			tagScores[i] = 0
 			continue
 		}
-
-		// Parse tag_feas JSON string: {"tag1": 0.5, "tag2": 0.3}
-		tagFeaMap := parseTagFeasRerank(tagFeaStr)
 		// Sort keys for deterministic float accumulation
 		tagFeaKeys := make([]string, 0, len(tagFeaMap))
 		for k := range tagFeaMap {
@@ -816,14 +815,13 @@ func applyRankFeatureScoresForIDs(ids []string, field map[string]map[string]inte
 			tagScores[i] = 0
 			continue
 		}
-		tagFeaStr, ok := chunk[common.TAG_FLD].(string)
-		if !ok || tagFeaStr == "" {
+		// tag_feas may be a JSON string (legacy) or an object as stored by the
+		// "rank_features" ES mapping; normalize to map[string]float64 either way.
+		tagFeaMap := extractTagFeasMap(chunk[common.TAG_FLD])
+		if len(tagFeaMap) == 0 {
 			tagScores[i] = 0
 			continue
 		}
-
-		// Parse tag_feas JSON string: {"tag1": 0.5, "tag2": 0.3}
-		tagFeaMap := parseTagFeasRerank(tagFeaStr)
 		// Sort keys for deterministic float accumulation
 		tagFeaKeys := make([]string, 0, len(tagFeaMap))
 		for k := range tagFeaMap {
@@ -1048,3 +1046,37 @@ func parseTagFeasRerank(tagFeasStr string) map[string]float64 {
 	}
 	return result
 }
+
+// extractTagFeasMap normalizes a tag_feas value into map[string]float64.
+// Depending on the search backend, tag_feas is stored either as an object of
+// numeric values (e.g. the ES "rank_features" type) or as a JSON string (e.g.
+// Infinity's varchar rankfeatures column), so both forms are accepted.
+func extractTagFeasMap(v interface{}) map[string]float64 {
+	switch t := v.(type) {
+	case string:
+		return parseTagFeasRerank(t)
+	case map[string]interface{}:
+		out := make(map[string]float64, len(t))
+		for k, val := range t {
+			if f, ok := toFloat64(val); ok {
+				out[k] = f
+			}
+		}
+		return out
+	case map[string]float64:
+		return t
+	case map[string]int:
+		out := make(map[string]float64, len(t))
+		for k, val := range t {
+			out[k] = float64(val)
+		}
+		return out
+	case nil:
+		return nil
+	default:
+		if b, err := json.Marshal(t); err == nil {
+			return parseTagFeasRerank(string(b))
+		}
+		return nil
+	}
+}