feat(vector): pluggable multimodal vector space

核心暴露 MultimodalEmbedder 接口,两条路径共享同一套 L0/L2/L3
向量缓存、media.Store 坐标、QueryMemoryMediaScored 检索:
  - onnx:内嵌 ONNX 模型(CLIP 等),通过 build tag 编译
  - http:外部向量 API 服务(Jina v5 / OpenAI / 自建)

跨模态融合权重改为 CrossModalFusionConfig 可配置结构体,
移除所有模型特定硬编码(CLIP/Jina),版本切换只需改配置。

模型切换自动迁移:
  - StaleVecDigestsAll 支持全模态(image+audio+video)
  - 启动时并发重算(ONNX 4 workers / API 8 workers)
  - 修复 SQL 运算符优先级导致 kind 过滤失效的 bug

实测对比(492 篇生产文档 + 3 张真实图片):
  - TF-IDF:MRR 0.457(精确匹配快,语义差)
  - fastText:MRR 0.530(语义中等,延迟 8ms)
  - Jina v5-omni:MRR 0.900(全面领先,延迟 40ms)
  - 中文文本→图片:Jina MRR 0.833 vs CLIP 0.611

See docs/embedding-comparison.md for full benchmark.
This commit is contained in:
JianFeeeee
2026-09-09 17:38:34 +08:00
parent fba85be396
commit e02a672772
15 changed files with 1480 additions and 94 deletions

View File

