Redis 向量检索实战：RediSearch、HNSW 与 Embedding 管道集成

# 检查 RediSearch 是否可用
redis-cli FT._LIST
# (empty array)      # 尚未建索引
redis-cli MODULE LIST | grep search
# 2) "name" 3) "search" 4) "ver" 5) "20808"

# FT.CREATE 时的 HNSW 参数
# VECTOR HNSW {参数个数} TYPE FLOAT32 DIM {维度} DISTANCE_METRIC {度量} M {边数} EF_CONSTRUCTION {建图} EF_RUNTIME {查询}

# 建索引示例：768 维、余弦距离、M=40
redis-cli FT.CREATE product_idx ON HASH PREFIX 1 product: SCHEMA \
  embedding VECTOR HNSW 6 TYPE FLOAT32 DIM 768 DISTANCE_METRIC COSINE \
  name TEXT category TAG price NUMERIC

# 1. 建索引（Hash 前缀 product:）
redis-cli FT.CREATE product_idx ON HASH PREFIX 1 product: SCHEMA \
  embedding VECTOR HNSW 6 TYPE FLOAT32 DIM 1536 DISTANCE_METRIC COSINE \
  name TEXT category TAG

# 2. 写入带向量的数据（向量是 float32 二进制，通常由程序写入）
# 程序侧: HSET product:1 embedding <1536*4字节> name "无线耳机" category electronics

# 3. 向量相似度 Top 10（DIALECT 2 必须开启）
redis-cli FT.SEARCH product_idx "*=>[KNN 10 @embedding $vec AS score]" \
  PARAMS 2 vec "<查询向量二进制>" \
  SORTBY score ASC DIALECT 2

# 4. 查看索引信息
redis-cli FT.INFO product_idx

import numpy as np
from redis import Redis
from redis.commands.search.query import Query

r = Redis(host="localhost", port=6379, decode_responses=True)

# 建索引
from redis.commands.search.field import VectorField, TextField, TagField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType

schema = [
    TextField("name"),
    TagField("category"),
    VectorField("embedding", "HNSW", {
        "TYPE": "FLOAT32", "DIM": 1536,
        "DISTANCE_METRIC": "COSINE", "M": 40,
        "EF_CONSTRUCTION": 200,
    }),
]
r.ft("product_idx").create_index(
    schema, definition=IndexDefinition(prefix=["product:"], index_type=IndexType.HASH))

# 写入向量（float32 little-endian 字节）
vec = np.random.rand(1536).astype(np.float32)
r.hset("product:1", mapping={"name": "无线耳机", "category": "electronics",
                             "embedding": vec.tobytes()})

# KNN 查询
q = Query("*=>[KNN 10 @embedding $vec AS score]") \
    .params({"vec": query_vec.tobytes()}) \
    .sort_by("score") \
    .return_fields("name", "score") \
    .dialect(2)
for doc in r.ft("product_idx").search(q).docs:
    print(doc.name, doc.score)

from openai import OpenAI
import numpy as np

client = OpenAI()

def embed_openai(text: str) -> bytes:
    resp = client.embeddings.create(
        model="text-embedding-3-small", input=text)
    vec = np.array(resp.data[0].embedding, dtype=np.float32)
    return vec.tobytes()

# 写入
r.hset("doc:1", mapping={"content": "Redis 向量检索实战",
                         "embedding": embed_openai("Redis 向量检索实战")})

# 查询
q_vec = embed_openai("如何使用 Redis 做语义搜索")

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-m3")   # 本地加载，离线可用

def embed_local(text: str) -> bytes:
    vec = model.encode(text, normalize_embeddings=True)
    return vec.astype(np.float32).tobytes()

# 建索引：带 TAG（category）与 NUMERIC（price）字段
redis-cli FT.CREATE product_idx ON HASH PREFIX 1 product: SCHEMA \
  embedding VECTOR HNSW 6 TYPE FLOAT32 DIM 1536 DISTANCE_METRIC COSINE \
  name TEXT category TAG price NUMERIC brand TAG

# 混合查询：先按标签/数值过滤，再在结果内做向量相似度 Top 10
redis-cli FT.SEARCH product_idx \
  "@category:{electronics} @price:[100 1000] =>[KNN 10 @embedding $vec AS score]" \
  PARAMS 2 vec "<向量>" SORTBY score ASC DIALECT 2

from redis.commands.search.query import Query

q = (Query("@category:{electronics} @price:[100 1000] "
           "=>[KNN 10 @embedding $vec AS score]")
     .params({"vec": q_vec})
     .sort_by("score")
     .return_fields("name", "price", "score")
     .dialect(2))

for doc in r.ft("product_idx").search(q).docs:
    print(doc.name, doc.price, doc.score)

# 以商品 10086 的向量为查询向量，找相似商品 Top 10
redis-cli FT.SEARCH product_idx "*=>[KNN 10 @embedding $vec AS score]" \
  PARAMS 2 vec "$(redis-cli HGET product:10086 embedding)" \
  SORTBY score ASC DIALECT 2

# 在线推荐：读一次源商品向量，复用多次查询
src_vec = r.hget("product:10086", "embedding")
q = (Query("*=>[KNN 10 @embedding $vec AS score]")
     .params({"vec": src_vec}).sort_by("score").dialect(2))

# 思路: 每条新内容 -> 向量 -> KNN 查询 -> 若 Top1 距离 < 阈值 判定重复
# 阈值经验值: COSINE 下 score < 0.15 视为高度相似（需按业务标定）
redis-cli FT.SEARCH post_idx "*=>[KNN 1 @embedding $vec AS score]" \
  PARAMS 2 vec "<新帖向量>" SORTBY score ASC DIALECT 2

def is_duplicate(text: str, threshold: float = 0.15) -> bool:
    vec = embed(text)
    q = Query("*=>[KNN 1 @embedding $vec AS score]") \
        .params({"vec": vec}).sort_by("score").dialect(2)
    res = r.ft("post_idx").search(q)
    return bool(res.docs) and float(res.docs[0].score) < threshold

# 观察向量索引内存
redis-cli FT.INFO product_idx | grep -E "indexing|hash_indexing_failures|total_indexing_time"
# 查看 Redis 内存，评估容量
redis-cli INFO memory | grep used_memory_human

# 用 redis-benchmark 测 KNN 查询吞吐（DIALECT 2）
redis-benchmark -h 127.0.0.1 -p 6379 -n 10000 -c 50 \
  -q -P 1 evalsha <搜索脚本> ...
# 更实用的是自写 Python 并发脚本测 p99 延迟

# 重建索引流程（模型升级/维度变化时）
redis-cli FT.DROPINDEX product_idx          # 删除索引（数据仍在）
# 重新 FT.CREATE 建新索引
redis-cli FT.CREATE product_idx ON HASH PREFIX 1 product: SCHEMA \
  embedding VECTOR HNSW 6 TYPE FLOAT32 DIM 1536 DISTANCE_METRIC COSINE \
  name TEXT category TAG price NUMERIC
# 全量重算向量并 HSET，或用 SCAN 遍历已有数据补充

# 给查询设置超时（毫秒），避免拖累其他命令
redis-cli CONFIG SET search-timeout 500