diff --git a/internal/agent/core/agent.go b/internal/agent/core/agent.go index 67ed8cc..f4b3437 100644 --- a/internal/agent/core/agent.go +++ b/internal/agent/core/agent.go @@ -105,20 +105,6 @@ type Agent struct { noMergeMarkers map[string]int noMergeMu sync.Mutex - // toolCallRing 保护最近 40 条工具调用记录不被上下文淘汰, - // 确保 LLM 不会重复调用同一工具、反复查询同一数据。 - toolCallRing []ToolCallRecord - toolCallRingMax int - toolCallRingMu sync.Mutex // 独立的锁,不与 a.mu 混用避免死锁 -} - -// ToolCallRecord 记录一次工具调用,保留元数据供后续 LLM 回合参考。 -type ToolCallRecord struct { - Timestamp time.Time `json:"timestamp"` - Name string `json:"name"` - Args string `json:"args,omitempty"` // 参数摘要(最多 200 字符) - ResultStub string `json:"result_stub"` // 结果摘要(具体内容通过文本记忆层获取) - FullResult string `json:"result_full,omitempty"` // 完整结果(仅保留最近 5 条,其余仅存 stub) } type AgentConfig struct { @@ -202,8 +188,7 @@ func New(cfg AgentConfig) *Agent { thinkingEnabled: cfg.ThinkingEnabled, inputCfg: cfg.InputProcessing, noMergeMarkers: make(map[string]int), - toolCallRing: make([]ToolCallRecord, 0, 40), - toolCallRingMax: 40, + } } diff --git a/internal/agent/core/context.go b/internal/agent/core/context.go index 1ed0568..a6bbf87 100644 --- a/internal/agent/core/context.go +++ b/internal/agent/core/context.go @@ -111,6 +111,25 @@ func textForVector(evt *ContextEvent, toolDefLookup func(name string) *sdk.ToolD return memory.CleanText(text) } +// toolOutputClean 根据工具定义的 NoMemory/Cleaner 清洗输出,用于计算层。 +// 返回 "" 表示跳过(NoMemory),否则返回清洗后文本(Cleaner 或原文)。 +func (c *RelevanceContext) toolOutputClean(name, output string) string { + if c.toolDefLookup == nil { + return output + } + def := c.toolDefLookup(name) + if def == nil { + return output + } + if def.NoMemory { + return "" + } + if def.Cleaner != nil { + return def.Cleaner(output) + } + return output +} + func (c *RelevanceContext) computeVector(evt *ContextEvent) vector.Vector { return c.embedder.Vectorize(textForVector(evt, c.toolDefLookup)) } @@ -255,7 +274,7 @@ func (c *RelevanceContext) Prune(currentInput string, topK int, docStore *docume ToolResults: convertToolResults(s.event.ToolResults), } } - doc, err := docStore.ContextToDoc("context_archived", entries, c.embedder) + doc, err := docStore.ContextToDoc("context_archived", entries, c.embedder, nil, c.toolOutputClean) if err == nil && doc != nil { archived = len(entries) } diff --git a/internal/agent/core/distill.go b/internal/agent/core/distill.go index 6e32c1c..60df1aa 100644 --- a/internal/agent/core/distill.go +++ b/internal/agent/core/distill.go @@ -46,7 +46,6 @@ func (a *Agent) distillLoop() { case <-ticker.C: log.Printf("[agent] heartbeat distill tick") a.distillContext() - a.syncGraphToDocs() a.reorgGraph() a.autoReloadPlugins() case <-a.ctx.Done(): @@ -68,74 +67,7 @@ func (a *Agent) distillContext() { } } -func (a *Agent) syncGraphToDocs() { - if a.memory == nil || a.docStore == nil { - return - } - stats, err := a.memory.Introspect() - if err != nil { - return - } - - entityCount, _ := stats["entity_count"].(int) - if entityCount == 0 { - return - } - - result, err := a.memory.Recall(nil, nil, 1, "") - if err != nil || result == nil { - return - } - - if len(result.Entities) == 0 && len(result.Relations) == 0 { - return - } - - var summaryParts []string - summaryParts = append(summaryParts, fmt.Sprintf("图记忆快照: %d 个热点实体", len(result.Entities))) - for _, e := range result.Entities { - summaryParts = append(summaryParts, fmt.Sprintf("- %s (%s, %d次)", e.Name, e.Type, e.MentionCount)) - } - if len(result.Relations) > 0 { - summaryParts = append(summaryParts, "关联关系:") - for i, r := range result.Relations { - if i >= 10 { - break - } - summaryParts = append(summaryParts, fmt.Sprintf(" %s →(%s)→ %s", r.SourceName, r.RelationType, r.TargetName)) - } - } - - summary := fmt.Sprintf("图记忆索引 (%d 实体, %d 关系)", len(result.Entities), len(result.Relations)) - content := strings.Join(summaryParts, "\n") - - recent := a.docStore.RecentDocs(1) - if len(recent) > 0 && recent[0].Source == "graph" && recent[0].Content == content { - return - } - - doc := &document.Doc{ - Summary: summary, - Content: content, - Tags: []string{"graph_memory", "auto_sync"}, - Entities: extractEntityNames(result.Entities), - Source: "graph", - } - if err := a.docStore.Insert(doc); err != nil { - log.Printf("[agent] graph→doc sync error: %v", err) - } else { - log.Printf("[agent] graph→doc synced: %s", doc.Summary) - } -} - -func extractEntityNames(entities []memory.Entity) []string { - names := make([]string, len(entities)) - for i, e := range entities { - names[i] = e.Name - } - return names -} func (a *Agent) reorgGraph() { if a.memory == nil { @@ -237,54 +169,69 @@ func (a *Agent) evaluateGraphQuality() { return } - result, err := a.memory.Recall(nil, nil, 1, "") - if err != nil || result == nil || len(result.Relations) == 0 { + pending, err := a.memory.RecallPending(10) + if err != nil { + log.Printf("[agent] recall pending relations error: %v", err) + return + } + if len(pending) == 0 { return } var lowQuality []string - for _, r := range result.Relations { + var pendingIDs []int64 + var skipIDs []int64 + for _, r := range pending { + isLow := false if (r.SourceName == "用户" || r.SourceName == "AI") && (r.RelationType == "提及" || r.RelationType == "回应") { - lowQuality = append(lowQuality, fmt.Sprintf("「%s」-「%s」→「%s」", r.SourceName, r.RelationType, r.TargetName)) + isLow = true + } else if r.RelationType == "关联" { + isLow = true + } else if r.Confidence < 0.3 && r.RelationType != "" { + isLow = true + } + if !isLow { + skipIDs = append(skipIDs, r.ID) continue } + pendingIDs = append(pendingIDs, r.ID) + label := fmt.Sprintf("「%s」-「%s」→「%s」", r.SourceName, r.RelationType, r.TargetName) if r.RelationType == "关联" { - lowQuality = append(lowQuality, fmt.Sprintf("「%s」-「%s」→「%s」(jieba 共现)", r.SourceName, r.RelationType, r.TargetName)) - continue - } - if r.Confidence < 0.3 && r.RelationType != "" { - lowQuality = append(lowQuality, fmt.Sprintf("「%s」-「%s」→「%s」(confidence=%.1f)", r.SourceName, r.RelationType, r.TargetName, r.Confidence)) + label += "(jieba 共现)" + } else if r.Confidence < 0.3 { + label += fmt.Sprintf("(confidence=%.1f)", r.Confidence) } + lowQuality = append(lowQuality, label) + } + + if len(skipIDs) > 0 { + a.memory.UpdateEvalStatusBatch(skipIDs, "approved") } if len(lowQuality) == 0 { return } - batchSize := 10 - for i := 0; i < len(lowQuality); i += batchSize { - end := i + batchSize - if end > len(lowQuality) { - end = len(lowQuality) - } - batch := lowQuality[i:end] - - a.enqueueConsolidationTask(ConsolidationTask{ - Type: "graph_quality", - Reason: fmt.Sprintf( - "图数据库中发现 %d 条低质量关系,请逐条判断是否应该删除(保留 = keep,删除 = discard):\n%s", - len(batch), - strings.Join(batch, "\n"), - ), - Data: map[string]interface{}{ - "candidates": batch, - "action": "evaluate_quality", - }, - }) + if err := a.memory.UpdateEvalStatusBatch(pendingIDs, "evaluating"); err != nil { + log.Printf("[agent] mark relations evaluating error: %v", err) + return } - log.Printf("[agent] graph quality: %d low-quality connection batches sent for LLM evaluation", (len(lowQuality)+batchSize-1)/batchSize) + a.enqueueConsolidationTask(ConsolidationTask{ + Type: "graph_quality", + Reason: fmt.Sprintf( + "图数据库中发现 %d 条低质量关系,请逐条判断是否应该删除(保留 = keep,删除 = discard):\n%s", + len(lowQuality), + strings.Join(lowQuality, "\n"), + ), + Data: map[string]interface{}{ + "candidates": lowQuality, + "action": "evaluate_quality", + }, + }) + + log.Printf("[agent] graph quality: %d pending relations sent for LLM evaluation", len(lowQuality)) } func entitySimilarity(a, b string) float64 { @@ -335,6 +282,10 @@ func docToTriples(doc *document.Doc) []memory.Triple { return triples } + if doc.Source == "graph" || doc.Source == "" { + return nil + } + triples = append(triples, memory.Triple{ Subject: "文档", SubjectType: "Concept", @@ -397,28 +348,19 @@ func (a *Agent) processConsolidation(evt *agentIO.InputEvent, input string) { stageCtx.Extra["output_channel"] = evt.OutputChannel a.injectSourceContext(stageCtx, evt) - archived := a.context.Prune(input, a.maxContextSize-1, a.docStore) - if archived > 0 { - log.Printf("[agent] consolidation: pruned %d low-relevance events", archived) - } - - a.context.Append(ContextEvent{ - Timestamp: start, - Source: "system", - Input: input, - }) - response, toolsUsed, toolResults, err := a.process(input, stageCtx) + _, toolsUsed, _, err := a.process(input, stageCtx) if err != nil { log.Printf("[agent] consolidation error: %v", err) return } - a.context.Append(ContextEvent{ - Timestamp: time.Now(), - Source: "agent", - Input: input, - Response: response, - ToolsUsed: toolsUsed, - ToolResults: toolResults, - }) + + if a.memory != nil { + if n, err := a.memory.ResolveEvaluating(); err != nil { + log.Printf("[agent] resolve evaluating relations error: %v", err) + } else if n > 0 { + log.Printf("[agent] resolved %d evaluating relations to approved", n) + } + } + log.Printf("[agent] consolidation done (%dms, tools=%v)", time.Since(start).Milliseconds(), toolsUsed) } diff --git a/internal/agent/core/process.go b/internal/agent/core/process.go index 16e64e3..cbce4a9 100644 --- a/internal/agent/core/process.go +++ b/internal/agent/core/process.go @@ -2,13 +2,10 @@ package core import ( "context" - "encoding/json" "errors" "fmt" "log" - "sort" "strings" - "time" agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" "gitcode.com/JianFeeeee/HomeAgent/internal/events" @@ -237,9 +234,6 @@ func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response stri } } - argsJSON, _ := json.Marshal(tc.Arguments) - a.recordToolCall(tc.Name, string(argsJSON), result) - msgContent := "" if contentOnce { msgContent = resp.Content @@ -301,117 +295,28 @@ func (a *Agent) docStoreSize() int { return 0 } -func (a *Agent) recordToolCall(name, args, result string) { - a.toolCallRingMu.Lock() - defer a.toolCallRingMu.Unlock() - - if len(args) > 200 { - args = args[:200] + "..." - } - - var resultStub string - var fullResult string - if len(a.toolCallRing) < 5 { - fullResult = result - } - if len(result) > 80 { - resultStub = result[:80] + "..." - } else { - resultStub = result - } - - rec := ToolCallRecord{ - Timestamp: time.Now(), - Name: name, - Args: args, - ResultStub: resultStub, - FullResult: fullResult, - } - - if len(a.toolCallRing) >= a.toolCallRingMax { - a.toolCallRing = a.toolCallRing[1:] - } - a.toolCallRing = append(a.toolCallRing, rec) -} - -func (a *Agent) formatToolCallRing() string { - a.toolCallRingMu.Lock() - defer a.toolCallRingMu.Unlock() - - if len(a.toolCallRing) == 0 { - return "" - } - - var sb strings.Builder - sb.WriteString("【已执行工具记录(最近40条)】\n") - start := 0 - if len(a.toolCallRing) > 40 { - start = len(a.toolCallRing) - 40 - } - for i, rec := range a.toolCallRing[start:] { - if len(rec.FullResult) > 0 { - sb.WriteString(fmt.Sprintf(" [%d] %s: %s(%s)=%s\n", i+1, - rec.Timestamp.Format("15:04:05"), rec.Name, rec.Args, - truncateStr(rec.FullResult, 120))) - } else { - sb.WriteString(fmt.Sprintf(" [%d] %s: %s(%s) → (已缓存,具体结果通过文本记忆层获取)\n", i+1, - rec.Timestamp.Format("15:04:05"), rec.Name, rec.Args)) - } - } - return sb.String() -} - func (a *Agent) formatMergedTimeline() string { a.context.mu.Lock() events := make([]*ContextEvent, len(a.context.events)) copy(events, a.context.events) a.context.mu.Unlock() - a.toolCallRingMu.Lock() - ring := make([]ToolCallRecord, len(a.toolCallRing)) - copy(ring, a.toolCallRing) - a.toolCallRingMu.Unlock() - - if len(events) == 0 && len(ring) == 0 { + if len(events) == 0 { return "" } - type timelineEntry struct { - ts time.Time - label string - text string - } - entries := make([]timelineEntry, 0, len(events)+len(ring)) - - for _, e := range events { - text := fmt.Sprintf("[对话] %s: %s", e.Source, e.Input) - if len(e.ToolsUsed) > 0 { - text += fmt.Sprintf(" → 调用工具: %s", strings.Join(e.ToolsUsed, ", ")) - } - if e.Response != "" { - text += fmt.Sprintf(" → %s", truncateStr(e.Response, 120)) - } - entries = append(entries, timelineEntry{ts: e.Timestamp, label: "对话", text: text}) - } - - for _, r := range ring { - text := fmt.Sprintf("[工具] %s(%s)", r.Name, r.Args) - if r.FullResult != "" { - text += fmt.Sprintf(" = %s", truncateStr(r.FullResult, 120)) - } else { - text += " → (结果已缓存,可通过文本记忆层获取)" - } - entries = append(entries, timelineEntry{ts: r.Timestamp, label: "工具", text: text}) - } - - sort.Slice(entries, func(i, j int) bool { - return entries[i].ts.Before(entries[j].ts) - }) - var sb strings.Builder sb.WriteString("【对话时序】\n") - for _, e := range entries { - sb.WriteString(fmt.Sprintf("[%s] %s\n", e.ts.Format("15:04:05"), e.text)) + for _, e := range events { + sb.WriteString(fmt.Sprintf("[%s] %s: %s", + e.Timestamp.Format("15:04:05"), e.Source, e.Input)) + if len(e.ToolsUsed) > 0 { + sb.WriteString(fmt.Sprintf(" → 调用工具: %s", strings.Join(e.ToolsUsed, ", "))) + } + if e.Response != "" { + sb.WriteString(fmt.Sprintf(" → %s", truncateStr(e.Response, 120))) + } + sb.WriteString("\n") } return sb.String() } diff --git a/internal/memory/document/document.go b/internal/memory/document/document.go index f86a9dc..660e6a8 100644 --- a/internal/memory/document/document.go +++ b/internal/memory/document/document.go @@ -124,15 +124,17 @@ func (s *Store) Insert(doc *Doc) error { } // ContextToDoc — 将一段上下文对话历史提炼为文档(带内容去重) -// cleanFn 可选,用于在计算层(摘要/标签/实体提取)前过滤文本,不影响原文存储。 -func (s *Store) ContextToDoc(source string, entries []ContextEntry, vec vector.Vectorizer, cleanFn ...func(string) string) (*Doc, error) { +// cleanFn 可选,在计算层前统一过滤文本,不影响原文存储。 +// toolCleanFn 可选,func(name, output string) string,按工具名对输出进行过滤/清洗: +// - 返回 "" → 跳过该工具输出(NoMemory) +// - 返回清洗后文本 → 用于计算层(Cleaner),原文不受影响 +func (s *Store) ContextToDoc(source string, entries []ContextEntry, vec vector.Vectorizer, cleanFn func(string) string, toolCleanFn func(name, output string) string) (*Doc, error) { if len(entries) == 0 { return nil, nil } - cleanText := func(text string) string { return text } - if len(cleanFn) > 0 && cleanFn[0] != nil { - cleanText = cleanFn[0] + if cleanFn == nil { + cleanFn = func(text string) string { return text } } var parts []string @@ -149,9 +151,9 @@ func (s *Store) ContextToDoc(source string, entries []ContextEntry, vec vector.V content := strings.Join(parts, "\n") contentHash := simpleHash(content) - summary := summarizeEntries(entries, cleanText) - tags := extractTags(entries, cleanText) - entities := extractEntities(entries, cleanText) + summary := summarizeEntries(entries, cleanFn, toolCleanFn) + tags := extractTags(entries, cleanFn, toolCleanFn) + entities := extractEntities(entries, cleanFn, toolCleanFn) s.mu.Lock() @@ -439,23 +441,26 @@ type ContextEntry struct { ToolResults []ToolResultItem } -func summarizeEntries(entries []ContextEntry, cleanText ...func(string) string) string { +func summarizeEntries(entries []ContextEntry, cleanText func(string) string, toolCleanFn func(name, output string) string) string { if len(entries) == 0 { return "" } - clean := func(text string) string { return text } - if len(cleanText) > 0 && cleanText[0] != nil { - clean = cleanText[0] - } sources := make(map[string]int) var topics []string for _, e := range entries { sources[e.Source]++ - words := memory.ExtractKeywords(clean(e.Content)) + words := memory.ExtractKeywords(cleanText(e.Content)) topics = append(topics, words...) for _, tr := range e.ToolResults { - cleaned := clean(tr.Output) - toolWords := memory.ExtractKeywords(cleaned) + out := tr.Output + if toolCleanFn != nil { + if c := toolCleanFn(tr.Name, tr.Output); c == "" { + continue + } else { + out = c + } + } + toolWords := memory.ExtractKeywords(out) topics = append(topics, toolWords...) } } @@ -485,18 +490,22 @@ func summarizeEntries(entries []ContextEntry, cleanText ...func(string) string) return summary } -func extractTags(entries []ContextEntry, cleanText ...func(string) string) []string { - clean := func(text string) string { return text } - if len(cleanText) > 0 && cleanText[0] != nil { - clean = cleanText[0] - } +func extractTags(entries []ContextEntry, cleanText func(string) string, toolCleanFn func(name, output string) string) []string { tagSet := make(map[string]bool) for _, e := range entries { - for _, kw := range memory.ExtractKeywords(clean(e.Content)) { + for _, kw := range memory.ExtractKeywords(cleanText(e.Content)) { tagSet[kw] = true } for _, tr := range e.ToolResults { - for _, kw := range memory.ExtractKeywords(clean(tr.Output)) { + out := tr.Output + if toolCleanFn != nil { + if c := toolCleanFn(tr.Name, tr.Output); c == "" { + continue + } else { + out = c + } + } + for _, kw := range memory.ExtractKeywords(out) { tagSet[kw] = true } } @@ -511,23 +520,26 @@ func extractTags(entries []ContextEntry, cleanText ...func(string) string) []str return tags } -func extractEntities(entries []ContextEntry, cleanText ...func(string) string) []string { - // 简易实体提取:提取引号内的内容、粗体/标记词 - clean := func(text string) string { return text } - if len(cleanText) > 0 && cleanText[0] != nil { - clean = cleanText[0] - } +func extractEntities(entries []ContextEntry, cleanText func(string) string, toolCleanFn func(name, output string) string) []string { var entities []string seen := make(map[string]bool) for _, e := range entries { - for _, kw := range memory.ExtractKeywords(clean(e.Content)) { + for _, kw := range memory.ExtractKeywords(cleanText(e.Content)) { if len(kw) >= 2 && !seen[kw] { seen[kw] = true entities = append(entities, kw) } } for _, tr := range e.ToolResults { - for _, kw := range memory.ExtractKeywords(clean(tr.Output)) { + out := tr.Output + if toolCleanFn != nil { + if c := toolCleanFn(tr.Name, tr.Output); c == "" { + continue + } else { + out = c + } + } + for _, kw := range memory.ExtractKeywords(out) { if len(kw) >= 2 && !seen[kw] { seen[kw] = true entities = append(entities, kw) diff --git a/internal/memory/document/document_test.go b/internal/memory/document/document_test.go index aa2e07b..b1523ed 100644 --- a/internal/memory/document/document_test.go +++ b/internal/memory/document/document_test.go @@ -82,7 +82,7 @@ func TestContextToDoc(t *testing.T) { {Timestamp: time.Now(), Source: "user", Content: "特别是Go语言", Response: "Go很棒"}, } - doc, err := s.ContextToDoc("test", entries, nil) + doc, err := s.ContextToDoc("test", entries, nil, nil, nil) if err != nil { t.Fatal(err) } @@ -178,7 +178,7 @@ func TestSummarizeEntries(t *testing.T) { {Source: "user", Content: "今天天气如何"}, {Source: "user", Content: "明天会下雨吗"}, } - summary := summarizeEntries(entries) + summary := summarizeEntries(entries, func(s string) string { return s }, nil) if summary == "" { t.Error("summary should not be empty") } @@ -198,7 +198,7 @@ func TestExtractTags(t *testing.T) { entries := []ContextEntry{ {Content: "我喜欢喝咖啡和编程"}, } - tags := extractTags(entries) + tags := extractTags(entries, func(s string) string { return s }, nil) if len(tags) == 0 { t.Error("should extract tags") } @@ -343,3 +343,127 @@ func TestRemoveNonexistent(t *testing.T) { t.Errorf("expected 1 doc after remove nonexistent, got %d", stats["doc_count"]) } } + +func TestSummarizeEntriesWithToolCleanFn(t *testing.T) { + tests := []struct { + name string + toolCleanFn func(name, output string) string + wantTopics []string + notTopics []string + }{ + { + name: "nil toolCleanFn uses raw output", + toolCleanFn: nil, + wantTopics: []string{"手机", "电脑"}, + notTopics: nil, + }, + { + name: "NoMemory returns empty skips tool output", + toolCleanFn: func(name, output string) string { + return "" + }, + wantTopics: nil, + notTopics: []string{"手机", "电脑"}, + }, + { + name: "Cleaner applies filter", + toolCleanFn: func(name, output string) string { + return "电脑 编程" + }, + wantTopics: []string{"电脑", "编程"}, + notTopics: nil, + }, + } + + entry := ContextEntry{ + Content: "今天天气", + ToolResults: []ToolResultItem{ + {Name: "test_tool", Output: "手机 电脑"}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + summary := summarizeEntries([]ContextEntry{entry}, func(s string) string { return s }, tc.toolCleanFn) + for _, w := range tc.wantTopics { + if !contains(summary, w) { + t.Errorf("summary should contain %q, got: %s", w, summary) + } + } + for _, n := range tc.notTopics { + if contains(summary, n) { + t.Errorf("summary should NOT contain %q, got: %s", n, summary) + } + } + }) + } +} + +func TestExtractTagsWithToolCleanFn(t *testing.T) { + entries := []ContextEntry{ + { + Content: "对话", + ToolResults: []ToolResultItem{ + {Name: "search", Output: "编程和咖啡"}, + }, + }, + } + + // toolCleanFn 返回 "" → NoMemory,工具输出被跳过 + tagsSkip := extractTags(entries, func(s string) string { return s }, func(name, output string) string { return "" }) + for _, tag := range tagsSkip { + if tag == "编程" || tag == "咖啡" { + t.Errorf("NoMemory tool should not contribute keywords, got tag: %s", tag) + } + } + + // toolCleanFn 返回清洗文本 → 用清洗后内容提取关键词 + tagsClean := extractTags(entries, func(s string) string { return s }, func(name, output string) string { return "咖啡 编程" }) + found := false + for _, tag := range tagsClean { + if tag == "编程" { + found = true + } + } + if !found { + t.Errorf("cleaner output keywords should appear in tags, got: %v", tagsClean) + } +} + +func TestContextToDocContentPreservesRawToolOutput(t *testing.T) { + dir, err := os.MkdirTemp("", "doc_toolclean_*") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(dir) + + s := NewStore(dir) + s.Start() + defer s.Stop() + + entries := []ContextEntry{ + { + Timestamp: time.Now(), + Source: "user", + Content: "查天气", + ToolResults: []ToolResultItem{ + {Name: "weather", Output: "{\"temp\": 25}"}, + }, + }, + } + + // toolCleanFn 返回清洗文本,但 Content 必须保留原始输出 + cleaner := func(name, output string) string { + return "天气 温度" + } + doc, err := s.ContextToDoc("test", entries, nil, nil, cleaner) + if err != nil { + t.Fatal(err) + } + if !contains(doc.Content, "{\"temp\": 25}") { + t.Errorf("Content should preserve raw tool output, got: %s", doc.Content) + } + if doc.Summary == "" { + t.Error("summary should not be empty") + } +} diff --git a/internal/memory/graph.go b/internal/memory/graph.go index 9d6829f..7631738 100644 --- a/internal/memory/graph.go +++ b/internal/memory/graph.go @@ -31,6 +31,9 @@ type Relation struct { TurnID int `json:"turn_id"` CreatedAt time.Time `json:"created_at"` DateBucket string `json:"date_bucket"` + EvalStatus string `json:"eval_status"` + EvalRound int `json:"eval_round"` + EvalAt time.Time `json:"eval_at,omitempty"` } type Triple struct { @@ -93,6 +96,9 @@ func (g *GraphDB) initSchema() error { created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, date_bucket TEXT, + eval_status TEXT DEFAULT 'pending', + eval_round INTEGER DEFAULT 0, + eval_at TIMESTAMP, FOREIGN KEY (source_id) REFERENCES entities(id), FOREIGN KEY (target_id) REFERENCES entities(id) )`, @@ -111,7 +117,20 @@ func (g *GraphDB) initSchema() error { } } - return tx.Commit() + if err := tx.Commit(); err != nil { + return err + } + + migrations := []string{ + `ALTER TABLE relations ADD COLUMN eval_status TEXT DEFAULT 'pending'`, + `ALTER TABLE relations ADD COLUMN eval_round INTEGER DEFAULT 0`, + `ALTER TABLE relations ADD COLUMN eval_at TIMESTAMP`, + } + for _, m := range migrations { + g.db.Exec(m) + } + + return nil } func (g *GraphDB) Commit(triples []Triple, sessionID string, turnID int) (int, int, error) { @@ -241,7 +260,7 @@ func (g *GraphDB) Recall(keywords []string, seedEntities []string, depth int, se if len(keywords) == 0 && len(seedEntities) == 0 { rows, err := g.db.Query( `SELECT id, name, type, mention_count, created_at, updated_at - FROM entities ORDER BY mention_count DESC LIMIT 50`, + FROM entities ORDER BY mention_count DESC`, ) if err != nil { return nil, err @@ -259,7 +278,8 @@ func (g *GraphDB) Recall(keywords []string, seedEntities []string, depth int, se relRows, err := g.db.Query( `SELECT r.id, r.source_id, r.target_id, e1.name, e2.name, r.relation_type, r.confidence, r.status, r.session_id, - r.turn_id, r.created_at, COALESCE(r.date_bucket, '') + r.turn_id, r.created_at, COALESCE(r.date_bucket, ''), + COALESCE(r.eval_status, 'pending'), COALESCE(r.eval_round, 0), r.eval_at FROM relations r JOIN entities e1 ON r.source_id = e1.id JOIN entities e2 ON r.target_id = e2.id @@ -275,7 +295,8 @@ func (g *GraphDB) Recall(keywords []string, seedEntities []string, depth int, se if err := relRows.Scan(&rel.ID, &rel.SourceID, &rel.TargetID, &rel.SourceName, &rel.TargetName, &rel.RelationType, &rel.Confidence, &rel.Status, &rel.SessionID, - &rel.TurnID, &rel.CreatedAt, &rel.DateBucket); err != nil { + &rel.TurnID, &rel.CreatedAt, &rel.DateBucket, + &rel.EvalStatus, &rel.EvalRound, &rel.EvalAt); err != nil { return nil, err } result.Relations = append(result.Relations, rel) @@ -340,7 +361,8 @@ func (g *GraphDB) Recall(keywords []string, seedEntities []string, depth int, se query := fmt.Sprintf( `SELECT r.id, r.source_id, r.target_id, e1.name, e2.name, r.relation_type, r.confidence, r.status, r.session_id, - r.turn_id, r.created_at, COALESCE(r.date_bucket, '') + r.turn_id, r.created_at, COALESCE(r.date_bucket, ''), + COALESCE(r.eval_status, 'pending'), COALESCE(r.eval_round, 0), r.eval_at FROM relations r JOIN entities e1 ON r.source_id = e1.id JOIN entities e2 ON r.target_id = e2.id @@ -367,7 +389,8 @@ func (g *GraphDB) Recall(keywords []string, seedEntities []string, depth int, se if err := relRows.Scan(&rel.ID, &rel.SourceID, &rel.TargetID, &rel.SourceName, &rel.TargetName, &rel.RelationType, &rel.Confidence, &rel.Status, &rel.SessionID, - &rel.TurnID, &rel.CreatedAt, &rel.DateBucket); err != nil { + &rel.TurnID, &rel.CreatedAt, &rel.DateBucket, + &rel.EvalStatus, &rel.EvalRound, &rel.EvalAt); err != nil { relRows.Close() return nil, err } @@ -747,6 +770,89 @@ func (g *GraphDB) Archive(days int) (int, error) { return int(n), nil } +func (g *GraphDB) RecallPending(limit int) ([]Relation, error) { + g.mu.RLock() + defer g.mu.RUnlock() + + rows, err := g.db.Query( + `SELECT r.id, r.source_id, r.target_id, e1.name, e2.name, + r.relation_type, r.confidence, r.status, r.session_id, + r.turn_id, r.created_at, COALESCE(r.date_bucket, ''), + COALESCE(r.eval_status, 'pending'), COALESCE(r.eval_round, 0), r.eval_at + FROM relations r + JOIN entities e1 ON r.source_id = e1.id + JOIN entities e2 ON r.target_id = e2.id + WHERE r.status = 'active' + AND (r.eval_status IS NULL OR r.eval_status = 'pending') + ORDER BY r.created_at DESC + LIMIT ?`, limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + + var relations []Relation + for rows.Next() { + var rel Relation + if err := rows.Scan(&rel.ID, &rel.SourceID, &rel.TargetID, + &rel.SourceName, &rel.TargetName, &rel.RelationType, + &rel.Confidence, &rel.Status, &rel.SessionID, + &rel.TurnID, &rel.CreatedAt, &rel.DateBucket, + &rel.EvalStatus, &rel.EvalRound, &rel.EvalAt); err != nil { + return nil, err + } + relations = append(relations, rel) + } + return relations, rows.Err() +} + +func (g *GraphDB) UpdateEvalStatus(id int64, status string) error { + g.mu.Lock() + defer g.mu.Unlock() + + _, err := g.db.Exec( + `UPDATE relations SET eval_status = ?, eval_round = eval_round + 1, eval_at = CURRENT_TIMESTAMP WHERE id = ?`, + status, id, + ) + return err +} + +func (g *GraphDB) UpdateEvalStatusBatch(ids []int64, status string) error { + g.mu.Lock() + defer g.mu.Unlock() + + if len(ids) == 0 { + return nil + } + + for _, id := range ids { + _, err := g.db.Exec( + `UPDATE relations SET eval_status = ?, eval_round = eval_round + 1, eval_at = CURRENT_TIMESTAMP WHERE id = ?`, + status, id, + ) + if err != nil { + return err + } + } + return nil +} + +func (g *GraphDB) ResolveEvaluating() (int, error) { + g.mu.Lock() + defer g.mu.Unlock() + + result, err := g.db.Exec( + `UPDATE relations SET eval_status = 'approved', eval_round = eval_round + 1, eval_at = CURRENT_TIMESTAMP + WHERE eval_status = 'evaluating' AND status = 'active'`, + ) + if err != nil { + return 0, err + } + n, _ := result.RowsAffected() + return int(n), nil +} + func (g *GraphDB) Close() error { return g.db.Close() }