diff --git a/cmd/ragflow_server.go b/cmd/ragflow_server.go index 6c4476cea1..4af6bf56ee 100644 --- a/cmd/ragflow_server.go +++ b/cmd/ragflow_server.go @@ -49,7 +49,7 @@ import ( "github.com/gin-gonic/gin" "go.uber.org/zap" - _ "ragflow/internal/agent/component" + "ragflow/internal/agent/component" "ragflow/internal/common" "ragflow/internal/dao" "ragflow/internal/engine" @@ -705,6 +705,10 @@ func startServer(config *server.Config) { mcpService := service.NewMCPService() modelProviderService := service.NewModelProviderService() + // Wire the real MemorySaver so the Message component can persist + // conversation turns to memory stores declared in the canvas DSL. + component.SetMemorySaver(service.NewMemorySaverAdapter(memoryService)) + // Initialize doc engine for skill search docEngine := engine.Get() documentDAO := dao.NewDocumentDAO() diff --git a/internal/agent/component/message.go b/internal/agent/component/message.go index 06a057f04a..d17b33cf02 100644 --- a/internal/agent/component/message.go +++ b/internal/agent/component/message.go @@ -56,6 +56,8 @@ type MessageComponent struct { autoPlay audio.Engine voice string lang string + memoryIDs []string + userID string } // NewMessageComponent constructs a Message component. The params map @@ -82,6 +84,8 @@ func NewMessageComponent(params map[string]any) (Component, error) { format = OutputFormat(v) } engine, voice, lang := extractAudioConfig(params) + memIDs := extractMemoryIDsFromAny(params["memory_ids"]) + userID, _ := params["user_id"].(string) return &MessageComponent{ name: componentNameMessage, text: tpl, @@ -89,6 +93,8 @@ func NewMessageComponent(params map[string]any) (Component, error) { autoPlay: engine, voice: voice, lang: lang, + memoryIDs: memIDs, + userID: userID, }, nil } @@ -285,22 +291,38 @@ func (m *MessageComponent) Invoke(ctx context.Context, inputs map[string]any) (m // memory service returns ErrMemoryServiceMissing which we // surface under outputs["memory_error"] so the message still // flows. - memSave, _ := inputs["memory_save"].(bool) - if memSave { - memIDs := extractMemoryIDs(inputs) - if len(memIDs) > 0 { - saver := GetMemorySaver() - saveErr := saver.Save(ctx, MemorySaveRequest{ - MemoryIDs: memIDs, - AgentID: state.TaskID, - SessionID: state.RunID, - UserInput: stringFromStateSys(state, "query"), - AgentResponse: rendered, - }) - if saveErr != nil { - out["memory_error"] = saveErr.Error() - common.Error("Message: memory_save failed", saveErr) - } + // + // The effective memory IDs come from inputs (runtime override) + // or fall back to the DSL-declared m.memoryIDs. This matches + // the Python Message component, which saves whenever + // memory_ids is non-empty. + memIDs := extractMemoryIDs(inputs) + if len(memIDs) == 0 { + memIDs = m.memoryIDs + } + if len(memIDs) > 0 { + userID := stringFromStateSys(state, "user_id") + if userID == "" { + userID = m.userID + } + // If userID is a canvas variable reference (e.g. "{cpn@user_id}"), + // resolve it against the current state. Mirrors Python's + // agent/component/message.py:569-571. + if userID != "" && runtime.VarRefPattern.MatchString(userID) { + userID = runtime.ResolveTemplateForDisplay(userID, state) + } + saver := GetMemorySaver() + saveErr := saver.Save(ctx, MemorySaveRequest{ + MemoryIDs: memIDs, + UserID: userID, + AgentID: state.TaskID, + SessionID: state.RunID, + UserInput: stringFromStateSys(state, "query"), + AgentResponse: rendered, + }) + if saveErr != nil { + out["memory_error"] = saveErr.Error() + common.Error("Message: memory_save failed", saveErr) } } @@ -310,10 +332,13 @@ func (m *MessageComponent) Invoke(ctx context.Context, inputs map[string]any) (m // extractMemoryIDs normalises a memory_ids value from inputs / // params. Accepts []string and []any[string]. func extractMemoryIDs(inputs map[string]any) []string { - v, ok := inputs["memory_ids"] - if !ok { - return nil - } + return extractMemoryIDsFromAny(inputs["memory_ids"]) +} + +// extractMemoryIDsFromAny normalises a memory_ids value from any +// source (DSL params or runtime inputs). Accepts []string and +// []any[string]. +func extractMemoryIDsFromAny(v any) []string { switch x := v.(type) { case []string: return x @@ -362,7 +387,7 @@ func fallbackMessageText(inputs map[string]any) string { func isMessageInfraInput(key string) bool { switch key { case "state", "__cpn_id__", "__legacy_noop__", "_created_time", "_elapsed_time", - "output_format", "voice", "lang", "auto_play", "memory_save", "stream": + "output_format", "voice", "lang", "auto_play", "memory_save", "memory_ids", "user_id", "stream": return true default: return false diff --git a/internal/agent/component/message_phase8b_test.go b/internal/agent/component/message_phase8b_test.go index fa646882d6..2064b5bcfa 100644 --- a/internal/agent/component/message_phase8b_test.go +++ b/internal/agent/component/message_phase8b_test.go @@ -285,6 +285,113 @@ func TestMessage_MemorySave_Success(t *testing.T) { } } +// TestMessage_MemorySave_FromDSLParams: memory_ids declared in the +// DSL params (constructor) are honoured even when the runtime inputs +// map does not carry memory_ids. This is the production path where +// the pipeline only provides upstream component outputs. +func TestMessage_MemorySave_FromDSLParams(t *testing.T) { + var saved MemorySaveRequest + saveFn := memSaverFunc(func(_ context.Context, req MemorySaveRequest) error { + saved = req + return nil + }) + SetMemorySaver(&saveFn) + defer SetMemorySaver(nil) + + c, _ := NewMessageComponent(map[string]any{ + "text": "hi", + "memory_ids": []string{"dsl-m1", "dsl-m2"}, + }) + state := canvas.NewCanvasState("run-dsl", "task-dsl") + state.Sys["query"] = "hello?" + ctx := withStateForTest(context.Background(), state) + + // Inputs simulate what the pipeline actually provides: only upstream + // outputs, NO memory_ids or memory_save keys. + _, err := c.Invoke(ctx, map[string]any{ + "text": "hi", + "stream": false, + }) + if err != nil { + t.Fatalf("Invoke: %v", err) + } + if len(saved.MemoryIDs) != 2 || saved.MemoryIDs[0] != "dsl-m1" || saved.MemoryIDs[1] != "dsl-m2" { + t.Errorf("MemoryIDs from DSL params: %+v", saved.MemoryIDs) + } + if saved.AgentResponse != "hi" { + t.Errorf("AgentResponse: got %q, want hi", saved.AgentResponse) + } +} + +// TestMessage_MemorySave_UserIDVariable: when user_id is a canvas +// variable reference (e.g. "{begin@user_id}"), it is resolved +// against the current state before being passed to the saver. +// Mirrors Python agent/component/message.py:569-571. +func TestMessage_MemorySave_UserIDVariable(t *testing.T) { + var saved MemorySaveRequest + saveFn := memSaverFunc(func(_ context.Context, req MemorySaveRequest) error { + saved = req + return nil + }) + SetMemorySaver(&saveFn) + defer SetMemorySaver(nil) + + c, _ := NewMessageComponent(map[string]any{ + "text": "hi", + "user_id": "{begin@user_id}", + }) + state := canvas.NewCanvasState("run-uid", "task-uid") + state.Sys["query"] = "hello?" + state.SetVar("begin", "user_id", "resolved-user-123") + ctx := withStateForTest(context.Background(), state) + + _, err := c.Invoke(ctx, map[string]any{ + "text": "hi", + "stream": false, + "memory_save": true, + "memory_ids": []string{"m1"}, + }) + if err != nil { + t.Fatalf("Invoke: %v", err) + } + if saved.UserID != "resolved-user-123" { + t.Errorf("UserID: got %q, want resolved-user-123", saved.UserID) + } +} + +// TestMessage_MemorySave_UserIDLiteral: a plain (non-variable) +// user_id is passed through as-is. +func TestMessage_MemorySave_UserIDLiteral(t *testing.T) { + var saved MemorySaveRequest + saveFn := memSaverFunc(func(_ context.Context, req MemorySaveRequest) error { + saved = req + return nil + }) + SetMemorySaver(&saveFn) + defer SetMemorySaver(nil) + + c, _ := NewMessageComponent(map[string]any{ + "text": "hi", + "user_id": "plain-user-456", + }) + state := canvas.NewCanvasState("run-uid2", "task-uid2") + state.Sys["query"] = "hello?" + ctx := withStateForTest(context.Background(), state) + + _, err := c.Invoke(ctx, map[string]any{ + "text": "hi", + "stream": false, + "memory_save": true, + "memory_ids": []string{"m1"}, + }) + if err != nil { + t.Fatalf("Invoke: %v", err) + } + if saved.UserID != "plain-user-456" { + t.Errorf("UserID: got %q, want plain-user-456", saved.UserID) + } +} + // memSaverFunc adapts a closure to the MemorySaver interface so // tests can record the call inline. type memSaverFunc func(ctx context.Context, req MemorySaveRequest) error diff --git a/internal/service/memory_saver_adapter.go b/internal/service/memory_saver_adapter.go new file mode 100644 index 0000000000..152aa713bc --- /dev/null +++ b/internal/service/memory_saver_adapter.go @@ -0,0 +1,63 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package service + +import ( + "context" + "fmt" + + "ragflow/internal/agent/component" +) + +// memorySaverAdapter bridges the component.MemorySaver interface to +// MemoryService.AddMessage. It is installed via component.SetMemorySaver +// at boot time so that the Message component can persist conversation +// turns to memory stores declared in the canvas DSL. +type memorySaverAdapter struct { + svc *MemoryService +} + +// NewMemorySaverAdapter returns a component.MemorySaver backed by the +// given MemoryService. +func NewMemorySaverAdapter(svc *MemoryService) component.MemorySaver { + return &memorySaverAdapter{svc: svc} +} + +// Save implements component.MemorySaver. It delegates to +// MemoryService.AddMessage which handles access filtering, message +// construction, embedding, and async task queueing — the same pipeline +// used by the REST API add_message endpoint. +func (a *memorySaverAdapter) Save(ctx context.Context, req component.MemorySaveRequest) error { + if a == nil || a.svc == nil { + return fmt.Errorf("memory: saver adapter not initialised") + } + msg := MemoryMessage{ + UserID: req.UserID, + AgentID: req.AgentID, + SessionID: req.SessionID, + UserInput: req.UserInput, + AgentResponse: req.AgentResponse, + } + ok, detail, err := a.svc.AddMessage(ctx, req.UserID, req.MemoryIDs, msg) + if err != nil { + return err + } + if !ok { + return fmt.Errorf("memory: %s", detail) + } + return nil +}