feat(agent): emit dialogue phases and classified SSE errors with request id

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
WeiCHEN
2026-09-09 16:16:25 +08:00
co-authored by Cursor
parent bfa68892ac
commit 8f29d84908
5 changed files with 354 additions and 11 deletions
+53 -8
View File
@@ -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
}
+218
View File
@@ -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
}
+15 -3
View File
@@ -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))
+21
View File
@@ -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 ""
}
+47
View File
@@ -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)
}
}