插入寫

若目標數據已存在，則不允許寫入。以寫入索引vector_test為例：

func writeDataCreate() {
	// 插入寫
	doc := strings.NewReader(`{
		"field1": 2,
		"vector1": [2.2, 2.3, 2.4]
	}`)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	req := opensearchapi.CreateRequest{
		Index:      "vector_test",
		DocumentID: "2",
		Body:       doc,
	}

	res, err := req.Do(ctx, client)
	if err != nil {
		fmt.Println("Write index error ", err)
	}
	defer res.Body.Close()
}

更新寫

以寫入索引vector_test為例：

func writeDataIndex() {
	// 更新寫
	doc := strings.NewReader(`{
		"field1": 1,
		"vector1": [1.2, 1.3, 1.4]
	}`)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	req := opensearchapi.IndexRequest{
		Index:      "vector_test",
		DocumentID: "1",
		Body:       doc,
	}
	res, err := req.Do(ctx, client)
	if err != nil {
		fmt.Println("Write index error ", err)
	}
        defer res.Body.Close()
}

通用寫入

_bulk寫入方式為指定JSON寫入。以寫入索引vector_test為例，具體如下：

// 人為指定json串
func writeDataBulk() {
	// bulk 寫入
	doc := strings.NewReader(`
		{ "index" : { "_index" : "vector_test", "_id" : "3" } }
		{ "field1" : 3, "vector1": [3.2, 3.3, 3.4] }
		{ "create" : { "_index" : "vector_test", "_id" : "4" } }
		{ "field1" : 4, "vector1": [4.2, 4.3, 4.4] }
		{ "delete" : { "_index" : "vector_test", "_id" : "2" } }
		{ "update" : {"_id" : "1", "_index" : "vector_test"} }
		{ "doc" : {"field1" : 3, "vector1": [2.21, 3.31, 4.41]} }
	`)

	// 確保每行后都有換行符
	var buffer bytes.Buffer
	scanner := bufio.NewScanner(doc)
	for scanner.Scan() {
		buffer.WriteString(scanner.Text())
		buffer.WriteString("\n")
	}

	res, err := client.Bulk(&buffer)
	fmt.Println(res.String())
	if err != nil {
		fmt.Println("Bulk write error ", err)
	}
	defer res.Body.Close()
}

您也可以定義數據寫入邏輯，通過_bulk一次性寫入大量數據。以寫入索引vector_ivfpq_test為例：

// 代碼樣例是隨機生成dim維度的向量，實際業務替換
// 生成格式 [0.7862035,0.9371029,0.50112325]
func randomDenseVector(dim int) string {
	random := rand.New(rand.NewSource(time.Now().UnixNano()))
	randomArray := make([]float32, dim)
	for i := range randomArray {
		randomArray[i] = random.Float32()
	}
	var formattedArray []string
	for _, num := range randomArray {
		formattedArray = append(formattedArray, fmt.Sprintf("%.8f", num))
	}
	result := "[" + strings.Join(formattedArray, ", ") + "]"
	return result
}

func bulkWriteDenseVector() {
	indexName := "vector_ivfpq_test"
	var buf bytes.Buffer
	// 添加多條數據，bulk寫入也要控制每次寫入的條數
	for i := 1; i <= 20; i++ {
		// 寫20次，每次寫入100條
		for j := (i-1)*100 + 1; j <= i*100; j++ {
			// id 可以根據業務自己指定，向量的random請替換為業務自己的。create可以替換為index
			meta := []byte(fmt.Sprintf(`{ "index": { "_index": "%s", "_id": "%d" } }`, indexName, j))
			data := map[string]interface{}{
				"vector1": randomDenseVector(3),
				"field1":  i,
			}
			dataJson, err := json.Marshal(data)
			if err != nil {
				fmt.Println("Error marshaling JSON:", err)
				continue
			}
			// 將操作寫入緩沖區
			buf.Grow(len(meta) + len(dataJson) + 2) // 預留空間
			buf.Write(meta)
			buf.WriteByte('\n') // 換行
			buf.Write(dataJson)
			buf.WriteByte('\n') // 換行
		}

		// 執行 Bulk 請求
		bulkReq := opensearchapi.BulkRequest{
			Body:  &buf,
			Index: indexName,
		}

		// 設置超時
		ctx, cancel := context.WithTimeout(context.Background(), 30*2*time.Second)
		defer cancel()

		res, err := bulkReq.Do(ctx, client)
		if err != nil {
			fmt.Println("happend err", err)
		}
		defer res.Body.Close()
	}
}

