通过Go（OpenSearch Go）访问向量引擎

插入写

若目标数据已存在，则不允许写入。以写入索引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)
	defer res.Body.Close()
	if err != nil {
		fmt.Println("Write index error ", err)
	}
}

更新写

以写入索引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)
	defer res.Body.Close()
	if err != nil {
		fmt.Println("Write index error ", err)
	}
}

通用写入

_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)
	defer res.Body.Close()
	fmt.Println(res.String())
	if err != nil {
		fmt.Println("Bulk write error ", err)
	}
}

您也可以定义数据写入逻辑，通过_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)
	defer searchResponse.Body.Close()
	if err != nil {
		fmt.Println("Knn err ", err)
		return
	}
	// 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)
	defer searchResponse.Body.Close()
	if err != nil {
		fmt.Println("Knn err ", err)
		return
	}
	// 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)
	defer searchResponse.Body.Close()
	if err != nil {
		fmt.Println("Knn err", err)
		return
	}
	// 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)
	defer searchResponse.Body.Close()
	if err != nil {
		fmt.Println("Knn err ", err)
		return
	}
	// 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)
	defer searchResponse.Body.Close()
	if err != nil {
		fmt.Println("Knn err", err)
		return
	}
	// 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)
	defer searchResponse.Body.Close()
	if err != nil {
		fmt.Println("Knn err %s", err)
		return
	}
	// 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)
	}
}