From 10f08ffb1465ad76f5033ee8bbc513b098f39cac Mon Sep 17 00:00:00 2001 From: Blizzard Date: Thu, 25 Jun 2026 15:26:19 +0800 Subject: [PATCH] =?UTF-8?q?feat(harness):=20=E8=AF=84=E6=B5=8B=E9=97=AD?= =?UTF-8?q?=E7=8E=AF=20=E2=80=94=E2=80=94=20=E8=AF=84=E6=B5=8B=E7=BB=93?= =?UTF-8?q?=E6=9E=9C=E8=90=BD=E5=BA=93=20+=20=E5=88=86=E7=BA=A7=20+=20?= =?UTF-8?q?=E5=91=8A=E8=AD=A6=20+=20=E5=8F=AF=E6=9F=A5=EF=BC=88=E6=B5=8B?= =?UTF-8?q?=E6=B8=A9=E8=AE=A1=E2=86=92=E6=81=92=E6=B8=A9=E5=99=A8=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 此前 eval 只打日志、不闭环。现在： - 分级：evalLevel 据综合分+忠实度 → ok(≥0.75) / warn(≥0.5 或忠实<0.6) / poor(<0.5)；poor 出 slog.Warn 告警。 - 落库：dispatcher 评完经 NATS(SubjectEval) 广播 EvalEvent → 网关订阅写 PG（新表 sundynix_eval， 按 task_id upsert）。沿用任务状态回写那套（dispatcher 无 DB，经 bus→gateway 落库）。 - 可查：GET /api/v1/tasks/:id/eval 返回 overall/rule/llm/faithful/level/flags/reason/sources。 - 契约 EvalEvent + EvalOK/Warn/Poor；bus PublishEval/SubscribeEval；dispatcher EvalSink（NewOrchestrator 第9参）。 验证：三模块 build+vet+test 全绿；live RAG 任务评测落库，端点返回 overall~1.0 / level=ok / faithful=1 / sources=1。 剩：桌面端质量面板、低分自动重试(P3)。project_analysis 勾掉该项。 Co-Authored-By: Claude Opus 4.8 (1M context) --- project_analysis.md | 3 +- sundynix-dispatcher/cmd/dispatcher/main.go | 4 +- .../internal/eino/orchestrator.go | 43 ++++++++++++++++--- .../internal/nats/subscriber.go | 5 +++ sundynix-gateway/cmd/server/main.go | 14 ++++++ .../internal/handler/task_handler.go | 15 +++++++ sundynix-gateway/internal/nats/publisher.go | 5 +++ sundynix-gateway/internal/router/router.go | 1 + sundynix-gateway/internal/store/models.go | 16 +++++++ sundynix-gateway/internal/store/pgsql.go | 28 +++++++++++- sundynix-shared/bus/bus.go | 25 +++++++++++ sundynix-shared/contract/task.go | 24 +++++++++++ 12 files changed, 173 insertions(+), 10 deletions(-) diff --git a/project_analysis.md b/project_analysis.md index 3f0e5aa..c91fe67 100644 --- a/project_analysis.md +++ b/project_analysis.md @@ -112,7 +112,8 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为 **已知短板 / 优化清单**（按性价比，逐项推进）： -- [ ] **P1 评测闭环**：eval 结果落库 + 阈值门控（低分标记/触发重试）+ 桌面端质量面板。现状只打日志、不闭环。 +- [x] **P1 评测闭环** ✅：eval 结果经 NATS 落 PG（sundynix_eval）+ 分级(ok/warn/poor) + poor 告警 + + `GET /tasks/:id/eval` 可查。dispatcher 评→广播→网关落库。剩：桌面端质量面板、低分自动重试(P3)。 - [x] **P1 RAG 忠实度评测** ✅：检索原文喂给 judge，一次评质量+忠实度，未被来源支持的说法进 Flags。 综合分(有来源)=0.3规则+0.35质量+0.35忠实。runGraph 透传 refs → evaluate。live 实测忠实 1.00/来源 1，单测覆盖。 - [ ] **P2 输出脱敏增强**：滑动窗口跨片检测（现流式逐片会漏跨 token 的密钥）+ PII 模式（手机号/邮箱/身份证）。 diff --git a/sundynix-dispatcher/cmd/dispatcher/main.go b/sundynix-dispatcher/cmd/dispatcher/main.go index fd1655a..ae3f014 100644 --- a/sundynix-dispatcher/cmd/dispatcher/main.go +++ b/sundynix-dispatcher/cmd/dispatcher/main.go @@ -46,8 +46,8 @@ func main() { go sub.FetchModelConfigWithRetry(context.Background(), pool.SetConfig) // sub 同时作为 Token 回流(TokenSink)、MCP 工具调用(ToolCaller)、执行事件(ExecSink)、 - // 任务状态回写(StatusSink)与 HITL 审批等待(ApprovalWaiter)出口。 - orch, err := eino.NewOrchestrator(pool, breaker, eval, sub, sub, sub, sub, sub) + // 任务状态回写(StatusSink)、HITL 审批等待(ApprovalWaiter)与评测落库(EvalSink)出口。 + orch, err := eino.NewOrchestrator(pool, breaker, eval, sub, sub, sub, sub, sub, sub) if err != nil { log.Fatalf("[dispatcher] build eino graph: %v", err) } diff --git a/sundynix-dispatcher/internal/eino/orchestrator.go b/sundynix-dispatcher/internal/eino/orchestrator.go index 6be6586..f8e75bd 100644 --- a/sundynix-dispatcher/internal/eino/orchestrator.go +++ b/sundynix-dispatcher/internal/eino/orchestrator.go @@ -46,6 +46,11 @@ type ApprovalWaiter interface { WaitApproval(ctx context.Context, taskID string, timeout time.Duration) (*contract.ApprovalDecision, error) } +// EvalSink 回写自动化评测结果供网关落库（由 NATS bus 实现；可为 nil → 仅打日志不落库）。 +type EvalSink interface { + PublishEval(ev *contract.EvalEvent) error +} + // errRejected 是审批节点拒绝（或超时）时图执行返回的哨兵错误：它是合法终态而非故障， // Handle 据此判 rejected 并优雅收尾（不计熔断失败）。 var errRejected = errors.New("approval rejected") @@ -82,16 +87,17 @@ type Orchestrator struct { exec ExecSink status StatusSink // 任务生命周期状态回写（可为 nil） approval ApprovalWaiter // HITL 审批等待（可为 nil → 审批节点自动放行） + evalSink EvalSink // 评测结果回写落库（可为 nil → 仅打日志） turnMu sync.Mutex // 保护 turns（攒批计数，多任务 goroutine 共享） turns map[string]int // sessionID → 累计轮次，用于每 N 轮触发 consolidate } // NewOrchestrator 持有依赖；图按任务的 DSL 在 Handle 内动态编译。 -// exec 为执行可视化事件出口（可为 nil，则不发轨迹事件）；eval 为自动化评测（可为 nil）； -// status 为任务生命周期状态回写出口（可为 nil）；approval 为 HITL 审批等待（可为 nil）。 -func NewOrchestrator(pool LLM, breaker *harness.CircuitBreaker, eval *harness.Evaluator, sink TokenSink, tools ToolCaller, exec ExecSink, status StatusSink, approval ApprovalWaiter) (*Orchestrator, error) { - return &Orchestrator{pool: pool, breaker: breaker, eval: eval, sink: sink, tools: tools, exec: exec, status: status, approval: approval}, nil +// exec 为执行可视化事件出口（可为 nil）；eval 为自动化评测（可为 nil）；status 为状态回写（可为 nil）； +// approval 为 HITL 审批等待（可为 nil）；evalSink 为评测结果落库出口（可为 nil）。 +func NewOrchestrator(pool LLM, breaker *harness.CircuitBreaker, eval *harness.Evaluator, sink TokenSink, tools ToolCaller, exec ExecSink, status StatusSink, approval ApprovalWaiter, evalSink EvalSink) (*Orchestrator, error) { + return &Orchestrator{pool: pool, breaker: breaker, eval: eval, sink: sink, tools: tools, exec: exec, status: status, approval: approval, evalSink: evalSink}, nil } // setStatus 回写一次任务状态流转（status 为 nil 时静默跳过）。 @@ -203,8 +209,33 @@ func (o *Orchestrator) evaluate(t *contract.Task, input, output string, sources ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() r := o.eval.Score(ctx, input, output, sources) - log.Printf("[eval] task %s 综合 %.2f（规则 %.2f / LLM %.2f / 忠实 %.2f，来源 %d）flags=%v %s", - t.ID, r.Overall, r.Rule, r.LLM, r.Faithful, len(sources), r.Flags, r.Reason) + 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) + } + // 闭环：评测结果落库（供 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(), + }); err != nil { + log.Printf("[eval] 回写评测结果失败 task=%s: %v", t.ID, err) + } + } +} + +// evalLevel 据综合分 + 忠实度把评测分级（闭环门控/告警用）。 +func evalLevel(r harness.Result) string { + switch { + case r.Overall < 0.5: + return contract.EvalPoor + case r.Overall < 0.75 || (r.Faithful > 0 && r.Faithful < 0.6): + return contract.EvalWarn + default: + return contract.EvalOK + } } // fetchMemory 经 MCP memory_get 工具召回用户常驻画像。 diff --git a/sundynix-dispatcher/internal/nats/subscriber.go b/sundynix-dispatcher/internal/nats/subscriber.go index c9b4f13..be5c40e 100644 --- a/sundynix-dispatcher/internal/nats/subscriber.go +++ b/sundynix-dispatcher/internal/nats/subscriber.go @@ -81,6 +81,11 @@ func (s *Subscriber) PublishTaskStatus(taskID, status, detail string) error { }) } +// PublishEval 让 Subscriber 满足 eino.EvalSink，把自动化评测结果回写给网关落库。 +func (s *Subscriber) PublishEval(ev *contract.EvalEvent) error { + return s.inner.PublishEval(ev) +} + // WaitApproval 让 Subscriber 满足 eino.ApprovalWaiter，阻塞等待审批节点的人工决定。 func (s *Subscriber) WaitApproval(ctx context.Context, taskID string, timeout time.Duration) (*contract.ApprovalDecision, error) { return s.inner.WaitApproval(ctx, taskID, timeout) diff --git a/sundynix-gateway/cmd/server/main.go b/sundynix-gateway/cmd/server/main.go index 0bebbb5..503ef48 100644 --- a/sundynix-gateway/cmd/server/main.go +++ b/sundynix-gateway/cmd/server/main.go @@ -3,6 +3,7 @@ package main import ( "context" + "encoding/json" "log" "os" @@ -64,6 +65,19 @@ func main() { log.Printf("[gateway] subscribe task status: %v", err) } + // 评测闭环：订阅 dispatcher 回写的自动化评测结果，落 PG 供 UI 查询 / 质量趋势 / 门控。 + if _, err := bus.SubscribeEval(func(ev *contract.EvalEvent) { + 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, + }); err != nil { + log.Printf("[gateway] 落库评测 %s 失败: %v", ev.TaskID, err) + } + }); err != nil { + log.Printf("[gateway] subscribe eval: %v", err) + } + r := router.New(db, cache, bus, blobStore) addr := envOr("GATEWAY_ADDR", ":8080") log.Printf("[gateway] listening on %s", addr) diff --git a/sundynix-gateway/internal/handler/task_handler.go b/sundynix-gateway/internal/handler/task_handler.go index dabdc2c..7132b7c 100644 --- a/sundynix-gateway/internal/handler/task_handler.go +++ b/sundynix-gateway/internal/handler/task_handler.go @@ -89,6 +89,21 @@ func (h *Handler) TaskStatus(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"task_id": id, "status": status, "detail": detail}) } +// TaskEval: GET /api/v1/tasks/:id/eval —— 取一次任务的自动化评测结果（综合/质量/忠实度/分级/flags）。 +func (h *Handler) TaskEval(c *gin.Context) { + e := h.db.GetEval(c.Request.Context(), c.Param("id")) + if e == nil { + c.JSON(http.StatusNotFound, gin.H{"error": "尚无评测结果（任务未完成或评测进行中）"}) + return + } + var flags []string + _ = 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, + }) +} + // ApproveTask: POST /api/v1/tasks/:id/approve {approved, node?, note?} —— 人工审批决定（HITL）。 // 把决定经 NATS 发给 dispatcher，解除审批节点的阻塞（批准放行 / 拒绝中止）。 func (h *Handler) ApproveTask(c *gin.Context) { diff --git a/sundynix-gateway/internal/nats/publisher.go b/sundynix-gateway/internal/nats/publisher.go index 4905d07..74f1eb2 100644 --- a/sundynix-gateway/internal/nats/publisher.go +++ b/sundynix-gateway/internal/nats/publisher.go @@ -69,6 +69,11 @@ func (b *Bus) PublishApproval(dec *contract.ApprovalDecision) error { return b.inner.PublishApproval(dec) } +// SubscribeEval 订阅 dispatcher 回写的自动化评测结果（落 PG）。 +func (b *Bus) SubscribeEval(onEvent func(*contract.EvalEvent)) (func() error, error) { + return b.inner.SubscribeEval(onEvent) +} + // ServeConfig 让网关作为配置控制面，响应某 kind 的配置请求。 func (b *Bus) ServeConfig(kind string, provide func() *contract.ModelConfig) (func() error, error) { return b.inner.ServeConfig(kind, provide) diff --git a/sundynix-gateway/internal/router/router.go b/sundynix-gateway/internal/router/router.go index f455292..496d3e1 100644 --- a/sundynix-gateway/internal/router/router.go +++ b/sundynix-gateway/internal/router/router.go @@ -53,6 +53,7 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob. p.POST("/tasks", h.SubmitTask) // 解析 DSL 并 Publish 到 NATS（带已验证 uid） p.GET("/tasks/:id", h.TaskStatus) // 任务生命周期状态（UI 轮询 submitted/running/done/failed/timeout/waiting/rejected） p.POST("/tasks/:id/approve", h.ApproveTask) // HITL 人工审批决定（批准/拒绝） + p.GET("/tasks/:id/eval", h.TaskEval) // 自动化评测结果（综合/质量/忠实度/分级） p.PUT("/memory", h.SetMemory) // 偏好记忆登记（→ mcp-go memory_upsert） p.GET("/memory", h.ListMemory) // 列出当前用户偏好（记忆面板） p.DELETE("/memory", h.DeleteMemory) // 软删一条偏好（?key=） diff --git a/sundynix-gateway/internal/store/models.go b/sundynix-gateway/internal/store/models.go index 81092f3..eb2a989 100644 --- a/sundynix-gateway/internal/store/models.go +++ b/sundynix-gateway/internal/store/models.go @@ -22,3 +22,19 @@ type Task struct { Status string `gorm:"size:32"` // submitted / running / done / failed / timeout Detail string `gorm:"type:text"` // 失败/超时原因等（状态机回写） } + +// 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 // 检索来源数 +} + +func (Eval) TableName() string { return "sundynix_eval" } diff --git a/sundynix-gateway/internal/store/pgsql.go b/sundynix-gateway/internal/store/pgsql.go index c4344cc..6c915cf 100644 --- a/sundynix-gateway/internal/store/pgsql.go +++ b/sundynix-gateway/internal/store/pgsql.go @@ -11,6 +11,7 @@ import ( "gorm.io/driver/postgres" "gorm.io/gorm" + "gorm.io/gorm/clause" "gorm.io/gorm/schema" "github.com/sundynix/sundynix-shared/contract" @@ -65,7 +66,7 @@ func OpenPostgres(dsn string) *Postgres { migrateLegacyIntIDs(db) migrateDocLinkToID(db) - if err := db.AutoMigrate(&User{}, &Task{}, &LLMModel{}, &KB{}, &Doc{}, &Agent{}, &DocLink{}, &Pricing{}); err != nil { + if err := db.AutoMigrate(&User{}, &Task{}, &Eval{}, &LLMModel{}, &KB{}, &Doc{}, &Agent{}, &DocLink{}, &Pricing{}); err != nil { log.Printf("[store] postgres AutoMigrate 失败，降级运行: %v", err) return &Postgres{} } @@ -144,6 +145,31 @@ func (p *Postgres) GetTaskStatus(ctx context.Context, id string) (status, detail return t.Status, t.Detail } +// SaveEval 落库一条评测结果（按 task_id upsert：重评覆盖）。 +func (p *Postgres) SaveEval(ctx context.Context, e *Eval) error { + if p.db == nil { + return nil + } + 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", + }), + }).Create(e).Error +} + +// GetEval 取一条任务的评测结果（不存在返回 nil）。 +func (p *Postgres) GetEval(ctx context.Context, taskID string) *Eval { + if p.db == nil { + return nil + } + var e Eval + if err := p.db.WithContext(ctx).Where("task_id = ?", taskID).First(&e).Error; err != nil { + return nil + } + return &e +} + // CountTasks 返回已提交任务数（降级模式返回 0）。 func (p *Postgres) CountTasks(ctx context.Context) (int64, error) { if p.db == nil { diff --git a/sundynix-shared/bus/bus.go b/sundynix-shared/bus/bus.go index 5b0e1ad..3ab20df 100644 --- a/sundynix-shared/bus/bus.go +++ b/sundynix-shared/bus/bus.go @@ -338,6 +338,31 @@ func (b *Bus) SubscribeTaskStatus(onEvent func(*contract.TaskStatusEvent)) (unsu return sub.Unsubscribe, nil } +// ---- 自动化评测结果回写（core NATS pub-sub）---- + +// PublishEval 广播一次评测结果（dispatcher 调用）。 +func (b *Bus) PublishEval(ev *contract.EvalEvent) error { + data, err := json.Marshal(ev) + if err != nil { + return err + } + return b.nc.Publish(contract.SubjectEval, data) +} + +// SubscribeEval 订阅评测结果（网关调用，落 PG）。 +func (b *Bus) SubscribeEval(onEvent func(*contract.EvalEvent)) (unsub func() error, err error) { + sub, err := b.nc.Subscribe(contract.SubjectEval, func(m *nats.Msg) { + var ev contract.EvalEvent + if json.Unmarshal(m.Data, &ev) == nil { + onEvent(&ev) + } + }) + if err != nil { + return nil, fmt.Errorf("subscribe eval: %w", err) + } + return sub.Unsubscribe, nil +} + // ---- 人工审批（HITL，core NATS pub-sub）---- // PublishApproval 广播一次人工审批决定（网关在收到 UI 的批准/拒绝后调用）。 diff --git a/sundynix-shared/contract/task.go b/sundynix-shared/contract/task.go index 4e4178a..5e22889 100644 --- a/sundynix-shared/contract/task.go +++ b/sundynix-shared/contract/task.go @@ -37,8 +37,32 @@ const ( // 人工审批（HITL）决定回传前缀：实际 sundynix.approval.。 // core NATS pub-sub：UI 点批准/拒绝 → 网关发到此 → dispatcher 解除审批节点的阻塞。 SubjectApproval = "sundynix.approval" + + // 自动化评测结果回写：dispatcher 评完经此广播，网关订阅落 PG 并供 UI 查询。core NATS pub-sub。 + SubjectEval = "sundynix.eval.task" ) +// 评测质量分级（据综合分 + 忠实度阈值，闭环门控/告警用）。 +const ( + EvalOK = "ok" // 综合 ≥ 0.75 且无忠实度风险 + EvalWarn = "warn" // 0.5 ≤ 综合 < 0.75，或忠实度偏低 + EvalPoor = "poor" // 综合 < 0.5 +) + +// 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 毫秒 +} + // 任务生命周期状态机：submitted（网关建任务）→ running（dispatcher 开跑） // → done / failed / timeout（dispatcher 收尾）。 // HITL：执行到审批节点 → waiting（等人工决定）→ 批准回 running / 拒绝→rejected。