/* * # Licensed to the LF AI & Data foundation under one * # or more contributor license agreements. See the NOTICE file * # distributed with this work for additional information * # regarding copyright ownership. The ASF licenses this file * # to you 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 highlight import ( "context" "encoding/json" "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" "github.com/milvus-io/milvus-proto/go-api/v3/schemapb" "github.com/milvus-io/milvus/internal/util/function/models" "github.com/milvus-io/milvus/pkg/v3/util/merr" ) type semanticHighlightProvider interface { highlight(ctx context.Context, query string, texts []string) ([][]string, [][]float32, error) maxBatch() int } type baseSemanticHighlightProvider struct { batchSize int } func (provider *baseSemanticHighlightProvider) maxBatch() int { return provider.batchSize } type SemanticHighlight struct { fieldNames map[int64]string // schema field: FieldID -> fieldName fieldIDs []int64 // schema field IDs for highlight processing dynamicFieldNames []string // dynamic field names (fields not in schema) dynamicFieldID int64 // $meta field's FieldID provider semanticHighlightProvider queries []string } const ( queryKeyName string = "queries" inputFieldKeyName string = "input_fields" ) func NewSemanticHighlight(collSchema *schemapb.CollectionSchema, params []*commonpb.KeyValuePair, conf map[string]string, extraInfo *models.ModelExtraInfo) (*SemanticHighlight, error) { queries := []string{} inputFields := []string{} for _, param := range params { switch param.Key { case queryKeyName: if err := json.Unmarshal([]byte(param.Value), &queries); err != nil { return nil, merr.Wrap(err, "parse queries failed") } case inputFieldKeyName: if err := json.Unmarshal([]byte(param.Value), &inputFields); err != nil { return nil, merr.Wrap(err, "parse input_field failed") } } } if len(queries) == 0 { return nil, merr.WrapErrParameterMissingMsg("queries is required") } if len(inputFields) == 0 { return nil, merr.WrapErrParameterMissingMsg("input_field is required") } fieldIDMap := make(map[string]*schemapb.FieldSchema) fieldIDNameMap := make(map[int64]string) var dynamicFieldID int64 = -1 for _, field := range collSchema.Fields { fieldIDMap[field.Name] = field fieldIDNameMap[field.FieldID] = field.Name if field.IsDynamic { dynamicFieldID = field.FieldID } } fieldIDs := []int64{} dynamicFieldNames := []string{} for _, fieldName := range inputFields { field, ok := fieldIDMap[fieldName] if ok { // schema field found if field.DataType != schemapb.DataType_VarChar && field.DataType != schemapb.DataType_Text { return nil, merr.WrapErrParameterInvalidMsg("input_field %s is not a VarChar or Text field", fieldName) } fieldIDs = append(fieldIDs, field.FieldID) } else { // field not in schema, check if dynamic field is enabled if !collSchema.EnableDynamicField { return nil, merr.WrapErrParameterInvalidMsg("input_field %s not found in schema", fieldName) } // Non-schema fields are dynamic fields dynamicFieldNames = append(dynamicFieldNames, fieldName) } } // TODO: support other providers if have more providers provider, err := newZillizHighlightProvider(params, conf, extraInfo) if err != nil { return nil, err } return &SemanticHighlight{ fieldNames: fieldIDNameMap, fieldIDs: fieldIDs, dynamicFieldNames: dynamicFieldNames, dynamicFieldID: dynamicFieldID, provider: provider, queries: queries, }, nil } // FieldIDs returns schema field IDs for highlight processing (not including $meta) func (highlight *SemanticHighlight) FieldIDs() []int64 { return highlight.fieldIDs } // RequiredFieldIDs returns all field IDs required for highlight processing (including $meta if has dynamic fields) func (highlight *SemanticHighlight) RequiredFieldIDs() []int64 { if highlight.HasDynamicFields() && highlight.dynamicFieldID > 0 { // Create a new slice to avoid mutating the original fieldIDs slice result := make([]int64, len(highlight.fieldIDs)+1) copy(result, highlight.fieldIDs) result[len(highlight.fieldIDs)] = highlight.dynamicFieldID return result } return highlight.fieldIDs } func (highlight *SemanticHighlight) GetFieldName(id int64) string { return highlight.fieldNames[id] } // DynamicFieldNames returns the list of dynamic field names func (highlight *SemanticHighlight) DynamicFieldNames() []string { return highlight.dynamicFieldNames } // HasDynamicFields returns true if there are any dynamic fields func (highlight *SemanticHighlight) HasDynamicFields() bool { return len(highlight.dynamicFieldNames) > 0 } // DynamicFieldID returns the $meta field ID func (highlight *SemanticHighlight) DynamicFieldID() int64 { return highlight.dynamicFieldID } func (highlight *SemanticHighlight) processOneQuery(ctx context.Context, query string, documents []string) ([][]string, [][]float32, error) { if len(documents) == 0 { return [][]string{}, [][]float32{}, nil } highlights, scores, err := highlight.provider.highlight(ctx, query, documents) if err != nil { return nil, nil, err } if len(highlights) != len(documents) || len(scores) != len(documents) { return nil, nil, merr.WrapErrFunctionFailedMsg("highlights size must equal to documents size, but got highlights size [%d], scores size [%d], documents size [%d]", len(highlights), len(scores), len(documents)) } return highlights, scores, nil } func (highlight *SemanticHighlight) Process(ctx context.Context, topks []int64, documents []string) ([][]string, [][]float32, error) { nq := len(topks) if len(highlight.queries) != nq { return nil, nil, merr.WrapErrParameterInvalidMsg("nq must equal to queries size, but got nq [%d], queries size [%d], queries: [%v]", nq, len(highlight.queries), highlight.queries) } if len(documents) == 0 { return [][]string{}, [][]float32{}, nil } highlights := make([][]string, 0, len(documents)) scores := make([][]float32, 0, len(documents)) start := int64(0) for i, query := range highlight.queries { size := topks[i] //nolint:gosec // bounds checked by loop condition singleQueryHighlights, singleQueryScores, err := highlight.processOneQuery(ctx, query, documents[start:start+size]) //nolint:gosec // bounds are guaranteed by topks summing to len(documents) if err != nil { return nil, nil, err } highlights = append(highlights, singleQueryHighlights...) scores = append(scores, singleQueryScores...) start += size } return highlights, scores, nil }