Files
sundynix-agentix/sundynix-gateway/cmd/server/main.go
T
Blizzard 93e9d3b195 feat(harness): 低分自动纠偏 —— poor 触发评语驱动重生成,取更优者(测温计→恒温器)
评测闭环延伸出自愈:当自动评测判定输出为 poor(综合<0.5),dispatcher 在热路径外
自动用「原问题+初版回答+评审短板(flags/评语)」(有来源则连来源一并喂回、要求严格基于来源)
让模型重写,重评后仅当新分严格更高才采纳(绝不退步);采纳的修订版落会话历史,
保证多轮上下文用的是好答案而非被判低分的初版。评测终值带 corrected 标记经 NATS→网关落库。

- maxRefineRounds=1:poor 稀少,1 轮重写+重评够用,防成本失控
- canRefine 门控:模型就绪且熔断未开才纠偏,避免后端抖时雪上加霜
- 单 goroutine 串 评测→纠偏→落历史,杜绝原两 goroutine 对答案版本的竞态
- 契约 EvalEvent / Eval 表 / upsert / GET /tasks/:id/eval 均加 corrected 字段
- refine_test.go:采纳更优 / 不退步 / 非低分不触发 三测;live 验证好答案不误触发

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-25 16:22:45 +08:00

95 lines
3.7 KiB
Go

// Command server 启动 sundynix-gateway —— 第 2 层业务网关 / 统一接入层。
package main
import (
"context"
"encoding/json"
"log"
"os"
"github.com/sundynix/sundynix-gateway/internal/blob"
"github.com/sundynix/sundynix-gateway/internal/nats"
"github.com/sundynix/sundynix-gateway/internal/router"
"github.com/sundynix/sundynix-gateway/internal/store"
"github.com/sundynix/sundynix-shared/contract"
"github.com/sundynix/sundynix-shared/otelx"
"github.com/sundynix/sundynix-shared/secrets"
)
func main() {
secrets.MustHaveKeyInProd() // 生产须配 SUNDYNIX_SECRET_KEY 以加密落库的 api_key
otelx.SetupSlog("sundynix-gateway") // 结构化日志 + 链路感知(日志带 trace_id)
// 链路追踪:HTTP 入口 + 跨 NATS 传播的根。退出前 flush 残留 span。
shutdownTrace, _ := otelx.Init(context.Background(), "sundynix-gateway")
defer func() { _ = shutdownTrace(context.Background()) }()
natsURL := envOr("NATS_URL", "nats://localhost:4222")
pgDSN := envOr("POSTGRES_DSN", "postgres://sundynix:sundynix@localhost:5432/sundynix?sslmode=disable")
redisAddr := envOr("REDIS_ADDR", "localhost:6379")
// 对象存储(大文档正文):默认连 docker-compose 暴露的 MinIO;连不上则降级内联存 PG。
blobStore := blob.Open(
envOr("MINIO_ENDPOINT", "localhost:9000"),
envOr("MINIO_ACCESS_KEY", "minioadmin"),
envOr("MINIO_SECRET_KEY", "minioadmin"),
envOr("MINIO_BUCKET", "sundynix-docs"),
)
db := store.OpenPostgres(pgDSN) // MainDB: Users / Billing / DSL(连不上则降级)
defer db.Close()
cache := store.OpenRedis(redisAddr) // CacheDB: Session / Rate Limit(连不上则降级)
defer cache.Close()
bus := nats.MustConnect(natsURL) // 接入 NATS 零拷贝骨干网 + 声明任务流
defer bus.Close()
// 配置控制面:按 kind 响应消费方(Dispatcher=chat / mcp-go=embedding)的配置请求。
for _, kind := range []string{contract.ConfigKindChat, contract.ConfigKindEmbedding} {
k := kind
if _, err := bus.ServeConfig(k, func() *contract.ModelConfig {
row, _ := db.GetActiveModel(context.Background(), k)
if row == nil {
return nil
}
return &contract.ModelConfig{Provider: row.Provider, BaseURL: row.BaseURL, APIKey: row.APIKey, Model: row.Model}
}); err != nil {
log.Printf("[gateway] serve %s config: %v", k, err)
}
}
// 任务生命周期:订阅 dispatcher 回写的状态流转(running/done/failed/timeout),落 PG 供 UI 查询。
if _, err := bus.SubscribeTaskStatus(func(ev *contract.TaskStatusEvent) {
if err := db.UpdateTaskStatus(context.Background(), ev.TaskID, ev.Status, ev.Detail); err != nil {
log.Printf("[gateway] 更新任务状态 %s=%s 失败: %v", ev.TaskID, ev.Status, err)
}
}); err != nil {
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, Corrected: ev.Corrected,
}); 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)
if err := r.Run(addr); err != nil {
log.Fatalf("[gateway] exit: %v", err)
}
}
func envOr(key, def string) string {
if v := os.Getenv(key); v != "" {
return v
}
return def
}