稀疏向量寫入

以寫入索引vector_sparse_test為例：

type Data struct {
	Indices []int     `json:"indices"`
	Values  []float32 `json:"values"`
}

/*
* 格式參考curl使用文檔，稀疏向量的格式說明
{"indices":[91,30,92],"values":[0.7862035,0.9371029,0.50112325]}
*/
func randomSparseVector() string {
	random := rand.New(rand.NewSource(time.Now().UnixNano()))
	// 隨機生成 indices
	sparsedim := random.Intn(16) + 2
	indices := make([]int, sparsedim)
	for i := range indices {
		indices[i] = random.Intn(100) // 生成 0-99 之間的隨機整數
	}
	// 隨機生成 values
	values := make([]float32, sparsedim)
	for i := range values {
		values[i] = random.Float32() // 生成 0.0-1.0 之間的隨機浮點數
	}
	// 創建數據結構
	data := Data{
		Indices: indices,
		Values:  values,
	}
	// 將 data 編碼為 JSON
	jsonData, err := json.Marshal(data)
	if err != nil {
		fmt.Println("Error marshaling to JSON:", err)
		return ""
	}
	// 打印生成的 JSON 字符串
	return string(jsonData)
}

func bulkWriteSparseVector() {
	indexName := "vector_sparse_test"
	var buf bytes.Buffer
	// 添加多條數據，bulk寫入也要控制每次寫入的條數，不能過大
	for i := 5; i <= 50; i++ {
		// id 可以根據業務自己指定，向量的random請替換為業務自己的。create可以替換為index
		meta := []byte(fmt.Sprintf(`{ "index": { "_index": "%s", "_id": "%d" } }`, indexName, i))
		data := map[string]interface{}{
			"vector1": randomSparseVector(),
			"field1":  i,
		}
		dataJson, err := json.Marshal(data)
		if err != nil {
			fmt.Println("Error marshaling JSON: ", err)
			continue
		}
		// 將操作寫入緩沖區
		buf.Grow(len(meta) + len(dataJson) + 2) // 預留空間
		buf.Write(meta)
		buf.WriteByte('\n') // 換行
		buf.Write(dataJson)
		buf.WriteByte('\n') // 換行
	}

	// 執行 Bulk 請求
	bulkReq := opensearchapi.BulkRequest{
		Body:  &buf,
		Index: indexName,
	}

	// 設置超時
	ctx, cancel := context.WithTimeout(context.Background(), 30*2*time.Second)
	defer cancel()

	res, err := bulkReq.Do(ctx, client)
	if err != nil {
		fmt.Println("happend err", err)
	}
	defer res.Body.Close()
}

返回指定字段

如果需要在查詢時返回指定字段，可以指定 "_source": ["field1", "field2"] 或使用"_source": true 返回非向量的全部字段。以查詢索引vector_test為例，使用方法如下：

func knnSearchHnswWithSource() {
	indexName := "vector_test"
	vectorField := "vector1"
	knnString := fmt.Sprintf(`{
	  "size": 10,
      "_source": ["field1"],
	  "query": {
		"knn": {
		  "%s": {
			"vector": %s,
			"k": 10
		  }
		}
	  }
	}`, vectorField, randomDenseVector(3))

	content := strings.NewReader(knnString)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	search := opensearchapi.SearchRequest{
		Index: []string{indexName},
		Body:  content,
	}
	searchResponse, err := search.Do(ctx, client)
	if err != nil {
		fmt.Println("Knn err ", err)
		return
	}
	defer searchResponse.Body.Close()
	// fmt.Println("knn result ", searchResponse.String())
	if searchResponse.IsError() {
		handlerCommonError(searchResponse)
		return
	}
	var responseBody KnnResponse
	if err := json.NewDecoder(searchResponse.Body).Decode(&responseBody); err != nil {
		fmt.Println("Knn result Decode err ", err)
		return
	}

	for _, hit := range responseBody.Hits.Hits {
		fmt.Printf("ID: %s, Score: %f, Field1 %d \n", hit.ID, hit.Score, hit.Source.Field1)
	}
}