@ -427,7 +427,8 @@ func (s *Store) Search(query string, kind Kind, limit int) ([]*Item, error) {
defer s.mu.RUnlock()
q := `SELECT digest, kind, mime, size, width, height, origin_path, tool,
description, described_by, ref_count, first_seen, last_seen
description, described_by, ref_count, first_seen, last_seen,
vec, vec_model
FROM media WHERE COALESCE(description,'') != ''`
args := []interface{}{}
if strings.TrimSpace(query) != "" {
@ -473,7 +474,8 @@ func (s *Store) Pending(limit int) ([]*Item, error) {
defer s.mu.RUnlock()
rows, err := s.db.Query(`
SELECT digest, kind, mime, size, width, height, origin_path, tool,
description, described_by, ref_count, first_seen, last_seen
description, described_by, ref_count, first_seen, last_seen,
vec, vec_model
FROM media
WHERE COALESCE(description,'') = '' AND COALESCE(described_by,'') = ''
ORDER BY last_seen DESC LIMIT ?`, limit)
@ -650,15 +652,36 @@ func (s *Store) SetVec(digest string, vec []float64, model string) error {
// vec_model 不等于 currentModel(模型切换)或 vec_model 为空(从未嵌入)。
// 调用方使用返回的 digest 列表调用 Get/EmbedImage/SetVec 完成重算。
func (s *Store) StaleVecDigests(currentModel string) ([]string, error) {
return s.staleVecDigests(currentModel, "image")
}
// StaleVecDigestsAll 返回所有需要重新嵌入的媒体 digest(不限 kind),
// 供模型切换后全量迁移向量空间(image + audio + video 等)。
func (s *Store) StaleVecDigestsAll(currentModel string) ([]string, error) {
return s.staleVecDigests(currentModel, "")
}
// staleVecDigests 是 StaleVecDigests 的核心实现,kind=” 时不按 kind 过滤。
// 废弃了"只迁移图片"的限定:模型切换后所有模态都应迁移到新向量空间。
func (s *Store) staleVecDigests(currentModel string, kind string) ([]string, error) {
s.mu.RLock()
defer s.mu.RUnlock()
rows, err := s.db.Query(`
query := `
SELECT digest FROM media
WHERE kind = 'image'
AND COALESCE(description,'') != ''
AND (COALESCE(vec_model,'') = '' OR vec_model != ?)
ORDER BY last_seen`, currentModel)
WHERE (COALESCE(vec_model,'') = '' OR vec_model != ?)`
if kind != "" {
query += ` AND kind = ?`
}
query += ` ORDER BY last_seen`
var args []interface{}
args = append(args, currentModel)
if kind != "" {
args = append(args, kind)
}
rows, err := s.db.Query(query, args...)
if err != nil {
return nil, err
}
@ -681,6 +704,46 @@ func (s *Store) StaleVecDigests(currentModel string) ([]string, error) {
// 谁的相似度更高就召回谁——不再区分「这是一张图的查询」还是「这是一段文字的查询」,
// 由向量空间的相似度自动判断。
func (s *Store) QueryMedia(queryVec []float64, model string, topK int) ([]*Item, error) {
hits, err := s.QueryMediaScored(queryVec, model, topK)
if err != nil {
return nil, err
}
if hits == nil {
return nil, nil
}
out := make([]*Item, len(hits))
for i, h := range hits {
out[i] = h.Item
}
return out, nil
}
// MediaHit 是一条媒体相似度候选及其分数。
// 跨模态融合需要原始分数做归一化,仅返回 Item 会丢掉尺度信息。
type MediaHit struct {
Item *Item
Score float64
}
// QueryMemoryMediaScored 只检索当前仍被 L0/L2/L3 记忆块引用的媒体。
// CAS 中 ref_count=0 的项是等待 GC 的孤儿缓存,不是可召回记忆;若把它们也查出,
// 已从三层记忆淘汰的图片会被视觉路“复活”,破坏与文本块一致的生命周期。
//
// 分数只做排序,不在存储层设绝对阈值:多模态文本→图像的绝对 cosine 随模型、
// 语言与数据域漂移,真实标定中有效命中可以低至 0.015。相关性门控在融合器中
// 使用当前候选集合的相对分布完成。
func (s *Store) QueryMemoryMediaScored(queryVec []float64, model string, topK int) ([]MediaHit, error) {
return s.queryMediaScored(queryVec, model, topK, true)
}
// QueryMediaScored 用查询向量对所有已嵌入媒体做余弦相似度检索,
// 返回 topK 个最相似的候选及其原始 cosine 分数(供跨模态归一化)。
// 这是媒体存储层的诊断/显式全库入口;记忆召回应调用 QueryMemoryMediaScored。
func (s *Store) QueryMediaScored(queryVec []float64, model string, topK int) ([]MediaHit, error) {
return s.queryMediaScored(queryVec, model, topK, false)
}
func (s *Store) queryMediaScored(queryVec []float64, model string, topK int, referencedOnly bool) ([]MediaHit, error) {
if topK <= 0 {
topK = 20
}
@ -690,10 +753,21 @@ func (s *Store) QueryMedia(queryVec []float64, model string, topK int) ([]*Item,
s.mu.RLock()
defer s.mu.RUnlock()
rows, err := s.db.Query(`SELECT digest, kind, mime, size, width, height,
query := `SELECT digest, kind, mime, size, width, height,
origin_path, tool, description, described_by, ref_count, first_seen, last_seen,
vec, vec_model
FROM media WHERE vec IS NOT NULL AND vec != ''`)
FROM media WHERE vec IS NOT NULL AND vec != ''`
var args []interface{}
if model != "" {
query += ` AND vec_model = ?`
args = append(args, model)
}
if referencedOnly {
query += ` AND ref_count > 0 AND EXISTS (
SELECT 1 FROM media_refs r WHERE r.digest = media.digest
)`
}
rows, err := s.db.Query(query, args...)
if err != nil {
return nil, err
}
@ -744,9 +818,9 @@ func (s *Store) QueryMedia(queryVec []float64, model string, topK int) ([]*Item,
if len(candidates) > topK {
candidates = candidates[:topK]
}
out := make([]*Item, len(candidates))
out := make([]MediaHit, len(candidates))
for i, c := range candidates {
out[i] = c.item
out[i] = MediaHit{Item: c.item, Score: c.score}
}
return out, nil
}

View File

@ -149,25 +149,26 @@ func TestStaleVecDigests(t *testing.T) {
// 有描述但从未嵌入(vec_model 空)→ stale
d3, _ := s.Put([]byte("img3"), Item{MIME: "image/png", Description: "图三"})
// 无描述 → 不参与(描述流程外)
s.Put([]byte("img4"), Item{MIME: "image/png"})
// 无描述但有图片 → 也应被迁移(描述是可选语义通道,图片应独立于描述参与向量空间)
d4, _ := s.Put([]byte("img4"), Item{MIME: "image/png"})
// 音频不属于图片 → 不算 stale
// 音频不参与图片迁移(StaleVecDigests 只查 kind='image')
s.Put([]byte("aud1"), Item{MIME: "audio/wav", Description: "语音"})
stale, err := s.StaleVecDigests("clip-vit-b32")
if err != nil {
t.Fatal(err)
}
if len(stale) != 2 {
t.Fatalf("expected 2 stale digests (d2 旧模型 + d3 未嵌入), got %d: %v", len(stale), stale)
// d1 匹配模型 → 非 stale;d2 旧模型 + d3 未嵌入 + d4 无描述图片 = 3 stale;aud1 不算
if len(stale) != 3 {
t.Fatalf("expected 3 stale digests (d2 旧模型 + d3 未嵌入 + d4 无描述), got %d: %v", len(stale), stale)
}
got := map[string]bool{}
for _, d := range stale {
got[d] = true
}
if !got[d2] || !got[d3] {
t.Errorf("expected d2 and d3 stale, got %v", stale)
if !got[d2] || !got[d3] || !got[d4] {
t.Errorf("expected d2, d3, d4 stale, got %v", stale)
}
if got[d1] {
t.Errorf("d1 (匹配模型) 不应 stale")

View File

@ -0,0 +1,131 @@
package vector
import (
"bytes"
"encoding/base64"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
"sync"
"time"
)
// HTTPEmbedderConfig 配置一个外部多模态向量服务。
// 服务契约刻意很小:POST Endpoint,输入 modality/data/mime/side,返回 embedding。
// 任何云 API 或自建服务只需适配这一个协议,即可复用内核全部向量存储与检索链路。
type HTTPEmbedderConfig struct {
Endpoint string
APIKey string
Model string
Dimension int
Timeout time.Duration
Fingerprint string
}
// HTTPEmbedder 是 MultimodalEmbedder 的外部 API 实现。
type HTTPEmbedder struct {
cfg HTTPEmbedderConfig
client *http.Client
mu sync.Mutex
closed bool
}
type httpEmbedRequest struct {
Model string `json:"model,omitempty"`
Modality string `json:"modality"`
Side string `json:"side"`
Text string `json:"text,omitempty"`
Data string `json:"data,omitempty"`
MIME string `json:"mime,omitempty"`
}
type httpEmbedResponse struct {
Embedding []float64 `json:"embedding"`
Data []struct {
Embedding []float64 `json:"embedding"`
} `json:"data,omitempty"`
}
func NewHTTPEmbedder(cfg HTTPEmbedderConfig) (*HTTPEmbedder, error) {
if strings.TrimSpace(cfg.Endpoint) == "" {
return nil, fmt.Errorf("vector: empty HTTP embedding endpoint")
}
if cfg.Dimension <= 0 {
return nil, fmt.Errorf("vector: invalid HTTP embedding dimension %d", cfg.Dimension)
}
if cfg.Timeout <= 0 {
cfg.Timeout = 30 * time.Second
}
if cfg.Fingerprint == "" {
cfg.Fingerprint = "http:" + cfg.Model + fmt.Sprintf(":%d", cfg.Dimension)
}
return &HTTPEmbedder{cfg: cfg, client: &http.Client{Timeout: cfg.Timeout}}, nil
}
func (e *HTTPEmbedder) VectorizeDense(text string) ([]float64, error) {
return e.embed(httpEmbedRequest{Model: e.cfg.Model, Modality: string(ModalityText), Side: "query", Text: text})
}
func (e *HTTPEmbedder) EmbedImageDense(img []byte, mime string) ([]float64, error) {
return e.embed(httpEmbedRequest{Model: e.cfg.Model, Modality: string(ModalityImage), Side: "document", Data: base64.StdEncoding.EncodeToString(img), MIME: mime})
}
func (e *HTTPEmbedder) embed(payload httpEmbedRequest) ([]float64, error) {
e.mu.Lock()
closed := e.closed
e.mu.Unlock()
if closed {
return nil, fmt.Errorf("vector: HTTP embedder closed")
}
body, err := json.Marshal(payload)
if err != nil {
return nil, err
}
req, err := http.NewRequest(http.MethodPost, e.cfg.Endpoint, bytes.NewReader(body))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/json")
if e.cfg.APIKey != "" {
req.Header.Set("Authorization", "Bearer "+e.cfg.APIKey)
}
resp, err := e.client.Do(req)
if err != nil {
return nil, fmt.Errorf("vector: HTTP embedding request: %w", err)
}
defer resp.Body.Close()
b, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
if err != nil {
return nil, err
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return nil, fmt.Errorf("vector: HTTP embedding status %d: %s", resp.StatusCode, strings.TrimSpace(string(b)))
}
var out httpEmbedResponse
if err := json.Unmarshal(b, &out); err != nil {
return nil, fmt.Errorf("vector: decode HTTP embedding: %w", err)
}
v := out.Embedding
if len(v) == 0 && len(out.Data) > 0 {
v = out.Data[0].Embedding
}
if len(v) != e.cfg.Dimension {
return nil, fmt.Errorf("vector: HTTP embedding dimension %d, want %d", len(v), e.cfg.Dimension)
}
return v, nil
}
func (e *HTTPEmbedder) Fingerprint() string { return e.cfg.Fingerprint }
func (e *HTTPEmbedder) Dim() int { return e.cfg.Dimension }
func (e *HTTPEmbedder) Loaded() bool {
e.mu.Lock()
defer e.mu.Unlock()
return !e.closed
}
func (e *HTTPEmbedder) Close() {
e.mu.Lock()
e.closed = true
e.mu.Unlock()
}

View File

@ -0,0 +1,178 @@
package vector
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
)
func TestHTTPEmbedder_RequiresEndpoint(t *testing.T) {
_, err := NewHTTPEmbedder(HTTPEmbedderConfig{Dimension: 512})
if err == nil {
t.Fatal("应拒绝空 endpoint")
}
}
func TestHTTPEmbedder_RequiresDimension(t *testing.T) {
_, err := NewHTTPEmbedder(HTTPEmbedderConfig{Endpoint: "http://localhost"})
if err == nil {
t.Fatal("应拒绝 dimension<=0")
}
}
func TestHTTPEmbedder_TextEmbedding(t *testing.T) {
// 模拟返回 4 维向量的外部服务
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
t.Errorf("期望 POST,实际 %s", r.Method)
}
var req httpEmbedRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
t.Fatal(err)
}
if req.Modality != "text" {
t.Errorf("期望 modality=text,实际 %s", req.Modality)
}
if req.Text == "" {
t.Fatal("text 不应为空")
}
w.Header().Set("Content-Type", "application/json")
w.Write([]byte(`{"embedding":[0.1,0.2,0.3,0.4]}`))
}))
defer srv.Close()
e, err := NewHTTPEmbedder(HTTPEmbedderConfig{
Endpoint: srv.URL,
Dimension: 4,
Model: "test-model",
})
if err != nil {
t.Fatal(err)
}
defer e.Close()
if !e.Loaded() {
t.Fatal("应处于 loaded 状态")
}
vec, err := e.VectorizeDense("hello world")
if err != nil {
t.Fatal(err)
}
if len(vec) != 4 || vec[0] != 0.1 || vec[3] != 0.4 {
t.Errorf("向量不符合预期: %v", vec)
}
if e.Fingerprint() != "http:test-model:4" {
t.Errorf("指纹不符合预期: %s", e.Fingerprint())
}
}
func TestHTTPEmbedder_ImageEmbedding(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var req httpEmbedRequest
json.NewDecoder(r.Body).Decode(&req)
if req.Modality != "image" {
t.Errorf("期望 modality=image,实际 %s", req.Modality)
}
if req.MIME != "image/png" {
t.Errorf("期望 mime=image/png,实际 %s", req.MIME)
}
w.Write([]byte(`{"embedding":[0.5,0.5,0.5]}`))
}))
defer srv.Close()
e, err := NewHTTPEmbedder(HTTPEmbedderConfig{
Endpoint: srv.URL,
Dimension: 3,
Model: "img-model",
})
if err != nil {
t.Fatal(err)
}
defer e.Close()
vec, err := e.EmbedImageDense([]byte("fake-png-data"), "image/png")
if err != nil {
t.Fatal(err)
}
if len(vec) != 3 {
t.Errorf("期望 3 维,实际 %d", len(vec))
}
}
func TestHTTPEmbedder_CustomFingerprint(t *testing.T) {
e, err := NewHTTPEmbedder(HTTPEmbedderConfig{
Endpoint: "http://localhost:1234",
Dimension: 512,
Fingerprint: "jina-v5-omni-nano:2026",
})
if err != nil {
t.Fatal(err)
}
defer e.Close()
if e.Fingerprint() != "jina-v5-omni-nano:2026" {
t.Errorf("自定义指纹未生效: %s", e.Fingerprint())
}
}
func TestHTTPEmbedder_DimensionMismatchReturnsError(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte(`{"embedding":[1,2]}`)) // 返回 2 维,配置期望 4
}))
defer srv.Close()
e, err := NewHTTPEmbedder(HTTPEmbedderConfig{
Endpoint: srv.URL,
Dimension: 4,
})
if err != nil {
t.Fatal(err)
}
defer e.Close()
_, err = e.VectorizeDense("test")
if err == nil {
t.Fatal("维度不匹配时应返回错误")
}
}
func TestHTTPEmbedder_ServerErrorReturnsError(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusBadGateway)
w.Write([]byte("gateway down"))
}))
defer srv.Close()
e, err := NewHTTPEmbedder(HTTPEmbedderConfig{
Endpoint: srv.URL,
Dimension: 4,
})
if err != nil {
t.Fatal(err)
}
defer e.Close()
_, err = e.VectorizeDense("test")
if err == nil {
t.Fatal("服务端错误时应返回错误")
}
}
func TestHTTPEmbedder_ClosePreventsFurtherCalls(t *testing.T) {
e, err := NewHTTPEmbedder(HTTPEmbedderConfig{
Endpoint: "http://localhost:1234",
Dimension: 4,
})
if err != nil {
t.Fatal(err)
}
e.Close()
if e.Loaded() {
t.Fatal("关闭后 Loaded() 应返回 false")
}
_, err = e.VectorizeDense("test")
if err == nil {
t.Fatal("关闭后应返回错误")
}
}

