diff --git a/project_analysis.md b/project_analysis.md index c91fe67..ee194e7 100644 --- a/project_analysis.md +++ b/project_analysis.md @@ -118,7 +118,9 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为 综合分(有来源)=0.3规则+0.35质量+0.35忠实。runGraph 透传 refs → evaluate。live 实测忠实 1.00/来源 1，单测覆盖。 - [ ] **P2 输出脱敏增强**：滑动窗口跨片检测（现流式逐片会漏跨 token 的密钥）+ PII 模式（手机号/邮箱/身份证）。 - [ ] **P2 输入护栏升级**：纯正则易被改写/编码绕过；加轻量 jailbreak 分类器或 LLM 兜底；`bannedTerms` 落地。 -- [ ] **P3 坏输出自动纠偏**：低 eval 分触发重生成/降级路由（让 harness 真闭环）。 +- [x] **P3 坏输出自动纠偏** ✅：poor(<0.5) 触发评语驱动的重生成，重评后**仅采纳更优者(不退步)**， + 采纳的修订版落会话历史 + 评测终值带 `corrected` 标记落库。`maxRefineRounds=1`、`canRefine`(模型就绪且熔断未开)门控； + 单 goroutine 串 评测→纠偏→落历史避竞态。单测覆盖 采纳/不退步/非低分不触发，live 验证好答案不误触发。**至此 harness 由「测温计」迈入「恒温器」。** - [ ] **P3 成本/Token 预算护栏**：单任务/单用户 token 上限与告警。 --- diff --git a/sundynix-dispatcher/internal/eino/orchestrator.go b/sundynix-dispatcher/internal/eino/orchestrator.go index f8e75bd..4280d82 100644 --- a/sundynix-dispatcher/internal/eino/orchestrator.go +++ b/sundynix-dispatcher/internal/eino/orchestrator.go @@ -193,37 +193,118 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error { o.breaker.Report(true) o.finishStatus(t.ID, nil) - // 写回阶段：离开热路径、异步落历史 + （TODO）抽取记忆。 - go o.memorize(t, answer) - // 自动化评测：离开热路径，对本轮输出打分并记录（规则 + LLM-as-judge + RAG 忠实度）。 - go o.evaluate(t, dsl.Compile(t.Graph).Query, answer, refs) + // 写回阶段（离热路径，单 goroutine 串起评测与落历史）： + // 先评测（低分则自动纠偏重生成、取更优者），再把最终采纳的答案落会话历史 —— 保证 + // 多轮上下文用的是纠偏后的好答案，而非被判低分的初版。 + go func() { + query := dsl.Compile(t.Graph).Query + final := o.evaluate(t, query, answer, refs) + o.memorize(t, final) + }() return nil } +// maxRefineRounds 是低分自动纠偏的最大重生成轮数（防成本失控/死循环）。 +// 每轮 = 1 次重写 + 1 次重评；poor 本就稀少，1 轮足以覆盖大多数偶发劣化。 +const maxRefineRounds = 1 + // evaluate 异步对一次输出做自动化评测并记录评分（off 热路径，不影响响应）。 +// 低分（poor）时触发"评语驱动"的自动纠偏：重生成更优答案、重评、取更优者（恒温器闭环）。 // sources 为本轮检索来源：非空时额外评忠实度（幻觉检测）。 -func (o *Orchestrator) evaluate(t *contract.Task, input, output string, sources []string) { +// 返回最终采纳的答案（纠偏成功则为修订版，否则为原文）——供调用方落会话历史。 +func (o *Orchestrator) evaluate(t *contract.Task, input, output string, sources []string) string { if o.eval == nil { - return + return output } - ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second) defer cancel() r := o.eval.Score(ctx, input, output, sources) level := evalLevel(r) log.Printf("[eval] task %s 综合 %.2f（规则 %.2f / LLM %.2f / 忠实 %.2f，来源 %d）level=%s flags=%v %s", t.ID, r.Overall, r.Rule, r.LLM, r.Faithful, len(sources), level, r.Flags, r.Reason) - if level == contract.EvalPoor { // 低质量告警（可观测 + 后续可挂报警） - slog.Warn("eval poor quality", "task_id", t.ID, "overall", r.Overall, "faithful", r.Faithful, "flags", r.Flags) + + // 低分自动纠偏：把"问题 + 初版回答 + 评审短板"交给模型重写，重评后仅采纳更优者（不退步）。 + corrected := false + for round := 0; level == contract.EvalPoor && round < maxRefineRounds && o.canRefine(); round++ { + newOut, newR, ok := o.refine(ctx, t, input, output, sources, r) + if !ok { + break // 重写失败或未改好：保留当前最优，停止 + } + prevOverall := r.Overall + output, r, level, corrected = newOut, newR, evalLevel(newR), true + log.Printf("[eval] task %s 自动纠偏第 %d 轮采纳：综合 %.2f→%.2f level=%s", t.ID, round+1, prevOverall, r.Overall, level) } - // 闭环：评测结果落库（供 UI 查询 / 质量趋势 / 门控）。 + + if level == contract.EvalPoor { // 纠偏后仍低质量：告警（可观测 + 后续可挂报警） + slog.Warn("eval poor quality", "task_id", t.ID, "overall", r.Overall, "faithful", r.Faithful, "flags", r.Flags, "corrected", corrected) + } + // 闭环：评测结果（纠偏后终值）落库（供 UI 查询 / 质量趋势 / 门控）。 if o.evalSink != nil { if err := o.evalSink.PublishEval(&contract.EvalEvent{ TaskID: t.ID, Overall: r.Overall, Rule: r.Rule, LLM: r.LLM, Faithful: r.Faithful, - Level: level, Flags: r.Flags, Reason: r.Reason, Sources: len(sources), TS: time.Now().UnixMilli(), + Level: level, Flags: r.Flags, Reason: r.Reason, Sources: len(sources), Corrected: corrected, + TS: time.Now().UnixMilli(), }); err != nil { log.Printf("[eval] 回写评测结果失败 task=%s: %v", t.ID, err) } } + return output +} + +// canRefine 报告是否具备自动纠偏条件：模型就绪且熔断未开（重写要再打一次模型，避雪上加霜）。 +func (o *Orchestrator) canRefine() bool { + return o.pool != nil && o.pool.Ready() && (o.breaker == nil || o.breaker.Allow()) +} + +// refine 对低分输出做一次"评语驱动"的重生成：把原问题 + 初版回答 + 评测短板（flags/评语） +// 交给模型重写，再重新评分。仅当新分严格更高时采纳（ok=true）；否则保留原文（不退步）。 +// 有检索来源时把来源一并喂回，要求严格基于来源作答（同时压低幻觉）。 +func (o *Orchestrator) refine(ctx context.Context, t *contract.Task, input, output string, sources []string, prev harness.Result) (string, harness.Result, bool) { + tr := o.tracer(t.ID) + end := tr.span("refine", "system", "低分自动纠偏") + + critique := strings.Join(prev.Flags, "；") + if prev.Reason != "" { + if critique != "" { + critique += "；" + } + critique += "评语：" + prev.Reason + } + if critique == "" { + critique = "整体质量不足（相关性/准确性/完整性欠佳）" + } + sys := "你是严谨的回答修订专家。下面给你一个用户问题、一份质量不足的初版回答、以及评审指出的问题。请针对这些问题重写一份更准确、相关、完整的回答。" + srcBlock := "" + if len(sources) > 0 { + srcBlock = "\n\n【可参考的检索资料（严格基于它作答，不要编造资料外的内容）】\n" + + truncate(strings.Join(sources, "\n---\n"), 3000) + } + user := fmt.Sprintf("【用户问题】%s\n\n【初版回答（质量不足）】%s\n\n【评审指出的问题】%s%s\n\n请直接输出修订后的回答正文，不要解释你做了哪些修改。", + truncate(input, 1000), truncate(output, 2000), critique, srcBlock) + + revised, err := o.pool.Chat(ctx, []llm.ChatMessage{ + {Role: "system", Content: sys}, + {Role: "user", Content: user}, + }) + if err != nil || strings.TrimSpace(revised) == "" { + end("重写失败", errOrEmpty(err)) + return output, prev, false + } + newR := o.eval.Score(ctx, input, revised, sources) + if newR.Overall <= prev.Overall { + end(fmt.Sprintf("重写未提升（%.2f≤%.2f），保留原版", newR.Overall, prev.Overall), nil) + return output, prev, false + } + end(fmt.Sprintf("已采纳修订：综合 %.2f→%.2f", prev.Overall, newR.Overall), nil) + return revised, newR, true +} + +// errOrEmpty 把空输出也表达成一个错误，供 exec error 事件着色。 +func errOrEmpty(err error) error { + if err != nil { + return err + } + return errors.New("空输出") } // evalLevel 据综合分 + 忠实度把评测分级（闭环门控/告警用）。 diff --git a/sundynix-dispatcher/internal/eino/refine_test.go b/sundynix-dispatcher/internal/eino/refine_test.go new file mode 100644 index 0000000..4826d23 --- /dev/null +++ b/sundynix-dispatcher/internal/eino/refine_test.go @@ -0,0 +1,107 @@ +package eino + +import ( + "context" + "strings" + "testing" + + "github.com/sundynix/sundynix-dispatcher/internal/harness" + "github.com/sundynix/sundynix-dispatcher/internal/llm" + "github.com/sundynix/sundynix-shared/contract" +) + +// fakeEvalSink 捕获回流的评测事件，供断言纠偏终值。 +type fakeEvalSink struct{ ev *contract.EvalEvent } + +func (f *fakeEvalSink) PublishEval(ev *contract.EvalEvent) error { f.ev = ev; return nil } + +// orchForEval 组一个仅评测/纠偏所需依赖的编排器（生成器 gen + 评审 judge + 评测出口 sink）。 +func orchForEval(gen *fakeLLM, judge func(ctx context.Context, sys, user string) (string, error), sink *fakeEvalSink) *Orchestrator { + return &Orchestrator{ + pool: gen, + breaker: harness.NewCircuitBreaker(), + eval: harness.NewEvaluator(func() bool { return true }, judge), + exec: &fakeExec{}, + evalSink: sink, + } +} + +// 低分 → 自动纠偏重生成 → 新答更优 → 采纳修订并标记 Corrected，落库终值升为 ok。 +func TestEvaluate_AutoRefine_AdoptsBetter(t *testing.T) { + // 评审：含 refusal 标记的初版给 1 分，重写后的正常回答给 5 分。 + judge := func(_ context.Context, _, user string) (string, error) { + if strings.Contains(user, "error:") { + return `{"score":1,"reason":"差"}`, nil + } + return `{"score":5,"reason":"好"}`, nil + } + gen := &fakeLLM{ready: true, chat: func(_ []llm.ChatMessage) (string, error) { + return "光合作用是植物利用光能把二氧化碳和水合成有机物并释放氧气的过程。", nil + }} + sink := &fakeEvalSink{} + o := orchForEval(gen, judge, sink) + + final := o.evaluate(&contract.Task{ID: "t1"}, "介绍光合作用", "error: 我无法回答", nil) + + if !strings.Contains(final, "光合作用") { + t.Fatalf("应采纳纠偏后的好答案，got %q", final) + } + if sink.ev == nil || !sink.ev.Corrected { + t.Fatalf("应标记 Corrected=true, got %+v", sink.ev) + } + if sink.ev.Level != contract.EvalOK { + t.Errorf("纠偏后应升为 ok，got %s (overall %.2f)", sink.ev.Level, sink.ev.Overall) + } +} + +// 低分 → 纠偏后未提升 → 保留原版（不退步），不标记 Corrected，终值仍 poor。 +func TestEvaluate_AutoRefine_KeepsOriginalWhenNoGain(t *testing.T) { + judge := func(_ context.Context, _, _ string) (string, error) { + return `{"score":1,"reason":"差"}`, nil // 怎么写都低分 + } + gen := &fakeLLM{ready: true, chat: func(_ []llm.ChatMessage) (string, error) { + return "另一个同样糟糕的 error: 回答", nil + }} + sink := &fakeEvalSink{} + o := orchForEval(gen, judge, sink) + + orig := "error: 初版很差的回答" + final := o.evaluate(&contract.Task{ID: "t2"}, "问题", orig, nil) + + if final != orig { + t.Fatalf("未提升应保留原版，got %q", final) + } + if sink.ev.Corrected { + t.Error("未提升不应标记 Corrected") + } + if sink.ev.Level != contract.EvalPoor { + t.Errorf("应仍为 poor，got %s", sink.ev.Level) + } +} + +// 非低分（ok/warn）不触发纠偏：生成器不应被调用。 +func TestEvaluate_NoRefineWhenNotPoor(t *testing.T) { + called := false + gen := &fakeLLM{ready: true, chat: func(_ []llm.ChatMessage) (string, error) { + called = true + return "不该被调用", nil + }} + judge := func(_ context.Context, _, _ string) (string, error) { + return `{"score":5,"reason":"好"}`, nil + } + sink := &fakeEvalSink{} + o := orchForEval(gen, judge, sink) + + orig := "一段质量不错的正常回答内容。" + final := o.evaluate(&contract.Task{ID: "t3"}, "问题", orig, nil) + + if called { + t.Error("非低分不应触发纠偏（生成器被调用了）") + } + if final != orig { + t.Errorf("无纠偏应原样返回，got %q", final) + } + if sink.ev.Corrected || sink.ev.Level != contract.EvalOK { + t.Errorf("应为未纠偏的 ok，got level=%s corrected=%v", sink.ev.Level, sink.ev.Corrected) + } +} diff --git a/sundynix-gateway/cmd/server/main.go b/sundynix-gateway/cmd/server/main.go index 503ef48..f7bedaa 100644 --- a/sundynix-gateway/cmd/server/main.go +++ b/sundynix-gateway/cmd/server/main.go @@ -70,7 +70,7 @@ func main() { flags, _ := json.Marshal(ev.Flags) if err := db.SaveEval(context.Background(), &store.Eval{ TaskID: ev.TaskID, Overall: ev.Overall, Rule: ev.Rule, LLM: ev.LLM, Faithful: ev.Faithful, - Level: ev.Level, Flags: string(flags), Reason: ev.Reason, Sources: ev.Sources, + Level: ev.Level, Flags: string(flags), Reason: ev.Reason, Sources: ev.Sources, Corrected: ev.Corrected, }); err != nil { log.Printf("[gateway] 落库评测 %s 失败: %v", ev.TaskID, err) } diff --git a/sundynix-gateway/internal/handler/task_handler.go b/sundynix-gateway/internal/handler/task_handler.go index 7132b7c..a6c18e1 100644 --- a/sundynix-gateway/internal/handler/task_handler.go +++ b/sundynix-gateway/internal/handler/task_handler.go @@ -100,7 +100,8 @@ func (h *Handler) TaskEval(c *gin.Context) { _ = json.Unmarshal([]byte(e.Flags), &flags) c.JSON(http.StatusOK, gin.H{ "task_id": e.TaskID, "overall": e.Overall, "rule": e.Rule, "llm": e.LLM, - "faithful": e.Faithful, "level": e.Level, "flags": flags, "reason": e.Reason, "sources": e.Sources, + "faithful": e.Faithful, "level": e.Level, "flags": flags, "reason": e.Reason, + "sources": e.Sources, "corrected": e.Corrected, }) } diff --git a/sundynix-gateway/internal/store/models.go b/sundynix-gateway/internal/store/models.go index eb2a989..f4f62d0 100644 --- a/sundynix-gateway/internal/store/models.go +++ b/sundynix-gateway/internal/store/models.go @@ -26,15 +26,16 @@ type Task struct { // Eval 是一次任务的自动化评测结果（dispatcher 评完经 NATS 回写，每任务一条，按 task_id upsert）。 type Eval struct { BaseModel - TaskID string `gorm:"uniqueIndex;size:64"` - Overall float64 // 综合分 [0,1] - Rule float64 // 规则分 - LLM float64 // LLM 质量分 - Faithful float64 // RAG 忠实度分（0=无来源未评） - Level string `gorm:"size:16"` // ok / warn / poor - Flags string `gorm:"type:text"` // 命中问题（JSON 数组字符串） - Reason string `gorm:"type:text"` // 评语 - Sources int // 检索来源数 + TaskID string `gorm:"uniqueIndex;size:64"` + Overall float64 // 综合分 [0,1] + Rule float64 // 规则分 + LLM float64 // LLM 质量分 + Faithful float64 // RAG 忠实度分（0=无来源未评） + Level string `gorm:"size:16"` // ok / warn / poor + Flags string `gorm:"type:text"` // 命中问题（JSON 数组字符串） + Reason string `gorm:"type:text"` // 评语 + Sources int // 检索来源数 + Corrected bool // 是否经低分自动纠偏重生成后采纳（恒温器闭环） } func (Eval) TableName() string { return "sundynix_eval" } diff --git a/sundynix-gateway/internal/store/pgsql.go b/sundynix-gateway/internal/store/pgsql.go index 6c915cf..0d15c59 100644 --- a/sundynix-gateway/internal/store/pgsql.go +++ b/sundynix-gateway/internal/store/pgsql.go @@ -153,7 +153,7 @@ func (p *Postgres) SaveEval(ctx context.Context, e *Eval) error { return p.db.WithContext(ctx).Clauses(clause.OnConflict{ Columns: []clause.Column{{Name: "task_id"}}, DoUpdates: clause.AssignmentColumns([]string{ - "overall", "rule", "llm", "faithful", "level", "flags", "reason", "sources", "updated_at", + "overall", "rule", "llm", "faithful", "level", "flags", "reason", "sources", "corrected", "updated_at", }), }).Create(e).Error } diff --git a/sundynix-shared/contract/task.go b/sundynix-shared/contract/task.go index 5e22889..2cd48f3 100644 --- a/sundynix-shared/contract/task.go +++ b/sundynix-shared/contract/task.go @@ -51,16 +51,17 @@ const ( // EvalEvent 是一次自动化评测的结果（经 SubjectEval 回流给网关落库）。 type EvalEvent struct { - TaskID string `json:"task_id"` - Overall float64 `json:"overall"` // 综合分 [0,1] - Rule float64 `json:"rule"` // 规则分 - LLM float64 `json:"llm"` // LLM 质量分 - Faithful float64 `json:"faithful"` // RAG 忠实度分（0=无来源未评） - Level string `json:"level"` // ok / warn / poor - Flags []string `json:"flags,omitempty"` // 命中问题（规则 + 未被来源支持） - Reason string `json:"reason,omitempty"` // 评语 - Sources int `json:"sources,omitempty"` // 检索来源数 - TS int64 `json:"ts"` // unix 毫秒 + TaskID string `json:"task_id"` + Overall float64 `json:"overall"` // 综合分 [0,1] + Rule float64 `json:"rule"` // 规则分 + LLM float64 `json:"llm"` // LLM 质量分 + Faithful float64 `json:"faithful"` // RAG 忠实度分（0=无来源未评） + Level string `json:"level"` // ok / warn / poor（纠偏后的终值） + Flags []string `json:"flags,omitempty"` // 命中问题（规则 + 未被来源支持） + Reason string `json:"reason,omitempty"` // 评语 + Sources int `json:"sources,omitempty"` // 检索来源数 + Corrected bool `json:"corrected,omitempty"` // 是否经低分自动纠偏重生成后采纳（恒温器闭环） + TS int64 `json:"ts"` // unix 毫秒 } // 任务生命周期状态机：submitted（网关建任务）→ running（dispatcher 开跑）