Merge pull request 'Add model identity and in-database vector top-k via pgvector (U6 slice 3)' (#159) from feature/collab-storage-u6-vector into main
ci / vet-and-test (push) Successful in 10m52s
ci / vet-and-test (push) Successful in 10m52s
This commit was merged in pull request #159.
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
+21
-3
@@ -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
|
||||
}
|
||||
|
||||
@@ -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`,
|
||||
}}},
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -77,6 +77,9 @@ type Store struct {
|
||||
|
||||
migrateMu sync.Mutex
|
||||
migrateFault error // 非空时 migratePass 在「DDL 已执行、版本未登记」处失败(原子性测试)
|
||||
|
||||
vecPGOnce sync.Once // pgvector 列探测(U6-3:一次性缓存)
|
||||
vecPG bool
|
||||
}
|
||||
|
||||
// OpenSQLite 打开(必要时创建)嵌入式存储并执行迁移。
|
||||
|
||||
+158
-7
@@ -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
|
||||
}
|
||||
|
||||
Binary file not shown.
Reference in New Issue
Block a user