返回結果如下：

{
  "took" : 35,
  "timed_out" : false,
  "terminated_early" : false,
  "num_reduce_phases" : 0,
  "_shards" : {
    "total" : 2,
    "successful" : 2,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 3,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "vector_test",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "field1" : 1
        }
      },
      {
        "_index" : "vector_test",
        "_id" : "1",
        "_score" : 0.25,
        "_source" : {
          "field1" : 1
        }
      },
      {
        "_index" : "vector_test",
        "_id" : "3",
        "_score" : 0.14285715,
        "_source" : {
          "field1" : 2
        }
      }
    ]
  }
}

func knnSearchHnsw() {
	indexName := "vector_test"
	vectorField := "vector1"
	knnString := fmt.Sprintf(`{
	  "size": 10,
	  "query": {
		"knn": {
		  "%s": {
			"vector": %s,
			"k": 10
		  }
		}
	  },
     "ext": {"lvector": {"ef_search": "100"}}
	}`, vectorField, randomDenseVector(3))

	content := strings.NewReader(knnString)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	search := opensearchapi.SearchRequest{
		Index: []string{indexName},
		Body:  content,
	}
	searchResponse, err := search.Do(ctx, client)
	if err != nil {
		fmt.Println("Knn err ", err)
		return
	}
	defer searchResponse.Body.Close()
	// fmt.Println("Knn result ", searchResponse.String())
	if searchResponse.IsError() {
		handlerCommonError(searchResponse)
		return
	}
	var responseBody KnnResponse
	if err := json.NewDecoder(searchResponse.Body).Decode(&responseBody); err != nil {
		fmt.Println("Knn result Decode err", err)
		return
	}
	for _, hit := range responseBody.Hits.Hits {
		fmt.Printf("ID: %s, Score: %f\n", hit.ID, hit.Score)
	}
}

func knnSearchIvfpq() {
	indexName := "vector_test"
	vectorField := "vector1"
	knnString := fmt.Sprintf(`{
	  "size": 10,
	  "query": {
		"knn": {
		  "%s": {
			"vector": %s,
			"k": 10
		  }
		}
	  },
      "ext": {"lvector": {"nprobe": "60", "reorder_factor": "5"}}
	}`, vectorField, "[2.21, 3.31, 4.41]")

	content := strings.NewReader(knnString)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	search := opensearchapi.SearchRequest{
		Index: []string{indexName},
		Body:  content,
	}
	searchResponse, err := search.Do(ctx, client)
	if err != nil {
		fmt.Println("Knn err", err)
		return
	}
	defer searchResponse.Body.Close()
	// fmt.Println("Knn result", searchResponse.String())
        if searchResponse.IsError() {
		handlerCommonError(searchResponse)
		return
	}
	var responseBody KnnResponse
	if err := json.NewDecoder(searchResponse.Body).Decode(&responseBody); err != nil {
		fmt.Println("Knn result Decode err", err)
		return
	}
	for _, hit := range responseBody.Hits.Hits {
		fmt.Printf("ID: %s, Score: %f\n", hit.ID, hit.Score)
	}
}

