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

feat(agent): 在小队编排与任务生命周期中打通阶段性状态与摘要流式通知

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
WeiCHEN
2026-09-13 02:50:19 +00:00
committed by weichen
co-authored by Cursor
parent 1bdea54022
commit 0f01f047d5
9 changed files with 248 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/bizdev/service.go
+26
View File
@@ -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,
internal/chat/session.go
+45 -1
View File
@@ -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,
internal/chat/session_stream_test.go
+78
View File
@@ -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")
}
}
internal/coordinator/engine.go
+16
View File
@@ -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 {
internal/squad/roles.go
+25
View File
@@ -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 + "专员"
}
}
internal/squad/service.go
+30
View File
@@ -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),
internal/squad/workflow.go
+26 -1
View File
@@ -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)
}
internal/stream/stream.go
+1 -3
View File
@@ -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
}
internal/stream/types.go
+1 -3
View File
@@ -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"`