通过rqlite sdk 快速访问sqlite-vec

sqlite-vec 的的轻量化以及标准sqlite 扩展的模式对于需要使用向量存储的场景（比如相似比较的），但是使用上可能并不是很友好（从轻量化来说够用了，但是我们很多时候还是需要server 以及轻量能力，同时将扩展隐藏起来，方便维护），rqlite 是一个不错的基于sqlite 但是支持分布式特性的db 平台

参考玩法

简单说: 通过rqlite server 直接加载sqlite-vec 扩展，之后启动通过rqlite 提供的sql 直接基于sql 模式方法

示例代码

启动服务

sudo rqlited -extensions-path=sqlite-vec.tar.gz -http-addr=127.0.0.1:8080  data

代码访问

直接使用rqlite 提供的python sdk pyrqlite

import pyrqlite.dbapi2 as dbapi2
import numpy as np
import uuid
from typing import List
from struct import pack
def serialize_float32(vector: List[float]) -> bytes:"""Serializes a list of floats into the "raw bytes" format sqlite-vec expects"""return pack("%sf" % len(vector), *vector)
# Connect to the database
connection = dbapi2.connect(host='localhost',port=8080,
)
is_first = True
first_embedding = np.random.rand(3).astype(np.float32)
test = [1.0,2.0,4.0]
try:with connection.cursor() as cursor:cursor.execute('create virtual table if not exists dalongrong_vec using vec0 (id text PRIMARY KEY, vector float[3] distance_metric=cosine, user_id text, type text, version text)')for i in range(20,30):if  is_first:cursor.execute("insert into dalongrong_vec (id, vector, user_id, type, version) values (?,?, ?, ?, ?)", (str(uuid.uuid4()), serialize_float32(first_embedding), f"user_{i}", "text", "v1"))print(first_embedding)is_first = Falseelse:embedding = np.random.rand(3).astype(np.float32)print(embedding)cursor.execute("insert into dalongrong_vec (id, vector, user_id, type, version) values (?,?, ?, ?, ?)", (str(uuid.uuid4()), serialize_float32(embedding), f"user_{i}", "text", "v1"))with connection.cursor() as cursor:search_result = cursor.execute("select id, user_id,vector, ROUND(vec_distance_cosine(vector,?), 7) distance from dalongrong_vec where vector match ? and type = ? and version = ?  and k = 10 order by distance", (serialize_float32(test),serialize_float32(test),"text","v1")).fetchall()for row in search_result:print(row)
finally:connection.close()