fix G7 E3: crystallize路由路径+GetSkillCandidates通过Rust IPC查LanceDB
- server.go: 修复mux路由路径（去掉"POST "前缀） - skill_crystallize.go: 添加POST方法检查 - models/memory.go: 添加safeTime类型避免空字符串解析报错 - lancedb_ipc.go: GetSkillCandidates改用Rust IPC的lancedb_query - Rust lancedb_ops.rs: 新增query_memories()支持min_recall过滤 - Rust main.rs: 新增lancedb_query IPC handler
This commit is contained in:
xiaowei
parent
243a2066a4
commit
b0059166a2
|
|
@ -73,6 +73,10 @@ func isMemoryLinked(memID string) bool {
|
|||
|
||||
// CrystallizeSkill 对指定记忆执行结晶（HTTP 路由）
|
||||
func CrystallizeSkill(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
respondError(w, 405, "POST required")
|
||||
return
|
||||
}
|
||||
memID := r.PathValue("id")
|
||||
if memID == "" {
|
||||
respondError(w, 400, "memory id required")
|
||||
|
|
|
|||
|
|
@ -693,7 +693,7 @@ func NewServer() http.Handler {
|
|||
mux.HandleFunc("/api/v1/skills/{name}/execute", routes.ExecuteSkill) // G7.3 skill执行
|
||||
// G7.2: 结晶路由
|
||||
mux.HandleFunc("/api/v1/crystallize/candidates", routes.GetSkillCandidates) // 获取候选
|
||||
mux.HandleFunc("POST /api/v1/crystallize/memory/{id}", routes.CrystallizeSkill) // 对记忆执行结晶
|
||||
mux.HandleFunc("/api/v1/crystallize/memory/{id}", routes.CrystallizeSkill) // 对记忆执行结晶
|
||||
|
||||
// 评估 + 自调参
|
||||
mux.HandleFunc("/api/v1/eval/run", evalAPI.Run)
|
||||
|
|
|
|||
|
|
@ -3,6 +3,27 @@ package models
|
|||
|
||||
import "time"
|
||||
|
||||
// Time 是 time.Time 的安全版本，支持空字符串解析
|
||||
type Time struct{ time.Time }
|
||||
|
||||
// UnmarshalJSON 实现 json.Unmarshaler：空字符串解析为当前时间（而非报错）
|
||||
func (t *Time) UnmarshalJSON(data []byte) error {
|
||||
s := string(data)
|
||||
if s == `""` || s == `"0001-01-01T00:00:00Z"` {
|
||||
t.Time = time.Time{}
|
||||
return nil
|
||||
}
|
||||
// 去掉前后引号
|
||||
if len(s) >= 2 && s[0] == '"' && s[len(s)-1] == '"' {
|
||||
s = s[1 : len(s)-1]
|
||||
}
|
||||
if s == "" || s == "null" {
|
||||
t.Time = time.Time{}
|
||||
return nil
|
||||
}
|
||||
return t.Time.UnmarshalJSON(data)
|
||||
}
|
||||
|
||||
// VersionRecord 记忆版本记录（溯源链中的单次修改）
|
||||
type VersionRecord struct {
|
||||
Version int `json:"version"`
|
||||
|
|
|
|||
|
|
@ -42,6 +42,8 @@ type ipcReq struct {
|
|||
Namespace string `json:"namespace,omitempty"`
|
||||
ID string `json:"id,omitempty"`
|
||||
Fields string `json:"fields,omitempty"`
|
||||
MinRecall int `json:"min_recall,omitempty"`
|
||||
QueryLimit int `json:"limit,omitempty"`
|
||||
}
|
||||
|
||||
type ipcResp struct {
|
||||
|
|
@ -465,20 +467,112 @@ func (rc *RustLanceDBClient) GetCandidatesForForgetting() ([]map[string]interfac
|
|||
return out, nil
|
||||
}
|
||||
|
||||
func (rc *RustLanceDBClient) GetSkillCandidates(minRecalls, limit int) ([]models.MemoryRecord, error) {
|
||||
_local.mu.RLock()
|
||||
defer _local.mu.RUnlock()
|
||||
var out []models.MemoryRecord
|
||||
for _, m := range _local.memories {
|
||||
if m.IsDeleted || m.Tier == "core" || m.RecallCount < minRecalls {
|
||||
continue
|
||||
}
|
||||
out = append(out, *m)
|
||||
if len(out) >= limit {
|
||||
break
|
||||
func mapStr(m map[string]interface{}, key string) string {
|
||||
if v, ok := m[key]; ok {
|
||||
if s, ok := v.(string); ok {
|
||||
return s
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
return ""
|
||||
}
|
||||
func mapFloat(m map[string]interface{}, key string) float64 {
|
||||
if v, ok := m[key]; ok {
|
||||
if f, ok := v.(float64); ok {
|
||||
return f
|
||||
}
|
||||
}
|
||||
return 0
|
||||
}
|
||||
func mapInt(m map[string]interface{}, key string) int {
|
||||
if v, ok := m[key]; ok {
|
||||
if f, ok := v.(float64); ok {
|
||||
return int(f)
|
||||
}
|
||||
}
|
||||
return 0
|
||||
}
|
||||
func mapBool(m map[string]interface{}, key string) bool {
|
||||
if v, ok := m[key]; ok {
|
||||
if b, ok := v.(bool); ok {
|
||||
return b
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (rc *RustLanceDBClient) GetSkillCandidates(minRecalls, limit int) ([]models.MemoryRecord, error) {
|
||||
resp, err := rc.rpc(ipcReq{
|
||||
Type: "lancedb_query",
|
||||
MinRecall: minRecalls,
|
||||
QueryLimit: limit,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("lancedb_query: %w", err)
|
||||
}
|
||||
raw := resp.ReportJSON
|
||||
if raw == "" {
|
||||
return []models.MemoryRecord{}, nil
|
||||
}
|
||||
|
||||
// 用 raw JSON 反序列化，避免 time.Time 空字符串报错
|
||||
var rawRecords []map[string]interface{}
|
||||
if err := json.Unmarshal([]byte(raw), &rawRecords); err != nil {
|
||||
return nil, fmt.Errorf("parse lancedb_query result: %w", err)
|
||||
}
|
||||
if len(rawRecords) == 0 {
|
||||
return []models.MemoryRecord{}, nil
|
||||
}
|
||||
|
||||
records := make([]models.MemoryRecord, 0, len(rawRecords))
|
||||
for _, r := range rawRecords {
|
||||
m := models.MemoryRecord{
|
||||
ID: mapStr(r, "id"),
|
||||
AgentID: mapStr(r, "agent_id"),
|
||||
Namespace: mapStr(r, "namespace"),
|
||||
Content: mapStr(r, "content"),
|
||||
Category: mapStr(r, "category"),
|
||||
Tier: mapStr(r, "tier"),
|
||||
Source: mapStr(r, "source"),
|
||||
Freshness: mapStr(r, "freshness"),
|
||||
DerivedFrom: mapStr(r, "derived_from"),
|
||||
VersionHistory: nil,
|
||||
DependsOn: nil,
|
||||
}
|
||||
m.QualityScore = mapFloat(r, "quality_score")
|
||||
m.Importance = mapFloat(r, "importance")
|
||||
m.UsefulCount = mapInt(r, "useful_count")
|
||||
m.NotUsefulCount = mapInt(r, "not_useful_count")
|
||||
m.RecallCount = mapInt(r, "recall_count")
|
||||
m.Version = mapInt(r, "version")
|
||||
m.VolatileFlag = mapBool(r, "volatile_flag")
|
||||
m.IsDeleted = mapBool(r, "is_deleted")
|
||||
// 时间字段：空字符串 → 零值 time.Time
|
||||
if t := mapStr(r, "last_recalled_at"); t != "" {
|
||||
if pt, err := time.Parse(time.RFC3339, t); err == nil {
|
||||
m.LastRecalledAt = pt
|
||||
}
|
||||
}
|
||||
if t := mapStr(r, "created_at"); t != "" {
|
||||
if pt, err := time.Parse(time.RFC3339, t); err == nil {
|
||||
m.CreatedAt = pt
|
||||
}
|
||||
}
|
||||
if t := mapStr(r, "updated_at"); t != "" {
|
||||
if pt, err := time.Parse(time.RFC3339, t); err == nil {
|
||||
m.UpdatedAt = pt
|
||||
}
|
||||
}
|
||||
records = append(records, m)
|
||||
}
|
||||
|
||||
// 同步到本地缓存
|
||||
_local.mu.Lock()
|
||||
for _, m := range records {
|
||||
cp := m
|
||||
_local.memories[m.ID] = &cp
|
||||
}
|
||||
_local.mu.Unlock()
|
||||
return records, nil
|
||||
}
|
||||
|
||||
func (rc *RustLanceDBClient) Backup(path string) error { return nil }
|
||||
|
|
|
|||
|
|
@ -131,6 +131,52 @@ impl LanceDBOps {
|
|||
Ok(records)
|
||||
}
|
||||
|
||||
/// 查询记忆（支持 min_recall_count 过滤），供 GetSkillCandidates 使用
|
||||
pub fn query_memories(&self, min_recall: i64, limit: usize) -> Result<Vec<MemoryRecord>, Box<dyn std::error::Error>> {
|
||||
let db = rt().block_on(lancedb::connect(self.data_dir.to_str().unwrap()).execute())?;
|
||||
let tbl = rt().block_on(db.open_table("memories").execute())?;
|
||||
|
||||
let filter = format!("recall_count >= {} AND is_deleted = false AND tier != 'core'", min_recall);
|
||||
let mut results = Box::pin(rt().block_on(
|
||||
tbl.query()
|
||||
.only_if(&filter)
|
||||
.limit(limit)
|
||||
.execute()
|
||||
)?);
|
||||
|
||||
let mut records = Vec::new();
|
||||
while let Some(Ok(batch)) = rt().block_on(results.next()) {
|
||||
for i in 0..batch.num_rows() {
|
||||
records.push(MemoryRecord {
|
||||
id: col_str(&batch, i, "id"),
|
||||
agent_id: col_str(&batch, i, "agent_id"),
|
||||
namespace: col_str(&batch, i, "namespace"),
|
||||
content: col_str(&batch, i, "content"),
|
||||
category: col_str(&batch, i, "category"),
|
||||
vector: col_vector(&batch, i, "vector"),
|
||||
tier: col_str(&batch, i, "tier"),
|
||||
importance: col_f64(&batch, i, "importance"),
|
||||
quality_score: col_f64(&batch, i, "quality_score"),
|
||||
recall_count: col_i64(&batch, i, "recall_count"),
|
||||
useful_count: col_i64(&batch, i, "useful_count"),
|
||||
not_useful_count: col_i64(&batch, i, "not_useful_count"),
|
||||
freshness: col_str(&batch, i, "freshness"),
|
||||
version: col_i64(&batch, i, "version"),
|
||||
version_history: String::new(),
|
||||
source: col_str(&batch, i, "source"),
|
||||
volatile_flag: col_bool(&batch, i, "volatile_flag"),
|
||||
is_deleted: col_bool(&batch, i, "is_deleted"),
|
||||
depends_on: String::new(),
|
||||
derived_from: String::new(),
|
||||
last_recalled_at: String::new(),
|
||||
created_at: col_str(&batch, i, "created_at"),
|
||||
updated_at: col_str(&batch, i, "updated_at"),
|
||||
});
|
||||
}
|
||||
}
|
||||
Ok(records)
|
||||
}
|
||||
|
||||
pub fn insert_batch(&self, table: &str, records_json: &str) -> Result<usize, Box<dyn std::error::Error>> {
|
||||
let values: serde_json::Value = serde_json::from_str(records_json)?;
|
||||
let records: Vec<serde_json::Value> = values.as_array().ok_or("expected JSON array")?.to_vec();
|
||||
|
|
|
|||
|
|
@ -300,6 +300,18 @@ fn handle_client(mut stream: UnixStream, args: &Args, lancedb: LanceDBOps, state
|
|||
}
|
||||
}
|
||||
}
|
||||
"lancedb_query" => {
|
||||
let min_recall = msg["min_recall"].as_i64().unwrap_or(5) as i64;
|
||||
let limit = msg["limit"].as_u64().unwrap_or(20) as usize;
|
||||
eprintln!("[lancedb] query: min_recall={} limit={}", min_recall, limit);
|
||||
match lancedb.query_memories(min_recall, limit) {
|
||||
Ok(r) => {
|
||||
eprintln!("[lancedb] query returned {} records", r.len());
|
||||
send_ok(&mut stream, &serde_json::to_string(&r).unwrap_or_default())
|
||||
}
|
||||
Err(e) => send_error(&mut stream, "lancedb_query", &e.to_string()),
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
let req: ConsolidateRequest = match serde_json::from_value(msg) {
|
||||
Ok(r) => r, Err(e) => {
|
||||
|
|
|
|||
Loading…
Reference in New Issue