mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-10-03 15:53:56 +00:00
fix: 修复记忆系统自循环与计算层污染
- 删 syncGraphToDocs(): Graph 快照不再写入 Document,避免污染向量索引和三层隔离 - 删 toolCallRing(): 已被工具 NoMemory/Cleaner 机制取代,不再需要独立环形缓冲 - 加 toolOutputClean 回调线程 Prune→ContextToDoc: 归档时按 NoMemory 跳过、Cleaner 清洗后再过 jieba,原文保留 - 加 eval_status 持久化 (RecallPending/UpdateEvalStatus/ResolveEvaluating): 避免重复 LLM 评估
This commit is contained in:
@ -105,20 +105,6 @@ type Agent struct {
|
|||||||
noMergeMarkers map[string]int
|
noMergeMarkers map[string]int
|
||||||
noMergeMu sync.Mutex
|
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 {
|
type AgentConfig struct {
|
||||||
@ -202,8 +188,7 @@ func New(cfg AgentConfig) *Agent {
|
|||||||
thinkingEnabled: cfg.ThinkingEnabled,
|
thinkingEnabled: cfg.ThinkingEnabled,
|
||||||
inputCfg: cfg.InputProcessing,
|
inputCfg: cfg.InputProcessing,
|
||||||
noMergeMarkers: make(map[string]int),
|
noMergeMarkers: make(map[string]int),
|
||||||
toolCallRing: make([]ToolCallRecord, 0, 40),
|
|
||||||
toolCallRingMax: 40,
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -111,6 +111,25 @@ func textForVector(evt *ContextEvent, toolDefLookup func(name string) *sdk.ToolD
|
|||||||
return memory.CleanText(text)
|
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 {
|
func (c *RelevanceContext) computeVector(evt *ContextEvent) vector.Vector {
|
||||||
return c.embedder.Vectorize(textForVector(evt, c.toolDefLookup))
|
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),
|
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 {
|
if err == nil && doc != nil {
|
||||||
archived = len(entries)
|
archived = len(entries)
|
||||||
}
|
}
|
||||||
|
|||||||
@ -46,7 +46,6 @@ func (a *Agent) distillLoop() {
|
|||||||
case <-ticker.C:
|
case <-ticker.C:
|
||||||
log.Printf("[agent] heartbeat distill tick")
|
log.Printf("[agent] heartbeat distill tick")
|
||||||
a.distillContext()
|
a.distillContext()
|
||||||
a.syncGraphToDocs()
|
|
||||||
a.reorgGraph()
|
a.reorgGraph()
|
||||||
a.autoReloadPlugins()
|
a.autoReloadPlugins()
|
||||||
case <-a.ctx.Done():
|
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() {
|
func (a *Agent) reorgGraph() {
|
||||||
if a.memory == nil {
|
if a.memory == nil {
|
||||||
@ -237,54 +169,69 @@ func (a *Agent) evaluateGraphQuality() {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
result, err := a.memory.Recall(nil, nil, 1, "")
|
pending, err := a.memory.RecallPending(10)
|
||||||
if err != nil || result == nil || len(result.Relations) == 0 {
|
if err != nil {
|
||||||
|
log.Printf("[agent] recall pending relations error: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if len(pending) == 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
var lowQuality []string
|
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") &&
|
if (r.SourceName == "用户" || r.SourceName == "AI") &&
|
||||||
(r.RelationType == "提及" || r.RelationType == "回应") {
|
(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
|
continue
|
||||||
}
|
}
|
||||||
|
pendingIDs = append(pendingIDs, r.ID)
|
||||||
|
label := fmt.Sprintf("「%s」-「%s」→「%s」", r.SourceName, r.RelationType, r.TargetName)
|
||||||
if r.RelationType == "关联" {
|
if r.RelationType == "关联" {
|
||||||
lowQuality = append(lowQuality, fmt.Sprintf("「%s」-「%s」→「%s」(jieba 共现)", r.SourceName, r.RelationType, r.TargetName))
|
label += "(jieba 共现)"
|
||||||
continue
|
} else if r.Confidence < 0.3 {
|
||||||
}
|
label += fmt.Sprintf("(confidence=%.1f)", r.Confidence)
|
||||||
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))
|
|
||||||
}
|
}
|
||||||
|
lowQuality = append(lowQuality, label)
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(skipIDs) > 0 {
|
||||||
|
a.memory.UpdateEvalStatusBatch(skipIDs, "approved")
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(lowQuality) == 0 {
|
if len(lowQuality) == 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
batchSize := 10
|
if err := a.memory.UpdateEvalStatusBatch(pendingIDs, "evaluating"); err != nil {
|
||||||
for i := 0; i < len(lowQuality); i += batchSize {
|
log.Printf("[agent] mark relations evaluating error: %v", err)
|
||||||
end := i + batchSize
|
return
|
||||||
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",
|
|
||||||
},
|
|
||||||
})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
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 {
|
func entitySimilarity(a, b string) float64 {
|
||||||
@ -335,6 +282,10 @@ func docToTriples(doc *document.Doc) []memory.Triple {
|
|||||||
return triples
|
return triples
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if doc.Source == "graph" || doc.Source == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
triples = append(triples, memory.Triple{
|
triples = append(triples, memory.Triple{
|
||||||
Subject: "文档",
|
Subject: "文档",
|
||||||
SubjectType: "Concept",
|
SubjectType: "Concept",
|
||||||
@ -397,28 +348,19 @@ func (a *Agent) processConsolidation(evt *agentIO.InputEvent, input string) {
|
|||||||
stageCtx.Extra["output_channel"] = evt.OutputChannel
|
stageCtx.Extra["output_channel"] = evt.OutputChannel
|
||||||
a.injectSourceContext(stageCtx, evt)
|
a.injectSourceContext(stageCtx, evt)
|
||||||
|
|
||||||
archived := a.context.Prune(input, a.maxContextSize-1, a.docStore)
|
_, toolsUsed, _, err := a.process(input, stageCtx)
|
||||||
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)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("[agent] consolidation error: %v", err)
|
log.Printf("[agent] consolidation error: %v", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
a.context.Append(ContextEvent{
|
|
||||||
Timestamp: time.Now(),
|
if a.memory != nil {
|
||||||
Source: "agent",
|
if n, err := a.memory.ResolveEvaluating(); err != nil {
|
||||||
Input: input,
|
log.Printf("[agent] resolve evaluating relations error: %v", err)
|
||||||
Response: response,
|
} else if n > 0 {
|
||||||
ToolsUsed: toolsUsed,
|
log.Printf("[agent] resolved %d evaluating relations to approved", n)
|
||||||
ToolResults: toolResults,
|
}
|
||||||
})
|
}
|
||||||
|
|
||||||
log.Printf("[agent] consolidation done (%dms, tools=%v)", time.Since(start).Milliseconds(), toolsUsed)
|
log.Printf("[agent] consolidation done (%dms, tools=%v)", time.Since(start).Milliseconds(), toolsUsed)
|
||||||
}
|
}
|
||||||
|
|||||||
@ -2,13 +2,10 @@ package core
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
"log"
|
||||||
"sort"
|
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
|
||||||
|
|
||||||
agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
|
agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
|
||||||
"gitcode.com/JianFeeeee/HomeAgent/internal/events"
|
"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 := ""
|
msgContent := ""
|
||||||
if contentOnce {
|
if contentOnce {
|
||||||
msgContent = resp.Content
|
msgContent = resp.Content
|
||||||
@ -301,117 +295,28 @@ func (a *Agent) docStoreSize() int {
|
|||||||
return 0
|
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 {
|
func (a *Agent) formatMergedTimeline() string {
|
||||||
a.context.mu.Lock()
|
a.context.mu.Lock()
|
||||||
events := make([]*ContextEvent, len(a.context.events))
|
events := make([]*ContextEvent, len(a.context.events))
|
||||||
copy(events, a.context.events)
|
copy(events, a.context.events)
|
||||||
a.context.mu.Unlock()
|
a.context.mu.Unlock()
|
||||||
|
|
||||||
a.toolCallRingMu.Lock()
|
if len(events) == 0 {
|
||||||
ring := make([]ToolCallRecord, len(a.toolCallRing))
|
|
||||||
copy(ring, a.toolCallRing)
|
|
||||||
a.toolCallRingMu.Unlock()
|
|
||||||
|
|
||||||
if len(events) == 0 && len(ring) == 0 {
|
|
||||||
return ""
|
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
|
var sb strings.Builder
|
||||||
sb.WriteString("【对话时序】\n")
|
sb.WriteString("【对话时序】\n")
|
||||||
for _, e := range entries {
|
for _, e := range events {
|
||||||
sb.WriteString(fmt.Sprintf("[%s] %s\n", e.ts.Format("15:04:05"), e.text))
|
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()
|
return sb.String()
|
||||||
}
|
}
|
||||||
|
|||||||
@ -124,15 +124,17 @@ func (s *Store) Insert(doc *Doc) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// ContextToDoc — 将一段上下文对话历史提炼为文档(带内容去重)
|
// ContextToDoc — 将一段上下文对话历史提炼为文档(带内容去重)
|
||||||
// cleanFn 可选,用于在计算层(摘要/标签/实体提取)前过滤文本,不影响原文存储。
|
// cleanFn 可选,在计算层前统一过滤文本,不影响原文存储。
|
||||||
func (s *Store) ContextToDoc(source string, entries []ContextEntry, vec vector.Vectorizer, cleanFn ...func(string) string) (*Doc, error) {
|
// 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 {
|
if len(entries) == 0 {
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
cleanText := func(text string) string { return text }
|
if cleanFn == nil {
|
||||||
if len(cleanFn) > 0 && cleanFn[0] != nil {
|
cleanFn = func(text string) string { return text }
|
||||||
cleanText = cleanFn[0]
|
|
||||||
}
|
}
|
||||||
|
|
||||||
var parts []string
|
var parts []string
|
||||||
@ -149,9 +151,9 @@ func (s *Store) ContextToDoc(source string, entries []ContextEntry, vec vector.V
|
|||||||
content := strings.Join(parts, "\n")
|
content := strings.Join(parts, "\n")
|
||||||
contentHash := simpleHash(content)
|
contentHash := simpleHash(content)
|
||||||
|
|
||||||
summary := summarizeEntries(entries, cleanText)
|
summary := summarizeEntries(entries, cleanFn, toolCleanFn)
|
||||||
tags := extractTags(entries, cleanText)
|
tags := extractTags(entries, cleanFn, toolCleanFn)
|
||||||
entities := extractEntities(entries, cleanText)
|
entities := extractEntities(entries, cleanFn, toolCleanFn)
|
||||||
|
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
|
|
||||||
@ -439,23 +441,26 @@ type ContextEntry struct {
|
|||||||
ToolResults []ToolResultItem
|
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 {
|
if len(entries) == 0 {
|
||||||
return ""
|
return ""
|
||||||
}
|
}
|
||||||
clean := func(text string) string { return text }
|
|
||||||
if len(cleanText) > 0 && cleanText[0] != nil {
|
|
||||||
clean = cleanText[0]
|
|
||||||
}
|
|
||||||
sources := make(map[string]int)
|
sources := make(map[string]int)
|
||||||
var topics []string
|
var topics []string
|
||||||
for _, e := range entries {
|
for _, e := range entries {
|
||||||
sources[e.Source]++
|
sources[e.Source]++
|
||||||
words := memory.ExtractKeywords(clean(e.Content))
|
words := memory.ExtractKeywords(cleanText(e.Content))
|
||||||
topics = append(topics, words...)
|
topics = append(topics, words...)
|
||||||
for _, tr := range e.ToolResults {
|
for _, tr := range e.ToolResults {
|
||||||
cleaned := clean(tr.Output)
|
out := tr.Output
|
||||||
toolWords := memory.ExtractKeywords(cleaned)
|
if toolCleanFn != nil {
|
||||||
|
if c := toolCleanFn(tr.Name, tr.Output); c == "" {
|
||||||
|
continue
|
||||||
|
} else {
|
||||||
|
out = c
|
||||||
|
}
|
||||||
|
}
|
||||||
|
toolWords := memory.ExtractKeywords(out)
|
||||||
topics = append(topics, toolWords...)
|
topics = append(topics, toolWords...)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -485,18 +490,22 @@ func summarizeEntries(entries []ContextEntry, cleanText ...func(string) string)
|
|||||||
return summary
|
return summary
|
||||||
}
|
}
|
||||||
|
|
||||||
func extractTags(entries []ContextEntry, cleanText ...func(string) string) []string {
|
func extractTags(entries []ContextEntry, cleanText func(string) string, toolCleanFn func(name, output string) string) []string {
|
||||||
clean := func(text string) string { return text }
|
|
||||||
if len(cleanText) > 0 && cleanText[0] != nil {
|
|
||||||
clean = cleanText[0]
|
|
||||||
}
|
|
||||||
tagSet := make(map[string]bool)
|
tagSet := make(map[string]bool)
|
||||||
for _, e := range entries {
|
for _, e := range entries {
|
||||||
for _, kw := range memory.ExtractKeywords(clean(e.Content)) {
|
for _, kw := range memory.ExtractKeywords(cleanText(e.Content)) {
|
||||||
tagSet[kw] = true
|
tagSet[kw] = true
|
||||||
}
|
}
|
||||||
for _, tr := range e.ToolResults {
|
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
|
tagSet[kw] = true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -511,23 +520,26 @@ func extractTags(entries []ContextEntry, cleanText ...func(string) string) []str
|
|||||||
return tags
|
return tags
|
||||||
}
|
}
|
||||||
|
|
||||||
func extractEntities(entries []ContextEntry, cleanText ...func(string) string) []string {
|
func extractEntities(entries []ContextEntry, cleanText func(string) string, toolCleanFn func(name, output string) string) []string {
|
||||||
// 简易实体提取:提取引号内的内容、粗体/标记词
|
|
||||||
clean := func(text string) string { return text }
|
|
||||||
if len(cleanText) > 0 && cleanText[0] != nil {
|
|
||||||
clean = cleanText[0]
|
|
||||||
}
|
|
||||||
var entities []string
|
var entities []string
|
||||||
seen := make(map[string]bool)
|
seen := make(map[string]bool)
|
||||||
for _, e := range entries {
|
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] {
|
if len(kw) >= 2 && !seen[kw] {
|
||||||
seen[kw] = true
|
seen[kw] = true
|
||||||
entities = append(entities, kw)
|
entities = append(entities, kw)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
for _, tr := range e.ToolResults {
|
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] {
|
if len(kw) >= 2 && !seen[kw] {
|
||||||
seen[kw] = true
|
seen[kw] = true
|
||||||
entities = append(entities, kw)
|
entities = append(entities, kw)
|
||||||
|
|||||||
@ -82,7 +82,7 @@ func TestContextToDoc(t *testing.T) {
|
|||||||
{Timestamp: time.Now(), Source: "user", Content: "特别是Go语言", Response: "Go很棒"},
|
{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 {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@ -178,7 +178,7 @@ func TestSummarizeEntries(t *testing.T) {
|
|||||||
{Source: "user", Content: "今天天气如何"},
|
{Source: "user", Content: "今天天气如何"},
|
||||||
{Source: "user", Content: "明天会下雨吗"},
|
{Source: "user", Content: "明天会下雨吗"},
|
||||||
}
|
}
|
||||||
summary := summarizeEntries(entries)
|
summary := summarizeEntries(entries, func(s string) string { return s }, nil)
|
||||||
if summary == "" {
|
if summary == "" {
|
||||||
t.Error("summary should not be empty")
|
t.Error("summary should not be empty")
|
||||||
}
|
}
|
||||||
@ -198,7 +198,7 @@ func TestExtractTags(t *testing.T) {
|
|||||||
entries := []ContextEntry{
|
entries := []ContextEntry{
|
||||||
{Content: "我喜欢喝咖啡和编程"},
|
{Content: "我喜欢喝咖啡和编程"},
|
||||||
}
|
}
|
||||||
tags := extractTags(entries)
|
tags := extractTags(entries, func(s string) string { return s }, nil)
|
||||||
if len(tags) == 0 {
|
if len(tags) == 0 {
|
||||||
t.Error("should extract tags")
|
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"])
|
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")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@ -31,6 +31,9 @@ type Relation struct {
|
|||||||
TurnID int `json:"turn_id"`
|
TurnID int `json:"turn_id"`
|
||||||
CreatedAt time.Time `json:"created_at"`
|
CreatedAt time.Time `json:"created_at"`
|
||||||
DateBucket string `json:"date_bucket"`
|
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 {
|
type Triple struct {
|
||||||
@ -93,6 +96,9 @@ func (g *GraphDB) initSchema() error {
|
|||||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||||
date_bucket TEXT,
|
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 (source_id) REFERENCES entities(id),
|
||||||
FOREIGN KEY (target_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) {
|
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 {
|
if len(keywords) == 0 && len(seedEntities) == 0 {
|
||||||
rows, err := g.db.Query(
|
rows, err := g.db.Query(
|
||||||
`SELECT id, name, type, mention_count, created_at, updated_at
|
`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 {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@ -259,7 +278,8 @@ func (g *GraphDB) Recall(keywords []string, seedEntities []string, depth int, se
|
|||||||
relRows, err := g.db.Query(
|
relRows, err := g.db.Query(
|
||||||
`SELECT r.id, r.source_id, r.target_id, e1.name, e2.name,
|
`SELECT r.id, r.source_id, r.target_id, e1.name, e2.name,
|
||||||
r.relation_type, r.confidence, r.status, r.session_id,
|
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
|
FROM relations r
|
||||||
JOIN entities e1 ON r.source_id = e1.id
|
JOIN entities e1 ON r.source_id = e1.id
|
||||||
JOIN entities e2 ON r.target_id = e2.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,
|
if err := relRows.Scan(&rel.ID, &rel.SourceID, &rel.TargetID,
|
||||||
&rel.SourceName, &rel.TargetName, &rel.RelationType,
|
&rel.SourceName, &rel.TargetName, &rel.RelationType,
|
||||||
&rel.Confidence, &rel.Status, &rel.SessionID,
|
&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
|
return nil, err
|
||||||
}
|
}
|
||||||
result.Relations = append(result.Relations, rel)
|
result.Relations = append(result.Relations, rel)
|
||||||
@ -340,7 +361,8 @@ func (g *GraphDB) Recall(keywords []string, seedEntities []string, depth int, se
|
|||||||
query := fmt.Sprintf(
|
query := fmt.Sprintf(
|
||||||
`SELECT r.id, r.source_id, r.target_id, e1.name, e2.name,
|
`SELECT r.id, r.source_id, r.target_id, e1.name, e2.name,
|
||||||
r.relation_type, r.confidence, r.status, r.session_id,
|
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
|
FROM relations r
|
||||||
JOIN entities e1 ON r.source_id = e1.id
|
JOIN entities e1 ON r.source_id = e1.id
|
||||||
JOIN entities e2 ON r.target_id = e2.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,
|
if err := relRows.Scan(&rel.ID, &rel.SourceID, &rel.TargetID,
|
||||||
&rel.SourceName, &rel.TargetName, &rel.RelationType,
|
&rel.SourceName, &rel.TargetName, &rel.RelationType,
|
||||||
&rel.Confidence, &rel.Status, &rel.SessionID,
|
&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()
|
relRows.Close()
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@ -747,6 +770,89 @@ func (g *GraphDB) Archive(days int) (int, error) {
|
|||||||
return int(n), nil
|
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 {
|
func (g *GraphDB) Close() error {
|
||||||
return g.db.Close()
|
return g.db.Close()
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user