mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-10-03 15:53:56 +00:00
fix(memory): 记忆层启动接线/并发/落盘一致性整备
按设计方案整顿记忆系统，收敛一批"单测照不出、只在长跑生产里暴露"的缺陷： - 启动接线：initMemoryStack 残留 `defer distiller.Stop()`，规则蒸馏 10min 心跳启动即死。改为由调用点 cleanup 停机，并补 Stopped() 探针 + TestInitMemoryStackKeepsDistillerRunning / TestStartKeepsLoopRunningUntilStop。 - L0 相关性上下文：SetDenseSpace 注入稠密空间时回填已有事件的稠密向量， 否则旧事件走稀疏余弦、新事件走稠密余弦，同一次 Prune 里两种尺度混排。 - 文档检索：QueryScored 访问计数从读锁内写移出（-race 竞争），更新后置脏， 优雅关停可落盘、FindColdDocs 冷度判据跨重启不再失真。 - 蒸馏管线：只 flush 未落盘记录（persisted 标记）、蒸馏成功后从 raw 文件 删除对应行、原子写文件，修重启重复蒸馏导致 mention_count 膨胀。 - 图库：全量 Recall 加实体上限（内部整备路径，防大图整表进内存）； ClearSentenceID 补写锁；Commit/upsertEntity 计数语义注释澄清。 - 索引器：recalled 去重集加 FIFO 上限，防长跑进程自动注入越来越沉默。 - 文档/媒体注释修正；README 记忆层流程对齐跨模态召回。 验证：go build ./...、go vet、go test -race ./internal/memory/... ./internal/agent/core/... ./cmd/homed/... 全绿。
This commit is contained in:
JianFeeeee
@ -24,6 +24,12 @@ type RawRecord struct {
|
||||
Content string `json:"content"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
Distilled bool `json:"distilled"`
|
||||
|
||||
// persisted 表示该记录已经写在磁盘 raw 文件里。
|
||||
// 未导出：只影响本进程的落盘行为，不进 JSON。
|
||||
// 作用：flush 只写**尚未落盘**的记录，否则 loadExisting 载入的记录
|
||||
// 会被再写一份，重启后同一批记录从新旧两个文件各读回一次。
|
||||
persisted bool
|
||||
}
|
||||
|
||||
type DistillerConfig struct {
|
||||
@ -33,16 +39,16 @@ type DistillerConfig struct {
|
||||
}
|
||||
|
||||
type Distiller struct {
|
||||
mu sync.Mutex
|
||||
db *memory.GraphDB
|
||||
rawPath string
|
||||
records []RawRecord
|
||||
nextID int64
|
||||
cfg DistillerConfig
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
onMemory func(input, response string)
|
||||
embedder nlp.Vectorizer
|
||||
mu sync.Mutex
|
||||
db *memory.GraphDB
|
||||
rawPath string
|
||||
records []RawRecord
|
||||
nextID int64
|
||||
cfg DistillerConfig
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
onMemory func(input, response string)
|
||||
embedder nlp.Vectorizer
|
||||
}
|
||||
|
||||
func (d *Distiller) SetEmbedder(ev nlp.Vectorizer) { d.embedder = ev }
|
||||
@ -76,6 +82,22 @@ func (d *Distiller) Stop() {
|
||||
d.flush()
|
||||
}
|
||||
|
||||
// Stopped 报告蒸馏循环是否已被 Stop() 取消。
|
||||
//
|
||||
// Stop() 里的 cancel() 是同步生效的，所以本方法在 Stop() 返回后立即为 true，
|
||||
// 不受循环 goroutine 何时退出的影响。启动自检、健康检查用它确认
|
||||
// 「Start 之后没有被立即 Stop 掉」——历史回归：main() 拆分时
|
||||
// initMemoryStack 里残留一句 defer distiller.Stop()，函数一返回就把刚起的
|
||||
// 循环杀了，10min 心跳从不运行。Start() 之前返回 false（尚未被停）。
|
||||
func (d *Distiller) Stopped() bool {
|
||||
select {
|
||||
case <-d.ctx.Done():
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Distiller) Append(sessionID string, role string, content string) {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
@ -89,19 +111,35 @@ func (d *Distiller) Append(sessionID string, role string, content string) {
|
||||
func (d *Distiller) flush() {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
if len(d.records) == 0 {
|
||||
d.flushLocked()
|
||||
}
|
||||
|
||||
// flushLocked 把**尚未落盘**的记录追加写入一个新的 raw 文件（原子写）。
|
||||
//
|
||||
// 只写 !persisted 的记录：loadExisting 载入的记录已经在磁盘上，若 flush 再把
|
||||
// 它们整体重写一份，重启后同一批记录会同时从旧文件与新文件被读回，
|
||||
// 实体 mention_count 与关系被重复蒸馏。
|
||||
func (d *Distiller) flushLocked() {
|
||||
var pending []RawRecord
|
||||
for _, r := range d.records {
|
||||
if !r.persisted {
|
||||
pending = append(pending, r)
|
||||
}
|
||||
}
|
||||
if len(pending) == 0 {
|
||||
return
|
||||
}
|
||||
path := filepath.Join(d.rawPath, fmt.Sprintf("raw_%d.tsv", time.Now().UnixNano()))
|
||||
f, err := os.Create(path)
|
||||
if err != nil {
|
||||
var sb strings.Builder
|
||||
for _, r := range pending {
|
||||
fmt.Fprintf(&sb, "%d\t%s\t%s\t%s\t%d\n", r.ID, r.SessionID, r.Role, r.Content, r.CreatedAt.Unix())
|
||||
}
|
||||
if err := writeFileAtomic(path, []byte(sb.String())); err != nil {
|
||||
log.Printf("[memory] flush error: %v", err)
|
||||
return
|
||||
}
|
||||
defer f.Close()
|
||||
for _, r := range d.records {
|
||||
line := fmt.Sprintf("%d\t%s\t%s\t%s\t%d\n", r.ID, r.SessionID, r.Role, r.Content, r.CreatedAt.Unix())
|
||||
f.WriteString(line)
|
||||
for i := range d.records {
|
||||
d.records[i].persisted = true
|
||||
}
|
||||
}
|
||||
|
||||
@ -167,6 +205,7 @@ func (d *Distiller) loadExisting() {
|
||||
}
|
||||
d.records = append(d.records, RawRecord{
|
||||
ID: d.nextID, SessionID: parts[1], Role: parts[2], Content: parts[3], CreatedAt: createdAt,
|
||||
persisted: true,
|
||||
})
|
||||
d.nextID++
|
||||
loaded++
|
||||
@ -222,6 +261,9 @@ func (d *Distiller) distillOnce() {
|
||||
}
|
||||
if d.distillBatch(toDistill[i:end]) {
|
||||
distilled += end - i
|
||||
// 成功即从磁盘 raw 文件里删除这些行，否则重启后 loadExisting
|
||||
// 会把它们当未蒸馏记录重新读回，每次启动重蒸同一批历史。
|
||||
d.removeRawRecords(toDistill[i:end])
|
||||
} else {
|
||||
// 蒸馏失败：记录写回待处理队列，下次 tick 重试
|
||||
d.mu.Lock()
|
||||
@ -279,6 +321,80 @@ func (d *Distiller) cleanupRawFiles() {
|
||||
}
|
||||
}
|
||||
|
||||
// writeFileAtomic 写临时文件再 rename，避免进程在写一半时崩溃留下半个文件。
|
||||
func writeFileAtomic(path string, data []byte) error {
|
||||
tmp := path + ".tmp"
|
||||
if err := os.WriteFile(tmp, data, 0644); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(tmp, path)
|
||||
}
|
||||
|
||||
// rawKey 唯一标识一条原始记录，用于在 raw 文件里按内容定位并删除。
|
||||
//
|
||||
// 为什么按内容而不是 ID：loadExisting 读回记录时会重新分配连续 ID，
|
||||
// 文件里的 ID 与内存 ID 并不一一对应。
|
||||
func rawKey(session, role, content string, ts int64) string {
|
||||
return session + "\x00" + role + "\x00" + content + "\x00" + strconv.FormatInt(ts, 10)
|
||||
}
|
||||
|
||||
// removeRawRecords 从磁盘 raw 文件中删除已成功蒸馏的记录。
|
||||
//
|
||||
// 蒸馏成功后记录若只从内存移除、磁盘文件不动，下次启动 loadExisting 会把
|
||||
// 它们当未蒸馏记录重新读回，导致每次重启都重蒸同一批历史（实体
|
||||
// mention_count 膨胀，且 distillBatch 的 sessionID 取自 map 首个键，
|
||||
// 不确定性会放大重复）。
|
||||
func (d *Distiller) removeRawRecords(batch []RawRecord) {
|
||||
if len(batch) == 0 {
|
||||
return
|
||||
}
|
||||
drop := make(map[string]bool, len(batch))
|
||||
for _, r := range batch {
|
||||
drop[rawKey(r.SessionID, r.Role, r.Content, r.CreatedAt.Unix())] = true
|
||||
}
|
||||
entries, err := os.ReadDir(d.rawPath)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
for _, entry := range entries {
|
||||
ext := filepath.Ext(entry.Name())
|
||||
if ext != ".tsv" && ext != ".jsonl" {
|
||||
continue
|
||||
}
|
||||
path := filepath.Join(d.rawPath, entry.Name())
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
var kept []string
|
||||
removed := false
|
||||
for _, line := range strings.Split(strings.TrimRight(string(data), "\n"), "\n") {
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
parts := splitLine(line)
|
||||
if len(parts) >= 5 {
|
||||
if ts, err := strconv.ParseInt(parts[4], 10, 64); err == nil &&
|
||||
drop[rawKey(parts[1], parts[2], parts[3], ts)] {
|
||||
removed = true
|
||||
continue
|
||||
}
|
||||
}
|
||||
kept = append(kept, line)
|
||||
}
|
||||
if !removed {
|
||||
continue
|
||||
}
|
||||
if len(kept) == 0 {
|
||||
os.Remove(path)
|
||||
continue
|
||||
}
|
||||
if err := writeFileAtomic(path, []byte(strings.Join(kept, "\n")+"\n")); err != nil {
|
||||
log.Printf("[memory] rewrite raw %s: %v", entry.Name(), err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func extractKeyTriples(userContent, assistantContent string, embedder nlp.Vectorizer) []memory.Triple {
|
||||
var triples []memory.Triple
|
||||
|
||||
@ -310,13 +426,6 @@ func truncate(s string, max int) string {
|
||||
return s
|
||||
}
|
||||
|
||||
func parseLines(data string) []string {
|
||||
if data == "" {
|
||||
return nil
|
||||
}
|
||||
return strings.Split(strings.TrimRight(data, "\n"), "\n")
|
||||
}
|
||||
|
||||
func splitLine(line string) []string {
|
||||
if line == "" {
|
||||
return nil
|
||||
|
||||
@ -185,9 +185,9 @@ func TestDistillOnceBatchLimit(t *testing.T) {
|
||||
|
||||
func TestExtractKeyTriples(t *testing.T) {
|
||||
tests := []struct {
|
||||
user string
|
||||
user string
|
||||
assistant string
|
||||
check func([]memory.Triple) bool
|
||||
check func([]memory.Triple) bool
|
||||
}{
|
||||
{
|
||||
user: "我住在北京",
|
||||
@ -201,7 +201,7 @@ func TestExtractKeyTriples(t *testing.T) {
|
||||
},
|
||||
},
|
||||
{
|
||||
user: "我在杭州读书",
|
||||
user: "我在杭州读书",
|
||||
assistant: "好的",
|
||||
check: func(triples []memory.Triple) bool {
|
||||
for _, tr := range triples {
|
||||
@ -264,3 +264,25 @@ func TestTruncate(t *testing.T) {
|
||||
t.Errorf("expected 'hi', got %q", truncate("hi", 10))
|
||||
}
|
||||
}
|
||||
|
||||
// TestStartKeepsLoopRunningUntilStop 锁死 Start/Stop 的接线契约：
|
||||
// Start 之后蒸馏循环必须处于运行态，只有显式 Stop 才退出。
|
||||
//
|
||||
// 这条回归的直接来源：main() 拆分时 initMemoryStack 里残留一句
|
||||
// defer distiller.Stop()，函数返回即 cancel，循环启动即死——而
|
||||
// TestDistillOnce* 直接调 distillOnce，绕过了 Start/Stop，照不出这个洞。
|
||||
func TestStartKeepsLoopRunningUntilStop(t *testing.T) {
|
||||
d := NewDistiller(nil, t.TempDir(), DistillerConfig{
|
||||
Interval: time.Hour, // 不依赖 tick，只验证循环存活
|
||||
RetentionDays: 7,
|
||||
BatchSize: 50,
|
||||
})
|
||||
d.Start()
|
||||
if d.Stopped() {
|
||||
t.Fatal("Start 之后蒸馏循环必须处于运行态（不可被 defer Stop 杀掉）")
|
||||
}
|
||||
d.Stop()
|
||||
if !d.Stopped() {
|
||||
t.Fatal("Stop 之后应处于已停止态")
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user