mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-10-15 12:15:16 +00:00
9036db9321
## 问题（生产实证）
星图的「跟随 agent 活动」此前靠**工具名词元猜节点**：把 `memory_recall`
拆成 [memory, recall] 去匹配实体名。而图谱里的实体名是**中文概念**
（小宅/对话/待命），工具名是英文 —— 两套命名空间永不交集。
实测（生产库）每个工具名命中 **0** 个节点：
hint=memory_recall → memory/recall → 0
hint=files_read → files/read → 0
hint=knowledge_search → knowledge/search → 0
匹配失败后走「补齐」分支，按 mention_count 取前 8 个 ⇒ **无论 agent 在干什么，
点亮的都是同一批无关节点**（小宅/对话/待命/投递/图片/文档/插件/老大）。
这就是「星图没有读写反馈」的真身。
## 改法：让图数据库在真正读写处上报
新增 `events.EventMemoryAccess` + `GraphDB.AccessHook`，在读写入口埋点上报
**真实块 ID**（不再猜）：
| 操作 | 埋点 | 上报 |
|---|---|---|
| 读 | RecallSorted | blocks |
| 写 | commit | blocks + **created**（真正新建的） |
| 删 | DeleteEntity | removed |
| 合 | MergeBlocks | merged(from→to) + removed |
三处关键设计：
1. **钩子用独立锁** `hookMu`。用 `g.mu` 会**自死锁** —— 埋点方法已持该锁，
而 Go 的 RWMutex 不可重入。
2. **上报在锁外**。拆 `commit`→`commitLocked` 两层解决 `defer Unlock` 的时序：
钩子内部可能要回调内存系统（如 SSE 订阅者补数据），持锁调用必死锁。
3. **区分「新建」与「更新」**（`putBlockTx` 返回 isNew）。用 SELECT 判存在性
而非 `RETURNING (xmax = 0)` —— 后者在**事务 + ON CONFLICT** 下报
`no such column: xmax`（实测：autocommit 可用、包进 tx 就不可用），
而本函数只在事务里被调用。重复提交同一事实不会重复播「生长」动画。
接线：`bootstrap.newMainAgent` 里 `memDB.SetAccessHook(...)` → `events.Bus`。
只在 `evBus != nil` 时接，把「内核进程」与「独立 CLI 工具」
（homed-graph-migrate / memgc 也用 NewGraphDB）区分开 —— 后者不该为一次
批量迁移向不存在的订阅者发几万条事件。
## 验证
- `access_hook_test.go`：钉住 commit/recall/delete 三类埋点真的触发、
且重复提交不误报新建
- `access_hook_race_test.go`：4 写 + 4 读 + 钩子热替换并发，`-race` 下无竞态、
无死锁；**断言事件数 > 0**（否则「钩子从未触发」也会假通过）
- 该测试起初是 flaky 的（读者可能在写者落库前查询 ⇒ 空结果不报事件 ⇒
断言随机失败）。修法是先播种确定数据再并发读，已连跑 3 次稳定
- 生产实例端到端：CLI 注入一句话 → SSE 收到 6 条 memory_access，
带真实块 ID 与 op/created
122 lines
4.0 KiB
Go
122 lines
4.0 KiB
Go
package events
|
||
|
||
import (
|
||
"fmt"
|
||
"log"
|
||
"sync"
|
||
)
|
||
|
||
type EventType string
|
||
|
||
const (
|
||
EventRawInput EventType = "raw_input"
|
||
EventAgentOutput EventType = "agent_output"
|
||
EventAgentLLMChain EventType = "agent_llm_chain"
|
||
EventToolCall EventType = "tool_call"
|
||
EventReasoning EventType = "reasoning"
|
||
EventStage EventType = "stage"
|
||
// EventScheduler 是输入调度器的状态变更事件（抢占/挂起/恢复），
|
||
// 供状态页与诊断订阅（设计文档 §11 O2）。
|
||
EventScheduler EventType = "scheduler"
|
||
EventSystem EventType = "system"
|
||
EventTerminalOutput EventType = "terminal_output"
|
||
|
||
// 流式增量事件（LLM token 级）：核心改为流式后每收到一个增量块发布。
|
||
// 订阅者可选订；不认识的旧订阅者自然忽略（Bus 按 EventType 精确匹配分发）。
|
||
// 聚合事件 EventReasoning / EventAgentLLMChain 仍照常在每轮结束时全文发布，
|
||
// 插件体系行为不变。
|
||
EventReasoningDelta EventType = "reasoning_delta"
|
||
EventContentDelta EventType = "content_delta"
|
||
|
||
// skill_detected：clawhubadapter（OpenClaw 兼容层）扫描 skills 目录时
|
||
// 发现纯 SKILL 类型插件后发布，由原生 skillmgr 插件订阅并接管注册。
|
||
// 职责链：发现者（兼容层）→ 移交事件 → 归属者（skillmgr）加载管理。
|
||
EventSkillDetected EventType = "skill_detected"
|
||
|
||
// EventMemoryAccess 是**图记忆读写事件**：谁读了/写了哪几个记忆块。
|
||
//
|
||
// 为什么需要它（2026-10-08）：星图此前的「跟随 agent 活动」靠**工具名猜节点**
|
||
// （把 "memory_recall" 拆成词元去匹配实体名）。而图谱里的实体名是中文概念
|
||
// （小宅/对话/待命），与英文工具名永不交集 —— 实测每个工具名命中 0 个节点，
|
||
// 于是每次都回退到「按 mention_count 取前 8 个」，即每次工具调用点亮的都是
|
||
// **同一批无关节点**。
|
||
//
|
||
// 正确的做法是让**图数据库自己在读写处上报**：它本来就知道自己碰了哪些块 ID。
|
||
// payload 约定：
|
||
//
|
||
// op: "recall" | "commit" | "merge" | "purge" | "delete"
|
||
// blocks: []string，本次实际读写到的块 ID
|
||
// created: []string，本次**新建**的块 ID（驱动「从小变大」生长动画）
|
||
// removed: []string，本次**删除/合入消失**的块 ID（驱动粒子消散）
|
||
// merged: [][2]string 或 []map，合入对（驱动「两节点消失又出现」）
|
||
// tool: 触发这次访问的工具名（可空，便于 UI 显示来源）
|
||
//
|
||
// 订阅者（webui SSE）据此按**真实块 ID** 精确高亮，不再猜。
|
||
EventMemoryAccess EventType = "memory_access"
|
||
|
||
EventAll EventType = "*"
|
||
)
|
||
|
||
type Event struct {
|
||
Type EventType `json:"type"`
|
||
Source string `json:"source"`
|
||
Payload map[string]interface{} `json:"payload"`
|
||
Timestamp int64 `json:"timestamp"`
|
||
}
|
||
|
||
type Handler func(event *Event)
|
||
|
||
type Bus struct {
|
||
mu sync.RWMutex
|
||
subs map[EventType][]Handler
|
||
}
|
||
|
||
func NewBus() *Bus {
|
||
return &Bus{
|
||
subs: make(map[EventType][]Handler),
|
||
}
|
||
}
|
||
|
||
func (b *Bus) Publish(evt *Event) {
|
||
b.mu.RLock()
|
||
allHandlers := make([]Handler, len(b.subs[EventAll]))
|
||
copy(allHandlers, b.subs[EventAll])
|
||
typeHandlers := make([]Handler, len(b.subs[evt.Type]))
|
||
copy(typeHandlers, b.subs[evt.Type])
|
||
b.mu.RUnlock()
|
||
|
||
for _, h := range allHandlers {
|
||
b.safeCall(h, evt)
|
||
}
|
||
for _, h := range typeHandlers {
|
||
b.safeCall(h, evt)
|
||
}
|
||
}
|
||
|
||
func (b *Bus) safeCall(h Handler, evt *Event) {
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
log.Printf("[bus] handler panic: %v", r)
|
||
}
|
||
}()
|
||
h(evt)
|
||
}
|
||
|
||
func (b *Bus) Subscribe(eventType EventType, handler Handler) func() {
|
||
b.mu.Lock()
|
||
b.subs[eventType] = append(b.subs[eventType], handler)
|
||
b.mu.Unlock()
|
||
|
||
return func() {
|
||
b.mu.Lock()
|
||
defer b.mu.Unlock()
|
||
list := b.subs[eventType]
|
||
for i, h := range list {
|
||
if fmt.Sprintf("%p", h) == fmt.Sprintf("%p", handler) {
|
||
b.subs[eventType] = append(list[:i], list[i+1:]...)
|
||
break
|
||
}
|
||
}
|
||
}
|
||
}
|