From 5736ad145e9821c9b131eaef6127ed92e4270b6a Mon Sep 17 00:00:00 2001 From: Blizzard Date: Tue, 30 Jun 2026 14:59:52 +0800 Subject: [PATCH] =?UTF-8?q?feat(prompts):=20v2=20DB=20=E6=8E=A7=E5=88=B6?= =?UTF-8?q?=E9=9D=A2=E7=83=AD=E5=88=87=E6=8D=A2=20=E2=80=94=E2=80=94=20?= =?UTF-8?q?=E7=89=88=E6=9C=AC=E7=95=99=E5=AD=98=20+=20=E4=B8=8D=E9=87=8D?= =?UTF-8?q?=E5=90=AF=E5=8D=B3=E7=94=9F=E6=95=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 在 v1(注册表+文件覆盖)上加 DB 管理层与热切换，镜像 model-config 控制面： - store: sundynix_prompt 表(key/version/content/active) + ActivePrompts/ListPrompts/ CreateVersion/Activate/Deactivate - 控制面: ServePrompts/RequestActivePrompts(+Retry)/PublishPromptsUpdated/SubscribePromptsUpdated； prompts.ApplyOverrides 整体替换覆盖集(DB 激活集为权威) - gateway API: GET/POST /api/v1/prompts、version/activate/deactivate；激活/撤销即广播 - dispatcher/mcp-go: 启动拉激活集 + 订阅热更新(不重启) - live: 建版本→激活→mcp-go 图谱抽取 2→0→回滚 2(全程不重启)；deactivate 回退代码默认；版本可回溯 Co-Authored-By: Claude Opus 4.8 --- sundynix-dispatcher/cmd/dispatcher/main.go | 6 ++ .../internal/nats/subscriber.go | 10 +++ sundynix-gateway/cmd/server/main.go | 7 ++ sundynix-gateway/internal/handler/prompt.go | 81 ++++++++++++++++++ sundynix-gateway/internal/nats/publisher.go | 12 +++ sundynix-gateway/internal/router/router.go | 6 ++ sundynix-gateway/internal/store/pgsql.go | 2 +- sundynix-gateway/internal/store/prompt.go | 82 +++++++++++++++++++ sundynix-mcp-go/cmd/server/main.go | 6 ++ sundynix-shared/bus/bus.go | 69 ++++++++++++++++ sundynix-shared/contract/prompt.go | 8 ++ sundynix-shared/prompts/prompts.go | 18 ++++ 12 files changed, 306 insertions(+), 1 deletion(-) create mode 100644 sundynix-gateway/internal/handler/prompt.go create mode 100644 sundynix-gateway/internal/store/prompt.go create mode 100644 sundynix-shared/contract/prompt.go diff --git a/sundynix-dispatcher/cmd/dispatcher/main.go b/sundynix-dispatcher/cmd/dispatcher/main.go index 1a8db5b..e9d4e83 100644 --- a/sundynix-dispatcher/cmd/dispatcher/main.go +++ b/sundynix-dispatcher/cmd/dispatcher/main.go @@ -53,6 +53,12 @@ func main() { } go sub.FetchModelConfigWithRetry(context.Background(), pool.SetConfig) + // Prompt 控制面：拉激活集覆盖内置默认 + 订阅热更新（管理端激活某版即生效，不重启）。 + go sub.FetchPromptsWithRetry(context.Background(), prompts.ApplyOverrides) + if _, err := sub.SubscribePromptsUpdated(prompts.ApplyOverrides); err != nil { + log.Printf("[dispatcher] subscribe prompts: %v", err) + } + // sub 同时作为 Token 回流(TokenSink)、MCP 工具调用(ToolCaller)、执行事件(ExecSink)、 // 任务状态回写(StatusSink)、HITL 审批等待(ApprovalWaiter)与评测落库(EvalSink)出口。 orch, err := eino.NewOrchestrator(pool, breaker, eval, sub, sub, sub, sub, sub, sub) diff --git a/sundynix-dispatcher/internal/nats/subscriber.go b/sundynix-dispatcher/internal/nats/subscriber.go index 551c6cf..e35e2d3 100644 --- a/sundynix-dispatcher/internal/nats/subscriber.go +++ b/sundynix-dispatcher/internal/nats/subscriber.go @@ -133,4 +133,14 @@ func (s *Subscriber) FetchModelConfigWithRetry(ctx context.Context, apply func(* s.inner.RequestConfigWithRetry(ctx, contract.ConfigKindChat, apply) } +// FetchPromptsWithRetry 后台重试拉取初始激活 prompt 集（覆盖内置默认）。 +func (s *Subscriber) FetchPromptsWithRetry(ctx context.Context, apply func(map[string]string)) { + s.inner.RequestPromptsWithRetry(ctx, apply) +} + +// SubscribePromptsUpdated 订阅 prompt 激活集热更新。 +func (s *Subscriber) SubscribePromptsUpdated(onUpdate func(map[string]string)) (func() error, error) { + return s.inner.SubscribePromptsUpdated(onUpdate) +} + func (s *Subscriber) Close() { s.inner.Close() } diff --git a/sundynix-gateway/cmd/server/main.go b/sundynix-gateway/cmd/server/main.go index 7dc0b5b..e2f501e 100644 --- a/sundynix-gateway/cmd/server/main.go +++ b/sundynix-gateway/cmd/server/main.go @@ -60,6 +60,13 @@ func main() { } } + // Prompt 控制面：响应各服务「取激活 prompt」请求（DB 激活集；空则服务用内置默认）。 + if _, err := bus.ServePrompts(func() map[string]string { + return db.ActivePrompts(context.Background()) + }); err != nil { + log.Printf("[gateway] serve prompts: %v", 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 { diff --git a/sundynix-gateway/internal/handler/prompt.go b/sundynix-gateway/internal/handler/prompt.go new file mode 100644 index 0000000..9a2e9e2 --- /dev/null +++ b/sundynix-gateway/internal/handler/prompt.go @@ -0,0 +1,81 @@ +package handler + +import ( + "net/http" + + "github.com/gin-gonic/gin" + + "github.com/sundynix/sundynix-shared/prompts" +) + +// PromptList: GET /api/v1/prompts —— 列出受管 prompt 的全部版本 + 可配键（管理端浏览/对比/回滚）。 +func (h *Handler) PromptList(c *gin.Context) { + rows, err := h.db.ListPrompts(c.Request.Context()) + if err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) + return + } + out := make([]gin.H, 0, len(rows)) + for _, r := range rows { + out = append(out, gin.H{"key": r.Key, "version": r.Version, "active": r.Active, "note": r.Note, "content": r.Content}) + } + c.JSON(http.StatusOK, gin.H{"keys": prompts.Known, "versions": out}) +} + +// PromptCreateVersion: POST /api/v1/prompts/version {key, content, note} —— 为某 key 建新版本（不自动激活）。 +func (h *Handler) PromptCreateVersion(c *gin.Context) { + var body struct { + Key string `json:"key"` + Content string `json:"content"` + Note string `json:"note"` + } + if err := c.ShouldBindJSON(&body); err != nil || body.Key == "" || body.Content == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "key/content required"}) + return + } + v, err := h.db.CreatePromptVersion(c.Request.Context(), body.Key, body.Content, body.Note) + if err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"key": body.Key, "version": v}) +} + +// PromptActivate: POST /api/v1/prompts/activate {key, version} —— 激活某版本 → 经控制面热下发各服务（不重启即生效）。 +func (h *Handler) PromptActivate(c *gin.Context) { + var body struct { + Key string `json:"key"` + Version int `json:"version"` + } + if err := c.ShouldBindJSON(&body); err != nil || body.Key == "" || body.Version <= 0 { + c.JSON(http.StatusBadRequest, gin.H{"error": "key/version required"}) + return + } + if err := h.db.ActivatePrompt(c.Request.Context(), body.Key, body.Version); err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) + return + } + // 广播最新激活集 → dispatcher/mcp-go 热更新覆盖。 + if err := h.bus.PublishPromptsUpdated(h.db.ActivePrompts(c.Request.Context())); err != nil { + c.JSON(http.StatusOK, gin.H{"key": body.Key, "version": body.Version, "warn": "已激活但广播失败: " + err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"key": body.Key, "version": body.Version, "activated": true}) +} + +// PromptDeactivate: POST /api/v1/prompts/deactivate {key} —— 撤销某 key 激活，回退代码内置默认（热下发）。 +func (h *Handler) PromptDeactivate(c *gin.Context) { + var body struct { + Key string `json:"key"` + } + if err := c.ShouldBindJSON(&body); err != nil || body.Key == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "key required"}) + return + } + if err := h.db.DeactivatePrompt(c.Request.Context(), body.Key); err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) + return + } + _ = h.bus.PublishPromptsUpdated(h.db.ActivePrompts(c.Request.Context())) + c.JSON(http.StatusOK, gin.H{"key": body.Key, "deactivated": true}) +} diff --git a/sundynix-gateway/internal/nats/publisher.go b/sundynix-gateway/internal/nats/publisher.go index eaf0c9a..300a4d5 100644 --- a/sundynix-gateway/internal/nats/publisher.go +++ b/sundynix-gateway/internal/nats/publisher.go @@ -119,4 +119,16 @@ func (b *Bus) ConsumeIngestJobs(ctx context.Context, h sharedbus.IngestHandler) return b.inner.ConsumeIngestJobs(ctx, h) } +// ---- Prompt 控制面 ---- + +// ServePrompts 让网关响应「取全部激活 prompt」请求。 +func (b *Bus) ServePrompts(provide func() map[string]string) (func() error, error) { + return b.inner.ServePrompts(provide) +} + +// PublishPromptsUpdated 广播最新激活集 → 各服务热更新。 +func (b *Bus) PublishPromptsUpdated(m map[string]string) error { + return b.inner.PublishPromptsUpdated(m) +} + func (b *Bus) Close() { b.inner.Close() } diff --git a/sundynix-gateway/internal/router/router.go b/sundynix-gateway/internal/router/router.go index fdd4e45..dfc2e3d 100644 --- a/sundynix-gateway/internal/router/router.go +++ b/sundynix-gateway/internal/router/router.go @@ -65,6 +65,12 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob. p.GET("/kb/vault", h.KbVault) // 文库列表 p.GET("/kb/doc", h.KbDoc) // 取单篇文档 p.DELETE("/kb/doc", h.KbDeleteDoc) // 级联删文档（三库+MinIO+PG） + + // Prompt 控制面（平台级配置：建版本 → 激活 → 控制面热下发各服务） + p.GET("/prompts", h.PromptList) // 列出全部版本 + 可配键 + p.POST("/prompts/version", h.PromptCreateVersion) // 建新版本（不自动激活） + p.POST("/prompts/activate", h.PromptActivate) // 激活某版本 → 广播热更新 + p.POST("/prompts/deactivate", h.PromptDeactivate) // 撤销激活 → 回退代码默认（热） p.GET("/kb/links", h.KbLinks) // 某库双链 p.POST("/kb/note", h.KbSaveNote) // 新建/编辑笔记 p.GET("/kb/graph", h.KbGraph) // 知识图谱三元组 diff --git a/sundynix-gateway/internal/store/pgsql.go b/sundynix-gateway/internal/store/pgsql.go index 6f538cc..7ae94f0 100644 --- a/sundynix-gateway/internal/store/pgsql.go +++ b/sundynix-gateway/internal/store/pgsql.go @@ -66,7 +66,7 @@ func OpenPostgres(dsn string) *Postgres { migrateLegacyIntIDs(db) migrateDocLinkToID(db) - if err := db.AutoMigrate(&User{}, &Task{}, &Eval{}, &LLMModel{}, &KB{}, &Doc{}, &Agent{}, &DocLink{}, &Pricing{}); err != nil { + if err := db.AutoMigrate(&User{}, &Task{}, &Eval{}, &LLMModel{}, &KB{}, &Doc{}, &Agent{}, &DocLink{}, &Pricing{}, &Prompt{}); err != nil { log.Printf("[store] postgres AutoMigrate 失败，降级运行: %v", err) return &Postgres{} } diff --git a/sundynix-gateway/internal/store/prompt.go b/sundynix-gateway/internal/store/prompt.go new file mode 100644 index 0000000..994d5ba --- /dev/null +++ b/sundynix-gateway/internal/store/prompt.go @@ -0,0 +1,82 @@ +package store + +import ( + "context" + + "gorm.io/gorm" +) + +// Prompt 是一条受管系统提示词的某个版本。表名 sundynix_prompt，(key,version) 唯一。 +// 平台级配置（非按 owner），由管理端维护：建版本 → 激活某版 → 控制面热下发各服务。 +type Prompt struct { + BaseModel + Key string `gorm:"size:64;uniqueIndex:idx_prompt_kv;index"` // 受管 prompt 键，如 graph.extract + Version int `gorm:"uniqueIndex:idx_prompt_kv"` // 该 key 下递增版本号 + Content string `gorm:"type:text"` // 提示词正文 + Active bool `gorm:"index"` // 是否为该 key 当前激活版（每 key 至多一条） + Note string `gorm:"size:200"` // 版本说明 +} + +func (Prompt) TableName() string { return "sundynix_prompt" } + +// ActivePrompts 返回全部激活版的 key→content（控制面下发用）。 +func (p *Postgres) ActivePrompts(ctx context.Context) map[string]string { + out := map[string]string{} + if p.db == nil { + return out + } + var rows []Prompt + if err := p.db.WithContext(ctx).Where("active = ?", true).Find(&rows).Error; err != nil { + return out + } + for _, r := range rows { + out[r.Key] = r.Content + } + return out +} + +// ListPrompts 返回全部 prompt 版本（按 key、version 排序，供管理端浏览/对比/回滚）。 +func (p *Postgres) ListPrompts(ctx context.Context) ([]Prompt, error) { + if p.db == nil { + return nil, nil + } + var rows []Prompt + err := p.db.WithContext(ctx).Order("key asc, version asc").Find(&rows).Error + return rows, err +} + +// CreatePromptVersion 为某 key 新建一个版本（version=该 key 现有最大+1，默认不激活），返回新版本号。 +func (p *Postgres) CreatePromptVersion(ctx context.Context, key, content, note string) (int, error) { + if p.db == nil { + return 0, nil + } + var maxV struct{ V int } + p.db.WithContext(ctx).Model(&Prompt{}).Select("coalesce(max(version),0) as v").Where("key = ?", key).Scan(&maxV) + v := maxV.V + 1 + row := Prompt{Key: key, Version: v, Content: content, Note: note, Active: false} + if err := p.db.WithContext(ctx).Create(&row).Error; err != nil { + return 0, err + } + return v, nil +} + +// DeactivatePrompt 撤销某 key 的全部激活（控制面下发后该 key 回退到代码内置默认）。 +func (p *Postgres) DeactivatePrompt(ctx context.Context, key string) error { + if p.db == nil { + return nil + } + return p.db.WithContext(ctx).Model(&Prompt{}).Where("key = ?", key).Update("active", false).Error +} + +// ActivatePrompt 激活某 key 的指定版本（同 key 其它版本置非激活）。事务保证每 key 至多一个激活版。 +func (p *Postgres) ActivatePrompt(ctx context.Context, key string, version int) error { + if p.db == nil { + return nil + } + return p.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Model(&Prompt{}).Where("key = ?", key).Update("active", false).Error; err != nil { + return err + } + return tx.Model(&Prompt{}).Where("key = ? AND version = ?", key, version).Update("active", true).Error + }) +} diff --git a/sundynix-mcp-go/cmd/server/main.go b/sundynix-mcp-go/cmd/server/main.go index 9c2c291..aa3e0c3 100644 --- a/sundynix-mcp-go/cmd/server/main.go +++ b/sundynix-mcp-go/cmd/server/main.go @@ -98,6 +98,12 @@ func main() { go b.RequestConfigWithRetry(ctx, contract.ConfigKindEmbedding, applyEmbed) go b.RequestConfigWithRetry(ctx, contract.ConfigKindChat, applyChat) + // Prompt 控制面：拉取激活集覆盖内置默认 + 订阅热更新（管理端激活某版即生效，不重启）。 + go b.RequestPromptsWithRetry(ctx, prompts.ApplyOverrides) + if _, err := b.SubscribePromptsUpdated(prompts.ApplyOverrides); err != nil { + log.Printf("[mcp_go] subscribe prompts: %v", err) + } + gw := mcp.NewGateway(b, engine, mem, hist, ragEngine, pgDSN) log.Println("[mcp_go] serving MCP over sundynix.tools.go.* (Ctrl-C to quit)") diff --git a/sundynix-shared/bus/bus.go b/sundynix-shared/bus/bus.go index 732107b..0bc412d 100644 --- a/sundynix-shared/bus/bus.go +++ b/sundynix-shared/bus/bus.go @@ -588,6 +588,75 @@ func (b *Bus) SubscribeConfigUpdated(kind string, onUpdate func(*contract.ModelC return sub.Unsubscribe, nil } +// ---- Prompt 控制面（core NATS request-reply + broadcast，镜像 config）---- + +// ServePrompts 让控制面响应「取全部激活 prompt」请求；provide 返回 key→content 映射（可空）。 +// 队列组：多网关副本下每个请求只由一个副本应答。 +func (b *Bus) ServePrompts(provide func() map[string]string) (unsub func() error, err error) { + sub, err := b.nc.QueueSubscribe(contract.SubjectPromptsGet, contract.QueueGateway, func(m *nats.Msg) { + data, _ := json.Marshal(provide()) + _ = m.Respond(data) + }) + if err != nil { + return nil, fmt.Errorf("serve prompts: %w", err) + } + return sub.Unsubscribe, nil +} + +// RequestActivePrompts 向控制面请求当前全部激活 prompt（key→content）。无人应答/空集返回 (nil,nil)，调用方用内置默认。 +func (b *Bus) RequestActivePrompts(ctx context.Context) (map[string]string, error) { + msg, err := b.nc.RequestWithContext(ctx, contract.SubjectPromptsGet, nil) + if err != nil || len(msg.Data) == 0 { + return nil, nil + } + var m map[string]string + if err := json.Unmarshal(msg.Data, &m); err != nil { + return nil, fmt.Errorf("unmarshal prompts: %w", err) + } + return m, nil +} + +// RequestPromptsWithRetry 后台重试拉取初始激活 prompt 集（容忍消费方早于网关启动），拿到非空即 apply。 +func (b *Bus) RequestPromptsWithRetry(ctx context.Context, apply func(map[string]string)) { + for i := 0; i < 60; i++ { + cctx, cancel := context.WithTimeout(ctx, 3*time.Second) + m, _ := b.RequestActivePrompts(cctx) + cancel() + if len(m) > 0 { + apply(m) + return + } + select { + case <-ctx.Done(): + return + case <-time.After(2 * time.Second): + } + } +} + +// PublishPromptsUpdated 广播激活集变更（携带全量 key→content，消费方据此整体热更新）。 +func (b *Bus) PublishPromptsUpdated(m map[string]string) error { + data, err := json.Marshal(m) + if err != nil { + return err + } + return b.nc.Publish(contract.SubjectPromptsUpdated, data) +} + +// SubscribePromptsUpdated 订阅激活集变更，回调拿到全量 key→content。 +func (b *Bus) SubscribePromptsUpdated(onUpdate func(map[string]string)) (unsub func() error, err error) { + sub, err := b.nc.Subscribe(contract.SubjectPromptsUpdated, func(m *nats.Msg) { + var mp map[string]string + if json.Unmarshal(m.Data, &mp) == nil { + onUpdate(mp) + } + }) + if err != nil { + return nil, fmt.Errorf("subscribe prompts updated: %w", err) + } + return sub.Unsubscribe, nil +} + // TaskHandler 处理一个消费到的任务。 type TaskHandler func(ctx context.Context, t *contract.Task) error diff --git a/sundynix-shared/contract/prompt.go b/sundynix-shared/contract/prompt.go new file mode 100644 index 0000000..3ab5685 --- /dev/null +++ b/sundynix-shared/contract/prompt.go @@ -0,0 +1,8 @@ +package contract + +// Prompt 控制面：DB 存各 prompt 的多版本，激活某版 → 经控制面热下发给各服务， +// 覆盖内置默认（不重启即生效）。镜像 model-config 控制面(ConfigGetSubject/ConfigUpdatedSubject)。 +const ( + SubjectPromptsGet = "sundynix.prompts.get" // request-reply：取全部激活 prompt（map key→content） + SubjectPromptsUpdated = "sundynix.prompts.updated" // 广播：激活集变更，消费方据此热更新 +) diff --git a/sundynix-shared/prompts/prompts.go b/sundynix-shared/prompts/prompts.go index 9d739cc..a2652b5 100644 --- a/sundynix-shared/prompts/prompts.go +++ b/sundynix-shared/prompts/prompts.go @@ -20,6 +20,9 @@ const ( MemoryExtract = "memory.extract" // dispatcher：长期记忆对账 ) +// Known 是平台受管的全部 prompt key（供管理端列出可配项；网关进程不登记默认，靠它枚举）。 +var Known = []string{GraphExtract, EvalQuality, EvalRefine, GuardJailbreak, CoordinatorLead, MemoryExtract} + // Registry 持有默认与覆盖；Get 取覆盖优先、否则默认。并发安全。 type Registry struct { mu sync.RWMutex @@ -53,6 +56,21 @@ func (r *Registry) SetOverride(key, text string) { r.overrides[key] = text } +// ApplyOverrides 用给定集合整体替换当前覆盖集（控制面热下发：DB 激活集为权威）。 +// 空值条目忽略；传空 map 即清空所有覆盖、全回退内置默认。 +func ApplyOverrides(m map[string]string) { global.ApplyOverrides(m) } +func (r *Registry) ApplyOverrides(m map[string]string) { + r.mu.Lock() + defer r.mu.Unlock() + next := make(map[string]string, len(m)) + for k, v := range m { + if v != "" { + next[k] = v + } + } + r.overrides = next +} + // Get 取生效内容：有覆盖用覆盖，否则用默认（都没有则空串）。 func Get(key string) string { return global.Get(key) } func (r *Registry) Get(key string) string {