From 0f01f047d5b1233b857e4a29f14aaf341200f9fd Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Tue, 8 Sep 2026 09:14:58 +0800 Subject: [PATCH] =?UTF-8?q?feat(agent):=20=E5=9C=A8=E5=B0=8F=E9=98=9F?= =?UTF-8?q?=E7=BC=96=E6=8E=92=E4=B8=8E=E4=BB=BB=E5=8A=A1=E7=94=9F=E5=91=BD?= =?UTF-8?q?=E5=91=A8=E6=9C=9F=E4=B8=AD=E6=89=93=E9=80=9A=E9=98=B6=E6=AE=B5?= =?UTF-8?q?=E6=80=A7=E7=8A=B6=E6=80=81=E4=B8=8E=E6=91=98=E8=A6=81=E6=B5=81?= =?UTF-8?q?=E5=BC=8F=E9=80=9A=E7=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Cursor --- internal/bizdev/service.go | 26 ++++++++++ internal/chat/session.go | 46 +++++++++++++++- internal/chat/session_stream_test.go | 78 ++++++++++++++++++++++++++++ internal/coordinator/engine.go | 16 ++++++ internal/squad/roles.go | 25 +++++++++ internal/squad/service.go | 30 +++++++++++ internal/squad/workflow.go | 27 +++++++++- internal/stream/stream.go | 4 +- internal/stream/types.go | 4 +- 9 files changed, 248 insertions(+), 8 deletions(-) create mode 100644 internal/chat/session_stream_test.go diff --git a/internal/bizdev/service.go b/internal/bizdev/service.go index b2fa1d59..1feb5e13 100644 --- a/internal/bizdev/service.go +++ b/internal/bizdev/service.go @@ -12,6 +12,7 @@ import ( "chat-agent-system/internal/config" "chat-agent-system/internal/kb" "chat-agent-system/internal/squad" + "chat-agent-system/internal/stream" ) type Service struct { @@ -82,7 +83,19 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun } searchQuery := strings.TrimSpace(req.ProjectName + " " + device + " " + market) if searchQuery != "" { + streamer := stream.FromContext(ctx) + if streamer != nil { + streamer.EmitPhase("knowledge_retrieval", fmt.Sprintf("法规专家正在检索权威法规知识库并筛选适用条款（关键词: %s）...", searchQuery), "regulatory", 0, 0, 10) + } if kbHits, err := s.kb.Search(ctx, searchQuery, kb.SearchOptions{Limit: 5}); err == nil && len(kbHits) > 0 { + if streamer != nil { + streamer.EmitPhase("knowledge_retrieved", fmt.Sprintf("法规知识库初筛完成，精准命中 %d 条国家与行业标准条款", len(kbHits)), "regulatory", 0, 0, 15) + var hitLines []string + for _, h := range kbHits { + hitLines = append(hitLines, fmt.Sprintf("- [%s] %s (%s)", h.DocCode, h.Title, h.CiteKey)) + } + streamer.EmitDetail("knowledge_hits", "knowledge", "regulatory", "命中法规标准清单", strings.Join(hitLines, "\n"), true) + } for _, kh := range kbHits { hits = append(hits, squad.KnowledgeHit{ DocCode: kh.DocCode, @@ -183,6 +196,11 @@ func (s *Service) executeStageInternal(ctx context.Context, req StageRunRequest) targetMeta = &SupportedStages[0] } + streamer := stream.FromContext(ctx) + if streamer != nil { + streamer.EmitPhase("stage_started", fmt.Sprintf("开始推进商务成熟度阶段【%s】（包含 %d 项任务序列）...", req.Stage, len(targetMeta.Tasks)), "coordinator", 0, len(targetMeta.Tasks), 5) + } + // 3. 汇总当前阶段累积的事实与前序文档 currentFields := make(map[string]string) if req.Fields != nil { @@ -310,6 +328,14 @@ func (s *Service) executeStageInternal(ctx context.Context, req StageRunRequest) } } + if streamer != nil { + streamer.EmitPhase("stage_completed", fmt.Sprintf("商务成熟度阶段【%s】全部任务协同推演完成！", req.Stage), "coordinator", len(targetMeta.Tasks), len(targetMeta.Tasks), 100) + if stageBrief != nil { + streamer.EmitSummary(targetMeta.Title+"完成", stageBrief.MaturityHeadline, stageBrief.MaturityScore, stageBrief.MaturityLevel) + streamer.EmitBriefing(stageBrief) + } + } + return &StageRunResult{ Stage: req.Stage, StageTitle: targetMeta.Title, diff --git a/internal/chat/session.go b/internal/chat/session.go index 4c2c82cb..93e227ea 100644 --- a/internal/chat/session.go +++ b/internal/chat/session.go @@ -8,6 +8,7 @@ import ( "chat-agent-system/internal/coordinator" "chat-agent-system/internal/intent" "chat-agent-system/internal/squad" + "chat-agent-system/internal/stream" ) // Message 历史消息结构 @@ -64,8 +65,24 @@ func NewSession(id string, userCtx *coordinator.UserProjectContext, rec intent.R } } -// SendMessage 处理一轮用户输入 +// SendMessage 处理一轮用户输入 (标准阻塞模式) func (s *Session) SendMessage(ctx context.Context, userInput string) (*ChatResponse, error) { + streamer := stream.FromContext(ctx) + return s.sendMessageInternal(ctx, userInput, streamer) +} + +// SendMessageStream 处理一轮用户输入并实时流式推送事件 (支持 SSE / 终端流式监听) +func (s *Session) SendMessageStream(ctx context.Context, userInput string, listener stream.Listener, includeDetails bool) (*ChatResponse, error) { + ctxWithS, streamer := stream.WithListener(ctx, includeDetails, listener) + return s.sendMessageInternal(ctxWithS, userInput, streamer) +} + +func (s *Session) sendMessageInternal(ctx context.Context, userInput string, streamer *stream.Streamer) (*ChatResponse, error) { + start := time.Now() + if streamer != nil { + streamer.EmitPhase("intent_recognizing", "正在研判您的意图与项目上下文...", "coordinator", 0, 0, 0) + } + userMsg := Message{ ID: fmt.Sprintf("msg_%d_u", time.Now().UnixNano()), Role: "user", @@ -81,12 +98,28 @@ func (s *Session) SendMessage(ctx context.Context, userInput string) (*ChatRespo } recRes, err := s.Recognizer.Recognize(ctx, userInput, snap) if err != nil { + if streamer != nil { + streamer.EmitError(err) + } return nil, fmt.Errorf("intent recognition failed: %w", err) } + if streamer != nil && recRes.TopIntent != nil { + streamer.EmitStatus( + string(recRes.TopIntent.Intent), + string(recRes.Branch), + string(s.UserCtx.Project.CurrentStage), + fmt.Sprintf("识别分支: %s (置信度: %.0f%%)", recRes.Branch, recRes.TopIntent.Confidence*100), + recRes.TopIntent.FilledSlots, + ) + } + // 2. 协调调度与校验 (L2 协调层) coordRes, err := s.Coordinator.ProcessIntent(ctx, recRes, s.UserCtx) if err != nil { + if streamer != nil { + streamer.EmitError(err) + } return nil, fmt.Errorf("coordination execution failed: %w", err) } @@ -115,6 +148,17 @@ func (s *Session) SendMessage(ctx context.Context, userInput string) (*ChatRespo } s.Messages = append(s.Messages, asstMsg) + // 4. 流式推送参谋自然语言回复与决策锦囊 + if streamer != nil { + if coordRes.NaturalReply != "" { + streamer.EmitDelta(coordRes.NaturalReply) + } + if coordRes.AdvisorBrief != nil { + streamer.EmitBriefing(coordRes.AdvisorBrief) + } + streamer.EmitDone(s.ID, time.Since(start)) + } + return &ChatResponse{ ReplyText: coordRes.NaturalReply, Branch: recRes.Branch, diff --git a/internal/chat/session_stream_test.go b/internal/chat/session_stream_test.go new file mode 100644 index 00000000..5dce631d --- /dev/null +++ b/internal/chat/session_stream_test.go @@ -0,0 +1,78 @@ +package chat + +import ( + "context" + "testing" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/coordinator" + "chat-agent-system/internal/intent" + "chat-agent-system/internal/stream" +) + +func TestChatSession_SendMessageStream(t *testing.T) { + u := coordinator.UserInfo{ID: "u1", Name: "李工", Role: "engineer", ProjectRole: "composer"} + userCtx := &coordinator.UserProjectContext{ + User: u, + Project: coordinator.ProjectInfo{ + Name: "内窥镜研发项目", + CurrentStage: bizdev.StageV01_Triage, + }, + BlackboardFacts: make(map[string]string), + } + + rec := intent.NewRecognizer() + engine := coordinator.NewCoordinatorEngine(nil) + session := NewSession("test_stream_sess", userCtx, rec, engine) + + var events []stream.StreamEvent + listener := func(ev stream.StreamEvent) { + events = append(events, ev) + } + + // 1. 发送查询进度消息 + resp, err := session.SendMessageStream(context.Background(), "查一下当前进展", listener, false) + if err != nil { + t.Fatalf("SendMessageStream failed: %v", err) + } + + if resp == nil { + t.Fatalf("expected non-nil response") + } + + if len(events) == 0 { + t.Fatalf("expected streamed events, got 0") + } + + // 验证事件流序列 + hasPhase := false + hasStatus := false + hasDelta := false + hasDone := false + + for _, ev := range events { + switch ev.Type { + case stream.EventPhase: + hasPhase = true + case stream.EventStatus: + hasStatus = true + case stream.EventDelta: + hasDelta = true + case stream.EventDone: + hasDone = true + } + } + + if !hasPhase { + t.Errorf("expected at least one phase event") + } + if !hasStatus { + t.Errorf("expected at least one status event") + } + if !hasDelta { + t.Errorf("expected at least one delta event") + } + if !hasDone { + t.Errorf("expected done event") + } +} diff --git a/internal/coordinator/engine.go b/internal/coordinator/engine.go index 55b5c6f9..3ad61dc0 100644 --- a/internal/coordinator/engine.go +++ b/internal/coordinator/engine.go @@ -11,6 +11,7 @@ import ( "chat-agent-system/internal/bizdev" "chat-agent-system/internal/intent" "chat-agent-system/internal/squad" + "chat-agent-system/internal/stream" ) // CoordinationStatus 协调裁决状态 @@ -71,6 +72,11 @@ func (e *CoordinatorEngine) ProcessIntent( } } + streamer := stream.FromContext(ctx) + if streamer != nil { + streamer.EmitStatus(string(top.Intent), string(recRes.Branch), string(userCtx.Project.CurrentStage), fmt.Sprintf("意图判定: %s (置信度: %.0f%%)", top.Intent, top.Confidence*100), top.FilledSlots) + } + // 1. 若为分支 3 (模糊/分忧引导)，触发参谋主动分析与建议 if recRes.Branch == intent.BranchAmbiguousFallback { return e.handleAmbiguousConsult(top, userCtx) @@ -292,6 +298,11 @@ func (e *CoordinatorEngine) handleSupplementInfo(cand *intent.CandidateIntent, u summaryKey = "已补充新的背景信息" } + streamer := stream.FromContext(context.Background()) + if streamer != nil { + streamer.EmitStatus(string(cand.Intent), string(cand.Branch), string(userCtx.Project.CurrentStage), fmt.Sprintf("已吸收最新事实设定：%s", summaryKey), cand.FilledSlots) + } + // 持久化到 blackboard.json if userCtx.WorkDir != "" { bbPath := filepath.Join(userCtx.WorkDir, "blackboard.json") @@ -348,6 +359,11 @@ func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent targetStage = bizdev.StageV05_Feasibility } + streamer := stream.FromContext(ctx) + if streamer != nil { + streamer.EmitPhase("stage_transition", fmt.Sprintf("准备推进项目成熟度阶段至【%s】...", targetStage), "coordinator", 0, 0, 0) + } + // 1. 若携带补丁槽位 (例如 "好，但目标市场改成欧洲CE")，先行持久化更新 patchSummary := "" if len(cand.PatchSlots) > 0 { diff --git a/internal/squad/roles.go b/internal/squad/roles.go index 437f0094..1bbefada 100644 --- a/internal/squad/roles.go +++ b/internal/squad/roles.go @@ -270,3 +270,28 @@ func (r *RoleRegistry) SpawnAdHocRole(req AdHocRoleRequest) (AgentSpec, error) { r.roles[adHocSpec.Role] = adHocSpec return adHocSpec, nil } + +// FriendlyRoleName 将内部角色代号转换为友好的中文专员称谓 +func FriendlyRoleName(role string) string { + switch role { + case "organizer": + return "材料整理专员" + case "searcher": + return "资料检索专员" + case "regulatory": + return "法规专员" + case "engineer": + return "系统工程师" + case "market": + return "市场情报专员" + case "finance": + return "商务报价专员" + case "writer": + return "撰写专员" + case "leader": + return "审核主管" + default: + return role + "专员" + } +} + diff --git a/internal/squad/service.go b/internal/squad/service.go index 086f5e1a..612a1c9d 100644 --- a/internal/squad/service.go +++ b/internal/squad/service.go @@ -12,6 +12,7 @@ import ( "chat-agent-system/internal/agent" "chat-agent-system/internal/config" "chat-agent-system/internal/quote" + "chat-agent-system/internal/stream" ) // TaskMeta 任务元信息 @@ -210,6 +211,11 @@ func (s *Service) ExecuteTask(ctx context.Context, req SquadRunRequest) (*SquadR } // 3. 执行动态工作流调度 + streamer := stream.FromContext(ctx) + if streamer != nil { + streamer.EmitPhase("squad_assembly", fmt.Sprintf("正在组建并启动【%s】智能体小队（后端: %s）...", TaskArtifactTitle(req.TaskKey), backendName), "leader", 0, 0, 5) + } + backend := agent.NewBackend(backendName, s.cfg) timeout := req.Timeout if timeout <= 0 { @@ -222,6 +228,9 @@ func (s *Service) ExecuteTask(ctx context.Context, req SquadRunRequest) (*SquadR execRes, err := s.driver.Execute(ctx, workDir, backend, execOpts) if err != nil { + if streamer != nil { + streamer.EmitPhase("task_failed", fmt.Sprintf("任务执行异常: %v", err), "leader", 0, 0, 0) + } return &SquadRunResult{ TaskKey: req.TaskKey, ArtifactTitle: TaskArtifactTitle(req.TaskKey), @@ -234,6 +243,9 @@ func (s *Service) ExecuteTask(ctx context.Context, req SquadRunRequest) (*SquadR } // 4. 自动化质量门禁与修复 (QualityGate) + if streamer != nil { + streamer.EmitPhase("quality_gate", "审核主管正在核验交付物规范并执行质量门禁把关...", "leader", 0, 0, 85) + } maxRetries := 0 if req.AutoRepair { maxRetries = 2 @@ -242,6 +254,11 @@ func (s *Service) ExecuteTask(ctx context.Context, req SquadRunRequest) (*SquadR if err != nil { // 记录门禁错误但容错返回当前正文 report = &QualityReport{Passed: false, Violations: []string{err.Error()}} + if streamer != nil { + streamer.EmitPhase("quality_gate", fmt.Sprintf("质量门禁审查告警: %v", err), "leader", 0, 0, 90) + } + } else if streamer != nil { + streamer.EmitPhase("quality_gate", fmt.Sprintf("质量门禁审核通过（评分: %.1f / 100）", report.Score), "leader", 0, 0, 95) } manifest, _ := s.gate.GenerateManifest(workDir, req.TaskKey, *report) @@ -286,6 +303,19 @@ func (s *Service) ExecuteTask(ctx context.Context, req SquadRunRequest) (*SquadR } } + if streamer != nil { + score := 0.0 + if report != nil { + score = report.Score + } + streamer.EmitDeliverable(req.TaskKey, TaskArtifactTitle(req.TaskKey), finalOutPath, len(strings.Fields(string(finalBytes))), score) + if advisorBrief != nil { + streamer.EmitPhase("advisor_briefing", "贴身业务参谋正在提炼关键决策摘要与风险雷达...", "coordinator", 0, 0, 98) + streamer.EmitSummary(TaskArtifactTitle(req.TaskKey)+"已生成", advisorBrief.MaturityHeadline, advisorBrief.MaturityScore, advisorBrief.MaturityLevel) + streamer.EmitBriefing(advisorBrief) + } + } + return &SquadRunResult{ TaskKey: req.TaskKey, ArtifactTitle: TaskArtifactTitle(req.TaskKey), diff --git a/internal/squad/workflow.go b/internal/squad/workflow.go index 2c91e22b..19718923 100644 --- a/internal/squad/workflow.go +++ b/internal/squad/workflow.go @@ -11,6 +11,7 @@ import ( "time" "chat-agent-system/internal/agent" + "chat-agent-system/internal/stream" ) // StepEvent 表示小队单步执行生命周期的事件 @@ -275,7 +276,7 @@ func (d *WorkflowDriver) Execute(ctx context.Context, wsDir string, backend agen _ = d.SaveWorkflow(wsDir, wf) if step.AllowDynamicAdjustment && d.evaluator != nil { - d.applyDynamicAdjustment(wsDir, wf, step, &mu, opts.StepNotify) + d.applyDynamicAdjustment(ctx, wsDir, wf, step, &mu, opts.StepNotify) } } } @@ -346,6 +347,11 @@ func (d *WorkflowDriver) runOneStep( Status: "started", }) } + streamer := stream.FromContext(ctx) + roleTitle := FriendlyRoleName(step.Role) + if streamer != nil { + streamer.EmitPhase("step_executing", fmt.Sprintf("%s正在进行：%s...", roleTitle, step.Title), step.Role, 0, 0, 0) + } stepStart := time.Now() // 若当前步骤为 leader 审核定稿，且依赖前序专员稿件，先预置 baseline 草稿，供审查与润色 @@ -380,6 +386,9 @@ func (d *WorkflowDriver) runOneStep( duration := time.Since(stepStart).Milliseconds() if err != nil { + if streamer != nil { + streamer.EmitPhase("step_failed", fmt.Sprintf("%s执行异常：%s", roleTitle, err.Error()), step.Role, 0, 0, 0) + } if notify != nil { notify(StepEvent{ StepID: step.ID, @@ -396,6 +405,9 @@ func (d *WorkflowDriver) runOneStep( if res != nil && !res.Success { errMsg := res.ErrorReason + if streamer != nil { + streamer.EmitPhase("step_failed", fmt.Sprintf("%s执行未成功：%s", roleTitle, errMsg), step.Role, 0, 0, 0) + } if notify != nil { notify(StepEvent{ StepID: step.ID, @@ -449,6 +461,13 @@ func (d *WorkflowDriver) runOneStep( *producedFiles = append(*producedFiles, filepath.ToSlash(filepath.Join(step.SubWorkspace, deliverableName))) mu.Unlock() + if streamer != nil { + streamer.EmitPhase("step_completed", fmt.Sprintf("%s完成任务：%s (耗时 %dms)", roleTitle, step.Title, duration), step.Role, 0, 0, 0) + if stepData, err := os.ReadFile(targetPath); err == nil && len(stepData) > 0 { + streamer.EmitDetail("step_output", step.ID, step.Role, fmt.Sprintf("%s阶段性产出: %s", roleTitle, step.Title), string(stepData), true) + } + } + if notify != nil { notify(StepEvent{ StepID: step.ID, @@ -464,6 +483,7 @@ func (d *WorkflowDriver) runOneStep( } func (d *WorkflowDriver) applyDynamicAdjustment( + ctx context.Context, wsDir string, wf *WorkflowPlan, step StepSpec, @@ -571,6 +591,11 @@ func (d *WorkflowDriver) applyDynamicAdjustment( newSteps = append(newSteps, addStep) newSteps = append(newSteps, wf.Steps[insertIdx:]...) wf.Steps = newSteps + + streamer := stream.FromContext(ctx) + if streamer != nil { + streamer.EmitPhase("dynamic_adjustment", fmt.Sprintf("单步评估发现缺口，动态追加任务步骤：%s (%s)", addStep.Title, FriendlyRoleName(addStep.Role)), addStep.Role, 0, 0, 0) + } } _ = d.SaveWorkflow(wsDir, wf) } diff --git a/internal/stream/stream.go b/internal/stream/stream.go index f193c175..571e7da3 100644 --- a/internal/stream/stream.go +++ b/internal/stream/stream.go @@ -6,8 +6,6 @@ import ( "sync" "sync/atomic" "time" - - "chat-agent-system/internal/squad" ) type contextKey struct{} @@ -125,7 +123,7 @@ func (s *Streamer) EmitSummary(title, content string, maturityScore int, maturit } // EmitBriefing 发送 B 轨业务参谋决策锦囊 -func (s *Streamer) EmitBriefing(brief *squad.AdvisorBriefing) { +func (s *Streamer) EmitBriefing(brief any) { if s == nil || brief == nil { return } diff --git a/internal/stream/types.go b/internal/stream/types.go index 43788a15..0558d561 100644 --- a/internal/stream/types.go +++ b/internal/stream/types.go @@ -2,8 +2,6 @@ package stream import ( "time" - - "chat-agent-system/internal/squad" ) // EventType 定义流式事件类型 @@ -83,7 +81,7 @@ type StreamEvent struct { Phase *PhasePayload `json:"phase,omitempty"` Status *StatusPayload `json:"status,omitempty"` Summary *SummaryPayload `json:"summary,omitempty"` - Briefing *squad.AdvisorBriefing `json:"briefing,omitempty"` + Briefing any `json:"briefing,omitempty"` Detail *DetailPayload `json:"detail,omitempty"` Delta string `json:"delta,omitempty"` Deliverable *DeliverablePayload `json:"deliverable,omitempty"`