From 8f29d84908b260bb090e346bb343d4b29b7927c7 Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Wed, 9 Sep 2026 16:16:25 +0800 Subject: [PATCH] feat(agent): emit dialogue phases and classified SSE errors with request id Co-authored-by: Cursor --- internal/chat/session.go | 61 +++++++- internal/coordinator/dialogue_llm.go | 218 +++++++++++++++++++++++++++ internal/server/sse.go | 18 ++- internal/stream/context.go | 21 +++ internal/stream/error_json_test.go | 47 ++++++ 5 files changed, 354 insertions(+), 11 deletions(-) create mode 100644 internal/coordinator/dialogue_llm.go create mode 100644 internal/stream/context.go create mode 100644 internal/stream/error_json_test.go diff --git a/internal/chat/session.go b/internal/chat/session.go index f9df68bd..84becbec 100644 --- a/internal/chat/session.go +++ b/internal/chat/session.go @@ -3,6 +3,7 @@ package chat import ( "context" "fmt" + "sync" "time" "chat-agent-system/internal/coordinator" @@ -27,6 +28,7 @@ type Message struct { // ChatResponse 会话单轮响应 type ChatResponse struct { ReplyText string `json:"reply_text"` + ReplySource string `json:"reply_source,omitempty"` // pi-agent | template Branch intent.ResolutionBranch `json:"branch"` IntentResult *intent.IntentRecognitionResult `json:"intent_result"` CoordinationResult *coordinator.CoordinationResult `json:"coordination_result"` @@ -42,10 +44,12 @@ type Session struct { ProjectID int64 `json:"project_id"` UserCtx *coordinator.UserProjectContext `json:"user_ctx"` Recognizer intent.Recognizer `json:"-"` - Coordinator *coordinator.CoordinatorEngine `json:"-"` + Coordinator *coordinator.CoordinatorEngine `json:"-"` CtxMgr *ctxmgr.ContextManager `json:"-"` Messages []Message `json:"messages"` CreatedAt time.Time `json:"created_at"` + + mu sync.Mutex `json:"-"` // 同 session 消息串行化，保障并发正确性 } // NewSession 创建新对话会话 @@ -86,6 +90,9 @@ func (s *Session) SendMessageStream(ctx context.Context, userInput string, liste } func (s *Session) sendMessageInternal(ctx context.Context, userInput string, streamer *stream.Streamer) (*ChatResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + start := time.Now() if streamer != nil { streamer.EmitPhase("intent_recognizing", "正在研判您的意图与项目上下文...", "coordinator", 0, 0, 0) @@ -100,7 +107,8 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str s.Messages = append(s.Messages, userMsg) if s.CtxMgr != nil { - _ = s.CtxMgr.RecordUserMessage(ctx, s.ID, s.UserID, "user", userInput, "", "", nil) + userMeta := map[string]any{"project_id": s.ProjectID} + _ = s.CtxMgr.RecordUserMessage(ctx, s.ID, s.UserID, "user", userInput, "", "", userMeta) } // 1. 意图识别 (L1 对话层，带非阻塞桌角备忘快照) @@ -111,7 +119,7 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str recRes, err := s.Recognizer.Recognize(ctx, userInput, snap) if err != nil { if streamer != nil { - streamer.EmitError(err) + streamer.EmitErrorInfo(stream.ClassifyError(err, "intent_recognizing", stream.RequestIDFromContext(ctx), s.ID, "")) } return nil, fmt.Errorf("intent recognition failed: %w", err) } @@ -126,16 +134,35 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str ) } - // 2. 协调调度与校验 (L2 协调层) + // 2. 注入近轮历史与用户习惯偏好，供对话 LLM（不含本轮刚追加的用户消息，本轮由 RawInput 传入） + if s.UserCtx != nil { + s.UserCtx.RecentDialogue = s.collectRecentDialogue(8) + if s.CtxMgr != nil && s.UserID > 0 { + if pref, err := s.CtxMgr.GetUserPreference(ctx, s.UserID); err == nil && pref != nil { + s.UserCtx.PersonalizedGuidance = ctxmgr.NewPreferenceReflector().GeneratePromptGuidance(pref) + if pref.Name != "" && s.UserCtx.User.Name == "" { + s.UserCtx.User.Name = pref.Name + } + if pref.Role != "" && s.UserCtx.User.Role == "" { + s.UserCtx.User.Role = pref.Role + } + } + } + } + + // 3. 协调调度与校验 (L2 协调层) + if streamer != nil { + streamer.EmitPhase("dialogue_generating", "正在整理回复...", "assistant", 0, 0, 40) + } coordRes, err := s.Coordinator.ProcessIntent(ctx, recRes, s.UserCtx) if err != nil { if streamer != nil { - streamer.EmitError(err) + streamer.EmitErrorInfo(stream.ClassifyError(err, "dialogue_generating", stream.RequestIDFromContext(ctx), s.ID, "")) } return nil, fmt.Errorf("coordination execution failed: %w", err) } - // 3. 更新上下文与挂起预期生命周期 + // 4. 更新上下文与挂起预期生命周期 if coordRes.UpdatedContext != nil { s.UserCtx = coordRes.UpdatedContext } @@ -162,10 +189,11 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str s.Messages = append(s.Messages, asstMsg) if s.CtxMgr != nil { - _ = s.CtxMgr.RecordUserMessage(ctx, s.ID, 0, "assistant", coordRes.NaturalReply, string(coordRes.Intent), string(recRes.Branch), nil) + asstMeta := map[string]any{"project_id": s.ProjectID} + _ = s.CtxMgr.RecordUserMessage(ctx, s.ID, 0, "assistant", coordRes.NaturalReply, string(coordRes.Intent), string(recRes.Branch), asstMeta) } - // 4. 流式推送参谋自然语言回复与决策锦囊 + // 5. 流式推送参谋自然语言回复与决策锦囊 if streamer != nil { if coordRes.NaturalReply != "" { streamer.EmitDelta(coordRes.NaturalReply) @@ -178,6 +206,7 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str return &ChatResponse{ ReplyText: coordRes.NaturalReply, + ReplySource: coordRes.ReplySource, Branch: recRes.Branch, IntentResult: recRes, CoordinationResult: coordRes, @@ -186,3 +215,19 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str PendingExpectation: s.UserCtx.PendingExpectation, }, nil } + +// collectRecentDialogue 收集本轮之前的近 N 条消息供 LLM 上下文（排除末尾刚写入的本轮用户消息） +func (s *Session) collectRecentDialogue(limit int) []coordinator.DialogueTurn { + if limit <= 0 || len(s.Messages) <= 1 { + return nil + } + prior := s.Messages[:len(s.Messages)-1] + if len(prior) > limit { + prior = prior[len(prior)-limit:] + } + out := make([]coordinator.DialogueTurn, 0, len(prior)) + for _, m := range prior { + out = append(out, coordinator.DialogueTurn{Role: m.Role, Content: m.Content}) + } + return out +} diff --git a/internal/coordinator/dialogue_llm.go b/internal/coordinator/dialogue_llm.go new file mode 100644 index 00000000..7d5bd117 --- /dev/null +++ b/internal/coordinator/dialogue_llm.go @@ -0,0 +1,218 @@ +package coordinator + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + "time" + + "chat-agent-system/internal/agent" + "chat-agent-system/internal/stream" +) + +const ( + ReplySourcePiAgent = "pi-agent" + ReplySourceTemplate = "template" + dialogueHistoryMax = 8 + dialogueTimeout = 3 * time.Minute +) + +// DialogueTurn 多轮对话历史片段（不落盘到 JSON 快照） +type DialogueTurn struct { + Role string `json:"role"` + Content string `json:"content"` +} + +// generateDialogueReply 通过 pi-agent 生成参谋自然语言答复；失败返回空串由调用方回退模板。 +func (e *CoordinatorEngine) generateDialogueReply( + ctx context.Context, + userInput string, + userCtx *UserProjectContext, +) (reply string, source string) { + if e == nil || e.dialogueBackend == nil || strings.TrimSpace(userInput) == "" { + return "", ReplySourceTemplate + } + + workRoot := e.dataDir + if workRoot == "" { + workRoot = "./data/dev" + } + sessionKey := "anon" + if userCtx != nil && userCtx.SnapshotID != "" { + sessionKey = userCtx.SnapshotID + } else if userCtx != nil && userCtx.Project.ID != "" { + sessionKey = userCtx.Project.ID + } + // 每轮独立子目录，避免复用上一轮残留的 reply.md（曾导致「问驳回却答疲劳寿命」） + workDir := filepath.Join( + workRoot, + "workspaces", + "dialogue", + sanitizePathSegment(sessionKey), + fmt.Sprintf("turn_%d", time.Now().UnixNano()), + ) + if err := os.MkdirAll(workDir, 0o755); err != nil { + return "", ReplySourceTemplate + } + + prompt := buildDialoguePrompt(userInput, userCtx) + instructions := `你是「水木东方」医疗器械「项目助手」——温和、靠谱、会替用户着想的项目助理。 +定位：以助理身份与用户友好协作。能力覆盖商务研判、工程方案与法规合规，但对外只以「项目助手」自称，不要分饰多角色，也不要自称「小队/参谋/总工/法规专家」。 + +协作立场（优先遵守）： +1. 站在用户视角：结合其姓名/角色、当前阶段、待办、近期对话与已知习惯偏好，揣摩本轮提问背后想达成的效果（如准备汇报、消解风险、推进文档、对外沟通、弄清技术点）。 +2. 先回应核心诉求，再补一两句温和的辅助内容：可落地的下一步、需留意的点，或「如果您接下来是想…，我可以帮您…」。 +3. 主动但轻量地给建议：通常给 1～2 个最贴合的选项，用商量口吻；不强推、不推销阶段、不堆选项清单。 +4. 若上下文含「用户个性化工作习惯与偏好」，优先遵循其沟通风格与关注重点来组织答复。 + +回复风格： +1. 语气温和、友好、自然；可用适度口语与关心式过渡（如「好的」「我帮您看一下」），保持专业分寸。 +2. 严谨有重点：先结论/直接回答，再补必要依据；篇幅适中——有实质内容，但不冗长堆砌。 +3. 直接回答本轮问题；闲聊可轻松回应，并可自然关联当前项目。 +4. 技术机理、法规、工程问题须给出有实质内容的专业解答，不要回避成阶段推销话术。 +5. 若用户询问受控文档驳回/退回原因或整改要点，必须紧扣本轮问题与已知驳回理由作答，禁止复述无关的上一轮技术长文。 +6. 不要声称已修改系统配置、用户身份或受控文档状态；不要伪造已执行的工作流结果。 +7. 只把最终答复写入 reply.md，使用中文，不要输出无关元数据。` + + if st := stream.FromContext(ctx); st != nil { + st.EmitPhase("dialogue_generating", "正在生成项目助手回复...", "assistant", 0, 0, 60) + } + + runCtx, cancel := context.WithTimeout(ctx, dialogueTimeout) + defer cancel() + + startedAt := time.Now() + res, err := e.dialogueBackend.Execute(runCtx, agent.Request{ + WorkDir: workDir, + Prompt: prompt, + Instructions: instructions, + Targets: []string{"reply.md"}, + Timeout: dialogueTimeout, + }) + if err != nil || res == nil || !res.Success { + return "", ReplySourceTemplate + } + + content := "" + if b, ok := res.Patches["reply.md"]; ok { + content = string(b) + } + // 仅当本轮执行后新写入的 reply.md 才可读盘，禁止吃到历史残留 + if strings.TrimSpace(content) == "" { + replyPath := filepath.Join(workDir, "reply.md") + if info, statErr := os.Stat(replyPath); statErr == nil && !info.ModTime().Before(startedAt.Add(-2*time.Second)) { + if raw, readErr := os.ReadFile(replyPath); readErr == nil { + content = string(raw) + } + } + } + content = strings.TrimSpace(content) + if content == "" { + return "", ReplySourceTemplate + } + return content, ReplySourcePiAgent +} + +func buildDialoguePrompt(userInput string, userCtx *UserProjectContext) string { + var sb strings.Builder + sb.WriteString("【项目上下文】\n") + if userCtx != nil { + proj := userCtx.Project + name := proj.Name + if name == "" { + name = "未命名医疗器械项目" + } + sb.WriteString(fmt.Sprintf("- 项目名称: %s\n", name)) + if userCtx.User.Name != "" { + sb.WriteString(fmt.Sprintf("- 当前用户: %s", userCtx.User.Name)) + if userCtx.User.Role != "" { + sb.WriteString(fmt.Sprintf("（%s）", userCtx.User.Role)) + } + if userCtx.User.ProjectRole != "" { + sb.WriteString(fmt.Sprintf(" / 项目角色: %s", userCtx.User.ProjectRole)) + } + sb.WriteString("\n") + } + if proj.CustomerName != "" { + sb.WriteString(fmt.Sprintf("- 客户: %s\n", proj.CustomerName)) + } + if proj.DeviceType != "" { + sb.WriteString(fmt.Sprintf("- 器械类型: %s\n", proj.DeviceType)) + } + if proj.CurrentStage != "" { + sb.WriteString(fmt.Sprintf("- 当前阶段: %s\n", proj.CurrentStage)) + } + if len(userCtx.Pending.ConfirmModifications) > 0 || + len(userCtx.Pending.PendingReviews) > 0 || + len(userCtx.Pending.PendingApprovals) > 0 || + len(userCtx.Pending.MyDraftsInProgress) > 0 { + sb.WriteString("- 用户侧待办线索:\n") + for _, x := range userCtx.Pending.ConfirmModifications { + sb.WriteString(fmt.Sprintf(" - 待确认修改: %s\n", x)) + } + for _, x := range userCtx.Pending.PendingReviews { + sb.WriteString(fmt.Sprintf(" - 待审核: %s\n", x)) + } + for _, x := range userCtx.Pending.PendingApprovals { + sb.WriteString(fmt.Sprintf(" - 待批准: %s\n", x)) + } + for _, x := range userCtx.Pending.MyDraftsInProgress { + sb.WriteString(fmt.Sprintf(" - 编制中: %s\n", x)) + } + } + if userCtx.PendingExpectation != nil && userCtx.PendingExpectation.ContextSnippet != "" { + sb.WriteString(fmt.Sprintf("- 上一轮挂起建议摘要: %s\n", userCtx.PendingExpectation.ContextSnippet)) + } + if len(userCtx.BlackboardFacts) > 0 { + sb.WriteString("- 已沉淀事实:\n") + n := 0 + for k, v := range userCtx.BlackboardFacts { + sb.WriteString(fmt.Sprintf(" - %s = %s\n", k, v)) + n++ + if n >= 12 { + break + } + } + } + if g := strings.TrimSpace(userCtx.PersonalizedGuidance); g != "" { + sb.WriteString("\n") + sb.WriteString(g) + if !strings.HasSuffix(g, "\n") { + sb.WriteString("\n") + } + } + if len(userCtx.RecentDialogue) > 0 { + sb.WriteString("\n【近期对话】\n") + turns := userCtx.RecentDialogue + if len(turns) > dialogueHistoryMax { + turns = turns[len(turns)-dialogueHistoryMax:] + } + for _, t := range turns { + role := t.Role + if role == "" { + role = "user" + } + sb.WriteString(fmt.Sprintf("%s: %s\n", role, t.Content)) + } + } + } + sb.WriteString("\n【本轮用户】\n") + sb.WriteString(userInput) + sb.WriteString("\n\n请结合上下文，以温和助理口吻作答，并写入 reply.md。") + return sb.String() +} + +func sanitizePathSegment(s string) string { + s = strings.TrimSpace(s) + if s == "" { + return "anon" + } + replacer := strings.NewReplacer("/", "_", "\\", "_", "..", "_", " ", "_") + out := replacer.Replace(s) + if len(out) > 80 { + out = out[:80] + } + return out +} diff --git a/internal/server/sse.go b/internal/server/sse.go index 0afd0ed7..fb167e8f 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -65,7 +65,9 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { return } - // 1. 设置 SSE 响应头 + // 1. 设置 SSE 响应头 + 请求追踪 ID + requestID := fmt.Sprintf("req_%d", time.Now().UnixNano()) + w.Header().Set("X-Request-Id", requestID) w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") @@ -73,6 +75,8 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { w.Header().Set("X-Accel-Buffering", "no") // 禁用 Nginx 等反代缓冲 flusher.Flush() + ctx := stream.WithRequestID(r.Context(), requestID) + // 2. 准备会话 sess := s.getOrCreateSession( req.SessionID, @@ -85,6 +89,12 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { req.UserName, req.UserRole, ) + if req.UserID > 0 { + sess.UserID = req.UserID + } + if req.ProjectID > 0 { + sess.ProjectID = req.ProjectID + } // 3. 构建线程安全的 SSE 监听分发闭包 var writeMu sync.Mutex @@ -102,13 +112,15 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { } // 4. 发起流式执行 - _, err := sess.SendMessageStream(r.Context(), req.Message, listener, req.IncludeDetails) + _, err := sess.SendMessageStream(ctx, req.Message, listener, req.IncludeDetails) if err != nil { writeMu.Lock() + info := stream.ClassifyError(err, "chat_stream", requestID, sess.ID, "") errEv := stream.StreamEvent{ Type: stream.EventError, Timestamp: time.Now(), - Error: err.Error(), + Error: info.UserMessage, + ErrorInfo: &info, } data, _ := json.Marshal(errEv) fmt.Fprintf(w, "event: %s\ndata: %s\n\n", errEv.Type, string(data)) diff --git a/internal/stream/context.go b/internal/stream/context.go new file mode 100644 index 00000000..07f374ed --- /dev/null +++ b/internal/stream/context.go @@ -0,0 +1,21 @@ +package stream + +import "context" + +type requestIDKey struct{} + +// WithRequestID 将请求追踪 ID 注入 Context，供下游分类错误与日志关联。 +func WithRequestID(ctx context.Context, id string) context.Context { + return context.WithValue(ctx, requestIDKey{}, id) +} + +// RequestIDFromContext 从 Context 提取请求追踪 ID，未设置时返回空串。 +func RequestIDFromContext(ctx context.Context) string { + if ctx == nil { + return "" + } + if v, ok := ctx.Value(requestIDKey{}).(string); ok { + return v + } + return "" +} diff --git a/internal/stream/error_json_test.go b/internal/stream/error_json_test.go new file mode 100644 index 00000000..989d1998 --- /dev/null +++ b/internal/stream/error_json_test.go @@ -0,0 +1,47 @@ +package stream + +import ( + "encoding/json" + "errors" + "testing" +) + +// TestErrorEventJSONIncludesErrorInfo 校验 ErrorInfo 经 StreamEvent 序列化后 +// 前端可从 error_info 字段读取到分类元数据（含 request_id）。 +func TestErrorEventJSONIncludesErrorInfo(t *testing.T) { + info := ClassifyError(errors.New("context deadline exceeded"), "dialogue_generating", "req_x", "sess_y", "") + ev := StreamEvent{Type: EventError, Error: info.UserMessage, ErrorInfo: &info} + + b, err := json.Marshal(ev) + if err != nil { + t.Fatal(err) + } + var raw map[string]any + if err := json.Unmarshal(b, &raw); err != nil { + t.Fatal(err) + } + ei, ok := raw["error_info"].(map[string]any) + if !ok { + t.Fatalf("missing error_info in %s", string(b)) + } + if ei["request_id"] != "req_x" { + t.Fatalf("request_id=%v", ei["request_id"]) + } + if ei["phase_id"] != "dialogue_generating" { + t.Fatalf("phase_id=%v", ei["phase_id"]) + } + if ei["class"] != string(ErrorClassTransient) { + t.Fatalf("class=%v", ei["class"]) + } +} + +// TestWithRequestID_RoundTrip 校验请求 ID 可经 Context 注入并读回。 +func TestWithRequestID_RoundTrip(t *testing.T) { + ctx := WithRequestID(t.Context(), "req_abc") + if got := RequestIDFromContext(ctx); got != "req_abc" { + t.Fatalf("got=%s want=req_abc", got) + } + if got := RequestIDFromContext(nil); got != "" { + t.Fatalf("nil ctx should return empty, got=%s", got) + } +}