// // 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. // package chunk import ( "context" "encoding/json" "fmt" "strconv" "strings" "ragflow/internal/common" "ragflow/internal/engine/types" "go.uber.org/zap" ) // vectorFetcher is the consumer-side interface for chunk vector hydration. type vectorFetcher interface { Search(ctx context.Context, req *types.SearchRequest) (*types.SearchResult, error) GetType() string } // FetchChunkVectors fetches embedding vectors for a set of chunk IDs. // This is used by citation insertion (insert_citations) to hydrate chunk // vectors on demand, since the main retrieval path skips vector transport. // // On Infinity / OceanBase the chunks already carry vectors, so we skip // the round-trip. On ES we query by chunk ID list. // // Degrades gracefully: if the engine returns an error, zero vectors are // returned for all chunk IDs rather than failing the caller. // // The returned map has an entry for every requested chunkID. Each vector // slice is independently allocated — callers may safely modify them. func FetchChunkVectors(ctx context.Context, engine vectorFetcher, chunkIDs, tenantIDs, kbIDs []string, dim int) map[string][]float64 { out := make(map[string][]float64, len(chunkIDs)) if len(chunkIDs) == 0 || dim <= 0 { return out } // Infinity already ships vectors with chunks; no need to fetch. // TODO: OceanBase engine is not yet implemented — add "oceanbase" here when it lands. if engine.GetType() == "infinity" || engine.GetType() == "oceanbase" { for _, cid := range chunkIDs { out[cid] = zeroVector(dim) } return out } vecField := fmt.Sprintf("q_%d_vec", dim) // Convert chunkIDs to []interface{} because the ES filter builder // (buildBoolQueryFromCondition) only handles []interface{} for the // "id" key — passing []string would be silently dropped. idList := make([]interface{}, len(chunkIDs)) for i, cid := range chunkIDs { idList[i] = cid } // Query each tenant index for the requested chunk vectors. for _, tid := range tenantIDs { idxName := fmt.Sprintf("ragflow_%s", tid) res, err := engine.Search(ctx, &types.SearchRequest{ IndexNames: []string{idxName}, KbIDs: kbIDs, SelectFields: []string{vecField}, Filter: map[string]interface{}{"id": idList}, Limit: len(chunkIDs), }) if err != nil { common.Warn("FetchChunkVectors search failed, using zero vectors", zap.String("index", idxName), zap.String("error", err.Error())) continue } for _, chunk := range res.Chunks { cid, _ := chunk["id"].(string) if cid == "" { continue } if _, exists := out[cid]; exists { continue } if v := parseVectorField(chunk, vecField, dim); v != nil { out[cid] = v } else { out[cid] = zeroVector(dim) } } } // Fill any chunk IDs not found across all indices with independently // allocated zero vectors so callers cannot corrupt each other. for _, cid := range chunkIDs { if _, exists := out[cid]; !exists { out[cid] = zeroVector(dim) } } return out } // zeroVector returns a freshly allocated zero vector of the given dimension. func zeroVector(dim int) []float64 { return make([]float64, dim) } // parseVectorField extracts a vector from a chunk map. ES stores vectors // as tab-separated strings; Infinity stores them as []float64 / []interface{}. // Returns nil when the vector cannot be extracted or has the wrong dimension. func parseVectorField(chunk map[string]interface{}, field string, dim int) []float64 { raw, ok := chunk[field] if !ok { return nil } switch v := raw.(type) { case string: return parseVectorString(v, dim) case []float64: if len(v) == dim { out := make([]float64, dim) copy(out, v) return out } case []interface{}: vec := make([]float64, len(v)) for i, val := range v { switch fv := val.(type) { case float64: vec[i] = fv case float32: vec[i] = float64(fv) case json.Number: f, err := fv.Float64() if err != nil { return nil } vec[i] = f case string: f, err := strconv.ParseFloat(fv, 64) if err != nil { return nil } vec[i] = f default: return nil } } if len(vec) == dim { return vec } } return nil } // parseVectorString parses a tab-separated vector string from ES. // Returns nil when parsing fails or the dimension does not match. func parseVectorString(s string, dim int) []float64 { parts := strings.Split(s, "\t") if len(parts) != dim { return nil } vec := make([]float64, dim) for i, p := range parts { f, err := strconv.ParseFloat(strings.TrimSpace(p), 64) if err != nil { return nil } vec[i] = f } return vec }