func knnSearchSparseVector() {
	indexName := "vector_sparse_test"
	vectorField := "vector1"
	knnString := fmt.Sprintf(`{
	  "size": 10,
	  "query": {
		"knn": {
		  "%s": {
			"vector": %s,
			"k": 10
		  }
		}
	  }
	}`, vectorField, randomSparseVector())

	content := strings.NewReader(knnString)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	search := opensearchapi.SearchRequest{
		Index: []string{indexName},
		Body:  content,
	}
	searchResponse, err := search.Do(ctx, client)
	if err != nil {
		fmt.Println("Knn err ", err)
		return
	}
	defer searchResponse.Body.Close()
	// fmt.Println("Knn result ", searchResponse.String())
	if searchResponse.IsError() {
		handlerCommonError(searchResponse)
		return
	}
	var responseBody KnnResponse
	if err := json.NewDecoder(searchResponse.Body).Decode(&responseBody); err != nil {
		fmt.Println("Knn result Decode err ", err)
		return
	}
	for _, hit := range responseBody.Hits.Hits {
		fmt.Printf("ID: %s, Score: %f\n", hit.ID, hit.Score)
	}
}

Pre-Filter近似查詢

通過在knn查詢結構內部添加過濾器filter，并指定filter_type參數的值為pre_filter，可實現先過濾結構化數據，再查詢向量數據。

以查詢索引vector_test為例：

func preFilterKnnSearch() {
	indexName := "vector_test"
	vectorField := "vector1"
	knnString := fmt.Sprintf(`{
	  "size": 10,
	  "query": {
		"knn": {
		  "%s": {
			"vector": %s,
			"k": 10,
			"filter": {
			  "range": {
				"field1": {
				  "gte": 1
				}
			  }
			}
		  }
		}
	  },
     "ext": {"lvector": {"filter_type": "pre_filter"}}
	}`, vectorField, randomDenseVector(3))

	content := strings.NewReader(knnString)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	search := opensearchapi.SearchRequest{
		Index: []string{indexName},
		Body:  content,
	}
	searchResponse, err := search.Do(ctx, client)
	if err != nil {
		fmt.Println("Knn err", err)
		return
	}
	defer searchResponse.Body.Close()
	// fmt.Println("Knn result", searchResponse.String())
        if searchResponse.IsError() {
		handlerCommonError(searchResponse)
		return
	}
	var responseBody KnnResponse
	if err := json.NewDecoder(searchResponse.Body).Decode(&responseBody); err != nil {
		fmt.Println("Knn result Decode err", err)
		return
	}
	for _, hit := range responseBody.Hits.Hits {
		fmt.Printf("ID: %s, Score: %f\n", hit.ID, hit.Score)
	}
}

Post-Filter近似查詢

通過在knn查詢結構內部添加過濾器filter，并指定filter_type參數的值為post_filter，可實現先查詢向量數據，再過濾結構化數據。

以查詢索引vector_ivfpq_test為例：

func postFilterKnnSearch() {
	indexName := "vector_ivfpq_test"
	vectorField := "vector1"
	knnTopk := 1000
	knnString := fmt.Sprintf(`{
	  "size": 10,
	  "query": {
		"knn": {
		  "%s": {
			"vector": %s,
			"k": %d,
	        "filter": {
			  "range": {
				"field1": {
				  "gte": 0
				}
			  }
			}
		  }
		}
	  },
      "ext": {"lvector": {"filter_type": "post_filter", "nprobe": "100", "reorder_factor": "1"}}
	}`, vectorField, "[2.21, 3.31, 4.41]", knnTopk)

	content := strings.NewReader(knnString)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	search := opensearchapi.SearchRequest{
		Index: []string{indexName},
		Body:  content,
	}
	searchResponse, err := search.Do(ctx, client)
	if err != nil {
		fmt.Println("Knn err %s", err)
		return
	}
	defer searchResponse.Body.Close()
	// fmt.Println("Knn result %s", searchResponse.String())
	if searchResponse.IsError() {
		handlerCommonError(searchResponse)
		return
	}
	var responseBody KnnResponse
	if err := json.NewDecoder(searchResponse.Body).Decode(&responseBody); err != nil {
		fmt.Println("Knn result Decode err %s", err)
		return
	}
	// 打印每個文檔的 ID 和得分
	for _, hit := range responseBody.Hits.Hits {
		fmt.Printf("ID: %s, Score: %f\n", hit.ID, hit.Score)
	}
}