From 83a2072b96048b71f555d17496b800e361172dd3 Mon Sep 17 00:00:00 2001 From: flash Date: Sat, 27 Jun 2026 23:59:56 +0200 Subject: [PATCH 1/2] feat(search): open_taki v2 protocol support with content routing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Auto-detect open_taki as drop-in Tika replacement. When detected, uses v2 protocol for LLM-powered extraction with routing: - Meta (title, entities, summary) → always bleve - Content → bleve or vector-only based on routing config - Embedding → returned for vector DB integration - Graceful fallback to classic Tika v1 if v2 fails Co-Authored-By: Claude Opus 4.6 (1M context) --- services/search/pkg/content/content.go | 25 ++++ services/search/pkg/content/tika.go | 153 ++++++++++++++++++++++++- services/search/pkg/search/service.go | 35 +++++- 3 files changed, 205 insertions(+), 8 deletions(-) diff --git a/services/search/pkg/content/content.go b/services/search/pkg/content/content.go index e185351678..5c7ead934c 100644 --- a/services/search/pkg/content/content.go +++ b/services/search/pkg/content/content.go @@ -26,6 +26,31 @@ type Document struct { Image *libregraph.Image `json:"image,omitempty"` Location *libregraph.GeoCoordinates `json:"location,omitempty"` Photo *libregraph.Photo `json:"photo,omitempty"` + + // Taki v2 extended fields (populated when extractor is open_taki with v2 protocol) + Taki *TakiExtraction `json:"taki,omitempty"` +} + +// TakiExtraction holds extended extraction results from open_taki v2 protocol. +type TakiExtraction struct { + Method string `json:"method,omitempty"` + Summary string `json:"summary,omitempty"` + Entities []Entity `json:"entities,omitempty"` + Embed []float64 `json:"embedding,omitempty"` + Routing *Routing `json:"routing,omitempty"` +} + +// Entity represents a named entity extracted by the LLM. +type Entity struct { + Type string `json:"type"` + Value string `json:"value"` +} + +// Routing tells the search service where to store content. +type Routing struct { + ContentTarget string `json:"content_target"` // bleve, vector, both, none + MetaTarget string `json:"meta_target"` // always bleve + VectorTarget string `json:"vector_target"` // always vector } func CleanString(content, langCode string) string { diff --git a/services/search/pkg/content/tika.go b/services/search/pkg/content/tika.go index 8b6ac5e60b..7d57787419 100644 --- a/services/search/pkg/content/tika.go +++ b/services/search/pkg/content/tika.go @@ -1,9 +1,13 @@ package content import ( + "bytes" "context" + "encoding/json" "fmt" + "io" "math" + "net/http" "strconv" "strings" "time" @@ -12,20 +16,25 @@ import ( provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" "github.com/google/go-tika/tika" "github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool" + "github.com/opencloud-eu/reva/v2/pkg/storagespace" libregraph "github.com/opencloud-eu/libre-graph-api-go" "github.com/opencloud-eu/opencloud/pkg/log" "github.com/opencloud-eu/opencloud/services/search/pkg/config" ) -// Tika is used to extract content from a resource, -// it uses apache tika to retrieve all the data. +// Tika is used to extract content from a resource. +// Supports both Apache Tika and open_taki (drop-in replacement). +// When open_taki is detected, uses the v2 protocol for enhanced extraction. type Tika struct { *Basic Retriever tika *tika.Client + tikaURL string + httpClient *http.Client ContentExtractionSizeLimit uint64 CleanStopWords bool + isTaki bool } // NewTikaExtractor creates a new Tika instance. @@ -35,23 +44,46 @@ func NewTikaExtractor(gatewaySelector pool.Selectable[gateway.GatewayAPIClient], return nil, err } - tk := tika.NewClient(nil, cfg.Extractor.Tika.TikaURL) + tikaURL := cfg.Extractor.Tika.TikaURL + + tk := tika.NewClient(nil, tikaURL) tkv, err := tk.Version(context.Background()) if err != nil { return nil, err } - logger.Info().Msgf("Tika version: %s", tkv) + + // Detect open_taki by checking the health endpoint + isTaki := false + hc := &http.Client{Timeout: 5 * time.Second} + if resp, err := hc.Get(tikaURL + "/tika"); err == nil { + defer resp.Body.Close() + var health map[string]interface{} + if json.NewDecoder(resp.Body).Decode(&health) == nil { + if name, ok := health["name"].(string); ok && name == "open_taki" { + isTaki = true + logger.Info().Msgf("open_taki detected (version: %v), using v2 protocol", health["version"]) + } + } + } + + if !isTaki { + logger.Info().Msgf("Tika version: %s", tkv) + } return &Tika{ Basic: basic, Retriever: newCS3Retriever(gatewaySelector, logger, cfg.Extractor.CS3AllowInsecure), - tika: tika.NewClient(nil, cfg.Extractor.Tika.TikaURL), + tika: tk, + tikaURL: tikaURL, + httpClient: &http.Client{Timeout: 5 * time.Minute}, ContentExtractionSizeLimit: cfg.ContentExtractionSizeLimit, CleanStopWords: cfg.Extractor.Tika.CleanStopWords, + isTaki: isTaki, }, nil } -// Extract loads a resource from its underlying storage, passes it to tika and processes the result into a Document. +// Extract loads a resource from its underlying storage, passes it to tika/taki +// and processes the result into a Document. func (t Tika) Extract(ctx context.Context, ri *provider.ResourceInfo) (Document, error) { doc, err := t.Basic.Extract(ctx, ri) if err != nil { @@ -77,6 +109,14 @@ func (t Tika) Extract(ctx context.Context, ri *provider.ResourceInfo) (Document, } defer data.Close() + if t.isTaki { + return t.extractTakiV2(ctx, ri, data, doc) + } + return t.extractTikaV1(ctx, ri, data, doc) +} + +// extractTikaV1 is the original Tika extraction path (unchanged behavior). +func (t Tika) extractTikaV1(ctx context.Context, ri *provider.ResourceInfo, data io.ReadCloser, doc Document) (Document, error) { metas, err := t.tika.MetaRecursive(ctx, data) if err != nil { return doc, err @@ -107,6 +147,107 @@ func (t Tika) Extract(ctx context.Context, ri *provider.ResourceInfo) (Document, return doc, nil } +// takiV2Response matches open_taki's v2 JSON response format. +type takiV2Response struct { + Content string `json:"X-TIKA:content"` + ContentType string `json:"Content-Type"` + Method string `json:"X-TAKI:method"` + Meta *struct { + Title string `json:"title"` + Author string `json:"author"` + Created string `json:"created"` + Language string `json:"language"` + DocType string `json:"doc_type"` + } `json:"X-TAKI:meta"` + Entities []Entity `json:"X-TAKI:entities"` + Summary string `json:"X-TAKI:summary"` + Embed []float64 `json:"X-TAKI:embedding"` + Routing *struct { + ContentTarget string `json:"content_target"` + MetaTarget string `json:"meta_target"` + VectorTarget string `json:"vector_target"` + SourceRef string `json:"source_ref"` + } `json:"X-TAKI:routing"` +} + +// extractTakiV2 uses the open_taki v2 protocol for enhanced extraction. +func (t Tika) extractTakiV2(ctx context.Context, ri *provider.ResourceInfo, data io.ReadCloser, doc Document) (Document, error) { + body, err := io.ReadAll(data) + if err != nil { + return doc, fmt.Errorf("reading resource data: %w", err) + } + + sourceRef := storagespace.FormatResourceID(ri.Id) + + req, err := http.NewRequestWithContext(ctx, "PUT", + t.tikaURL+"/rmeta/text", bytes.NewReader(body)) + if err != nil { + return doc, err + } + + req.Header.Set("Content-Type", ri.MimeType) + req.Header.Set("X-Taki-Protocol", "v2") + req.Header.Set("X-Taki-Features", "meta,entities,summary,embedding") + req.Header.Set("X-Taki-Source-Ref", sourceRef) + + resp, err := t.httpClient.Do(req) + if err != nil { + t.logger.Warn().Err(err).Msg("taki v2 request failed, falling back to v1") + return t.extractTikaV1(ctx, ri, io.NopCloser(bytes.NewReader(body)), doc) + } + defer resp.Body.Close() + + var results []takiV2Response + if err := json.NewDecoder(resp.Body).Decode(&results); err != nil { + t.logger.Warn().Err(err).Msg("taki v2 response parse failed, falling back to v1") + return t.extractTikaV1(ctx, ri, io.NopCloser(bytes.NewReader(body)), doc) + } + + if len(results) == 0 { + return doc, nil + } + + r := results[0] + + // Content — always populate (bleve needs it for fulltext) + doc.Content = strings.TrimSpace(r.Content) + + // Title — prefer taki meta over Tika's title field + if r.Meta != nil && r.Meta.Title != "" { + doc.Title = r.Meta.Title + } + + // Build TakiExtraction with routing + enrichments + taki := &TakiExtraction{ + Method: r.Method, + Summary: r.Summary, + } + + if r.Entities != nil { + taki.Entities = r.Entities + } + + if r.Embed != nil { + taki.Embed = r.Embed + } + + if r.Routing != nil { + taki.Routing = &Routing{ + ContentTarget: r.Routing.ContentTarget, + MetaTarget: r.Routing.MetaTarget, + VectorTarget: r.Routing.VectorTarget, + } + } + + doc.Taki = taki + + if t.CleanStopWords && r.Meta != nil && r.Meta.Language != "" { + doc.Content = CleanString(doc.Content, r.Meta.Language) + } + + return doc, nil +} + func (t Tika) getImage(meta map[string][]string) *libregraph.Image { var image *libregraph.Image initImage := func() { diff --git a/services/search/pkg/search/service.go b/services/search/pkg/search/service.go index a13ccd3b3b..6b3386f5e3 100644 --- a/services/search/pkg/search/service.go +++ b/services/search/pkg/search/service.go @@ -615,10 +615,29 @@ func (s *Service) doUpsertItem(ref *provider.Reference, batch BatchOperator) { r.ParentID = storagespace.FormatResourceID(parentID) } + // Taki v2 routing: decide what goes to bleve based on content_target + bleveDoc := doc + if doc.Taki != nil && doc.Taki.Routing != nil { + target := doc.Taki.Routing.ContentTarget + if target == "vector" || target == "none" { + // Content should NOT go to bleve — clear it for the bleve index + // but keep Title (meta always goes to bleve) + bleveDoc.Content = "" + s.logger.Debug(). + Str("name", doc.Name). + Str("target", target). + Str("method", doc.Taki.Method). + Msg("taki routing: content excluded from bleve index") + } + } + + bleveResource := r + bleveResource.Document = bleveDoc + if batch != nil { - err = batch.Upsert(r.ID, r) + err = batch.Upsert(r.ID, bleveResource) } else { - err = s.engine.Upsert(r.ID, r) + err = s.engine.Upsert(r.ID, bleveResource) } if err != nil { s.logger.Error().Err(err).Msg("error adding updating the resource in the index") @@ -626,6 +645,18 @@ func (s *Service) doUpsertItem(ref *provider.Reference, batch BatchOperator) { logDocCount(s.engine, s.logger) } + // Taki v2: log extraction summary + if doc.Taki != nil { + s.logger.Info(). + Str("name", doc.Name). + Str("method", doc.Taki.Method). + Int("content_len", len(doc.Content)). + Int("embedding_dims", len(doc.Taki.Embed)). + Int("entities", len(doc.Taki.Entities)). + Str("summary", doc.Taki.Summary). + Msg("taki v2 extraction complete") + } + // determine if metadata needs to be stored in storage as well metadata := map[string]string{} addAudioMetadata(metadata, doc.Audio) From 586ba113f42352e3d3652ca90be2e8b1fe7695c9 Mon Sep 17 00:00:00 2001 From: flash Date: Sun, 28 Jun 2026 12:20:25 +0200 Subject: [PATCH 2/2] feat(search): parallel indexing with taki v2 worker pool When open_taki is detected, IndexSpace uses a configurable worker pool for parallel extraction instead of sequential processing. This leverages the LLM backend's batch capacity (vLLM max-num-seqs=16). - 8 workers by default (configurable via SEARCH_EXTRACTOR_TIKA_MAX_WORKERS) - Only active with open_taki v2 (classic Tika stays sequential) - Workers use direct upsert (no batch, thread-safe) - IsTaki() exported on Tika extractor for runtime detection Co-Authored-By: Claude Opus 4.6 (1M context) --- services/search/pkg/config/content.go | 1 + .../pkg/config/defaults/defaultconfig.go | 1 + services/search/pkg/content/tika.go | 5 ++ services/search/pkg/search/service.go | 48 ++++++++++++++++++- 4 files changed, 53 insertions(+), 2 deletions(-) diff --git a/services/search/pkg/config/content.go b/services/search/pkg/config/content.go index 515db03b63..1ee17b53b1 100644 --- a/services/search/pkg/config/content.go +++ b/services/search/pkg/config/content.go @@ -11,4 +11,5 @@ type Extractor struct { type ExtractorTika struct { TikaURL string `yaml:"tika_url" env:"SEARCH_EXTRACTOR_TIKA_TIKA_URL" desc:"URL of the tika server." introductionVersion:"1.0.0"` CleanStopWords bool `yaml:"clean_stop_words" env:"SEARCH_EXTRACTOR_TIKA_CLEAN_STOP_WORDS" desc:"Defines if stop words should be cleaned or not. See the documentation for more details." introductionVersion:"1.0.0"` + MaxWorkers int `yaml:"max_workers" env:"SEARCH_EXTRACTOR_TIKA_MAX_WORKERS" desc:"Maximum number of parallel extraction workers. Only effective with open_taki v2. Defaults to 8." introductionVersion:"7.1.0"` } diff --git a/services/search/pkg/config/defaults/defaultconfig.go b/services/search/pkg/config/defaults/defaultconfig.go index 3e504350fa..1e823c249f 100644 --- a/services/search/pkg/config/defaults/defaultconfig.go +++ b/services/search/pkg/config/defaults/defaultconfig.go @@ -51,6 +51,7 @@ func DefaultConfig() *config.Config { Tika: config.ExtractorTika{ TikaURL: "http://127.0.0.1:9998", CleanStopWords: false, + MaxWorkers: 8, }, }, Events: config.Events{ diff --git a/services/search/pkg/content/tika.go b/services/search/pkg/content/tika.go index 7d57787419..b97c39297f 100644 --- a/services/search/pkg/content/tika.go +++ b/services/search/pkg/content/tika.go @@ -37,6 +37,11 @@ type Tika struct { isTaki bool } +// IsTaki returns true if open_taki was detected as the extraction backend. +func (t *Tika) IsTaki() bool { + return t.isTaki +} + // NewTikaExtractor creates a new Tika instance. func NewTikaExtractor(gatewaySelector pool.Selectable[gateway.GatewayAPIClient], logger log.Logger, cfg *config.Config) (*Tika, error) { basic, err := NewBasicExtractor(logger) diff --git a/services/search/pkg/search/service.go b/services/search/pkg/search/service.go index 6b3386f5e3..44754ac67d 100644 --- a/services/search/pkg/search/service.go +++ b/services/search/pkg/search/service.go @@ -7,6 +7,7 @@ import ( "sort" "strconv" "strings" + "sync" "time" gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1" @@ -64,6 +65,7 @@ type Service struct { engine Engine extractor content.Extractor metrics *metrics.Metrics + cfg *config.Config serviceAccountID string serviceAccountSecret string @@ -81,6 +83,7 @@ func NewService(gatewaySelector pool.Selectable[gateway.GatewayAPIClient], eng E logger: logger, extractor: extractor, metrics: metrics, + cfg: cfg, serviceAccountID: cfg.ServiceAccount.ServiceAccountID, serviceAccountSecret: cfg.ServiceAccount.ServiceAccountSecret, @@ -485,6 +488,34 @@ func (s *Service) IndexSpace(spaceID *provider.StorageSpaceId, forceRescan bool) } logDocCount(s.engine, s.logger) }() + + // Parallel extraction when taki v2 is detected (LLM benefits from concurrent calls) + isTaki := false + if tikaExtractor, ok := s.extractor.(*content.Tika); ok { + isTaki = tikaExtractor.IsTaki() + } + + indexWorkers := s.cfg.Extractor.Tika.MaxWorkers + if indexWorkers < 1 { + indexWorkers = 8 + } + var workCh chan *provider.Reference + var indexWg sync.WaitGroup + + if isTaki { + s.logger.Info().Int("workers", indexWorkers).Msg("taki v2 detected: parallel indexing enabled") + workCh = make(chan *provider.Reference, indexWorkers*2) + for i := 0; i < indexWorkers; i++ { + indexWg.Add(1) + go func() { + defer indexWg.Done() + for ref := range workCh { + s.doUpsertItem(ref, nil) + } + }() + } + } + err = w.Walk(ownerCtx, &rootID, func(wd string, info *provider.ResourceInfo, err error) error { if err != nil { s.logger.Error().Err(err).Msg("error walking the tree") @@ -502,7 +533,11 @@ func (s *Service) IndexSpace(spaceID *provider.StorageSpaceId, forceRescan bool) s.logger.Debug().Str("path", ref.Path).Msg("Walking tree") if forceRescan { - s.doUpsertItem(ref, batch) + if isTaki { + workCh <- ref + } else { + s.doUpsertItem(ref, batch) + } return nil } @@ -519,11 +554,20 @@ func (s *Service) IndexSpace(spaceID *provider.StorageSpaceId, forceRescan bool) return nil } - s.doUpsertItem(ref, batch) + if isTaki { + workCh <- ref + } else { + s.doUpsertItem(ref, batch) + } return nil }) + if isTaki { + close(workCh) + indexWg.Wait() + } + if err != nil { return err }