Hologres搜索（Search）SDK

连接Hologres

import holo_search_sdk as holo

# 连接到数据库
client = holo.connect(
    host="<HOLO_HOST>",
    port=<HOLO_PORT>,
    database="<HOLO_DBNAME>",
    access_key_id="<ACCESS_KEY_ID>",
    access_key_secret="<ACCESS_KEY_SECRET>",
    schema="public" #可根据实际schema修改
)

# 建立连接
client.connect()

变量说明：

创建表

需要通过DDL创建，下面是创建的示例：

create_table_sql = """
    CREATE TABLE IF NOT EXISTS <TABLE_NAME> (
        id BIGINT PRIMARY KEY,
        content TEXT,
        vector_column FLOAT4[] CHECK (array_ndims(vector_column) = 1 AND array_length(vector_column, 1) = 3),
        publish_date TIMESTAMP
    );
"""
_ = client.execute(create_table_sql, fetch_result=False)

打开表

columns = {
    "id": ("INTEGER", "PRIMARY KEY"),
    "content": "TEXT",
    "vector_column": "FLOAT4[]",
    "publish_date": "TIMESTAMP"
}
table = client.open_table("<TABLE_NAME>")

导入文本/向量数据

data = [
    [1, "Hello world", [0.1, 0.2, 0.3], "2023-01-01"],
    [2, "Python SDK", [0.4, 0.5, 0.6], "2024-01-01"],
    [3, "Vector search", [0.7, 0.8, 0.9], "2025-01-01"]
]
table.insert_multi(data, ["id", "content", "vector_column", "publish_date"])

设置索引

• 设置向量索引

  table.set_vector_index(
      column="vector_column",
      distance_method="Cosine",
      base_quantization_type="rabitq",
      use_reorder=True,
      max_degree=64,
      ef_construction=400
  )

• 设置全文索引

  # 创建全文索引
  table.create_text_index(
      index_name="ft_idx_content",
      column="content",
      tokenizer="jieba"
  )
  
  # 修改全文索引
  table.set_text_index(
      index_name="ft_idx_content",
      tokenizer="ik"
  )
  
  # 删除全文索引
  table.drop_text_index(index_name="ft_idx_content")

查询数据

# 向量搜索
query_vector = [0.2, 0.3, 0.4]

# 限制结果数量
results = table.search_vector(
    vector=query_vector, 
    column="vector_column",
    distance_method="Cosine"
).limit(10).fetchall()

# 设置最小距离
results = table.search_vector(
    vector=query_vector, 
    column="vector_column",
    distance_method="Cosine"
).min_distance(0.5).fetchall()

# 带输出别名的搜索
results = table.search_vector(
    vector=query_vector,
    column="vector_column",
    output_name="similarity_score",
    distance_method="Cosine"
).fetchall()

# 基本全文检索
results = table.search_text(
    column="content",
    expression="机器学习",
    return_all_columns=True
).fetchall()

# 返回BM25相似性分数的全文检索
results = table.search_text(
    column="content",
    expression="深度学习",
    return_score=True,
    return_score_name="relevance_score"
).select(["id", "vector_column", "content"]).fetchall()

# 使用不同的检索模式
    # 关键词模式（默认）
results = table.search_text(
    column="content",
    expression="python programming",
    mode="match",
    operator="AND"  # 要求同时包含所有关键词
).fetchall()

    # 短语模式
results = table.search_text(
    column="content",
    expression="machine learning",
    mode="phrase"  # 精确短语匹配
).fetchall()

    # 自然语言模式
results = table.search_text(
    column="content",
    expression="+python -java",  # 必须包含python，不能包含java
    mode="natural_language"
).fetchall()

    # 术语检索
results = table.search_text(
    column="content",
    expression="python",
    mode="term" # 对expression不做分词或其他处理，直接去索引中精确匹配
).fetchall()

# 全文+标量检索
results = (
    table.search_text(
        column="content",
        expression="人工智能",
        return_score=True,
        return_score_name="score"
    )
    .where("publish_date > '2023-01-01'")
    .order_by("score", "desc")
    .limit(10)
    .fetchall()
)

# 向量+标量检索
results = (
    table.search_vector(
        vector=query_vector, 
        column="vector_column",
        output_name="similarity_score",
        distance_method="Cosine"
    )
    .where("publish_date > '2023-01-01'")
    .order_by("similarity_score", "desc")
    .limit(10)
    .fetchall()
)

# 根据主键查询单条记录
result = table.get_by_key(
    key_column="id",
    key_value=1,
    return_columns=["id", "content", "vector_column"]  # 可选，不指定则返回所有列
).fetchone()

# 根据主键列表批量查询
results = table.get_multi_by_keys(
    key_column="id", 
    key_values=[1, 2, 3],
    return_columns=["id", "content"]  # 可选，不指定则返回所有列
).fetchall()

关闭连接

# 关闭连接
client.disconnect()