feat(memory): P1 长期记忆升级 —— 异步攒批 Consolidate + 软删 + importance/last_seen #1

Merged
Blizzard merged 181 commits from feat/wails3 into main 2026-07-17 01:12:32 +00:00
9 changed files with 148 additions and 44 deletions
Showing only changes of commit 446b784fc6 - Show all commits
+2 -1
View File
@@ -113,7 +113,8 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为
**已知短板 / 优化清单**(按性价比,逐项推进):
- [ ] **P1 评测闭环**:eval 结果落库 + 阈值门控(低分标记/触发重试)+ 桌面端质量面板。现状只打日志、不闭环。
- [ ] **P1 RAG 忠实度评测**检索原文喂给 judge真评幻觉/引用一致性。现状 judge 看不到来源,幻觉其实没评(已知洞)
- [x] **P1 RAG 忠实度评测**:检索原文喂给 judge一次评质量+忠实度,未被来源支持的说法进 Flags
综合分(有来源)=0.3规则+0.35质量+0.35忠实。runGraph 透传 refs → evaluate。live 实测忠实 1.00/来源 1,单测覆盖。
- [ ] **P2 输出脱敏增强**:滑动窗口跨片检测(现流式逐片会漏跨 token 的密钥)+ PII 模式(手机号/邮箱/身份证)。
- [ ] **P2 输入护栏升级**:纯正则易被改写/编码绕过;加轻量 jailbreak 分类器或 LLM 兜底;`bannedTerms` 落地。
- [ ] **P3 坏输出自动纠偏**:低 eval 分触发重生成/降级路由(让 harness 真闭环)。
@@ -33,7 +33,7 @@ func TestAgentCollaborationPassesOutput(t *testing.T) {
return "研究产出XYZ"
}}
o := &Orchestrator{pool: ll, breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}}
ans, err := o.runGraph(context.Background(), &contract.Task{ID: "tc", Graph: []byte(graph)}, &execTracer{})
ans, _, err := o.runGraph(context.Background(), &contract.Task{ID: "tc", Graph: []byte(graph)}, &execTracer{})
if err != nil {
t.Fatal(err)
}
@@ -24,9 +24,11 @@ func registerFlowMerge() {
}
// executeGraph 按灰度开关选编排实现:compose.GraphPhase C)或自研 graph.go(默认/权威)。
func (o *Orchestrator) executeGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, error) {
// 返回 (成稿, 检索来源, error);来源供忠实度评测(compose 路径暂不提供来源 → 返回 nil)。
func (o *Orchestrator) executeGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, []string, error) {
if composeEnabled() {
return o.runComposeGraph(ctx, t, tr)
ans, err := o.runComposeGraph(ctx, t, tr)
return ans, nil, err
}
return o.runGraph(ctx, t, tr)
}
@@ -159,7 +161,8 @@ func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr
r, cerr := g.Compile(ctx, compose.WithNodeTriggerMode(compose.AllPredecessor))
if cerr != nil {
tr.info("task", "system", "compose 编译失败", "退回自研 graph.go"+cerr.Error())
return o.runGraph(ctx, t, tr)
ans, _, gerr := o.runGraph(ctx, t, tr) // 降级路径丢弃 refs(compose 路径暂不评忠实度)
return ans, gerr
}
if _, ierr := r.Invoke(ctx, flowSignal{}); ierr != nil {
tr.info("task", "system", "compose 执行告警", ierr.Error()) // 副作用已落 board;下方按需补一段答复
@@ -29,7 +29,7 @@ func runBoth(t *testing.T, graph string) (interp, comp string) {
task := &contract.Task{ID: "t_eq", Graph: []byte(graph)}
o1 := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}}
a1, err := o1.runGraph(context.Background(), task, &execTracer{})
a1, _, err := o1.runGraph(context.Background(), task, &execTracer{})
if err != nil {
t.Fatalf("runGraph: %v", err)
}
+12 -6
View File
@@ -42,7 +42,7 @@ type board struct {
// branch 按条件只激活选中的下游(剪枝)→ agent 节点流式回流 token。
//
// 逐节点点亮"运行·观测"。返回终端 agent 的完整产出(供写回历史)。
func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, error) {
func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, []string, error) {
flow, ferr := dsl.Parse(t.Graph)
plan := dsl.Compile(t.Graph)
b := &board{
@@ -57,7 +57,7 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT
b.profile = o.fetchMemory(ctx, b.uid, b.query)
b.history = o.fetchHistory(ctx, b.sid)
o.runConversation(ctx, t.ID, b, plan.System, tr, "agent")
return b.answer, b.fatalErr // 模型失败 → 上抛判 failed
return b.answer, refsOf(b), b.fatalErr // 模型失败 → 上抛判 failed
}
// 建邻接与入度(只认两端都存在的边)。保留整条边以便 branch 按 true/false 标签选路。
@@ -165,10 +165,10 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT
}
if b.rejected {
return b.answer, errRejected // 合法终态,Handle 据此判 rejected 并优雅收尾
return b.answer, nil, errRejected // 合法终态,Handle 据此判 rejected 并优雅收尾
}
if b.fatalErr != nil {
return b.answer, b.fatalErr // 上抛 → Handle 判 failed(带原因)
return b.answer, nil, b.fatalErr // 上抛 → Handle 判 failed(带原因)
}
// 图里无 agent 节点(纯工具/检索图)也要出一段模型答复,否则没有输出。
@@ -176,9 +176,15 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT
o.runConversation(ctx, t.ID, b, plan.System, tr, "agent")
}
if b.fatalErr != nil {
return b.answer, b.fatalErr
return b.answer, nil, b.fatalErr
}
return b.answer, nil
return b.answer, refsOf(b), nil // 成功:带回检索来源供忠实度评测
}
// refsOf 汇总本次执行的检索来源(检索资料 + 工具产出),供忠实度评测。
func refsOf(b *board) []string {
out := append([]string{}, b.refs...)
return append(out, b.toolOut...)
}
// retrieverNode 执行检索节点:kb 按 owner 作用域 → kb_search → 累计参考资料。
@@ -134,7 +134,7 @@ func TestRunGraph_BranchRouting(t *testing.T) {
ll := &fakeLLM{ready: true, stream: func(m []llm.ChatMessage) string { return m[0].Content }}
run := func(cond string) string {
o := newOrch(ll, &fakeTools{}, &fakeSink{}, &fakeExec{})
ans, err := o.runGraph(context.Background(), task(strings.Replace(g, "%s", cond, 1)), o.tracer("t1"))
ans, _, err := o.runGraph(context.Background(), task(strings.Replace(g, "%s", cond, 1)), o.tracer("t1"))
if err != nil {
t.Fatal(err)
}
@@ -160,7 +160,7 @@ func TestRunGraph_ToolFeedsAgent(t *testing.T) {
return &contract.ToolResult{OK: true, Content: "TOOLDATA"}
}}
o := newOrch(ll, ft, &fakeSink{}, &fakeExec{})
ans, err := o.runGraph(context.Background(), task(g), o.tracer("t1"))
ans, _, err := o.runGraph(context.Background(), task(g), o.tracer("t1"))
if err != nil {
t.Fatal(err)
}
@@ -186,7 +186,7 @@ func TestRunGraph_MapFanout(t *testing.T) {
return "正文XYZ", nil
}}
o := newOrch(ll, &fakeTools{}, &fakeSink{}, &fakeExec{})
ans, err := o.runGraph(context.Background(), task(g), o.tracer("t1"))
ans, _, err := o.runGraph(context.Background(), task(g), o.tracer("t1"))
if err != nil {
t.Fatal(err)
}
@@ -203,7 +203,7 @@ func TestRunGraph_OutputRedaction(t *testing.T) {
}}
fs := &fakeSink{}
o := newOrch(ll, &fakeTools{}, fs, &fakeExec{})
if _, err := o.runGraph(context.Background(), task(g), o.tracer("t1")); err != nil {
if _, _, err := o.runGraph(context.Background(), task(g), o.tracer("t1")); err != nil {
t.Fatal(err)
}
out := fs.text()
@@ -158,7 +158,7 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
tr.info("task", "system", "任务受理", fmt.Sprintf("DSL %d 字节,按图执行", len(t.Graph)))
// 按 DSL 图执行:compose.GraphEINO_COMPOSE=1)或自研 graph.go(默认);agent 节点流式回流 token。
answer, err := o.executeGraph(tctx, t, tr)
answer, refs, err := o.executeGraph(tctx, t, tr)
if errors.Is(err, errRejected) {
// HITL 拒绝:合法终态,非故障。收尾流 + 置 rejected,不计熔断、不重投。
slog.InfoContext(ctx, "task rejected by approval", "task_id", t.ID)
@@ -189,21 +189,22 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
// 写回阶段:离开热路径、异步落历史 + (TODO)抽取记忆。
go o.memorize(t, answer)
// 自动化评测:离开热路径,对本轮输出打分并记录(规则 + LLM-as-judge)。
go o.evaluate(t, dsl.Compile(t.Graph).Query, answer)
// 自动化评测:离开热路径,对本轮输出打分并记录(规则 + LLM-as-judge + RAG 忠实度)。
go o.evaluate(t, dsl.Compile(t.Graph).Query, answer, refs)
return nil
}
// evaluate 异步对一次输出做自动化评测并记录评分(off 热路径,不影响响应)。
func (o *Orchestrator) evaluate(t *contract.Task, input, output string) {
// sources 为本轮检索来源:非空时额外评忠实度(幻觉检测)。
func (o *Orchestrator) evaluate(t *contract.Task, input, output string, sources []string) {
if o.eval == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
r := o.eval.Score(ctx, input, output)
log.Printf("[eval] task %s 综合 %.2f(规则 %.2f / LLM %.2fflags=%v %s",
t.ID, r.Overall, r.Rule, r.LLM, r.Flags, r.Reason)
r := o.eval.Score(ctx, input, output, sources)
log.Printf("[eval] task %s 综合 %.2f(规则 %.2f / LLM %.2f / 忠实 %.2f,来源 %dflags=%v %s",
t.ID, r.Overall, r.Rule, r.LLM, r.Faithful, len(sources), r.Flags, r.Reason)
}
// fetchMemory 经 MCP memory_get 工具召回用户常驻画像。
+57 -8
View File
@@ -7,12 +7,13 @@ import (
"strings"
)
// Result 是一次输出评测的结果。Overall ∈ [0,1]。
// Result 是一次输出评测的结果。各分 ∈ [0,1]。
type Result struct {
Overall float64 // 综合分(有 LLM 评审则 0.4*规则 + 0.6*LLM,否则=规则分
Overall float64 // 综合分(见 Score 的加权
Rule float64 // 规则分
LLM float64 // LLM-as-judge 分(0 表示未评/失败)
Flags []string // 命中的规则问题
LLM float64 // LLM-as-judge 质量分(0=未评/失败)
Faithful float64 // RAG 忠实度分(有检索来源时才评;0=未评/无来源)
Flags []string // 命中的问题(规则 + 未被来源支持的说法)
Reason string // LLM 评语
}
@@ -28,17 +29,31 @@ func NewEvaluator(ready func() bool, chat func(ctx context.Context, sys, user st
return &Evaluator{ready: ready, chat: chat}
}
// Score 对一次推理输出综合打分。
func (e *Evaluator) Score(ctx context.Context, input, output string) Result {
// Score 对一次推理输出综合打分。sources 为检索到的资料(RAG 来源):非空时额外评忠实度
// (回答是否基于来源、有无编造),并把"未被来源支持的说法"加进 Flags。
func (e *Evaluator) Score(ctx context.Context, input, output string, sources []string) Result {
rule, flags := ruleScore(output)
res := Result{Rule: rule, Flags: flags, Overall: rule}
if e.ready != nil && e.chat != nil && e.ready() {
if e.ready == nil || e.chat == nil || !e.ready() {
return res
}
if len(sources) > 0 {
// RAG 路径:一次 judge 同时评质量 + 忠实度 + 列未支持说法。
if q, f, unsupported, reason, ok := e.llmJudgeGrounded(ctx, input, output, sources); ok {
res.LLM, res.Faithful, res.Reason = q, f, reason
res.Overall = 0.3*rule + 0.35*q + 0.35*f
for _, u := range unsupported {
res.Flags = append(res.Flags, "未被来源支持:"+u)
}
}
return res
}
// 无来源:仅评质量(相关/准确/完整)。
if s, reason, ok := e.llmJudge(ctx, input, output); ok {
res.LLM = s
res.Reason = reason
res.Overall = 0.4*rule + 0.6*s
}
}
return res
}
@@ -114,6 +129,40 @@ func (e *Evaluator) llmJudge(ctx context.Context, input, output string) (float64
return s, j.Reason, true
}
// llmJudgeGrounded 给 judge 同时喂用户问题、检索资料、模型回答,一次返回:
// quality 质量分、faithful 忠实度分(均归一到 [0,1])、unsupported 未被资料支持的说法、reason 评语。
func (e *Evaluator) llmJudgeGrounded(ctx context.Context, input, output string, sources []string) (quality, faithful float64, unsupported []string, reason string, ok bool) {
sys := "你是严格的 RAG 回答评审,重点核查回答是否严格基于检索资料、有无编造(幻觉)。"
src := evalTruncate(strings.Join(sources, "\n---\n"), 3000)
user := fmt.Sprintf(
"【用户问题】%s\n\n【检索资料】\n%s\n\n【模型回答】%s\n\n"+
"请评估两项:quality=回答质量(相关/准确/完整)faithfulness=忠实度(回答是否严格基于检索资料、有无编造)。"+
"并列出 unsupported:回答中未被检索资料支持的具体说法(无则空数组)。"+
"只输出 JSON{\"quality\":1到5整数,\"faithfulness\":1到5整数,\"unsupported\":[\"...\"],\"reason\":\"一句话\"},不要多余文字。",
evalTruncate(input, 400), src, evalTruncate(output, 1500))
txt, err := e.chat(ctx, sys, user)
if err != nil {
return 0, 0, nil, "", false
}
var j struct {
Quality float64 `json:"quality"`
Faithfulness float64 `json:"faithfulness"`
Unsupported []string `json:"unsupported"`
Reason string `json:"reason"`
}
if json.Unmarshal([]byte(evalStripFence(txt)), &j) != nil || j.Quality <= 0 || j.Faithfulness <= 0 {
return 0, 0, nil, "", false
}
norm := func(v float64) float64 {
s := v / 5.0
if s > 1 {
s = 1
}
return s
}
return norm(j.Quality), norm(j.Faithfulness), j.Unsupported, j.Reason, true
}
func evalTruncate(s string, n int) string {
r := []rune(s)
if len(r) <= n {
@@ -40,7 +40,7 @@ func TestRuleScore_HeavyRepeat(t *testing.T) {
func TestScore_RuleOnly(t *testing.T) {
e := NewEvaluator(nil, nil) // 无 LLM → 仅规则
r := e.Score(context.Background(), "问题", "一段质量不错的较完整回答内容,长度足够,没有任何问题。")
r := e.Score(context.Background(), "问题", "一段质量不错的较完整回答内容,长度足够,没有任何问题。", nil)
if r.LLM != 0 || r.Overall != r.Rule {
t.Errorf("无 LLM 时 Overall 应等于规则分, got overall=%.2f rule=%.2f llm=%.2f", r.Overall, r.Rule, r.LLM)
}
@@ -54,7 +54,7 @@ func TestScore_WithLLMJudge(t *testing.T) {
return "```json\n{\"score\":4,\"reason\":\"相关且较完整\"}\n```", nil
},
)
r := e.Score(context.Background(), "介绍杭州", "杭州是浙江省会,西湖闻名,历史与现代交融,电商发达。")
r := e.Score(context.Background(), "介绍杭州", "杭州是浙江省会,西湖闻名,历史与现代交融,电商发达。", nil)
if r.LLM <= 0 {
t.Fatalf("应有 LLM 分, got %.2f", r.LLM)
}
@@ -75,12 +75,56 @@ func TestScore_LLMJudgeBadJSONFallsBack(t *testing.T) {
func() bool { return true },
func(ctx context.Context, sys, user string) (string, error) { return "我觉得还行吧", nil },
)
r := e.Score(context.Background(), "q", "一段足够长且正常的回答内容用于评测。")
r := e.Score(context.Background(), "q", "一段足够长且正常的回答内容用于评测。", nil)
if r.LLM != 0 || r.Overall != r.Rule {
t.Errorf("LLM 返回非 JSON 应回退到规则分, got overall=%.2f llm=%.2f", r.Overall, r.LLM)
}
}
func TestScore_GroundedFaithfulness(t *testing.T) {
// 有检索来源 → 走忠实度评测:judge 返回 quality/faithfulness/unsupported。
var gotPrompt string
e := NewEvaluator(
func() bool { return true },
func(ctx context.Context, sys, user string) (string, error) {
gotPrompt = user
return `{"quality":4,"faithfulness":2,"unsupported":["该产品支持离线模式"],"reason":"部分说法无资料支撑"}`, nil
},
)
r := e.Score(context.Background(), "这产品支持什么", "它支持在线与离线模式。",
[]string{"产品支持在线协作。", "产品提供云端存储。"})
if !strings.Contains(gotPrompt, "检索资料") || !strings.Contains(gotPrompt, "产品支持在线协作") {
t.Fatalf("judge 提示词应包含检索资料, got: %q", gotPrompt)
}
if r.LLM < 0.79 || r.LLM > 0.81 { // quality 4/5
t.Errorf("quality 应为 0.8, got %.2f", r.LLM)
}
if r.Faithful < 0.39 || r.Faithful > 0.41 { // faithfulness 2/5
t.Errorf("忠实度应为 0.4, got %.2f", r.Faithful)
}
if !contains(r.Flags, "未被来源支持:该产品支持离线模式") {
t.Errorf("未支持说法应进 Flags, got %v", r.Flags)
}
want := 0.3*r.Rule + 0.35*r.LLM + 0.35*r.Faithful
if r.Overall < want-1e-9 || r.Overall > want+1e-9 {
t.Errorf("综合分应为 0.3规则+0.35质量+0.35忠实=%.3f, got %.3f", want, r.Overall)
}
}
func TestScore_NoSourcesSkipsFaithfulness(t *testing.T) {
e := NewEvaluator(
func() bool { return true },
func(ctx context.Context, sys, user string) (string, error) {
return `{"score":5,"reason":"好"}`, nil
},
)
r := e.Score(context.Background(), "q", "一段足够长且正常的回答内容用于评测。", nil)
if r.Faithful != 0 {
t.Errorf("无来源不应评忠实度, got %.2f", r.Faithful)
}
}
func contains(ss []string, want string) bool {
for _, s := range ss {
if s == want {