From be659cba1178585472d1e01069b85d8290ba8bd7 Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Thu, 10 Sep 2026 11:05:14 +0800 Subject: [PATCH] feat(chat): wire multi-turn dialogue broken links (G1/G2/G3/G9) with session restore --- internal/chat/session.go | 92 +++++++++++++++---- internal/chat/session_test.go | 71 +++++++++++++++ internal/ctxmgr/manager.go | 15 ++++ internal/ctxmgr/manager_test.go | 9 ++ internal/server/chat_restore_test.go | 127 +++++++++++++++++++++++++++ internal/server/handlers.go | 7 +- internal/server/server.go | 72 +++++++++++++++ 7 files changed, 370 insertions(+), 23 deletions(-) create mode 100644 internal/server/chat_restore_test.go diff --git a/internal/chat/session.go b/internal/chat/session.go index 4c0e638a..99f3bd7f 100644 --- a/internal/chat/session.go +++ b/internal/chat/session.go @@ -3,6 +3,7 @@ package chat import ( "context" "fmt" + "strings" "time" "chat-agent-system/internal/coordinator" @@ -37,15 +38,16 @@ type ChatResponse struct { // Session 多轮对话与协同会话管理器 type Session struct { - ID string `json:"id"` - UserID int64 `json:"user_id"` - ProjectID int64 `json:"project_id"` - UserCtx *coordinator.UserProjectContext `json:"user_ctx"` - Recognizer intent.Recognizer `json:"-"` - Coordinator *coordinator.CoordinatorEngine `json:"-"` - CtxMgr *ctxmgr.ContextManager `json:"-"` - Messages []Message `json:"messages"` - CreatedAt time.Time `json:"created_at"` + ID string `json:"id"` + UserID int64 `json:"user_id"` + ProjectID int64 `json:"project_id"` + UserCtx *coordinator.UserProjectContext `json:"user_ctx"` + Recognizer intent.Recognizer `json:"-"` + Coordinator *coordinator.CoordinatorEngine `json:"-"` + CtxMgr *ctxmgr.ContextManager `json:"-"` + Messages []Message `json:"messages"` + CreatedAt time.Time `json:"created_at"` + LastAccessedAt time.Time `json:"last_accessed_at"` } // NewSession 创建新对话会话 @@ -61,15 +63,59 @@ func NewSession(id string, userCtx *coordinator.UserProjectContext, rec intent.R userCtx.PendingExpectation = coordinator.BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name) } - return &Session{ - ID: id, - UserID: 1, - ProjectID: 1, - UserCtx: userCtx, - Recognizer: rec, - Coordinator: coord, - Messages: make([]Message, 0), - CreatedAt: time.Now(), + sess := &Session{ + ID: id, + UserID: 1, + ProjectID: 1, + UserCtx: userCtx, + Recognizer: rec, + Coordinator: coord, + Messages: make([]Message, 0), + CreatedAt: time.Now(), + LastAccessedAt: time.Now(), + } + sess.SyncRecentDialogue() + return sess +} + +// SyncRecentDialogue 将会话内存消息转换为 DialogueTurn 同步到 UserCtx.RecentDialogue (供 Phase A 对话 LLM 与路由画像使用) +func (s *Session) SyncRecentDialogue() { + if s.UserCtx == nil { + return + } + const maxTurns = 16 // 保持合理窗口 (略宽于 dialogueHistoryMax=8，保留完整轮次) + msgs := s.Messages + if len(msgs) > maxTurns { + msgs = msgs[len(msgs)-maxTurns:] + } + turns := make([]coordinator.DialogueTurn, 0, len(msgs)) + for _, m := range msgs { + if strings.TrimSpace(m.Content) == "" { + continue + } + turns = append(turns, coordinator.DialogueTurn{Role: m.Role, Content: m.Content}) + } + s.UserCtx.RecentDialogue = turns +} + +// syncPersonalizedGuidance 将用户偏好画像编译为 Prompt 指导语并注入 UserCtx +func (s *Session) syncPersonalizedGuidance(ctx context.Context, userInput string) { + if s.UserCtx == nil || s.CtxMgr == nil || s.UserID <= 0 { + return + } + // 首次注入，或用户显式表达定制要求（"记住..." / "以后请..." / "默认都..."），或周期性刷新（每10条消息） + needsRefresh := s.UserCtx.PersonalizedGuidance == "" || + strings.Contains(userInput, "记住") || + strings.Contains(userInput, "以后请") || + strings.Contains(userInput, "默认都") || + len(s.Messages)%10 == 0 + + if !needsRefresh { + return + } + + if guidance, err := s.CtxMgr.CompilePromptGuidance(ctx, s.UserID); err == nil && guidance != "" { + s.UserCtx.PersonalizedGuidance = guidance } } @@ -87,10 +133,14 @@ func (s *Session) SendMessageStream(ctx context.Context, userInput string, liste func (s *Session) sendMessageInternal(ctx context.Context, userInput string, streamer *stream.Streamer) (*ChatResponse, error) { start := time.Now() + s.LastAccessedAt = time.Now() if streamer != nil { streamer.EmitPhase("intent_recognizing", "正在研判您的意图与项目上下文...", "coordinator", 0, 0, 0) } + // 0. 同步已有前序对话到 UserCtx，供意图快照、路由评分与对话 LLM 感知历史轮次 + s.SyncRecentDialogue() + userMsg := Message{ ID: fmt.Sprintf("msg_%d_u", time.Now().UnixNano()), Role: "user", @@ -103,6 +153,9 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str _ = s.CtxMgr.RecordUserMessage(ctx, s.ID, s.UserID, "user", userInput, "", "", nil) } + // 偏好画像同步 + s.syncPersonalizedGuidance(ctx, userInput) + // 1. 意图识别 (L1 对话层，带非阻塞桌角备忘快照) var snap *intent.ContextSnapshot if s.UserCtx != nil { @@ -168,6 +221,9 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str _ = s.CtxMgr.RecordUserMessage(ctx, s.ID, 0, "assistant", coordRes.NaturalReply, string(coordRes.Intent), string(recRes.Branch), nil) } + // 本轮全部处理完毕后，再次同步，保持 UserCtx 与 Messages 一致 + s.SyncRecentDialogue() + // 4. 流式推送参谋自然语言回复与决策锦囊 if streamer != nil { if coordRes.NaturalReply != "" { diff --git a/internal/chat/session_test.go b/internal/chat/session_test.go index 7dac71b1..00b7a363 100644 --- a/internal/chat/session_test.go +++ b/internal/chat/session_test.go @@ -9,6 +9,7 @@ import ( "chat-agent-system/internal/chat" "chat-agent-system/internal/config" "chat-agent-system/internal/coordinator" + "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/docgov" "chat-agent-system/internal/intent" "chat-agent-system/internal/squad" @@ -312,3 +313,73 @@ func TestChatSession_AdvisorCaseRecommendation(t *testing.T) { t.Errorf("unexpected doc ID: %s", recCase.DocID) } } + +func TestChatSession_RecentDialogueAndGuidanceSync(t *testing.T) { + ctx := context.Background() + rec := intent.NewRecognizer() + engine := coordinator.NewCoordinatorEngine(nil) + ctxMgr := ctxmgr.NewContextManager(nil, nil, ctxmgr.DefaultTimeDecayConfig()) + + userCtx := &coordinator.UserProjectContext{ + SnapshotID: "sess_ctx_sync", + User: coordinator.UserInfo{ + ID: "u_sync_1", + Name: "李工", + Role: "engineer", + ProjectRole: "composer", + }, + Project: coordinator.ProjectInfo{ + Name: "数字化微创内窥镜", + CurrentStage: bizdev.StageV01_Triage, + }, + BlackboardFacts: make(map[string]string), + } + + session := chat.NewSession("sess_sync_001", userCtx, rec, engine) + session.UserID = 42 + session.ProjectID = 100 + session.CtxMgr = ctxMgr + + // Turn 1: 问候 + _, err := session.SendMessage(ctx, "你好，我是内窥镜项目负责人李工") + if err != nil { + t.Fatalf("turn 1 failed: %v", err) + } + if len(session.UserCtx.RecentDialogue) != 2 { + t.Fatalf("expected RecentDialogue to have 2 turns after turn 1, got %d", len(session.UserCtx.RecentDialogue)) + } + if session.UserCtx.RecentDialogue[0].Role != "user" || !strings.Contains(session.UserCtx.RecentDialogue[0].Content, "李工") { + t.Errorf("unexpected user dialogue turn: %+v", session.UserCtx.RecentDialogue[0]) + } + if session.UserCtx.RecentDialogue[1].Role != "assistant" { + t.Errorf("unexpected assistant dialogue turn: %+v", session.UserCtx.RecentDialogue[1]) + } + + // Turn 2: 用户表达偏好指令 ("以后请...") + _, err = session.SendMessage(ctx, "以后请直接说重点并优先关注法规标准合规") + if err != nil { + t.Fatalf("turn 2 failed: %v", err) + } + if len(session.UserCtx.RecentDialogue) != 4 { + t.Fatalf("expected RecentDialogue to have 4 turns after turn 2, got %d", len(session.UserCtx.RecentDialogue)) + } + if session.UserCtx.PersonalizedGuidance == "" { + t.Errorf("expected PersonalizedGuidance to be populated after custom instruction") + } + if !strings.Contains(session.UserCtx.PersonalizedGuidance, "Memory Injection") { + t.Errorf("expected Memory Injection header in PersonalizedGuidance, got: %s", session.UserCtx.PersonalizedGuidance) + } + + // Turn 3: 连续对话验证 + _, err = session.SendMessage(ctx, "现在项目进展到哪一步了？") + if err != nil { + t.Fatalf("turn 3 failed: %v", err) + } + if len(session.UserCtx.RecentDialogue) != 6 { + t.Fatalf("expected RecentDialogue to have 6 turns after turn 3, got %d", len(session.UserCtx.RecentDialogue)) + } + if session.UserCtx.PersonalizedGuidance == "" { + t.Errorf("PersonalizedGuidance lost on subsequent turn") + } +} + diff --git a/internal/ctxmgr/manager.go b/internal/ctxmgr/manager.go index f95b4aa1..4d7d8dba 100644 --- a/internal/ctxmgr/manager.go +++ b/internal/ctxmgr/manager.go @@ -436,6 +436,21 @@ func (m *ContextManager) BuildBlendedSquadContext(ctx context.Context, projectID }, nil } +// CompilePromptGuidance 编译指定用户的个性化偏好指导语 (供对话 LLM 或智能体 Prompt 注入) +func (m *ContextManager) CompilePromptGuidance(ctx context.Context, userID int64) (string, error) { + if userID <= 0 { + return "", errors.New("invalid userID") + } + pref, err := m.GetUserPreference(ctx, userID) + if err != nil { + return "", err + } + if pref == nil || m.reflector == nil { + return "", nil + } + return m.reflector.GeneratePromptGuidance(pref), nil +} + // GetUserPreference 获取用户偏好画像 func (m *ContextManager) GetUserPreference(ctx context.Context, userID int64) (*UserPreference, error) { if userID <= 0 { diff --git a/internal/ctxmgr/manager_test.go b/internal/ctxmgr/manager_test.go index 8e7c9e26..3628b46a 100644 --- a/internal/ctxmgr/manager_test.go +++ b/internal/ctxmgr/manager_test.go @@ -67,6 +67,15 @@ func TestContextManager_PreferenceReflection(t *testing.T) { if !strings.Contains(guidance, "PPT") || !strings.Contains(guidance, "条目清单") { t.Errorf("guidance missing ppt format instruction: %s", guidance) } + + // 5. 验证 CompilePromptGuidance 公共接口 + compiledGuidance, err := mgr.CompilePromptGuidance(ctx, userID) + if err != nil { + t.Fatalf("CompilePromptGuidance failed: %v", err) + } + if compiledGuidance != guidance { + t.Errorf("expected compiled guidance to match reflector output, got %s", compiledGuidance) + } } func TestContextManager_TimeDecayActivities(t *testing.T) { diff --git a/internal/server/chat_restore_test.go b/internal/server/chat_restore_test.go new file mode 100644 index 00000000..31215f4d --- /dev/null +++ b/internal/server/chat_restore_test.go @@ -0,0 +1,127 @@ +package server_test + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/config" + "chat-agent-system/internal/server" +) + +func TestServer_ChatRestoreAfterRestart(t *testing.T) { + cfg := &config.Config{ + HTTPHost: "127.0.0.1", + HTTPPort: 8889, + DataDir: t.TempDir(), + } + svc := bizdev.NewService(cfg) + srv := server.NewServer(cfg, svc) + handler := srv.Routes() + + sessID := "sess_restore_test_001" + userID := int64(88) + + // Turn 1: 用户发送第一条消息 + chatReq1 := map[string]any{ + "session_id": sessID, + "user_id": userID, + "project_name": "高频超声手术刀系统", + "device_type": "超声手术设备", + "message": "你好，我们正在规划立项一款高频超声手术刀系统，预算200万", + } + bodyBytes1, _ := json.Marshal(chatReq1) + req1 := httptest.NewRequest("POST", "/api/v1/chat", bytes.NewReader(bodyBytes1)) + req1.Header.Set("Content-Type", "application/json") + rr1 := httptest.NewRecorder() + handler.ServeHTTP(rr1, req1) + + if rr1.Code != http.StatusOK { + t.Fatalf("turn 1 failed with status %d: %s", rr1.Code, rr1.Body.String()) + } + + var resp1 map[string]any + if err := json.Unmarshal(rr1.Body.Bytes(), &resp1); err != nil { + t.Fatalf("unmarshal resp1 failed: %v", err) + } + reply1, _ := resp1["reply_text"].(string) + if reply1 == "" { + t.Fatalf("turn 1 reply text is empty") + } + t.Logf("✓ Turn 1 完成: %s", reply1) + + // 验证消息已被持久化至 ContextManager + msgs, err := svc.ContextManager().GetChatHistory(context.Background(), sessID, 10, 0) + if err != nil || len(msgs) != 2 { + t.Fatalf("expected 2 messages in ContextManager, got %d (err: %v)", len(msgs), err) + } + + // 模拟服务重启 / 内存淘汰 (G9 CleanStaleSessions 触发清空内存 map) + time.Sleep(5 * time.Millisecond) + cleaned := srv.CleanStaleSessions(1 * time.Millisecond) + if cleaned < 1 { + t.Fatalf("expected at least 1 session to be cleaned from memory, got %d", cleaned) + } + t.Logf("✓ 成功模拟会话从内存完全淘汰 (释放会话数=%d)", cleaned) + + // Turn 2: 服务重启后，用户携相同 session_id 再次发起追问 (G3 自动恢复) + chatReq2 := map[string]any{ + "session_id": sessID, + "user_id": userID, + "message": "关于刚才说的这个项目，国内型检和审评审批有哪些核心强标？", + } + bodyBytes2, _ := json.Marshal(chatReq2) + req2 := httptest.NewRequest("POST", "/api/v1/chat", bytes.NewReader(bodyBytes2)) + req2.Header.Set("Content-Type", "application/json") + rr2 := httptest.NewRecorder() + handler.ServeHTTP(rr2, req2) + + if rr2.Code != http.StatusOK { + t.Fatalf("turn 2 failed with status %d: %s", rr2.Code, rr2.Body.String()) + } + + var resp2 map[string]any + if err := json.Unmarshal(rr2.Body.Bytes(), &resp2); err != nil { + t.Fatalf("unmarshal resp2 failed: %v", err) + } + reply2, _ := resp2["reply_text"].(string) + if reply2 == "" { + t.Fatalf("turn 2 reply text is empty") + } + t.Logf("✓ Turn 2 完成: %s", reply2) + + // 验证完整历史记录包含全部 4 条消息 (Turn 1 user+asst, Turn 2 user+asst) + allMsgs, err := svc.ContextManager().GetChatHistory(context.Background(), sessID, 10, 0) + if err != nil { + t.Fatalf("failed to query chat history: %v", err) + } + if len(allMsgs) != 4 { + t.Fatalf("expected 4 total messages after restore, got %d", len(allMsgs)) + } + if allMsgs[0].Role != "user" || !strings.Contains(allMsgs[0].Content, "高频超声手术刀系统") { + t.Errorf("turn 1 user message mismatch: %+v", allMsgs[0]) + } + if allMsgs[2].Role != "user" || !strings.Contains(allMsgs[2].Content, "核心强标") { + t.Errorf("turn 2 user message mismatch: %+v", allMsgs[2]) + } + + // 通过 /api/v1/projects 检查恢复后的上下文事实 + reqCtx := httptest.NewRequest("GET", "/api/v1/projects?session_id="+sessID, nil) + rrCtx := httptest.NewRecorder() + handler.ServeHTTP(rrCtx, reqCtx) + if rrCtx.Code != http.StatusOK { + t.Fatalf("GET /api/v1/projects failed: %d", rrCtx.Code) + } + var ctxMap map[string]any + _ = json.Unmarshal(rrCtx.Body.Bytes(), &ctxMap) + facts, _ := ctxMap["blackboard_facts"].(map[string]any) + if facts != nil { + t.Logf("✓ 恢复后的项目黑板事实: %v", facts) + } +} diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 6887a0ed..79d0772d 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -83,11 +83,8 @@ func (s *Server) handleGetProject(w http.ResponseWriter, r *http.Request) { return } - s.sessionsMu.RLock() - sess, exists := s.sessions[sessionID] - s.sessionsMu.RUnlock() - - if !exists || sess == nil || sess.UserCtx == nil { + sess := s.getOrCreateSession(sessionID, "", "", "", "", "", "", "", "") + if sess == nil || sess.UserCtx == nil { http.Error(w, "session or project context not found", http.StatusNotFound) return } diff --git a/internal/server/server.go b/internal/server/server.go index 669f68b1..ce04afc9 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -16,6 +16,7 @@ import ( "chat-agent-system/internal/chat" "chat-agent-system/internal/config" "chat-agent-system/internal/coordinator" + "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/intent" ) @@ -193,6 +194,28 @@ func (s *Server) Shutdown(ctx context.Context) error { return nil } +// CleanStaleSessions 清理超过空闲 TTL 的内存会话 (释放内存，后续请求可通过 RestoreSessionContext 自动从 DB 恢复) +func (s *Server) CleanStaleSessions(ttl time.Duration) int { + s.sessionsMu.Lock() + defer s.sessionsMu.Unlock() + return s.cleanStaleSessionsLocked(ttl) +} + +func (s *Server) cleanStaleSessionsLocked(ttl time.Duration) int { + if ttl <= 0 { + ttl = 120 * time.Minute + } + now := time.Now() + cleaned := 0 + for id, sess := range s.sessions { + if sess == nil || now.Sub(sess.LastAccessedAt) > ttl { + delete(s.sessions, id) + cleaned++ + } + } + return cleaned +} + // getOrCreateSession 获取或动态构建会话 func (s *Server) getOrCreateSession( sessionID string, @@ -213,6 +236,7 @@ func (s *Server) getOrCreateSession( existing, exists := s.sessions[sessionID] s.sessionsMu.RUnlock() if exists && existing != nil { + existing.LastAccessedAt = time.Now() return existing } @@ -220,9 +244,23 @@ func (s *Server) getOrCreateSession( defer s.sessionsMu.Unlock() // 双重检查 if existing, exists := s.sessions[sessionID]; exists && existing != nil { + existing.LastAccessedAt = time.Now() return existing } + // 惰性内存防泄漏保护 (会话数超过 200 时清理空闲超 120 分钟会话) + if len(s.sessions) > 200 { + s.cleanStaleSessionsLocked(120 * time.Minute) + } + + // 尝试从持久层恢复会话上下文 (服务重启后跨进程保持 G3) + var restored *ctxmgr.RestoredSessionContext + if s.svc != nil && s.svc.ContextManager() != nil { + if r, err := s.svc.ContextManager().RestoreSessionContext(context.Background(), sessionID); err == nil && r != nil && len(r.Messages) > 0 { + restored = r + } + } + if userName == "" { userName = "用户" } @@ -279,6 +317,40 @@ func (s *Server) getOrCreateSession( if s.svc != nil { sess.CtxMgr = s.svc.ContextManager() } + + // 若从持久层恢复了会话历史，回填消息列表、事实黑板与近期对话 (G3) + if restored != nil { + if restored.UserID > 0 { + sess.UserID = restored.UserID + } + if restored.ProjectID > 0 { + sess.ProjectID = restored.ProjectID + } + for k, v := range restored.Facts { + if _, exists := userCtx.BlackboardFacts[k]; !exists { + userCtx.BlackboardFacts[k] = v + } + } + for _, rm := range restored.Messages { + sess.Messages = append(sess.Messages, chat.Message{ + ID: fmt.Sprintf("msg_%d", rm.ID), + Role: rm.Role, + Content: rm.Content, + Intent: intent.IntentType(rm.Intent), + Branch: intent.ResolutionBranch(rm.Branch), + Timestamp: rm.CreatedAt, + }) + } + sess.SyncRecentDialogue() + + if sess.UserID > 0 && sess.CtxMgr != nil { + if guidance, err := sess.CtxMgr.CompilePromptGuidance(context.Background(), sess.UserID); err == nil && guidance != "" { + userCtx.PersonalizedGuidance = guidance + } + } + } + + sess.LastAccessedAt = time.Now() s.sessions[sessionID] = sess return sess }