From 4928ffc0f73519bafea19fedfe9026c9ec231873 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Wed, 10 Jun 2026 14:18:45 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=9F=AD=E6=9C=9F=E5=A4=9A=E8=BD=AE?= =?UTF-8?q?=E5=8E=86=E5=8F=B2=E6=8E=A5=E5=85=A5=20Eino=20=E5=9B=BE=20Messa?= =?UTF-8?q?gesPlaceholder=20(=E2=91=A8)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 会话历史(Redis,易失,与长期画像分开)经 MCP 工具进出 Eino 图： recall 召回历史填 MessagesPlaceholder，写回把本轮 user/assistant 落历史。 - mcp-go: internal/history(go-redis, sundynix:history:, LPUSH+LTRIM 保留近20条, 24h TTL) + 工具 history_get(返回JSON turns)/history_append; main 开 Redis(降级) - dispatcher Eino: 模板加 MessagesPlaceholder('history'); recall 调 history_get→转 schema.Message; Handle 累积 answer; memorize 异步 history_append(user+assistant) - shared: contract.MetaSessionID; gateway: SubmitTask 注入 Meta[session_id](X-Session-ID 头,缺省 default) - demo.sh: 同会话两轮提交，验证第2轮召回第1轮历史 - 验证: 4 模块 build✓ + 3 e2e PASS; live 跑通——轮1=0轮历史→落库, 轮2 history_get 命中→注入 Co-Authored-By: Claude Opus 4.8 --- scripts/demo.sh | 32 ++++--- sundynix-dispatcher/internal/eino/graph.go | 18 ++-- sundynix-dispatcher/internal/eino/model.go | 22 ++++- .../internal/eino/orchestrator.go | 78 ++++++++++++++--- sundynix-dispatcher/internal/eino/state.go | 9 +- .../internal/handler/task_handler.go | 13 ++- sundynix-mcp-go/cmd/server/main.go | 8 +- sundynix-mcp-go/go.mod | 7 +- sundynix-mcp-go/go.sum | 18 +++- sundynix-mcp-go/internal/history/store.go | 84 +++++++++++++++++++ sundynix-mcp-go/internal/mcp/gateway.go | 44 ++++++++-- sundynix-shared/contract/task.go | 2 + 12 files changed, 288 insertions(+), 47 deletions(-) create mode 100644 sundynix-mcp-go/internal/history/store.go diff --git a/scripts/demo.sh b/scripts/demo.sh index 15c91aa..187e46e 100755 --- a/scripts/demo.sh +++ b/scripts/demo.sh @@ -46,17 +46,29 @@ curl -s -X PUT http://127.0.0.1:8080/api/v1/memory \ -H 'Content-Type: application/json' -H "X-User-ID: $USER" \ -d '{"key":"回答偏好","value":"简洁、中文、多给要点"}'; echo -echo "== 提交 DSL 任务 (带 X-User-ID，Dispatcher 将召回其画像) ==" -RESP=$(curl -s -X POST http://127.0.0.1:8080/api/v1/tasks \ - -H 'Content-Type: application/json' -H "X-User-ID: $USER" \ - -d '{"nodes":[{"id":"n1","type":"agent","data":{"prompt":"hello"}}],"edges":[]}') -echo "$RESP" -TASK_ID=$(echo "$RESP" | sed -n 's/.*"task_id":"\([^"]*\)".*/\1/p') +SESSION="sess-demo" +# 清掉上一次 demo 的会话历史，让本次轮次计数从 0 起（best-effort，无 Redis 容器则跳过）。 +docker exec sundynix_agentix-redis-1 redis-cli DEL "sundynix:history:$SESSION" >/dev/null 2>&1 || true -echo "== 订阅 SSE Token 流 (Gateway ← NATS ← Dispatcher) ==" -# 客户端在 TTFT(700ms) 内连上即可收全部 token；--max-time 超时(exit 28) 属正常，不让 set -e 中断 -curl -sN --max-time 10 "http://127.0.0.1:8080/api/v1/tasks/$TASK_ID/stream" || true -echo +submit_and_stream() { + local prompt="$1" + local resp task_id + resp=$(curl -s -X POST http://127.0.0.1:8080/api/v1/tasks \ + -H 'Content-Type: application/json' -H "X-User-ID: $USER" -H "X-Session-ID: $SESSION" \ + -d "{\"nodes\":[{\"id\":\"n1\",\"type\":\"agent\",\"data\":{\"prompt\":\"$prompt\"}}],\"edges\":[]}") + echo "$resp" + task_id=$(echo "$resp" | sed -n 's/.*"task_id":"\([^"]*\)".*/\1/p') + # 客户端在 TTFT(700ms) 内连上即可收全部 token；--max-time 超时(exit 28) 属正常 + curl -sN --max-time 12 "http://127.0.0.1:8080/api/v1/tasks/$task_id/stream" || true + echo +} + +echo "== 第 1 轮提交 (带 X-User-ID + X-Session-ID，召回画像) ==" +submit_and_stream "你好" +sleep 1 # 等写回把第 1 轮落进会话历史 + +echo "== 第 2 轮提交 (同会话，应召回到第 1 轮历史) ==" +submit_and_stream "继续" echo "== mcp-go 日志 (工具被调用) ==" cat .bin/mcp-go.log diff --git a/sundynix-dispatcher/internal/eino/graph.go b/sundynix-dispatcher/internal/eino/graph.go index d5b601f..5ac7102 100644 --- a/sundynix-dispatcher/internal/eino/graph.go +++ b/sundynix-dispatcher/internal/eino/graph.go @@ -14,36 +14,42 @@ import ( // memoryFetcher 召回某用户与本次输入相关的偏好记忆（经 MCP memory_get 工具）。 type memoryFetcher func(ctx context.Context, userID, query string) string +// historyFetcher 召回某会话的短期多轮历史（经 MCP history_get 工具）。 +type historyFetcher func(ctx context.Context, sessionID string) []*schema.Message + // buildGraph 编译这套"记忆增强"图： // -// START → recall(召回画像→写State) → prompt(注入system) → model(流式) → END +// START → recall(召回画像+历史→写State) → prompt(注入system+history) → model(流式) → END // // 返回可流式执行的 Runnable。 -func buildGraph(ctx context.Context, pool *llm.Pool, fetch memoryFetcher) (compose.Runnable[*contract.Task, *schema.Message], error) { +func buildGraph(ctx context.Context, pool *llm.Pool, fetch memoryFetcher, fetchHist historyFetcher) (compose.Runnable[*contract.Task, *schema.Message], error) { g := compose.NewGraph[*contract.Task, *schema.Message]( compose.WithGenLocalState(func(context.Context) *AgentState { return &AgentState{} }), ) - // 1) recall：取 user_id → memory_get 召回画像 → 写入 State，并输出模板变量。 + // 1) recall：取 user_id/session_id → 召回画像(memory_get)+历史(history_get) → 写 State，输出模板变量。 if err := g.AddLambdaNode("recall", compose.InvokableLambda( func(ctx context.Context, t *contract.Task) (map[string]any, error) { uid, _ := t.Meta[contract.MetaUserID].(string) + sid, _ := t.Meta[contract.MetaSessionID].(string) profile := fetch(ctx, uid, string(t.Graph)) + hist := fetchHist(ctx, sid) _ = compose.ProcessState(ctx, func(_ context.Context, s *AgentState) error { - s.UserID, s.Profile, s.Input = uid, profile, string(t.Graph) + s.UserID, s.SessionID, s.Profile, s.Input = uid, sid, profile, string(t.Graph) return nil }) if profile == "" { profile = "（暂无该用户的偏好记忆）" } - return map[string]any{"profile": profile, "query": string(t.Graph)}, nil + return map[string]any{"profile": profile, "query": string(t.Graph), "history": hist}, nil })); err != nil { return nil, err } - // 2) prompt：把画像注入 system message，用户输入作为 user message。 + // 2) prompt：画像注入 system，历史用占位符插入，用户输入作为 user message。 tpl := prompt.FromMessages(schema.FString, schema.SystemMessage("你在与特定用户对话。关于该用户的已知信息：\n{profile}\n请据此个性化作答并保持其偏好。"), + schema.MessagesPlaceholder("history", true), schema.UserMessage("{query}"), ) if err := g.AddChatTemplateNode("prompt", tpl); err != nil { diff --git a/sundynix-dispatcher/internal/eino/model.go b/sundynix-dispatcher/internal/eino/model.go index 8806c3d..8afc1f8 100644 --- a/sundynix-dispatcher/internal/eino/model.go +++ b/sundynix-dispatcher/internal/eino/model.go @@ -48,18 +48,36 @@ func (pm *poolModel) Stream(ctx context.Context, input []*schema.Message, _ ...m // 真实模型不需要本函数。 func replyFor(msgs []*schema.Message) string { var profile, user string - for _, m := range msgs { + priorTurns := 0 + for i, m := range msgs { switch m.Role { case schema.System: profile = m.Content + case schema.Assistant: + priorTurns++ // 历史里的助手消息 = 过往轮次 case schema.User: - user = m.Content + if i == len(msgs)-1 { + user = m.Content // 最后一条 user 才是本轮输入 + } } } return "【已注入用户画像】" + condense(profile, 80) + + "（本会话已有 " + itoa(priorTurns) + " 轮历史）" + " ｜ 据此为你个性化作答：已编排执行该 Agent 图（输入「" + condense(user, 30) + "」）。" } +func itoa(n int) string { + if n == 0 { + return "0" + } + var b []byte + for n > 0 { + b = append([]byte{byte('0' + n%10)}, b...) + n /= 10 + } + return string(b) +} + func condense(s string, max int) string { s = strings.TrimSpace(strings.ReplaceAll(s, "\n", " ")) r := []rune(s) diff --git a/sundynix-dispatcher/internal/eino/orchestrator.go b/sundynix-dispatcher/internal/eino/orchestrator.go index 16c7628..2514d00 100644 --- a/sundynix-dispatcher/internal/eino/orchestrator.go +++ b/sundynix-dispatcher/internal/eino/orchestrator.go @@ -3,9 +3,11 @@ package eino import ( "context" + "encoding/json" "errors" "io" "log" + "strings" "time" "github.com/cloudwego/eino/compose" @@ -41,7 +43,7 @@ type Orchestrator struct { // NewOrchestrator 构建并编译记忆增强图。 func NewOrchestrator(pool *llm.Pool, breaker *harness.CircuitBreaker, sink TokenSink, tools ToolCaller) (*Orchestrator, error) { o := &Orchestrator{breaker: breaker, sink: sink, tools: tools} - run, err := buildGraph(context.Background(), pool, o.fetchMemory) + run, err := buildGraph(context.Background(), pool, o.fetchMemory, o.fetchHistory) if err != nil { return nil, err } @@ -67,6 +69,7 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error { defer stream.Close() n := 0 + var answer strings.Builder for { chunk, rerr := stream.Recv() if errors.Is(rerr, io.EOF) { @@ -83,6 +86,7 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error { log.Printf("[eino] publish token failed: %v", perr) break } + answer.WriteString(chunk.Content) n++ } @@ -92,9 +96,9 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error { log.Printf("[eino] task %s done, %d tokens streamed", t.ID, n) o.breaker.Report(true) - // 写回阶段：流已排空（= 模型生成结束），此处离开热路径、异步抽取记忆。 + // 写回阶段：流已排空（= 模型生成结束），此处离开热路径、异步落历史 + 抽取记忆。 // 注：流式节点用 OnEndWithStreamOutput 而非 OnEndFn，故不走回调而在此触发。 - go o.memorize(t) + go o.memorize(t, answer.String()) return nil } @@ -122,13 +126,65 @@ func (o *Orchestrator) fetchMemory(ctx context.Context, userID, _ string) string return res.Content } -// memorize 写回阶段：从本轮对话抽取并更新偏好记忆。 -// 目前发占位日志；真实实现应跑抽取 LLM → 去重/更新 → memory_upsert（异步，离开热路径）。 -func (o *Orchestrator) memorize(t *contract.Task) { - uid, _ := t.Meta[contract.MetaUserID].(string) - if uid == "" { - return +// fetchHistory 经 MCP history_get 工具召回会话短期多轮历史，转为 Eino 消息。 +// 工具不可用/无 session 时返回空，降级为无历史（不阻断主流程）。 +func (o *Orchestrator) fetchHistory(ctx context.Context, sessionID string) []*schema.Message { + if o.tools == nil || sessionID == "" { + return nil + } + cctx, cancel := context.WithTimeout(ctx, toolCallTimeout) + defer cancel() + res, err := o.tools.CallTool(cctx, contract.ToolSubjectGo("history_get"), &contract.ToolCall{ + Tool: "history_get", + Args: map[string]any{"session_id": sessionID}, + }) + if err != nil || res == nil || !res.OK || res.Content == "" { + return nil + } + var turns []struct { + Role string `json:"role"` + Content string `json:"content"` + } + if json.Unmarshal([]byte(res.Content), &turns) != nil { + return nil + } + msgs := make([]*schema.Message, 0, len(turns)) + for _, tn := range turns { + if tn.Role == "assistant" { + msgs = append(msgs, schema.AssistantMessage(tn.Content, nil)) + } else { + msgs = append(msgs, schema.UserMessage(tn.Content)) + } + } + if len(msgs) > 0 { + log.Printf("[eino] history_get ok for %s: %d 条历史", sessionID, len(msgs)) + } + return msgs +} + +// memorize 写回阶段：把本轮对话落进短期历史，并（TODO）抽取长期偏好记忆。 +// 异步执行，离开热路径。 +func (o *Orchestrator) memorize(t *contract.Task, answer string) { + uid, _ := t.Meta[contract.MetaUserID].(string) + sid, _ := t.Meta[contract.MetaSessionID].(string) + if sid != "" && o.tools != nil { + o.appendHistory(sid, "user", string(t.Graph)) + o.appendHistory(sid, "assistant", answer) + log.Printf("[eino] (writeback) task %s 已落会话历史 session=%s", t.ID, sid) + } + if uid != "" { + log.Printf("[eino] (writeback) task %s 待抽取 user=%s 的新偏好记忆", t.ID, uid) + // TODO: 抽取 LLM → 去重/更新 → memory_upsert + } +} + +func (o *Orchestrator) appendHistory(sessionID, role, content string) { + cctx, cancel := context.WithTimeout(context.Background(), toolCallTimeout) + defer cancel() + if _, err := o.tools.CallTool(cctx, contract.ToolSubjectGo("history_append"), &contract.ToolCall{ + Tool: "history_append", + Args: map[string]any{"session_id": sessionID, "role": role, "content": content}, + }); err != nil { + log.Printf("[eino] history_append failed: %v", err) } - log.Printf("[eino] (writeback) task %s 完成，待抽取 user=%s 的新偏好记忆", t.ID, uid) - // TODO: 发 sundynix.memory.extract 事件 → memory worker 抽取 → memory_upsert } diff --git a/sundynix-dispatcher/internal/eino/state.go b/sundynix-dispatcher/internal/eino/state.go index 2949863..70486d1 100644 --- a/sundynix-dispatcher/internal/eino/state.go +++ b/sundynix-dispatcher/internal/eino/state.go @@ -3,8 +3,9 @@ package eino // AgentState 是 Eino 图的全局状态，贯穿 recall→prompt→model 各节点。 // 偏好记忆经 recall 节点写入，供模板注入与写回抽取使用。 type AgentState struct { - UserID string // 来自 Task.Meta["user_id"] - Profile string // 召回到的常驻画像（always-on 偏好记忆） - Input string // 本次输入（DSL 原文） - Answer string // 累积输出，供写回阶段抽取新记忆 + UserID string // 来自 Task.Meta["user_id"] + SessionID string // 来自 Task.Meta["session_id"] + Profile string // 召回到的常驻画像（always-on 偏好记忆） + Input string // 本次输入（DSL 原文） + Answer string // 累积输出，供写回阶段抽取新记忆 } diff --git a/sundynix-gateway/internal/handler/task_handler.go b/sundynix-gateway/internal/handler/task_handler.go index 34454d4..5bc4af2 100644 --- a/sundynix-gateway/internal/handler/task_handler.go +++ b/sundynix-gateway/internal/handler/task_handler.go @@ -37,9 +37,10 @@ func (h *Handler) SubmitTask(c *gin.Context) { c.JSON(http.StatusUnprocessableEntity, gin.H{"error": err.Error()}) return } - // 附上已登录用户标识，供 Dispatcher 召回其偏好记忆。 - // 真实场景由鉴权中间件注入；此处用 X-User-ID 头，缺省匿名。 + // 附上用户标识（召回偏好记忆）与会话标识（召回短期多轮历史）。 + // 真实场景由鉴权/会话中间件注入；此处用请求头，缺省匿名/默认会话。 task.Meta[contract.MetaUserID] = userID(c) + task.Meta[contract.MetaSessionID] = sessionID(c) // 持久化任务提交（best-effort：降级模式下静默跳过，不阻断发布）。 if err := h.db.SaveTask(c.Request.Context(), task.ID, string(task.Graph)); err != nil { log.Printf("[gateway] save task %s failed: %v", task.ID, err) @@ -124,6 +125,14 @@ func userID(c *gin.Context) string { return "anonymous" } +// sessionID 从请求取会话标识（真实场景应由会话中间件注入）。 +func sessionID(c *gin.Context) string { + if s := c.GetHeader("X-Session-ID"); s != "" { + return s + } + return "default" +} + func (h *Handler) Billing(c *gin.Context) { // TODO: 商业化与计费模块；暂以已提交任务计数演示真实读库。 n, err := h.db.CountTasks(c.Request.Context()) diff --git a/sundynix-mcp-go/cmd/server/main.go b/sundynix-mcp-go/cmd/server/main.go index 138920d..63eca57 100644 --- a/sundynix-mcp-go/cmd/server/main.go +++ b/sundynix-mcp-go/cmd/server/main.go @@ -10,6 +10,7 @@ import ( sharedbus "github.com/sundynix/sundynix-shared/bus" + "github.com/sundynix/sundynix-mcp-go/internal/history" "github.com/sundynix/sundynix-mcp-go/internal/mcp" "github.com/sundynix/sundynix-mcp-go/internal/memory" "github.com/sundynix/sundynix-mcp-go/internal/search" @@ -18,6 +19,7 @@ import ( func main() { 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") b, err := sharedbus.Connect(natsURL) if err != nil { @@ -27,9 +29,11 @@ func main() { log.Printf("[mcp_go] connected %s", natsURL) engine := search.NewHybrid() // LLM Wiki 混合检索：Bleve + Milvus + Neo4j - mem := memory.Open(pgDSN) // 偏好记忆：sundynix_user_profile（连不上则降级） + mem := memory.Open(pgDSN) // 偏好记忆：sundynix_user_profile（连不上则降级） defer mem.Close() - gw := mcp.NewGateway(b, engine, mem) + hist := history.Open(redisAddr) // 会话短期历史：Redis（连不上则降级） + defer hist.Close() + gw := mcp.NewGateway(b, engine, mem, hist) ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() diff --git a/sundynix-mcp-go/go.mod b/sundynix-mcp-go/go.mod index 5e32980..74bd705 100644 --- a/sundynix-mcp-go/go.mod +++ b/sundynix-mcp-go/go.mod @@ -1,14 +1,16 @@ module github.com/sundynix/sundynix-mcp-go -go 1.23 +go 1.24 require ( + github.com/redis/go-redis/v9 v9.20.0 github.com/sundynix/sundynix-shared v0.0.0 gorm.io/driver/postgres v1.6.0 gorm.io/gorm v1.31.1 ) require ( + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/pgx/v5 v5.6.0 // indirect @@ -19,9 +21,10 @@ require ( github.com/nats-io/nats.go v1.37.0 // indirect github.com/nats-io/nkeys v0.4.7 // indirect github.com/nats-io/nuid v1.0.1 // indirect + go.uber.org/atomic v1.11.0 // indirect golang.org/x/crypto v0.31.0 // indirect golang.org/x/sync v0.10.0 // indirect - golang.org/x/sys v0.28.0 // indirect + golang.org/x/sys v0.30.0 // indirect golang.org/x/text v0.21.0 // indirect ) diff --git a/sundynix-mcp-go/go.sum b/sundynix-mcp-go/go.sum index 1308d84..bef4478 100644 --- a/sundynix-mcp-go/go.sum +++ b/sundynix-mcp-go/go.sum @@ -1,3 +1,9 @@ +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -15,6 +21,8 @@ github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA= github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= +github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/minio/highwayhash v1.0.3 h1:kbnuUMoHYyVl7szWjSxJnxw11k2U709jqFPPmIUyD6Q= github.com/minio/highwayhash v1.0.3/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= github.com/nats-io/jwt/v2 v2.5.8 h1:uvdSzwWiEGWGXf+0Q+70qv6AQdvcvxrv9hPM0RiPamE= @@ -29,17 +37,23 @@ github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/redis/go-redis/v9 v9.20.0 h1:WnQYxLkgO2xiXTCJY0ldIiI8dNqCDlQAG+AtaH7a2a0= +github.com/redis/go-redis/v9 v9.20.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U= golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk= golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ= golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= -golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA= -golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= +golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo= golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ= golang.org/x/time v0.6.0 h1:eTDhh4ZXt5Qf0augr54TN6suAUudPcawVZeIAPU7D4U= diff --git a/sundynix-mcp-go/internal/history/store.go b/sundynix-mcp-go/internal/history/store.go new file mode 100644 index 0000000..7df755e --- /dev/null +++ b/sundynix-mcp-go/internal/history/store.go @@ -0,0 +1,84 @@ +// Package history 是会话短期多轮历史的存储后端（Redis CacheDB）。 +// 与长期偏好记忆（Postgres）分开：历史是易失的、按会话滚动保留最近 N 轮。 +package history + +import ( + "context" + "encoding/json" + "log" + "time" + + "github.com/redis/go-redis/v9" +) + +// maxTurns 是单会话保留的最近消息条数（user/assistant 各算一条）。 +const maxTurns = 20 + +// Turn 是一条历史消息。 +type Turn struct { + Role string `json:"role"` // user / assistant + Content string `json:"content"` +} + +// Store 封装会话历史读写。rdb 为 nil 表示降级（无 Redis 时历史为空，不阻断）。 +type Store struct{ rdb *redis.Client } + +// Open 连接 Redis。失败不 fatal：返回降级实例。 +func Open(addr string) *Store { + rdb := redis.NewClient(&redis.Options{Addr: addr}) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := rdb.Ping(ctx).Err(); err != nil { + log.Printf("[history] redis 不可用，会话历史降级（为空）: %v", err) + _ = rdb.Close() + return &Store{} + } + log.Println("[history] redis connected") + return &Store{rdb: rdb} +} + +func key(session string) string { return "sundynix:history:" + session } + +// Get 返回某会话最近的历史（按时间正序）。 +func (s *Store) Get(ctx context.Context, session string) ([]Turn, error) { + if s.rdb == nil || session == "" { + return nil, nil + } + // 列表头插尾老：取全部后反转为正序。 + vals, err := s.rdb.LRange(ctx, key(session), 0, maxTurns-1).Result() + if err != nil { + return nil, err + } + turns := make([]Turn, 0, len(vals)) + for i := len(vals) - 1; i >= 0; i-- { // 反转 → 正序 + var t Turn + if json.Unmarshal([]byte(vals[i]), &t) == nil { + turns = append(turns, t) + } + } + return turns, nil +} + +// Append 追加一条消息并裁剪到最近 maxTurns 条，刷新 TTL。 +func (s *Store) Append(ctx context.Context, session, role, content string) error { + if s.rdb == nil || session == "" { + return nil + } + data, err := json.Marshal(Turn{Role: role, Content: content}) + if err != nil { + return err + } + pipe := s.rdb.TxPipeline() + pipe.LPush(ctx, key(session), data) + pipe.LTrim(ctx, key(session), 0, maxTurns-1) + pipe.Expire(ctx, key(session), 24*time.Hour) + _, err = pipe.Exec(ctx) + return err +} + +// Close 释放连接。 +func (s *Store) Close() { + if s.rdb != nil { + _ = s.rdb.Close() + } +} diff --git a/sundynix-mcp-go/internal/mcp/gateway.go b/sundynix-mcp-go/internal/mcp/gateway.go index 534a812..9335854 100644 --- a/sundynix-mcp-go/internal/mcp/gateway.go +++ b/sundynix-mcp-go/internal/mcp/gateway.go @@ -3,25 +3,28 @@ package mcp import ( "context" + "encoding/json" "fmt" "log" sharedbus "github.com/sundynix/sundynix-shared/bus" "github.com/sundynix/sundynix-shared/contract" + "github.com/sundynix/sundynix-mcp-go/internal/history" "github.com/sundynix/sundynix-mcp-go/internal/memory" "github.com/sundynix/sundynix-mcp-go/internal/search" ) // Gateway 暴露 MCP 协议端点，经共享 bus 订阅 sundynix.tools.go.* 响应调用。 type Gateway struct { - bus *sharedbus.Bus - search *search.Hybrid - memory *memory.Store + bus *sharedbus.Bus + search *search.Hybrid + memory *memory.Store + history *history.Store } -func NewGateway(b *sharedbus.Bus, s *search.Hybrid, m *memory.Store) *Gateway { - return &Gateway{bus: b, search: s, memory: m} +func NewGateway(b *sharedbus.Bus, s *search.Hybrid, m *memory.Store, h *history.Store) *Gateway { + return &Gateway{bus: b, search: s, memory: m, history: h} } // Serve 以队列组通配订阅 sundynix.tools.go.>，按工具名分发并阻塞。 @@ -31,7 +34,7 @@ func (g *Gateway) Serve(ctx context.Context) error { return err } defer func() { _ = unsub() }() - log.Printf("[mcp_go] tools ready on %s (queue=%s): wiki_search, memory_get, memory_upsert, echo", + log.Printf("[mcp_go] tools ready on %s (queue=%s): wiki_search, memory_get, memory_upsert, history_get, history_append, echo", contract.SubjectToolsGoAll, contract.QueueToolsGo) <-ctx.Done() return ctx.Err() @@ -47,6 +50,10 @@ func (g *Gateway) dispatch(ctx context.Context, call *contract.ToolCall) *contra return g.memoryGet(ctx, call) case "memory_upsert": return g.memoryUpsert(ctx, call) + case "history_get": + return g.historyGet(ctx, call) + case "history_append": + return g.historyAppend(ctx, call) case "echo": return &contract.ToolResult{OK: true, Content: fmt.Sprint(call.Args["text"])} default: @@ -64,6 +71,31 @@ func (g *Gateway) memoryGet(ctx context.Context, call *contract.ToolCall) *contr return &contract.ToolResult{OK: true, Content: profile} } +// historyGet 召回某会话最近多轮历史，Content 为 JSON 数组 [{role,content},...]（正序）。 +func (g *Gateway) historyGet(ctx context.Context, call *contract.ToolCall) *contract.ToolResult { + session, _ := call.Args["session_id"].(string) + turns, err := g.history.Get(ctx, session) + if err != nil { + return &contract.ToolResult{OK: false, Error: "history_get: " + err.Error()} + } + data, _ := json.Marshal(turns) + return &contract.ToolResult{OK: true, Content: string(data)} +} + +// historyAppend 追加一条会话消息（session_id + role + content）。 +func (g *Gateway) historyAppend(ctx context.Context, call *contract.ToolCall) *contract.ToolResult { + session, _ := call.Args["session_id"].(string) + role, _ := call.Args["role"].(string) + content, _ := call.Args["content"].(string) + if session == "" || role == "" { + return &contract.ToolResult{OK: false, Error: "history_append: session_id 和 role 必填"} + } + if err := g.history.Append(ctx, session, role, content); err != nil { + return &contract.ToolResult{OK: false, Error: "history_append: " + err.Error()} + } + return &contract.ToolResult{OK: true} +} + // memoryUpsert 写入/更新一条画像偏好（user_id + key + value）。 func (g *Gateway) memoryUpsert(ctx context.Context, call *contract.ToolCall) *contract.ToolResult { uid, _ := call.Args["user_id"].(string) diff --git a/sundynix-shared/contract/task.go b/sundynix-shared/contract/task.go index 5db9c40..1358b95 100644 --- a/sundynix-shared/contract/task.go +++ b/sundynix-shared/contract/task.go @@ -27,6 +27,8 @@ const ( // MetaUserID 是 Task.Meta 中承载已登录用户标识的键（用于偏好记忆召回）。 MetaUserID = "user_id" + // MetaSessionID 是 Task.Meta 中承载会话标识的键（用于短期多轮历史）。 + MetaSessionID = "session_id" ) // Task 是 DSL 解析组装后的可调度任务，在 NATS 上以 JSON 传输。