mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-10-03 07:43:58 +00:00
refactor(memory): 拆除描述式媒体索引,媒体成为一等块并按原生向量融合
背景:此前媒体是靠「生成的描述文本」将就进记忆的——写 marker 进正文、 再由正则反解成 media_refs 与图库里的 type=Media 实体。这条链路有三个 致命缺陷:描述由异步模型生成(未生成前媒体等于不存在)、语义检索实质上 只搜描述文字、图库里的「媒体节点」是描述文本的投影而不是媒体本身。 本提交把这条链路整体拆除,媒体改为按自己的原生向量参与记忆: 一、描述链彻底删除(无残留、无兼容分支) - media.Item 去掉 Description/DescribedBy 与对应列; - 删除 Store.Describe / Store.Search / Store.Pending; - 删除 Agent.mediaDescribeLoop / describePendingMedia 与配置项 core.memory.media.describe_on_ingest; - SDK 侧 MediaAttachment 去掉 Description(见 SDK 仓独立提交)。 二、marker 机制删除,媒体归属改为结构化块边 - 删除 mediaMarkerLine/parseMediaMarkers/mediaEntityName/mediaTriplesFromText/ extractMediaDigests/sentenceWithMediaMarkers/docMediaContext; - memory.Triple 新增 MediaDigests 结构化字段;句子文本保持原样, 不再被 marker 污染; - 块以 sentence --contains--> block / document --contains--> block 结构边 挂到承载节点(新增 documents 表与 document 节点种类); - 模型未给原句时用「主谓宾。」拼一句自然语言作落点,不造 marker 文本。 三、旧数据迁移(幂等) - 新增 GraphDB.MigrateLegacyMediaEntities:把 type=Media 的旧实体按短 digest 还原成原生块、挂回原句子、删除旧实体与描述关系;Agent 启动时执行; - CleanupOrphanedSentences 同时看关系引用与块边,避免把只靠块存活的句子 连同块边一起删掉。 四、向量融合:媒体按图本身被召回 - 新增 vector.FuseVectors(逐维求和 + L2 归一化); - Doc.DenseVec = 文本向量 ⊕ 文档块的媒体向量(同 fingerprint 才融合), 新增 Doc.DenseFP,指纹变化触发重算; - ContextEvent.DenseVec 同理融合事件块;事件新增 DenseFP,Prune 只在 同一统一空间内比稠密余弦; - 跨模态视觉路只召回「仍被某层记忆块持有」的媒体,CAS 全库字节不再 直接充当记忆检索结果。 五、同时纳入本分支既有的嵌入基础改造(此前工作区未提交,缺它 HEAD 不可构建) - internal/tfidf 懒回退包、千问三段式多模态 ONNX 空间的 Go 侧 (qwen/embedder.go、image.go、model_input.go)、CLIP 移除、 sdk.NewStore 分词器签名与调用点、embed 侧车 systemd 单元。 验证:go build ./... 、go vet ./...(含 -tags medialive)均通过; 在 HEAD 的独立 worktree 上重放本次暂存集后 go test -short ./internal/... 全部通过(端口冲突类用例在隔离环境中亦通过)。未提交工作区中与本改造 无关的改动(HarmonyOS、waiter、devicebridge、plan.md 等)。
This commit is contained in:
@ -69,11 +69,6 @@ type Item struct {
|
||||
OriginPath string `json:"origin_path,omitempty"`
|
||||
// Tool 是注入这条媒体的工具名(如 multimodal_see_picture)。
|
||||
Tool string `json:"tool,omitempty"`
|
||||
// Description 是视觉/音频模型生成的文字描述,供 L2/L3 检索。
|
||||
// 空表示未描述(未开启描述、模型不可用或描述失败)。
|
||||
Description string `json:"description,omitempty"`
|
||||
// DescribedBy 记录描述来自哪个源,让后续读者能判断可靠性。
|
||||
DescribedBy string `json:"described_by,omitempty"`
|
||||
// FirstSeen/LastSeen 是首末次入库时间。
|
||||
FirstSeen time.Time `json:"first_seen"`
|
||||
LastSeen time.Time `json:"last_seen"`
|
||||
@ -127,8 +122,6 @@ func (s *Store) initSchema() error {
|
||||
height INTEGER DEFAULT 0,
|
||||
origin_path TEXT,
|
||||
tool TEXT,
|
||||
description TEXT,
|
||||
described_by TEXT,
|
||||
first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)`,
|
||||
@ -196,20 +189,15 @@ func (s *Store) Put(data []byte, meta Item) (string, error) {
|
||||
}
|
||||
_, err := s.db.Exec(`
|
||||
INSERT INTO media (digest, kind, mime, size, width, height,
|
||||
origin_path, tool, description, described_by,
|
||||
first_seen, last_seen)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
origin_path, tool, first_seen, last_seen)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(digest) DO UPDATE SET
|
||||
last_seen = excluded.last_seen,
|
||||
-- 只在原值为空时补写:先到的描述可能来自更强的模型,
|
||||
-- 后到的空值不该把它冲掉。
|
||||
description = CASE WHEN COALESCE(media.description,'') = '' THEN excluded.description ELSE media.description END,
|
||||
described_by = CASE WHEN COALESCE(media.described_by,'') = '' THEN excluded.described_by ELSE media.described_by END,
|
||||
width = CASE WHEN media.width = 0 THEN excluded.width ELSE media.width END,
|
||||
height = CASE WHEN media.height = 0 THEN excluded.height ELSE media.height END,
|
||||
tool = CASE WHEN COALESCE(media.tool,'') = '' THEN excluded.tool ELSE media.tool END
|
||||
`, digest, string(meta.Kind), meta.MIME, int64(len(data)), meta.Width, meta.Height,
|
||||
meta.OriginPath, meta.Tool, meta.Description, meta.DescribedBy, now, now)
|
||||
meta.OriginPath, meta.Tool, now, now)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("media: upsert meta: %w", err)
|
||||
}
|
||||
@ -242,115 +230,12 @@ func (s *Store) Stat(digest string) (*Item, error) {
|
||||
defer s.mu.RUnlock()
|
||||
return s.scanOne(s.db.QueryRow(`
|
||||
SELECT digest, kind, mime, size, width, height, origin_path, tool,
|
||||
description, described_by, first_seen, last_seen,
|
||||
first_seen, last_seen,
|
||||
vec, vec_model
|
||||
FROM media WHERE digest = ?`, digest))
|
||||
}
|
||||
|
||||
// Describe 写入(或覆盖)文字描述。
|
||||
//
|
||||
// 与 Put 的"只在空时补写"不同:Describe 是显式操作,调用方明确想要这份
|
||||
// 描述生效(例如换了更强的视觉模型重新描述)。
|
||||
func (s *Store) Describe(digest, description, describedBy string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
res, err := s.db.Exec(`UPDATE media SET description = ?, described_by = ? WHERE digest = ?`,
|
||||
description, describedBy, digest)
|
||||
if err != nil {
|
||||
return fmt.Errorf("media: describe: %w", err)
|
||||
}
|
||||
if n, _ := res.RowsAffected(); n == 0 {
|
||||
return fmt.Errorf("media: describe: unknown digest %s", shortDigest(digest))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Search 按描述文本做 LIKE 匹配,返回最近的若干条。
|
||||
//
|
||||
// 刻意不在这里做向量检索:媒体的语义检索走 L2 文档层的既有索引
|
||||
// (描述文字随记忆条目一起进 Doc.Content,复用那套 TF-IDF/embedding),
|
||||
// 本方法只是"按关键词直接翻媒体库"的补充入口。
|
||||
func (s *Store) Search(query string, kind Kind, limit int) ([]*Item, error) {
|
||||
if limit <= 0 {
|
||||
limit = 20
|
||||
}
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
q := `SELECT digest, kind, mime, size, width, height, origin_path, tool,
|
||||
description, described_by, first_seen, last_seen,
|
||||
vec, vec_model
|
||||
FROM media WHERE COALESCE(description,'') != ''`
|
||||
args := []interface{}{}
|
||||
if strings.TrimSpace(query) != "" {
|
||||
q += ` AND description LIKE ?`
|
||||
args = append(args, "%"+query+"%")
|
||||
}
|
||||
if kind != "" {
|
||||
q += ` AND kind = ?`
|
||||
args = append(args, string(kind))
|
||||
}
|
||||
q += ` ORDER BY last_seen DESC LIMIT ?`
|
||||
args = append(args, limit)
|
||||
|
||||
rows, err := s.db.Query(q, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []*Item
|
||||
for rows.Next() {
|
||||
it, err := s.scanRows(rows)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
out = append(out, it)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// Pending 返回尚无描述的媒体,供后台描述任务消费。
|
||||
// Pending 返回尚无描述的媒体,供后台描述任务消费。
|
||||
//
|
||||
// 不只看 description 为空,还要求 described_by 也为空。
|
||||
// 因为“已尝试但无法描述”的项(如 kind=other 的二进制、blob 已丢失)
|
||||
// 会被标记为 described_by=unsupported/content-missing 而 description 仍为空——
|
||||
// 若只看 description,这些项每轮都会被取出来重试,永远卡在队列头部,
|
||||
// 真正需要描述的新项永远轮不到(LIMIT 只取前 N 条)。
|
||||
func (s *Store) Pending(limit int) ([]*Item, error) {
|
||||
if limit <= 0 {
|
||||
limit = 10
|
||||
}
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
rows, err := s.db.Query(`
|
||||
SELECT digest, kind, mime, size, width, height, origin_path, tool,
|
||||
description, described_by, first_seen, last_seen,
|
||||
vec, vec_model
|
||||
FROM media
|
||||
WHERE COALESCE(description,'') = '' AND COALESCE(described_by,'') = ''
|
||||
ORDER BY last_seen DESC LIMIT ?`, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []*Item
|
||||
for rows.Next() {
|
||||
it, err := s.scanRows(rows)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
out = append(out, it)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// GC 清理已不被任何记忆块持有的内容。
|
||||
//
|
||||
// keep 是当前仍被 Context/Document/Graph 里一等记忆块持有的 digest 集合,
|
||||
// 由调用方从三层记忆节点计算得出;media.Store 不再自己维护引用账本。
|
||||
// 不在 keep 中且早于 minAge 的项被清理;超出 maxBytes 时也只淘汰不在 keep 中的项。
|
||||
// Delete 删除一份媒体内容(元数据 + blob)。
|
||||
// Stat 返回元数据,不读内容。
|
||||
//
|
||||
// 这不是 GC,也不看引用计数:调用方是记忆系统本身——当它把一个记忆块
|
||||
// 永久地从三层记忆中删掉(而非在层间迁移)时,媒体作为块的内容一并删除。
|
||||
@ -376,13 +261,11 @@ func (s *Store) Stats() map[string]interface{} {
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
out := map[string]interface{}{"blob_dir": s.blobDir}
|
||||
var count, described int
|
||||
var count int
|
||||
var total int64
|
||||
s.db.QueryRow(`SELECT COUNT(*), COALESCE(SUM(size),0) FROM media`).Scan(&count, &total)
|
||||
s.db.QueryRow(`SELECT COUNT(*) FROM media WHERE COALESCE(description,'') != ''`).Scan(&described)
|
||||
out["count"] = count
|
||||
out["total_bytes"] = total
|
||||
out["described"] = described
|
||||
|
||||
byKind := map[string]int{}
|
||||
rows, err := s.db.Query(`SELECT kind, COUNT(*) FROM media GROUP BY kind`)
|
||||
@ -524,7 +407,7 @@ func (s *Store) queryMediaScored(queryVec []float64, model string, topK int) ([]
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
query := `SELECT digest, kind, mime, size, width, height,
|
||||
origin_path, tool, description, described_by, first_seen, last_seen,
|
||||
origin_path, tool, first_seen, last_seen,
|
||||
vec, vec_model
|
||||
FROM media WHERE vec IS NOT NULL AND vec != ''`
|
||||
var args []interface{}
|
||||
@ -546,17 +429,15 @@ func (s *Store) queryMediaScored(queryVec []float64, model string, topK int) ([]
|
||||
for rows.Next() {
|
||||
var it Item
|
||||
var kind string
|
||||
var origin, tool, desc, by, vecJSON, vecModel sql.NullString
|
||||
var origin, tool, vecJSON, vecModel sql.NullString
|
||||
if err := rows.Scan(&it.Digest, &kind, &it.MIME, &it.Size, &it.Width, &it.Height,
|
||||
&origin, &tool, &desc, &by, &it.FirstSeen, &it.LastSeen,
|
||||
&origin, &tool, &it.FirstSeen, &it.LastSeen,
|
||||
&vecJSON, &vecModel); err != nil {
|
||||
continue
|
||||
}
|
||||
it.Kind = Kind(kind)
|
||||
it.OriginPath = origin.String
|
||||
it.Tool = tool.String
|
||||
it.Description = desc.String
|
||||
it.DescribedBy = by.String
|
||||
if !vecJSON.Valid || vecJSON.String == "" {
|
||||
continue
|
||||
}
|
||||
@ -623,17 +504,15 @@ func (s *Store) scanRows(r rowScanner) (*Item, error) { return scanItem(r) }
|
||||
func scanItem(r rowScanner) (*Item, error) {
|
||||
var it Item
|
||||
var kind string
|
||||
var origin, tool, desc, by, vecJSON, vecModel sql.NullString
|
||||
var origin, tool, vecJSON, vecModel sql.NullString
|
||||
if err := r.Scan(&it.Digest, &kind, &it.MIME, &it.Size, &it.Width, &it.Height,
|
||||
&origin, &tool, &desc, &by, &it.FirstSeen, &it.LastSeen,
|
||||
&origin, &tool, &it.FirstSeen, &it.LastSeen,
|
||||
&vecJSON, &vecModel); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
it.Kind = Kind(kind)
|
||||
it.OriginPath = origin.String
|
||||
it.Tool = tool.String
|
||||
it.Description = desc.String
|
||||
it.DescribedBy = by.String
|
||||
if vecJSON.Valid && vecJSON.String != "" {
|
||||
var v []float64
|
||||
if err := json.Unmarshal([]byte(vecJSON.String), &v); err == nil {
|
||||
|
||||
@ -142,50 +142,13 @@ func TestDelete_UnknownDigestIsNoop(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestDescribe_OverwritesExplicitly(t *testing.T) {
|
||||
s := newTestStore(t, 0)
|
||||
d, _ := s.Put([]byte("img"), Item{MIME: "image/png"})
|
||||
|
||||
if err := s.Describe(d, "一只橘猫", "vis-a"); err != nil {
|
||||
func TestDescribe_Removed(t *testing.T) {
|
||||
// 媒体不再有文字描述:描述式索引是废弃的就机制。
|
||||
// 这里只保留一个编译期断言,确保 API 不会静默回归。
|
||||
s := newTestStore(t)
|
||||
if _, err := s.Put([]byte("img"), Item{MIME: "image/png"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
it, _ := s.Stat(d)
|
||||
if it.Description != "一只橘猫" || it.DescribedBy != "vis-a" {
|
||||
t.Fatalf("描述未写入: %+v", it)
|
||||
}
|
||||
|
||||
// Describe 是显式操作,允许覆盖(换更强模型重描述)
|
||||
if err := s.Describe(d, "一只橘色虎斑猫坐在窗台", "vis-b"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
it, _ = s.Stat(d)
|
||||
if !strings.Contains(it.Description, "虎斑") || it.DescribedBy != "vis-b" {
|
||||
t.Fatalf("Describe 应覆盖旧描述: %+v", it)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDescribe_UnknownDigestErrors(t *testing.T) {
|
||||
s := newTestStore(t, 0)
|
||||
err := s.Describe("deadbeef", "x", "y")
|
||||
if err == nil {
|
||||
t.Fatal("未知 digest 应报错而非静默成功")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPut_DoesNotClobberExistingDescription(t *testing.T) {
|
||||
// 先到的描述可能来自更强的模型;后到的空值不该把它冲掉。
|
||||
s := newTestStore(t, 0)
|
||||
data := []byte("img")
|
||||
d, _ := s.Put(data, Item{MIME: "image/png", Description: "详细描述", DescribedBy: "strong-model"})
|
||||
|
||||
// 第二次 Put 同内容但不带描述
|
||||
if _, err := s.Put(data, Item{MIME: "image/png"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
it, _ := s.Stat(d)
|
||||
if it.Description != "详细描述" || it.DescribedBy != "strong-model" {
|
||||
t.Fatalf("重复 Put 的空描述不该冲掉已有描述: %+v", it)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPut_BackfillsMissingDimensions(t *testing.T) {
|
||||
@ -206,47 +169,22 @@ func TestPut_BackfillsMissingDimensions(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSearch_FiltersByDescriptionAndKind(t *testing.T) {
|
||||
s := newTestStore(t, 0)
|
||||
di, _ := s.Put([]byte("chart-img"), Item{MIME: "image/png"})
|
||||
da, _ := s.Put([]byte("speech-aud"), Item{MIME: "audio/wav"})
|
||||
dn, _ := s.Put([]byte("no-desc"), Item{MIME: "image/png"})
|
||||
s.Describe(di, "一张蓝色的柱状图表", "vis")
|
||||
s.Describe(da, "一段关于图表的讲解录音", "aud")
|
||||
func TestStats_CountsByKind(t *testing.T) {
|
||||
s := newTestStore(t)
|
||||
s.Put([]byte("i1"), Item{MIME: "image/png"})
|
||||
s.Put([]byte("i2"), Item{MIME: "image/jpeg"})
|
||||
s.Put([]byte("a1"), Item{MIME: "audio/wav"})
|
||||
|
||||
all, err := s.Search("图表", "", 10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
st := s.Stats()
|
||||
if st["count"].(int) != 3 {
|
||||
t.Fatalf("count 应为 3,实际 %v", st["count"])
|
||||
}
|
||||
if len(all) != 2 {
|
||||
t.Fatalf("两条描述都含「图表」,应返回 2,实际 %d", len(all))
|
||||
if _, ok := st["described"]; ok {
|
||||
t.Fatal("媒体已不再有描述计数")
|
||||
}
|
||||
|
||||
imgs, _ := s.Search("图表", KindImage, 10)
|
||||
if len(imgs) != 1 || imgs[0].Digest != di {
|
||||
t.Fatalf("按 image 过滤应只剩图片,实际 %d 条", len(imgs))
|
||||
}
|
||||
|
||||
// 无描述的项不该出现在语义检索结果里
|
||||
for _, it := range all {
|
||||
if it.Digest == dn {
|
||||
t.Fatal("无描述的项不该被 Search 返回")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestPending_ReturnsUndescribed(t *testing.T) {
|
||||
s := newTestStore(t, 0)
|
||||
described, _ := s.Put([]byte("has-desc"), Item{MIME: "image/png"})
|
||||
undescribed, _ := s.Put([]byte("needs-desc"), Item{MIME: "image/png"})
|
||||
s.Describe(described, "已有描述", "vis")
|
||||
|
||||
pending, err := s.Pending(10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(pending) != 1 || pending[0].Digest != undescribed {
|
||||
t.Fatalf("应只返回未描述项,实际 %d 条", len(pending))
|
||||
byKind := st["by_kind"].(map[string]int)
|
||||
if byKind["image"] != 2 || byKind["audio"] != 1 {
|
||||
t.Fatalf("by_kind 不对: %v", byKind)
|
||||
}
|
||||
}
|
||||
|
||||
@ -278,28 +216,8 @@ func TestParseDataURL(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestStats_CountsByKindAndDescription(t *testing.T) {
|
||||
s := newTestStore(t, 4096)
|
||||
d1, _ := s.Put([]byte("i1"), Item{MIME: "image/png"})
|
||||
s.Put([]byte("i2"), Item{MIME: "image/jpeg"})
|
||||
s.Put([]byte("a1"), Item{MIME: "audio/wav"})
|
||||
s.Describe(d1, "描述", "vis")
|
||||
|
||||
st := s.Stats()
|
||||
if st["count"].(int) != 3 {
|
||||
t.Fatalf("count 应为 3,实际 %v", st["count"])
|
||||
}
|
||||
if st["described"].(int) != 1 {
|
||||
t.Fatalf("described 应为 1,实际 %v", st["described"])
|
||||
}
|
||||
byKind := st["by_kind"].(map[string]int)
|
||||
if byKind["image"] != 2 || byKind["audio"] != 1 {
|
||||
t.Fatalf("by_kind 不对: %v", byKind)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPut_RejectsEmpty(t *testing.T) {
|
||||
s := newTestStore(t, 0)
|
||||
s := newTestStore(t)
|
||||
if _, err := s.Put(nil, Item{MIME: "image/png"}); err == nil {
|
||||
t.Fatal("空内容应报错")
|
||||
}
|
||||
@ -313,7 +231,7 @@ func TestReopen_PersistsAcrossRestart(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
d, _ := s1.Put([]byte("persistent-img"), Item{MIME: "image/png", OriginPath: "/tmp/x.png"})
|
||||
s1.Describe(d, "跨重启的描述", "vis")
|
||||
s1.SetVec(d, []float64{0.1, 0.2}, "test-space")
|
||||
s1.Close()
|
||||
|
||||
s2, err := New(dir)
|
||||
@ -326,8 +244,8 @@ func TestReopen_PersistsAcrossRestart(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("重开后应能查到: %v", err)
|
||||
}
|
||||
if it.Description != "跨重启的描述" {
|
||||
t.Fatalf("元数据应持久化: %+v", it)
|
||||
if it.OriginPath != "/tmp/x.png" || len(it.Vec) != 2 || it.VecModel != "test-space" {
|
||||
t.Fatalf("元数据与向量应持久化: %+v", it)
|
||||
}
|
||||
data, err := s2.Get(d)
|
||||
if err != nil || string(data) != "persistent-img" {
|
||||
@ -335,38 +253,26 @@ func TestReopen_PersistsAcrossRestart(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestPending_ExcludesAttemptedButUndescribable(t *testing.T) {
|
||||
// 「已尝试但无法描述」的项必须退出待描述队列。
|
||||
//
|
||||
// 这些项被标记为 described_by=unsupported/content-missing 而 description
|
||||
// 仍为空。若 Pending 只看 description,它们每轮都会被取出来重试、
|
||||
// 永久占着 LIMIT 的名额,真正需要描述的新项永远轮不到。
|
||||
s := newTestStore(t, 0)
|
||||
func TestStaleVecDigests_TracksModelSwitch(t *testing.T) {
|
||||
// 模型切换后旧向量必须被重算:StaleVecDigests 是启动迁移的入口。
|
||||
s := newTestStore(t)
|
||||
d1, _ := s.Put([]byte("a"), Item{MIME: "image/png"})
|
||||
d2, _ := s.Put([]byte("b"), Item{MIME: "image/png"})
|
||||
s.SetVec(d1, []float64{0.1}, "space-a")
|
||||
|
||||
fresh, _ := s.Put([]byte("needs-describe"), Item{MIME: "image/png"})
|
||||
unsupported, _ := s.Put([]byte("cannot-describe"), Item{MIME: "application/octet-stream"})
|
||||
described, _ := s.Put([]byte("已描述"), Item{MIME: "image/png"})
|
||||
|
||||
// 标记「尝试过但不支持」:description 空,described_by 非空
|
||||
if err := s.Describe(unsupported, "", "unsupported"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := s.Describe(described, "一张图", "visionllm"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
pending, err := s.Pending(10)
|
||||
stale, err := s.StaleVecDigestsAll("space-a")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(pending) != 1 {
|
||||
var names []string
|
||||
for _, p := range pending {
|
||||
names = append(names, shortDigest(p.Digest))
|
||||
}
|
||||
t.Fatalf("应只剩 1 条待描述,实际 %d 条: %v", len(pending), names)
|
||||
if len(stale) != 1 || stale[0] != d2 {
|
||||
t.Fatalf("只有未嵌入的 d2 需重算,实际 %v", stale)
|
||||
}
|
||||
if pending[0].Digest != fresh {
|
||||
t.Fatalf("待描述的应是未处理项,实际 %s", shortDigest(pending[0].Digest))
|
||||
|
||||
stale, err = s.StaleVecDigestsAll("space-b")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(stale) != 2 {
|
||||
t.Fatalf("换空间后两条都需重算,实际 %v", stale)
|
||||
}
|
||||
}
|
||||
|
||||
@ -10,9 +10,9 @@ func TestQueryMedia_BasicSimilarity(t *testing.T) {
|
||||
defer s.Close()
|
||||
|
||||
// 入库三张带向量的媒体:两张图、一段音频
|
||||
d1, _ := s.Put([]byte("img1"), Item{MIME: "image/png", Description: "紫蓝红三色带"})
|
||||
d2, _ := s.Put([]byte("img2"), Item{MIME: "image/jpeg", Description: "蓝紫红渐变"})
|
||||
d3, _ := s.Put([]byte("aud1"), Item{MIME: "audio/wav", Description: "一段语音"})
|
||||
d1, _ := s.Put([]byte("img1"), Item{MIME: "image/png"})
|
||||
d2, _ := s.Put([]byte("img2"), Item{MIME: "image/jpeg"})
|
||||
d3, _ := s.Put([]byte("aud1"), Item{MIME: "audio/wav"})
|
||||
|
||||
// 模拟视觉嵌入:img1 和 img2 向量接近,aud1 远离
|
||||
vec1 := []float64{0.9, 0.1, 0.0, 0.0}
|
||||
@ -138,24 +138,24 @@ func TestStaleVecDigests(t *testing.T) {
|
||||
s := newTestStore(t, 0)
|
||||
defer s.Close()
|
||||
|
||||
// 有描述且 vec_model 匹配 → 非 stale
|
||||
d1, _ := s.Put([]byte("img1"), Item{MIME: "image/png", Description: "图一"})
|
||||
s.SetVec(d1, []float64{0.1}, "clip-vit-b32")
|
||||
// 有向量且 vec_model 匹配 → 非 stale
|
||||
d1, _ := s.Put([]byte("img1"), Item{MIME: "image/png"})
|
||||
s.SetVec(d1, []float64{0.1}, "space-a")
|
||||
|
||||
// 有描述但 vec_model 旧 → stale
|
||||
d2, _ := s.Put([]byte("img2"), Item{MIME: "image/png", Description: "图二"})
|
||||
s.SetVec(d2, []float64{0.2}, "clip-vit-b14")
|
||||
// 有向量但 vec_model 旧 → stale
|
||||
d2, _ := s.Put([]byte("img2"), Item{MIME: "image/png"})
|
||||
s.SetVec(d2, []float64{0.2}, "space-old")
|
||||
|
||||
// 有描述但从未嵌入(vec_model 空)→ stale
|
||||
d3, _ := s.Put([]byte("img3"), Item{MIME: "image/png", Description: "图三"})
|
||||
// 从未嵌入(vec_model 空)→ stale
|
||||
d3, _ := s.Put([]byte("img3"), Item{MIME: "image/png"})
|
||||
|
||||
// 无描述但有图片 → 也应被迁移(描述是可选语义通道,图片应独立于描述参与向量空间)
|
||||
// 与向量/描述无关的图片同样应被迁移:图片独立参与向量空间
|
||||
d4, _ := s.Put([]byte("img4"), Item{MIME: "image/png"})
|
||||
|
||||
// 音频不参与图片迁移(StaleVecDigests 只查 kind='image')
|
||||
s.Put([]byte("aud1"), Item{MIME: "audio/wav", Description: "语音"})
|
||||
s.Put([]byte("aud1"), Item{MIME: "audio/wav"})
|
||||
|
||||
stale, err := s.StaleVecDigests("clip-vit-b32")
|
||||
stale, err := s.StaleVecDigests("space-a")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@ -123,31 +123,44 @@ func TestSmoke_VideoFramesDistinct(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSmoke_DescribeThenRetrieve(t *testing.T) {
|
||||
// 场景 C:视觉模型描述落库后,描述文字成为可检索的语义入口。
|
||||
// 这是本方案最关键的一环——blob 可能被淘汰,描述会长期留在记忆里。
|
||||
s := newTestStore(t, 50*1024*1024)
|
||||
func TestSmoke_NearestNeighborVectorRetrieve(t *testing.T) {
|
||||
// 场景 C:图片只按自己的原生向量被检索。
|
||||
// 没有描述文本参与——描述式索引是废弃的就机制。
|
||||
s := newTestStore(t)
|
||||
|
||||
pic, _ := s.Put(makePNG(400, 400, 0), Item{MIME: "image/png", Tool: "multimodal_see_picture"})
|
||||
if err := s.Describe(pic, "一张 400x400 的三色带图:上红、中绿、下蓝", "visionllm"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s.SetVec(pic, []float64{1, 0, 0, 0}, "space")
|
||||
var frames []string
|
||||
for i := 0; i < 6; i++ {
|
||||
d, _ := s.Put(makePNG(320, 240, i), Item{MIME: "image/jpeg", Tool: "multimodal_see_video"})
|
||||
if err := s.Describe(d, fmt.Sprintf("视频第 %d 帧:测试图卡,含彩条与计数器", i+1), "visionllm"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s.SetVec(d, []float64{1, 1, float64(i) / 10, 0}, "space")
|
||||
frames = append(frames, d)
|
||||
}
|
||||
|
||||
if hits, _ := s.Search("三色带", KindImage, 10); len(hits) != 1 {
|
||||
t.Fatalf("搜「三色带」应命中 1 条,实际 %d", len(hits))
|
||||
hits, err := s.QueryMediaScored([]float64{1, 0, 0, 0}, "space", 10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if hits, _ := s.Search("计数器", KindImage, 10); len(hits) != 6 {
|
||||
t.Fatalf("搜「计数器」应命中 6 帧,实际 %d", len(hits))
|
||||
if len(hits) != 7 {
|
||||
t.Fatalf("7 份媒体都有同空间向量,应全部可召,实际 %d", len(hits))
|
||||
}
|
||||
pend, _ := s.Pending(100)
|
||||
if len(pend) != 0 {
|
||||
t.Fatalf("应全部已描述,仍有 %d 条待描述", len(pend))
|
||||
if hits[0].Item.Digest != pic {
|
||||
t.Fatalf("与查询同向的应是第一命中,实际 %s", shortDigest(hits[0].Item.Digest))
|
||||
}
|
||||
|
||||
// 不同向量空间/模型的条目不得参与:坐标系不同,余弦无意义。
|
||||
foreign := frames[0]
|
||||
if err := s.SetVec(foreign, []float64{1, 0, 0, 0}, "other-space"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
hits, err = s.QueryMediaScored([]float64{1, 0, 0, 0}, "space", 10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, h := range hits {
|
||||
if h.Item.Digest == foreign {
|
||||
t.Fatal("另一套空间(other-space)的向量不该被 space 查询召回")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -204,7 +217,7 @@ func TestSmoke_DeleteRemovesOnlyThatContent(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestSmoke_FullLifecycleAcrossRestart(t *testing.T) {
|
||||
// 端到端:入库 → 描述 → 删除一些内容 → 重启 → 检索,
|
||||
// 端到端:入库 → 嵌入 → 删除一些内容 → 重启 → 向量检索,
|
||||
// 并确认磁盘与元数据不出现双向孤儿。记忆的意义就在于跨重启还在。
|
||||
dir := t.TempDir()
|
||||
s, err := New(dir)
|
||||
@ -214,10 +227,10 @@ func TestSmoke_FullLifecycleAcrossRestart(t *testing.T) {
|
||||
|
||||
png := makePNG(400, 400, 0)
|
||||
pic, _ := s.Put(png, Item{MIME: "image/png", Width: 400, Height: 400, Tool: "multimodal_see_picture"})
|
||||
s.Describe(pic, "一张 400x400 的三色带图:上红、中绿、下蓝", "visionllm")
|
||||
s.SetVec(pic, []float64{1, 0, 0}, "space")
|
||||
for i := 0; i < 6; i++ {
|
||||
d, _ := s.Put(makePNG(320, 240, i), Item{MIME: "image/jpeg", Tool: "multimodal_see_video"})
|
||||
s.Describe(d, fmt.Sprintf("视频第 %d 帧", i+1), "visionllm")
|
||||
s.SetVec(d, []float64{0, 1, float64(i)}, "space")
|
||||
}
|
||||
for i := 0; i < 10; i++ {
|
||||
d, _ := s.Put(makePNG(64, 64, 2000+i), Item{MIME: "image/png", Tool: "cmd_run"})
|
||||
@ -241,20 +254,24 @@ func TestSmoke_FullLifecycleAcrossRestart(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("重开后查不到: %v", err)
|
||||
}
|
||||
if it.Description == "" {
|
||||
t.Fatalf("元数据未持久化: %+v", it)
|
||||
if len(it.Vec) != 3 || it.VecModel != "space" {
|
||||
t.Fatalf("向量未持久化: %+v", it)
|
||||
}
|
||||
data, err := s2.Get(pic)
|
||||
if err != nil || !bytes.Equal(data, png) {
|
||||
t.Fatalf("重开后内容不一致: %v", err)
|
||||
}
|
||||
if hits, _ := s2.Search("三色带", KindImage, 10); len(hits) != 1 {
|
||||
t.Fatal("重开后描述应仍可检索")
|
||||
hits, err := s2.QueryMediaScored([]float64{1, 0, 0}, "space", 10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(hits) == 0 || hits[0].Item.Digest != pic {
|
||||
t.Fatal("重开后向量检索应仍能命中")
|
||||
}
|
||||
|
||||
// 磁盘文件数 == 元数据条数:无「元数据在文件没了」也无「文件在元数据没了」
|
||||
if n := blobFileCount(t, s2); n != beforeCount {
|
||||
t.Fatalf("磁盘 blob=%d 与元数据=%d 不一致", n, beforeCount)
|
||||
}
|
||||
t.Logf("跨重启:%d 条目、描述与内容全部完好", beforeCount)
|
||||
t.Logf("跨重启:%d 条目、向量与内容全部完好", beforeCount)
|
||||
}
|
||||
|
||||
@ -39,7 +39,7 @@ func TestSoak_SustainedMixedLoad(t *testing.T) {
|
||||
|
||||
stop := make(chan struct{})
|
||||
var wg sync.WaitGroup
|
||||
var puts, gets, deletes, describes, searches atomic.Int64
|
||||
var puts, gets, deletes, embeds, searches atomic.Int64
|
||||
var fatal atomic.Int64
|
||||
|
||||
worker := func(name string, fn func(iter int) error) {
|
||||
@ -109,25 +109,26 @@ func TestSoak_SustainedMixedLoad(t *testing.T) {
|
||||
return nil
|
||||
})
|
||||
|
||||
// 描述者
|
||||
worker("describe", func(i int) error {
|
||||
pend, err := s.Pending(5)
|
||||
// 向量写入者:持续给新内容嵌入并删除(模拟启动迁移/短命媒体)
|
||||
worker("embed", func(i int) error {
|
||||
b := make([]byte, 1024)
|
||||
rand.Read(b)
|
||||
b = append([]byte(fmt.Sprintf("emb-%d-", i)), b...)
|
||||
d, err := s.Put(b, Item{MIME: "image/png", Tool: "cmd_run"})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, it := range pend {
|
||||
// 忽略 unknown digest:GC 可能在 Pending 与 Describe 之间清掉它,
|
||||
// 这是正常竞态而非缺陷。
|
||||
_ = s.Describe(it.Digest, fmt.Sprintf("描述 %d 含图表与文字", i), "vis")
|
||||
describes.Add(1)
|
||||
if err := s.SetVec(d, []float64{1, float64(i % 7)}, "soak-space"); err != nil {
|
||||
return err
|
||||
}
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
embeds.Add(1)
|
||||
time.Sleep(time.Millisecond)
|
||||
return nil
|
||||
})
|
||||
|
||||
// 检索者
|
||||
worker("search", func(i int) error {
|
||||
if _, err := s.Search("图表", KindImage, 20); err != nil {
|
||||
if _, err := s.QueryMediaScored([]float64{1, 0}, "soak-space", 20); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := s.Stat(keep[i%keepN]); err != nil {
|
||||
@ -146,8 +147,8 @@ func TestSoak_SustainedMixedLoad(t *testing.T) {
|
||||
t.Fatalf("%d 个 worker 报致命错误", n)
|
||||
}
|
||||
|
||||
t.Logf("%v 内: put=%d get=%d delete=%d describe=%d search=%d",
|
||||
dur, puts.Load(), gets.Load(), deletes.Load(), describes.Load(), searches.Load())
|
||||
t.Logf("%v 内: put=%d get=%d delete=%d embed=%d search=%d",
|
||||
dur, puts.Load(), gets.Load(), deletes.Load(), embeds.Load(), searches.Load())
|
||||
|
||||
// 收尾断言
|
||||
for i, d := range keep {
|
||||
@ -161,6 +162,5 @@ func TestSoak_SustainedMixedLoad(t *testing.T) {
|
||||
}
|
||||
|
||||
st := s.Stats()
|
||||
t.Logf("收尾: 条目=%v 字节=%v 已描述=%v",
|
||||
st["count"], st["total_bytes"], st["described"])
|
||||
t.Logf("收尾: 条目=%v 字节=%v 类型=%v", st["count"], st["total_bytes"], st["by_kind"])
|
||||
}
|
||||
|
||||
@ -306,9 +306,9 @@ func TestStress_DeleteConcurrentWithReads(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestStress_DescribeConcurrentWithSearch(t *testing.T) {
|
||||
// 描述写入与检索并发。C 部分的后台描述任务会长期这样跑。
|
||||
s := newTestStore(t, 0)
|
||||
func TestStress_SetVecConcurrentWithQuery(t *testing.T) {
|
||||
// 嵌入写入与向量检索并发(启动时的向量迁移就会长期这样跑)。
|
||||
s := newTestStore(t)
|
||||
const n = 60
|
||||
digests := make([]string, n)
|
||||
for i := range digests {
|
||||
@ -320,55 +320,52 @@ func TestStress_DescribeConcurrentWithSearch(t *testing.T) {
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
var descErr, searchErr atomic.Int64
|
||||
var writeErr, queryErr atomic.Int64
|
||||
|
||||
// 描述写入者
|
||||
// 向量写入者
|
||||
for w := 0; w < 4; w++ {
|
||||
wg.Add(1)
|
||||
go func(wid int) {
|
||||
defer wg.Done()
|
||||
for i := wid; i < n; i += 4 {
|
||||
desc := fmt.Sprintf("第 %d 张图,含蓝色图表与文字", i)
|
||||
if err := s.Describe(digests[i], desc, "vis-src"); err != nil {
|
||||
descErr.Add(1)
|
||||
vec := []float64{1, float64(i) / 100, 0}
|
||||
if err := s.SetVec(digests[i], vec, "space"); err != nil {
|
||||
writeErr.Add(1)
|
||||
}
|
||||
}
|
||||
}(w)
|
||||
}
|
||||
|
||||
// 检索者 + Pending 消费者
|
||||
// 检索者
|
||||
for r := 0; r < 3; r++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for i := 0; i < 50; i++ {
|
||||
if _, err := s.Search("图表", KindImage, 20); err != nil {
|
||||
searchErr.Add(1)
|
||||
}
|
||||
if _, err := s.Pending(10); err != nil {
|
||||
searchErr.Add(1)
|
||||
if _, err := s.QueryMediaScored([]float64{1, 0, 0}, "space", 20); err != nil {
|
||||
queryErr.Add(1)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
if v := descErr.Load(); v > 0 {
|
||||
t.Fatalf("Describe 失败 %d 次", v)
|
||||
if v := writeErr.Load(); v > 0 {
|
||||
t.Fatalf("SetVec 失败 %d 次", v)
|
||||
}
|
||||
if v := searchErr.Load(); v > 0 {
|
||||
t.Fatalf("Search/Pending 失败 %d 次", v)
|
||||
if v := queryErr.Load(); v > 0 {
|
||||
t.Fatalf("QueryMediaScored 失败 %d 次", v)
|
||||
}
|
||||
|
||||
// 全部应已描述完
|
||||
pending, err := s.Pending(1000)
|
||||
// 全部应已嵌入,且都在同一空间
|
||||
stale, err := s.StaleVecDigestsAll("space")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(pending) != 0 {
|
||||
t.Fatalf("应全部描述完,仍有 %d 条未描述", len(pending))
|
||||
if len(stale) != 0 {
|
||||
t.Fatalf("应全部已嵌入,仍有 %d 条未嵌入", len(stale))
|
||||
}
|
||||
got, err := s.Search("图表", KindImage, 1000)
|
||||
got, err := s.QueryMediaScored([]float64{1, 0, 0}, "space", 1000)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user