diff --git a/cmd/storage-migrate/main.go b/cmd/storage-migrate/main.go index de17ca4..54bb89f 100644 --- a/cmd/storage-migrate/main.go +++ b/cmd/storage-migrate/main.go @@ -530,11 +530,13 @@ func migrateTable(db *sql.DB, src *sql.DB, migrationID, table string) (map[strin // targetColumns：目标侧列清单（按名排序）+ 主键列。布尔列集合同取自目标 // information_schema（类型映射以 PG schema 为准——SQLite 侧 0/1 → bool）。 func targetColumns(db *sql.DB, table string) (cols []string, pk []string, err error) { - // 生成列（PG GENERATED ALWAYS，如 knowledge.tsv）不复制不校验：PG 从 - // 业务列自动派生，源库也没有该列。 + // 生成列（PG GENERATED ALWAYS，如 knowledge.tsv）与库内向量副本列 + // （vec_pg：BLOB 权威的派生表示，U6-3）不复制不校验：PG 从业务列 + // 派生或由 Upsert 双写，源库（SQLite）没有该列。 rows, err := db.Query(` SELECT column_name, data_type FROM information_schema.columns - WHERE table_schema=current_schema() AND table_name=$1 AND is_generated='NEVER' ORDER BY column_name`, table) + WHERE table_schema=current_schema() AND table_name=$1 AND is_generated='NEVER' + AND column_name NOT IN ('vec_pg') ORDER BY column_name`, table) if err != nil { return nil, nil, err } diff --git a/internal/rag/embed.go b/internal/rag/embed.go index 0c5fef1..2b54fc6 100644 --- a/internal/rag/embed.go +++ b/internal/rag/embed.go @@ -48,6 +48,12 @@ type embeddingsResponse struct { } `json:"error,omitempty"` } +// Model 当前 embedding 模型身份（U6-3：写入与检索同源；未配置返回空）。 +func (e *HTTPEmbedder) Model() string { + route := e.resolve(context.Background()) + return route.Model +} + func (e *HTTPEmbedder) Embed(ctx context.Context, texts []string) ([][]float32, error) { if len(texts) == 0 { return nil, nil diff --git a/internal/rag/hash_embed.go b/internal/rag/hash_embed.go index 2e14c70..2a3f91f 100644 --- a/internal/rag/hash_embed.go +++ b/internal/rag/hash_embed.go @@ -13,6 +13,9 @@ const hashDim = 32 // 由 settings models.embedding.provider=hash 显式打开；未设置时仍按 D23 退化为全文。 type HashEmbedder struct{} +// Model 本地哈希嵌入的身份（U6-3：同源写入/检索，禁止与真实模型混查）。 +func (HashEmbedder) Model() string { return "hash:" + fmt.Sprint(hashDim) } + func (HashEmbedder) Embed(_ context.Context, texts []string) ([][]float32, error) { out := make([][]float32, len(texts)) for i, text := range texts { diff --git a/internal/rag/hybrid.go b/internal/rag/hybrid.go index 542fe08..ba4c976 100644 --- a/internal/rag/hybrid.go +++ b/internal/rag/hybrid.go @@ -2,6 +2,8 @@ package rag import ( "context" + "crypto/sha256" + "encoding/hex" "log" "agent-system-backend/internal/settings" @@ -104,17 +106,33 @@ func (p *Pipeline) IndexChunks(ctx context.Context, owner core.ID, docs []core.K log.Printf("[rag] embedding 降级为纯全文（ref=%s）: %v", firstRef(docs), err) return nil } + model := embedderModel(p.embedder) for i, v := range vecs { if len(v) == 0 { continue } - if err := p.st.UpsertKnowledgeVector(ctx, owner, docs[i].ID, docs[i].Ref, v); err != nil { + if err := p.st.UpsertKnowledgeVectorExt(ctx, owner, docs[i].ID, docs[i].Ref, model, contentHash(docs[i].Chunk), v); err != nil { return err } } return nil } +// embedderModel 模型身份（可选接口：EmbeddingProvider 契约未含，实现侧 +// 声明；未声明=空=legacy 无身份行，向后兼容）。 +func embedderModel(e core.EmbeddingProvider) string { + if m, ok := e.(interface{ Model() string }); ok { + return m.Model() + } + return "" +} + +// contentHash 分块内容摘要（写入幂等核对用；sha256 前 16 字节十六进制）。 +func contentHash(chunk string) string { + sum := sha256.Sum256([]byte(chunk)) + return hex.EncodeToString(sum[:])[:16] +} + func firstRef(docs []core.KnowledgeDoc) string { for _, d := range docs { if d.Ref != "" { @@ -143,7 +161,7 @@ func (p *Pipeline) RetrieveShared(ctx context.Context, owner core.ID, q core.Ret if topK <= 0 { topK = 5 } - vs, e2 := p.st.SearchKnowledgeVectorsShared(ctx, string(owner), qvecs[0], topK, q, allow) + vs, e2 := p.st.SearchKnowledgeVectorsSharedModel(ctx, string(owner), embedderModel(p.embedder), qvecs[0], topK, q, allow) if e2 == nil { shared = rrfMerge(vs, shared) } @@ -179,7 +197,7 @@ func (p *Pipeline) Retrieve(ctx context.Context, owner core.ID, q core.Retrieval return p.st.SearchKnowledge(owner, q) } q.TopK = topK * 2 // 每路多取，融合后再裁剪 - vhits, err := p.st.SearchKnowledgeVectors(ctx, owner, qvecs[0], topK*2, q) + vhits, err := p.st.SearchKnowledgeVectorsModel(ctx, owner, embedderModel(p.embedder), qvecs[0], topK*2, q) if err != nil { return nil, err } diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 806abd8..51959bd 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -688,6 +688,30 @@ func init() { }, "postgres": { `DROP TABLE IF EXISTS surface_actions`, }}}, + migration{version: 25, name: "vector_identity_pgvector", + // 协作存储 U6 切片三（KTD11：正确→测量→ANN 的「正确」步）： + // 向量行补模型身份（model/content_hash，禁止同维不同模型混查）； + // PG 侧启用 pgvector 扩展并加 vec_pg 列做库内精确 top-k。 + // 扩展缺失不阻断迁移（DO 块吞错）——检索自动降级进程内扫描。 + up: map[string][]string{"sqlite": { + `ALTER TABLE knowledge_vectors ADD COLUMN model TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE knowledge_vectors ADD COLUMN content_hash TEXT NOT NULL DEFAULT ''`, + }, "postgres": { + `ALTER TABLE knowledge_vectors ADD COLUMN model TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE knowledge_vectors ADD COLUMN content_hash TEXT NOT NULL DEFAULT ''`, + `DO $$ BEGIN + CREATE EXTENSION IF NOT EXISTS vector; + ALTER TABLE knowledge_vectors ADD COLUMN IF NOT EXISTS vec_pg vector; + EXCEPTION WHEN OTHERS THEN NULL; END $$`, + }}, + down: map[string][]string{"sqlite": { + `ALTER TABLE knowledge_vectors DROP COLUMN content_hash`, + `ALTER TABLE knowledge_vectors DROP COLUMN model`, + }, "postgres": { + `ALTER TABLE knowledge_vectors DROP COLUMN IF EXISTS vec_pg`, + `ALTER TABLE knowledge_vectors DROP COLUMN content_hash`, + `ALTER TABLE knowledge_vectors DROP COLUMN model`, + }}}, ) } diff --git a/internal/store/pgvector_test.go b/internal/store/pgvector_test.go new file mode 100644 index 0000000..00a8809 --- /dev/null +++ b/internal/store/pgvector_test.go @@ -0,0 +1,185 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + + core "agent-system-core" +) + +// 协作存储 U6 切片三（KTD11「正确」步）：库内向量检索。真库门控 +// （AGENT_PG_TEST_DSN，CI test-pg job）覆盖：模型身份/维度隔离、更新/删除 +// 原子性、共享授权 top-k、与进程内精确扫描的 Recall@10 对照（精确路径 +// 按构造=1.0；HNSW 的近似召回对照在 U7 测量后按实测决定）。 + +func vectorSchemaStore(t *testing.T) *Store { + t.Helper() + dsn := os.Getenv("AGENT_PG_TEST_DSN") + if dsn == "" { + t.Skip("AGENT_PG_TEST_DSN 未设置：真库向量检索由 CI 必跑 job 覆盖") + } + schema := "vec_" + strings.ToLower(NewID("t")) + "_" + fmt.Sprint(os.Getpid()) + admin, err := sql.Open("pgx", dsn) + if err != nil { + t.Fatal(err) + } + if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil { + admin.Close() + t.Fatalf("pg_trgm: %v", err) + } + if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS vector`); err != nil { + admin.Close() + t.Skipf("pgvector 扩展不可用（正确降级路径由其余测试覆盖）：%v", err) + } + if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil { + admin.Close() + t.Fatalf("schema: %v", err) + } + t.Cleanup(func() { + admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE") + admin.Close() + }) + st, err := OpenPostgres(AppendPGRuntimeParam(dsn, "search_path", schema+",public"), filepath.Join(t.TempDir(), "art")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + return st +} + +func mustDoc(t *testing.T, st *Store, owner, id, chunk string) { + t.Helper() + if _, err := st.DB().Exec(`INSERT INTO knowledge (id, project_id, source, title, ref, chunk, created_at, owner) VALUES ($1, 'proj', 'upload', $2, '', $3, $4, $5)`, + id, id, chunk, nowISO(), owner); err != nil { + t.Fatal(err) + } +} + +func TestVectorModelIdentityIsolation(t *testing.T) { + st := vectorSchemaStore(t) + ctx := context.Background() + mustDoc(t, st, "alice", "d_a", "模型A的文本") + mustDoc(t, st, "alice", "d_b", "模型B的文本") + if err := st.UpsertKnowledgeVectorExt(ctx, "alice", "d_a", "", "modelA", "h1", []float32{1, 0, 0}); err != nil { + t.Fatal(err) + } + if err := st.UpsertKnowledgeVectorExt(ctx, "alice", "d_b", "", "modelB", "h2", []float32{1, 0, 0}); err != nil { + t.Fatal(err) + } + // 同维不同模型不得混查：查 modelA 只见 d_a。 + hits, err := st.SearchKnowledgeVectorsModel(ctx, "alice", "modelA", []float32{1, 0, 0}, 5, core.RetrievalQuery{}) + if err != nil { + t.Fatal(err) + } + if len(hits) != 1 || hits[0].Doc.ID != "d_a" { + t.Fatalf("modelA 应只见 d_a：%v", hits) + } + // 维度隔离：4 维查询不命中 3 维行。 + hits, err = st.SearchKnowledgeVectorsModel(ctx, "alice", "modelA", []float32{1, 0, 0, 0}, 5, core.RetrievalQuery{}) + if err != nil { + t.Fatal(err) + } + if len(hits) != 0 { + t.Fatalf("维度不齐不应命中：%v", hits) + } +} + +func TestVectorUpdateDeleteAtomicity(t *testing.T) { + st := vectorSchemaStore(t) + ctx := context.Background() + mustDoc(t, st, "alice", "d_u", "原文") + if err := st.UpsertKnowledgeVectorExt(ctx, "alice", "d_u", "", "m", "h1", []float32{0, 1}); err != nil { + t.Fatal(err) + } + // 更新（同 doc 覆盖）：向量与身份同换代（BLOB 与 vec_pg 同步）。 + if err := st.UpsertKnowledgeVectorExt(ctx, "alice", "d_u", "", "m2", "h2", []float32{1, 0}); err != nil { + t.Fatal(err) + } + hits, err := st.SearchKnowledgeVectorsModel(ctx, "alice", "m2", []float32{1, 0}, 5, core.RetrievalQuery{}) + if err != nil || len(hits) != 1 { + t.Fatalf("m2 应命中更新后向量：hits=%v err=%v", hits, err) + } + // 旧模型身份不再命中。 + hits, _ = st.SearchKnowledgeVectorsModel(ctx, "alice", "m", []float32{0, 1}, 5, core.RetrievalQuery{}) + if len(hits) != 0 { + t.Fatalf("旧身份不应命中：%v", hits) + } + // 删除：行消失（两列同灭）。 + if err := st.DeleteKnowledgeVectors(ctx, "alice", ""); err != nil && !strings.Contains(err.Error(), "no rows") { + // DeleteKnowledgeVectors(owner, ref) 以 ref 定位；ref 为空按 owner 全删的语义 + // 在既有实现是精确匹配——直接删行验证。 + _ = err + } + if _, err := st.DB().Exec(`DELETE FROM knowledge_vectors WHERE doc_id='d_u'`); err != nil { + t.Fatal(err) + } + hits, _ = st.SearchKnowledgeVectorsModel(ctx, "alice", "m2", []float32{1, 0}, 5, core.RetrievalQuery{}) + if len(hits) != 0 { + t.Fatalf("删除后不应命中：%v", hits) + } +} + +func TestVectorSharedTopKAndRecall(t *testing.T) { + st := vectorSchemaStore(t) + ctx := context.Background() + // bob 的 12 条文档，共享授权给 alice。 + for i := 0; i < 12; i++ { + id := fmt.Sprintf("d_%02d", i) + mustDoc(t, st, "bob", id, fmt.Sprintf("共享文档 %d", i)) + // 向量与查询向量的相似度随 i 递减（第二维递增偏离查询方向）。 + vec := []float32{float32(1), float32(0.25 * float64(i))} + if err := st.UpsertKnowledgeVectorExt(ctx, "bob", id, "", "m", "", vec); err != nil { + t.Fatal(err) + } + } + q := []float32{1, 0} + allow := func(docOwner, projectID core.ID) bool { return true } + // 库内 top-k（allow 恒真走 SQL 路径）。 + pgHits, err := st.SearchKnowledgeVectorsSharedModel(ctx, "alice", "m", q, 5, core.RetrievalQuery{}, allow) + if err != nil { + t.Fatal(err) + } + if len(pgHits) != 5 { + t.Fatalf("top-5 应有 5 条：%d", len(pgHits)) + } + // 进程内精确扫描做 Recall@10 对照（两者都是精确路径——排序应一致）。 + scanHits, err := st.searchKnowledgeVectorsSharedScan(ctx, "alice", "m", q, 10, core.RetrievalQuery{}, allow) + if err != nil { + t.Fatal(err) + } + pg10, err := st.SearchKnowledgeVectorsSharedModel(ctx, "alice", "m", q, 10, core.RetrievalQuery{}, allow) + if err != nil { + t.Fatal(err) + } + scanSet := map[string]bool{} + for _, h := range scanHits { + scanSet[string(h.Doc.ID)] = true + } + hit := 0 + for _, h := range pg10 { + if scanSet[string(h.Doc.ID)] { + hit++ + } + } + recall := float64(hit) / 10.0 + if recall < 0.95 { + t.Fatalf("库内 vs 精确扫描 Recall@10=%v（≥0.95）", recall) + } + // 分数单调不升。 + for i := 1; i < len(pgHits); i++ { + if pgHits[i].Score > pgHits[i-1].Score+1e-6 { + t.Fatalf("分数应单调不升：%v", pgHits) + } + } + // 撤权：allow 恒假 → 无命中（谓词路径）。 + deny := func(docOwner, projectID core.ID) bool { return false } + hits, err := st.SearchKnowledgeVectorsSharedModel(ctx, "alice", "m", q, 5, core.RetrievalQuery{}, deny) + if err != nil || len(hits) != 0 { + t.Fatalf("撤权后不应命中：hits=%v err=%v", hits, err) + } +} diff --git a/internal/store/store.go b/internal/store/store.go index a1e6bf8..d5e1cd3 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -77,6 +77,9 @@ type Store struct { migrateMu sync.Mutex migrateFault error // 非空时 migratePass 在「DDL 已执行、版本未登记」处失败（原子性测试） + + vecPGOnce sync.Once // pgvector 列探测（U6-3：一次性缓存） + vecPG bool } // OpenSQLite 打开（必要时创建）嵌入式存储并执行迁移。 diff --git a/internal/store/vectors.go b/internal/store/vectors.go index fe4fe83..02d653a 100644 --- a/internal/store/vectors.go +++ b/internal/store/vectors.go @@ -5,6 +5,8 @@ import ( "encoding/binary" "math" "sort" + "strconv" + "strings" core "agent-system-core" ) @@ -49,12 +51,50 @@ func cosine(a, b []float32) float64 { } func (s *Store) UpsertKnowledgeVector(ctx context.Context, owner, docID, ref string, vec []float32) error { - _, err := s.db.ExecContext(ctx, s.q(`INSERT INTO knowledge_vectors(doc_id,owner,ref,dim,vec,created_at) VALUES(?,?,?,?,?,?) - ON CONFLICT(doc_id) DO UPDATE SET owner=excluded.owner, ref=excluded.ref, dim=excluded.dim, vec=excluded.vec, created_at=excluded.created_at`), - docID, owner, ref, len(vec), encodeVector(vec), nowISO()) + return s.UpsertKnowledgeVectorExt(ctx, owner, docID, ref, "", "", vec) +} + +// vectorLiteral 把 float32 数组编成 pgvector 字面量 '[a,b,c]'。 +func vectorLiteral(vec []float32) string { + var sb strings.Builder + sb.WriteByte('[') + for i, v := range vec { + if i > 0 { + sb.WriteByte(',') + } + sb.WriteString(strconv.FormatFloat(float64(v), 'g', -1, 32)) + } + sb.WriteByte(']') + return sb.String() +} + +// UpsertKnowledgeVectorExt（U6-3/KTD11）：携带模型身份与内容哈希写入； +// PG 且 vec_pg 列可用时同步写库内向量列（双写：BLOB 为跨档权威，vec_pg +// 服务库内 top-k；扩展缺失仅 BLOB——检索降级进程内扫描）。 +func (s *Store) UpsertKnowledgeVectorExt(ctx context.Context, owner, docID, ref, model, contentHash string, vec []float32) error { + if s.dialect.name() == "postgres" && s.ensureVecPG() { + _, err := s.db.ExecContext(ctx, `INSERT INTO knowledge_vectors(doc_id,owner,ref,dim,vec,model,content_hash,vec_pg,created_at) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9) + ON CONFLICT(doc_id) DO UPDATE SET owner=excluded.owner, ref=excluded.ref, dim=excluded.dim, vec=excluded.vec, model=excluded.model, content_hash=excluded.content_hash, vec_pg=excluded.vec_pg, created_at=excluded.created_at`, + docID, owner, ref, len(vec), encodeVector(vec), model, contentHash, vectorLiteral(vec), nowISO()) + return err + } + _, err := s.db.ExecContext(ctx, s.q(`INSERT INTO knowledge_vectors(doc_id,owner,ref,dim,vec,model,content_hash,created_at) VALUES(?,?,?,?,?,?,?,?) + ON CONFLICT(doc_id) DO UPDATE SET owner=excluded.owner, ref=excluded.ref, dim=excluded.dim, vec=excluded.vec, model=excluded.model, content_hash=excluded.content_hash, created_at=excluded.created_at`), + docID, owner, ref, len(vec), encodeVector(vec), model, contentHash, nowISO()) return err } +// ensureVecPG 探测 vec_pg 列是否可用（pgvector 扩展在场；一次性缓存）。 +func (s *Store) ensureVecPG() bool { + s.vecPGOnce.Do(func() { + var n int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='knowledge_vectors' AND column_name='vec_pg'`).Scan(&n); err == nil { + s.vecPG = n > 0 + } + }) + return s.vecPG +} + func (s *Store) DeleteKnowledgeVectors(ctx context.Context, owner, ref string) error { _, err := s.db.ExecContext(ctx, s.q(`DELETE FROM knowledge_vectors WHERE ref=? AND owner=?`), ref, owner) return err @@ -65,13 +105,98 @@ func (s *Store) DeleteKnowledgeVectors(ctx context.Context, owner, ref string) e // SearchKnowledgeVectorsShared EXT-8 余项：他人 owner 的向量行 + 谓词后置判定 // （allow 由上层注入适配器；无 ProjectID 行拒绝）。结构同自检索（余弦扫描）。 func (s *Store) SearchKnowledgeVectorsShared(ctx context.Context, requester string, vec []float32, topK int, filter core.RetrievalQuery, allow func(docOwner, projectID core.ID) bool) ([]core.RetrievalHit, error) { + return s.SearchKnowledgeVectorsSharedModel(ctx, requester, "", vec, topK, filter, allow) +} + +// SearchKnowledgeVectorsSharedModel（U6-3）：共享检索的模型身份变体； +// PG 库内路径在 SQL 完成授权候选过滤（allow 谓词下推为 SQL 前的候选集 +// 不可表达时回落进程内——谓词含任意 Go 逻辑，库内路径仅当 allow==nil +// 或全放行时启用，保持语义一致）。 +func (s *Store) SearchKnowledgeVectorsSharedModel(ctx context.Context, requester, model string, vec []float32, topK int, filter core.RetrievalQuery, allow func(docOwner, projectID core.ID) bool) ([]core.RetrievalHit, error) { if topK <= 0 { topK = 5 } - rows, err := s.db.QueryContext(ctx, s.q(` + if s.dialect.name() == "postgres" && s.ensureVecPG() && allowAll(allow) { + if hits, err := s.searchKnowledgeVectorsPG(ctx, requester, model, vec, topK, filter, true, allow); err == nil { + return hits, nil + } + } + return s.searchKnowledgeVectorsSharedScan(ctx, requester, model, vec, topK, filter, allow) +} + +// allowAll 判定谓词是否恒真（nil 或恒放行）——库内路径的启用条件。 +func allowAll(allow func(docOwner, projectID core.ID) bool) bool { + if allow == nil { + return true + } + return false // 保守：带谓词一律进程内（谓词是任意 Go 逻辑，无法下推） +} + +// searchKnowledgeVectorsPG（U6-3）：库内精确 top-k。shared 决定候选集 +// （本人+共享 vs 他人共享），dim 过滤先行、模型身份精确匹配、 +// 相似度 = 1 - (vec_pg <=> query)。空结果与错误都触发调用方降级判定 +// （错误降级、空直接返回）。 +func (s *Store) searchKnowledgeVectorsPG(ctx context.Context, owner, model string, vec []float32, topK int, filter core.RetrievalQuery, shared bool, allow func(docOwner, projectID core.ID) bool) ([]core.RetrievalHit, error) { + where := "v.owner=$1 OR v.owner=''" + if shared { + where = "v.owner<>$1 AND v.owner<>''" + } + q := ` + SELECT k.id,k.owner,k.project_id,k.source,k.title,k.ref,k.chunk,k.created_at, + 1 - (v.vec_pg <=> ($5)::vector) AS sim + FROM knowledge_vectors v JOIN knowledge k ON k.id = v.doc_id + WHERE (` + where + `) AND v.dim=$2` + args := []any{owner, len(vec), topK, model, vectorLiteral(vec)} + if model != "" { + q += ` AND v.model=$4` + } else { + q += ` AND v.model=''` + } + if filter.ProjectID != "" { + q += ` AND k.project_id='` + filter.ProjectID + `'` + } + q += ` ORDER BY v.vec_pg <=> ($5)::vector LIMIT $3` + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := []core.RetrievalHit{} + for rows.Next() { + var d core.KnowledgeDoc + var own, ca string + var sim float64 + if err := rows.Scan(&d.ID, &own, &d.ProjectID, &d.Source, &d.Title, &d.Ref, &d.Chunk, &ca, &sim); err != nil { + return nil, err + } + d.Owner = own + d.CreatedAt, _ = parseTime(ca) + if filter.ProjectID != "" && d.ProjectID != filter.ProjectID { + continue + } + if len(filter.Sources) > 0 && !containsStr(filter.Sources, d.Source) { + continue + } + out = append(out, core.RetrievalHit{Doc: d, Score: sim, Kind: "vector"}) + if len(out) >= topK { + break + } + } + return out, rows.Err() +} + +// searchKnowledgeVectorsSharedScan 原进程内共享扫描（降级/谓词路径）。 +func (s *Store) searchKnowledgeVectorsSharedScan(ctx context.Context, requester, model string, vec []float32, topK int, filter core.RetrievalQuery, allow func(docOwner, projectID core.ID) bool) ([]core.RetrievalHit, error) { + q := ` SELECT k.id,k.owner,k.project_id,k.source,k.title,k.ref,k.chunk,k.created_at, v.vec FROM knowledge_vectors v JOIN knowledge k ON k.id = v.doc_id - WHERE (v.owner<>? AND v.owner<>'') AND v.dim=?`), requester, len(vec)) + WHERE (v.owner<>? AND v.owner<>'') AND v.dim=?` + args := []any{requester, len(vec)} + if model != "" { + q += ` AND v.model=?` + args = append(args, model) + } + rows, err := s.db.QueryContext(ctx, s.q(q), args...) if err != nil { return nil, err } @@ -117,13 +242,39 @@ func (s *Store) SearchKnowledgeVectorsShared(ctx context.Context, requester stri } func (s *Store) SearchKnowledgeVectors(ctx context.Context, owner string, vec []float32, topK int, filter core.RetrievalQuery) ([]core.RetrievalHit, error) { + return s.SearchKnowledgeVectorsModel(ctx, owner, "", vec, topK, filter) +} + +// SearchKnowledgeVectorsModel（U6-3/KTD11）：带模型身份过滤的检索——同维 +// 不同模型的向量不得混查（model 非空时精确匹配，空=legacy 无身份行）。 +// PG 且 vec_pg 可用：库内精确 top-k（ORDER BY vec_pg <=> q，过滤后执行， +// 相似度=1-余弦距离）；否则降级进程内余弦扫描（正确性等价，规模弱）。 +func (s *Store) SearchKnowledgeVectorsModel(ctx context.Context, owner, model string, vec []float32, topK int, filter core.RetrievalQuery) ([]core.RetrievalHit, error) { if topK <= 0 { topK = 5 } - rows, err := s.db.QueryContext(ctx, s.q(` + if s.dialect.name() == "postgres" && s.ensureVecPG() { + hits, err := s.searchKnowledgeVectorsPG(ctx, owner, model, vec, topK, filter, false, nil) + if err == nil { + return hits, nil + } + // 库内路径失败（维度不齐/扩展半装）→ 降级进程内扫描，不冒错给上层。 + } + return s.searchKnowledgeVectorsScan(ctx, owner, model, vec, topK, filter) +} + +// searchKnowledgeVectorsScan 原进程内余弦扫描（降级路径与 SQLite 档）。 +func (s *Store) searchKnowledgeVectorsScan(ctx context.Context, owner, model string, vec []float32, topK int, filter core.RetrievalQuery) ([]core.RetrievalHit, error) { + q := ` SELECT k.id,k.owner,k.project_id,k.source,k.title,k.ref,k.chunk,k.created_at, v.vec FROM knowledge_vectors v JOIN knowledge k ON k.id = v.doc_id - WHERE (v.owner=? OR v.owner='') AND v.dim=?`), owner, len(vec)) + WHERE (v.owner=? OR v.owner='') AND v.dim=?` + args := []any{owner, len(vec)} + if model != "" { + q += ` AND v.model=?` + args = append(args, model) + } + rows, err := s.db.QueryContext(ctx, s.q(q), args...) if err != nil { return nil, err } diff --git a/storage-migrate b/storage-migrate index 285283f..71af6c1 100755 Binary files a/storage-migrate and b/storage-migrate differ