feat: 短期多轮历史接入 Eino 图 MessagesPlaceholder (⑨)

会话历史(Redis,易失,与长期画像分开)经 MCP 工具进出 Eino 图:
recall 召回历史填 MessagesPlaceholder,写回把本轮 user/assistant 落历史。

- mcp-go: internal/history(go-redis, sundynix:history:<session>, 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 <noreply@anthropic.com>
This commit is contained in:
Blizzard
2026-06-10 14:18:45 +08:00
parent cbd130ecae
commit 4928ffc0f7
12 changed files with 288 additions and 47 deletions
+6 -2
View File
@@ -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()
+5 -2
View File
@@ -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
)
+16 -2
View File
@@ -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=
+84
View File
@@ -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()
}
}
+38 -6
View File
@@ -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)