weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(chat): wire multi-turn dialogue broken links (G1/G2/G3/G9) with session restore

Browse Source
This commit is contained in:
WeiCHEN
2026-09-13 02:50:20 +00:00
committed by weichen
parent e590dcfd7b
commit be659cba11
7 changed files with 370 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/chat/session.go
+74 -18
View File
@@ -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 != "" {
internal/chat/session_test.go
+71
View File
@@ -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")
}
}
internal/ctxmgr/manager.go
+15
View File
@@ -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 {
internal/ctxmgr/manager_test.go
+9
View File
@@ -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) {
internal/server/chat_restore_test.go
+127
View File
@@ -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)
}
}
internal/server/handlers.go
+2 -5
View File
@@ -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
}
internal/server/server.go
+72
View File
@@ -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
}