View File

@ -17,11 +17,15 @@ type Vectorizer interface {
EmbedImage(img []byte, mime string) (Vector, error)
}
// MultimodalEmbedder 是稠密多模态编码器的接口(CLIP 等视觉-文本联合模型)。
// MultimodalEmbedder 是稠密多模态编码器的接口。
//
// 与 Vectorizer(稀疏词向量,供 TF-IDF/倒排检索)刻意区分:多模态模型产出的
// 是共享稠密空间(如 CLIP 512 维),直接用于 media.Store 的稠密余弦检索,
// 是共享稠密空间,直接用于 media.Store 的稠密余弦检索,
// **不得**塞进文档/知识层的稀疏 vector.Store(会破坏倒排剪枝与 TF-IDF 语义)。
//
// 实现不限:可以是内嵌 ONNX,也可以是外部 HTTP 向量服务——
// 内核只依赖本接口,两条路径共享同一套检索/存储基础设施。Fingerprint 是模型
// 空间标识(如模型文件指纹),作为 vec_model 持久化用于切换后重算。
type MultimodalEmbedder interface {
VectorizeDense(text string) ([]float64, error)
EmbedImageDense(img []byte, mime string) ([]float64, error)
@ -31,6 +35,18 @@ type MultimodalEmbedder interface {
Close()
}
// MultimodalModality 是统一向量空间支持的输入模态。
// 现内核只消费 text/image;外部 API 路径可能扩展 audio/video,
// 通过类型断言在接口外按需扩展,不破坏现有契约。
type MultimodalModality string
const (
ModalityText MultimodalModality = "text"
ModalityImage MultimodalModality = "image"
ModalityAudio MultimodalModality = "audio"
ModalityVideo MultimodalModality = "video"
)
// ErrNotSupported 表示 Vectorizer 不支持图像嵌入,调用方按文本描述降级。
var ErrNotSupported = fmt.Errorf("vectorizer does not support image embedding")
@ -83,6 +99,26 @@ func (s *Store) Remove(id string) {
}
func (s *Store) Search(query Vector, topK int) []DocVector {
hits := s.SearchScored(query, topK)
if len(hits) == 0 {
return nil
}
out := make([]DocVector, len(hits))
for i, h := range hits {
out[i] = h.Doc
}
return out
}
// DocVectorHit 是一篇文档的相似度候选及其原始 cosine 分数。
// 跨模态融合需要分数做归一化;纯排序的 Search 不暴露它。
type DocVectorHit struct {
Doc DocVector
Score float64
}
// SearchScored 与 Search 同语义,但返回带原始 cosine 分数的候选。
func (s *Store) SearchScored(query Vector, topK int) []DocVectorHit {
s.mu.RLock()
defer s.mu.RUnlock()
@ -122,9 +158,9 @@ func (s *Store) Search(query Vector, topK int) []DocVector {
results = results[:topK]
}
out := make([]DocVector, len(results))
out := make([]DocVectorHit, len(results))
for i, r := range results {
out[i] = r.doc
out[i] = DocVectorHit{Doc: r.doc, Score: r.score}
}
return out
}