fix: IPC 字段名 + recall_count 持久化 + feedback 端点
1. IPC 字段名修复 (lancedb_ipc.go) - Rust lancedb_search 返回 'result' 字段，Go ipcResp 用 'report_json' - 两边字段名不匹配导致 ReportJSON 永远为空，Search() 降级到 localSearch - 后果：recall_count 只写 Go 缓存，不写 LanceDB，重启后丢失 - 修复：ipcResp 加 'Result' 字段 (json:"result")，Search() 改用 resp.Result 2. Update 扩展 支持 (lancedb_ipc.go) - 原来只有 recall_count 支持 ，useful_count/not_useful_count 被丢弃 - 新增 getIncCachedValue() 统一处理所有 counter 字段 - 加日志：Update OK / Update FAILED 可见 - 修复 Feedback 端点 传递问题（map[string]int 而非 map[string]string） 3. /api/v1/feedback 端点 (core.go + server.go) - POST memory_id + useful/not_useful，通过 LanceDB.Update() 持久化 - 之前返回 404 破坏性：无（Search 降级是隐式行为，原代码已有 fallback）
This commit is contained in:
xiaowei
parent
190e700876
commit
8d3582046c
|
|
@ -253,6 +253,35 @@ func (a *API) Recall(w http.ResponseWriter, r *http.Request) {
|
|||
respond(w, 200, map[string]interface{}{"results": results, "count": len(results)})
|
||||
}
|
||||
|
||||
// POST /api/v1/feedback — 用户反馈记忆是否有用，同时更新 useful_count/not_useful_count
|
||||
func (a *API) Feedback(w http.ResponseWriter, r *http.Request) {
|
||||
var req struct {
|
||||
MemoryID string `json:"memory_id"`
|
||||
Useful bool `json:"useful"`
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
respondError(w, 400, "invalid body")
|
||||
return
|
||||
}
|
||||
if req.MemoryID == "" {
|
||||
respondError(w, 400, "memory_id required")
|
||||
return
|
||||
}
|
||||
field := "useful_count"
|
||||
val := map[string]int{"$inc": 1}
|
||||
if !req.Useful {
|
||||
field = "not_useful_count"
|
||||
}
|
||||
err := a.LanceDB.Update("memories", req.MemoryID, map[string]any{
|
||||
field: val,
|
||||
})
|
||||
if err != nil {
|
||||
respondError(w, 500, "feedback update failed: "+err.Error())
|
||||
return
|
||||
}
|
||||
respond(w, 200, map[string]string{"status": "ok"})
|
||||
}
|
||||
|
||||
// POST /api/v1/recall/debug — recall 诊断端点（优化），返回各阶段耗时和状态
|
||||
func (a *API) RecallDebug(w http.ResponseWriter, r *http.Request) {
|
||||
var req struct {
|
||||
|
|
|
|||
|
|
@ -305,6 +305,7 @@ func NewServer() http.Handler {
|
|||
})
|
||||
mux.HandleFunc("/api/v1/recall", api.Recall)
|
||||
mux.HandleFunc("/api/v1/recall/debug", api.RecallDebug)
|
||||
mux.HandleFunc("/api/v1/feedback", api.Feedback)
|
||||
mux.HandleFunc("/api/v1/bootstrap", api.Bootstrap)
|
||||
mux.HandleFunc("/api/v1/stats", api.Stats)
|
||||
mux.HandleFunc("/api/v1/batch-commit", api.BatchCommit)
|
||||
|
|
|
|||
|
|
@ -45,10 +45,11 @@ type ipcReq struct {
|
|||
}
|
||||
|
||||
type ipcResp struct {
|
||||
Status string `json:"status"`
|
||||
ReportJSON string `json:"report_json"`
|
||||
FailureStep string `json:"failure_step"`
|
||||
ErrorDetail string `json:"error_detail"`
|
||||
Status string `json:"status"`
|
||||
ReportJSON string `json:"report_json"` // consolidator uses this
|
||||
Result string `json:"result"` // lancedb search uses this
|
||||
FailureStep string `json:"failure_step"`
|
||||
ErrorDetail string `json:"error_detail"`
|
||||
}
|
||||
|
||||
func (rc *RustLanceDBClient) rpc(req ipcReq) (*ipcResp, error) {
|
||||
|
|
@ -156,7 +157,12 @@ func (rc *RustLanceDBClient) Search(table string, vector []float32, topK int, na
|
|||
Tier string `json:"tier"`
|
||||
LastRecalledAt string `json:"last_recalled_at"`
|
||||
}
|
||||
json.Unmarshal([]byte(resp.ReportJSON), &raw)
|
||||
json.Unmarshal([]byte(resp.Result), &raw)
|
||||
if len(raw) == 0 {
|
||||
log.Printf("[ipc] Search: resp.Result unmarshal gave 0 results (ReportJSON=%q), falling back to localSearch", resp.ReportJSON)
|
||||
return rc.localSearch(vector, topK, namespaceFilter), nil
|
||||
}
|
||||
log.Printf("[ipc] Search: got %d results from Rust IPC (ReportJSON=%q)", len(raw), resp.ReportJSON)
|
||||
|
||||
// 解析时间（Rust 返回 RFC3339 字符串 → time.Time）
|
||||
parseTime := func(s string) time.Time {
|
||||
|
|
@ -293,6 +299,21 @@ func (rc *RustLanceDBClient) Insert(table string, record any) error {
|
|||
return fmt.Errorf("unknown type")
|
||||
}
|
||||
|
||||
// helper: get a counter field's current value from cache, returns cache+1 or 1 if not cached
|
||||
func getIncCachedValue(id, field string, cache map[string]*models.MemoryRecord) int {
|
||||
if m, ok := cache[id]; ok {
|
||||
switch field {
|
||||
case "recall_count":
|
||||
return m.RecallCount + 1
|
||||
case "useful_count":
|
||||
return m.UsefulCount + 1
|
||||
case "not_useful_count":
|
||||
return m.NotUsefulCount + 1
|
||||
}
|
||||
}
|
||||
return 1
|
||||
}
|
||||
|
||||
// Update 同步执行：整个操作在锁内（cache read + IPC call）
|
||||
// 保证并发调用时不会 read stale value 再写回
|
||||
func (rc *RustLanceDBClient) Update(table, id string, fields map[string]any) error {
|
||||
|
|
@ -310,6 +331,36 @@ func (rc *RustLanceDBClient) Update(table, id string, fields map[string]any) err
|
|||
if inc["$inc"] == "1" {
|
||||
m.RecallCount++
|
||||
}
|
||||
case int:
|
||||
m.RecallCount = inc
|
||||
}
|
||||
}
|
||||
if v, ok := fields["useful_count"]; ok {
|
||||
switch inc := v.(type) {
|
||||
case map[string]string:
|
||||
if inc["$inc"] == "1" {
|
||||
m.UsefulCount++
|
||||
}
|
||||
case map[string]interface{}:
|
||||
if inc["$inc"] == "1" {
|
||||
m.UsefulCount++
|
||||
}
|
||||
case int:
|
||||
m.UsefulCount = inc
|
||||
}
|
||||
}
|
||||
if v, ok := fields["not_useful_count"]; ok {
|
||||
switch inc := v.(type) {
|
||||
case map[string]string:
|
||||
if inc["$inc"] == "1" {
|
||||
m.NotUsefulCount++
|
||||
}
|
||||
case map[string]interface{}:
|
||||
if inc["$inc"] == "1" {
|
||||
m.NotUsefulCount++
|
||||
}
|
||||
case int:
|
||||
m.NotUsefulCount = inc
|
||||
}
|
||||
}
|
||||
if v, ok := fields["last_recalled_at"]; ok {
|
||||
|
|
@ -322,7 +373,6 @@ func (rc *RustLanceDBClient) Update(table, id string, fields map[string]any) err
|
|||
}
|
||||
|
||||
// 2. 通过 IPC 持久化到 Rust LanceDB（仍在锁内，防止并发写同一行）
|
||||
// 构建 fields JSON: recall_count 发送实际值（非 "1" 而是当前 cache 值 +1）
|
||||
type updateField struct {
|
||||
Column string `json:"column"`
|
||||
Value string `json:"value"`
|
||||
|
|
@ -333,27 +383,14 @@ func (rc *RustLanceDBClient) Update(table, id string, fields map[string]any) err
|
|||
case string:
|
||||
fieldList = append(fieldList, updateField{k, "'" + strings.ReplaceAll(val, "'", "''") + "'"})
|
||||
case map[string]string:
|
||||
if k == "recall_count" {
|
||||
if inc, ok := val["$inc"]; ok {
|
||||
// 读 cache 当前值并 +1，保证每条 update 发送正确的递增结果
|
||||
if m, ok := _local.memories[id]; ok {
|
||||
newVal := m.RecallCount
|
||||
fieldList = append(fieldList, updateField{k, fmt.Sprintf("%d", newVal)})
|
||||
_ = inc // suppress unused
|
||||
} else {
|
||||
fieldList = append(fieldList, updateField{k, inc})
|
||||
}
|
||||
}
|
||||
if val["$inc"] == "1" {
|
||||
newVal := getIncCachedValue(id, k, _local.memories)
|
||||
fieldList = append(fieldList, updateField{k, fmt.Sprintf("%d", newVal)})
|
||||
}
|
||||
case map[string]interface{}:
|
||||
if k == "recall_count" {
|
||||
if inc, ok := val["$inc"]; ok {
|
||||
if m, ok := _local.memories[id]; ok {
|
||||
fieldList = append(fieldList, updateField{k, fmt.Sprintf("%d", m.RecallCount)})
|
||||
} else {
|
||||
fieldList = append(fieldList, updateField{k, fmt.Sprintf("%v", inc)})
|
||||
}
|
||||
}
|
||||
if val["$inc"] == float64(1) || val["$inc"] == 1 {
|
||||
newVal := getIncCachedValue(id, k, _local.memories)
|
||||
fieldList = append(fieldList, updateField{k, fmt.Sprintf("%d", newVal)})
|
||||
}
|
||||
case float64:
|
||||
fieldList = append(fieldList, updateField{k, fmt.Sprintf("%f", val)})
|
||||
|
|
@ -370,14 +407,16 @@ func (rc *RustLanceDBClient) Update(table, id string, fields map[string]any) err
|
|||
log.Printf("[ipc] Update: id=%s fields_count=%d", id, len(fieldList))
|
||||
log.Printf("[ipc] Update fields_json: %s", string(fj))
|
||||
_, err := rc.rpc(ipcReq{
|
||||
Type: "lancedb_update",
|
||||
Table: table,
|
||||
ID: id,
|
||||
Fields: string(fj),
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("[ipc] update failed (memory only): %v", err)
|
||||
}
|
||||
Type: "lancedb_update",
|
||||
Table: table,
|
||||
ID: id,
|
||||
Fields: string(fj),
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("[ipc] Update FAILED: id=%s fields=%s error=%v", id, string(fj), err)
|
||||
} else {
|
||||
log.Printf("[ipc] Update OK: id=%s fields=%s", id, string(fj))
|
||||
}
|
||||
}
|
||||
_local.mu.Unlock()
|
||||
return nil
|
||||
|
|
|
|||
Loading…
Reference in New Issue