Workflow-Skills 架构 vs 普通管道推进—多 Agent 协作视角的架构解析

// ===== 多 Agent 注册 =====
// 每个 Agent 通过 skillId 注册到 SkillFlowEngine
// 引擎运行时通过 skillId 查找到对应的 ActivityExecutor 并执行

@PostConstruct
public void init() {
    // RAG 知识查询 Agent（本地）
    skillFlowEngine.registerActivityExecutor("rag-knowledge-query",
        this::executeRagQuery);

    // DDD 设计 Agent（本地）
    skillFlowEngine.registerActivityExecutor("ddd-infer",
        this::executeDddInfer);

    // 代码生成 Agent（远程—通过 RemoteWorkflowBridge 桥接）
    skillFlowEngine.registerActivityExecutor("codegen-entity",
        ctx -> remoteCodeGenAgent(ctx, "entity"));

    // 代码审查 Agent（远程）
    skillFlowEngine.registerActivityExecutor("code-review",
        ctx -> remoteCodeReviewAgent(ctx));
}

// ===== Agent 执行器实现 =====
private Map<String, Object> executeRagQuery(SkillExecutionContext ctx) {
    String query = ctx.getContextValue("userInput");
    // 调用 RAG 知识库查询
    List<TableSchema> schemas = ragService.queryTableSchema(query);
    // 将结果写入 VFS 共享上下文
    vfsClient.write("/agents/rag/analysis.json", schemas);
    return Map.of("status", "completed", "result", schemas);
}

private Map<String, Object> executeDddInfer(SkillExecutionContext ctx) {
    // 从 VFS 读取 RAG Agent 的产出
    String analysisJson = vfsClient.read("/agents/rag/analysis.json");
    // 执行 DDD 领域分析
    DddDesign design = dddService.inferDesign(analysisJson);
    // 写入 VFS 供下一个 Agent 读取
    vfsClient.write("/agents/ddd/design.json", design);
    return Map.of("status", "completed");
}

// ===== 普通管道：Agent 间硬编码紧耦合 =====
// 所有 Agent 逻辑在同一个 Orchestrator 中硬编码串联
// 无法跨进程，无法独立部署，一个 Agent 升级要改整个编排器

public void orchestrate(...) {
    // Stage 0: 意图分析（内嵌在编排器中）
    String intent = analyzeIntent(userInput);

    // Stage 1: 数据库连接（硬编码调用）
    Connection conn = connectDatabase(config);

    // Stage 2: 库表分析（同步阻塞）
    List<TableSchema> tables = analyzeTables(conn);

    // Stage 3: 关系分析（方法直接调用）
    List<Relation> relations = analyzeRelations(tables);

    // Stage 4: DDD 建议（紧耦合类调用）
    DddDesign design = dddService.suggestDesign(tables, relations);
    // ... 后续阶段同样硬编码
}

// ===== VFS 共享上下文 =====
// Agent A 写入产物
public class RagAgent {
    public void execute(SkillExecutionContext ctx) {
        // 查询数据库表结构知识
        AnalysisResult result = ragService.query(ctx.getUserInput());
        // 写入 VFS（路径约定：/agents/{agentId}/{output-name}.json）
        vfsClient.write("/agents/rag/analysis.json", result.toJson());
        // 引擎自动推进阶段，下一个 Agent 将读取此文件
    }
}

// Agent B 读取 Agent A 的产物（可能在不同进程/服务中）
public class DddAgent {
    public void execute(SkillExecutionContext ctx) {
        // 从 VFS 读取 RAG Agent 的分析结果
        String json = vfsClient.read("/agents/rag/analysis.json");
        AnalysisResult ragResult = AnalysisResult.fromJson(json);
        // 基于 RAG 分析结果进行 DDD 设计
        DddDesign design = dddService.design(ragResult);
        vfsClient.write("/agents/ddd/design.json", design.toJson());
    }
}

// Agent C 读取 Agent B 的产物（可能在不同机器上）
public class CodeGenAgent {
    public void execute(SkillExecutionContext ctx) {
        // 从 VFS 读取 DDD 设计产出
        String json = vfsClient.read("/agents/ddd/design.json");
        DddDesign design = DddDesign.fromJson(json);
        // 基于设计生成代码
        List<GeneratedFile> files = codeGenService.generate(design);
        // 写入生成的代码文件
        for (GeneratedFile f : files) {
            vfsClient.write("/agents/codegen/" + f.getPath(), f.getContent());
        }
    }
}

// ===== RemoteWorkflowBridge 远程调用 =====
// Agent A 通过远程桥接调用 Agent B 的服务

public class ReviewAgent {
    @Autowired
    private RemoteWorkflowBridge remoteBridge;

    public Map<String, Object> execute(SkillExecutionContext ctx) {
        // 读取代码生成 Agent 的产物
        String code = vfsClient.read("/agents/codegen/Entity.java");

        // 远程调用代码审查 Agent（可能运行在 aiserver 上）
        Map<String, Object> reviewResult = remoteBridge.executeActivity(
            "code-review-service",  // 目标服务名
            Map.of(
                "code", code,
                "language", "java",
                "standards", List.of("naming", "structure", "security")
            ),
            RetryConfig.builder()
                .maxAttempts(3)
                .exponentialBackoff(500, 1500)  // 500ms/1000ms/1500ms
                .build()
        );

        // 将审查结果写入 VFS
        vfsClient.write("/agents/review/review-result.json", reviewResult);
        return Map.of("status", "completed", "review", reviewResult);
    }
}

// RemoteWorkflowBridge 实现要点
// 1. 封装 HTTP/WS 远程调用，对 Agent 透明
// 2. 内置重试机制（最多 3 次，指数退避）
// 3. 健康检查：连续 3 次失败标记为不可用
// 4. 降级：远程不可用时走本地 fallback 逻辑

// ===== definition.json 中的阶段定义 =====
// 每个阶段绑定一个或多个 Agent 的活动
// PhaseManager 自动按阶段顺序推进，确保 Agent 间协作有序

"phases": [
  {
    "phaseId": "phase_understand",
    "displayName": "理解阶段",
    "stage": 0,
    "autoAdvance": true,
    "activities": ["rag-knowledge-query", "intent-analyze"]
  },
  {
    "phaseId": "phase_design",
    "displayName": "设计阶段",
    "stage": 1,
    "autoAdvance": true,
    "exitHumanActivityId": "human_design_confirm",
    "activities": ["ddd-infer", "entity-lock"]
  },
  {
    "phaseId": "phase_build",
    "displayName": "构建阶段",
    "stage": 2,
    "autoAdvance": true,
    "activities": ["codegen-entity", "codegen-repo", "codegen-api"]
  },
  {
    "phaseId": "phase_review",
    "displayName": "审查阶段",
    "stage": 3,
    "autoAdvance": false,
    "exitHumanActivityId": "human_review_approve",
    "activities": ["code-review", "quality-check"]
  }
]

// ===== PhaseManager 自动管理 =====
// PhaseManager.onProcessStarted() → 注入 phase_understand 为初始阶段
// PhaseManager.onActivityCompleted() → 检测所有活动完成后推进到下一阶段
// PhaseManager.onActivityPaused() → 检测出口 HUMAN，暂停阶段等待用户确认
// 自动推送 phase_started / phase_advanced / phase_paused 事件

// ===== PhaseManager 核心接口 =====
// 三个关键 hook 点，分别在流程启动、活动完成、活动暂停时触发

public class PhaseManager {
    /** context 中的阶段状态字段 */
    public static final String CTX_CURRENT_PHASE = "_current_phase";
    public static final String CTX_PHASE_STATUSES = "_phase_statuses";

    /**
     * ★ 流程启动后调用 — 注入初始阶段，推送 phase_started 事件
     * 若流程未启用分阶段(phases 为空)，则跳过(兼容旧流程)
     */
    public void onProcessStarted(ProcessInstance instance,
                                  ProcessDefinition definition) {
        if (definition == null || !definition.hasPhases()) return;
        PhaseDefinition firstPhase = definition.getOrderedPhases().get(0);
        // 注入当前阶段到 context
        instance.getContext().put(CTX_CURRENT_PHASE, firstPhase.getPhaseId());
        // 初始化所有阶段状态：第一个为 running，其余为 pending
        Map<String, String> statuses = initPhaseStatuses(definition);
        instance.getContext().put(CTX_PHASE_STATUSES, statuses);
        // 推送 phase_started 事件
        eventPublisher.firePhaseStarted(instance, firstPhase, definition);
    }

    /**
     * ★ 活动完成后调用 — 检测是否需要阶段推进
     * Case 1: 出口 HUMAN 完成 → 若 autoAdvance 则推进
     * Case 2: 无出口 HUMAN + 当前阶段所有活动完成 → 推进
     */
    public boolean onActivityCompleted(ProcessInstance instance,
                                        ActivityDefinition actDef,
                                        ProcessDefinition definition) {
        if (!hasPhases(definition)) return false;
        PhaseDefinition currentPhase = getCurrentPhase(instance, definition);
        if (isExitHumanCompleted(currentPhase, actDef)
            || isAllActivitiesDone(currentPhase, instance)) {
            advanceToNextPhase(instance, currentPhase, definition);
            return true;
        }
        return false;
    }

    /**
     * ★ 活动暂停后调用 — 检测是否为阶段出口 HUMAN
     * 匹配出口 HUMAN 活动 ID 时，标记阶段为 paused 并推送事件
     */
    public void onActivityPaused(ProcessInstance instance,
                                  ActivityInstance actInstance,
                                  ActivityDefinition actDef,
                                  ProcessDefinition definition) {
        if (!hasPhases(definition)) return;
        PhaseDefinition currentPhase = getCurrentPhase(instance, definition);
        if (actDef.getActivityId().equals(currentPhase.getExitHumanActivityId())) {
            setPhaseStatus(instance, currentPhase.getPhaseId(), "paused");
            eventPublisher.firePhasePaused(instance, currentPhase,
                actInstance, definition);
        }
    }
}

用户输入
    │
    ▼
IntentDispatchScene
    │  [isThreeBranchFlow=true] 绕过标准引擎
    ▼
RadChatScene (降级路径)
    │
    ▼
XxxFirstOrchestrator.orchestrate()
    │
    ├── Stage 0: 同步执行 → fire专用事件
    ├── Stage 1: 同步执行 → fire专用事件
    │   ...
    └── Stage 9: 同步执行 → fire专用事件
         │
         ▼
    FlowEventPublisher
         │
         ├── 路径A: fireDbFirstEvent → flow_dbfirst_* (专用事件)
         │                → SseEventPushListener → 前端 _handleDbFirstEvent()
         │
         └── 路径B: pushStep → flow_step (通用事件)
                          → DBFirstFlowListener (冗余增强)
                          → SseEventPushListener → 前端 _handleFlowSSEEvent()

用户输入
    │
    ▼
IntentDispatchScene
    │  [统一 switch_scene_flow]
    ▼
SkillFlowEngine (Activity Loop)
    │
    ├── PhaseManager（多 Agent 编排调度）
    │   ├── onProcessStarted → 注入初始阶段
    │   ├── onActivityCompleted → 检测阶段推进
    │   ├── onActivityPaused → 检测出口 HUMAN
    │   └── 推送 phase_started/advanced/paused/rolled_back
    │
    ├── Activity Loop（Agent 调度循环）
    │   ├── 取下一个待执行的 Agent 活动
    │   ├── 通过 skillId 找到对应的 ActivityExecutor
    │   │   └── Agent 执行器（本地或远程）
    │   │       ├── RAG Agent → 写入 VFS
    │   │       ├── DDD Agent → 读取 VFS，写入 VFS
    │   │       ├── CodeGen Agent → 远程桥接(aiserver)
    │   │       └── Review Agent → 读取 VFS，远程桥接
    │   └── 活动完成 → PhaseManager 检测阶段推进
    │
    ├── VFS 虚拟文件系统（Agent 间共享上下文）
    │   └── /agents/{agentId}/{output}.json
    │
    └── 统一事件通道
        ├── flow_step (活动级进度)
        ├── phase_* (阶段级生命周期)
        ├── flow_plan (流程骨架)
        └── flow_complete (流程结束)
             → SseEventPushListener (单一透传)
             → 前端 _handleFlowSSEEvent()
             → _renderActivitiesGroupedByPhase()

时间线 →
───────
orchestrate() 开始
  │
  ├── Stage0: intentAndPath()
  │     ├── pushStep("executing")
  │     ├── fireStage0IntentAndPath()  → flow_dbfirst_intent_detected
  │     ├── human_confirm (阻塞) ← 此时所有 Agent 都挂起!
  │     └── pushStep("completed")
  │
  ├── Stage1: connect()
  │     ├── pushStep("executing")
  │     ├── fireStage1Connect()
  │     └── pushStep("completed")
  │
  │  ... 阶段 2-9 依次同步执行 ...
  │
  └── orchestrate() 结束

问题：
1. 同步阻塞：human_confirm 卡住所有 Agent，无法处理其他流程
2. 紧耦合 Agent：Agent 间通过方法参数传递，无法跨进程
3. 无法回滚：一个 Agent 失败，已完成的 Agent 工作无法回退
4. 无阶段语义：flow_step 只有 executing/completed，无阶段概念

时间线 →
───────
SkillFlowEngine 启动 Activity Loop（多 Agent 调度循环）
  │
  ├── Activity #1: RAG Agent (rag-knowledge-query)
  │     ├── PhaseManager.onActivityStarted → phase_started(phase_0)
  │     ├── Agent 执行：查询知识库 → 写入 VFS
  │     ├── PhaseManager.onActivityCompleted → phase_advanced(phase_0→phase_1)
  │     └── Activity #1 完成，Loop 取下一个
  │
  ├── Activity #2: DDD Agent (ddd-infer)
  │     ├── PhaseManager.onActivityStarted → phase_started(phase_1)
  │     ├── Agent 执行：读取 VFS → DDD 设计 → 写入 VFS
  │     ├── human_confirm (仅暂停此 Agent，Loop 继续运行)
  │     └── 恢复后 → PhaseManager.phase_advanced(phase_1→phase_2)
  │
  │  ⚡ Loop 在 DDD Agent 暂停期间，可调度其他流程实例
  │
  ├── Activity #3: CodeGen Agent (codegen-entity)
  │     ├── PhaseManager.onActivityStarted → phase_started(phase_2)
  │     ├── Agent 执行：远程桥接(aiserver) → 代码生成
  │     └── 写入 VFS → PhaseManager 推进阶段
  │
  ├── Activity #4: Review Agent (code-review)
  │     ├── PhaseManager.onActivityStarted → phase_started(phase_3)
  │     ├── Agent 执行：读取 VFS → 远程审查 → 写入 VFS
  │     └── PhaseManager 推进 → flow_complete
  │
  └── 所有 Agent 完成

优势：
1. 异步非阻塞：human_confirm 仅暂停单个 Agent，Loop 可调度其他流程
2. 松耦合 Agent：通过 VFS 共享上下文，Agent 可独立部署
3. 可回滚：PhaseManager 支持 phase_rolled_back，按 Agent 阶段回退
4. 五种阶段语义：phase_started/advanced/paused/rolled_back/summary

// 普通管道：3 个管道 × 17+ 事件 = 50+ 个 addEventListener
// 每个 Agent 对应一套专用事件类型，前端需要为每个 Agent 注册监听器
eventSource.addEventListener('flow_dbfirst_intent_detected', handler);
eventSource.addEventListener('flow_dbfirst_connect_start', handler);
eventSource.addEventListener('flow_dbfirst_connect_done', handler);
// ... 17+ 个 flow_dbfirst_* 事件

eventSource.addEventListener('flow_viewfirst_*', handler);     // 另一套
eventSource.addEventListener('flow_designerfirst_*', handler);  // 第三套

// 每个事件回调专用处理器，内部分支逻辑
_handleDbFirstEvent(data)       // 根据 eventType 做不同渲染
_handleViewFirstEvent(data)     // 另一套逻辑
_handleDesignerFirstEvent(data) // 第三套逻辑

// 问题：Agent 的专用事件类型与前端代码紧耦合
// 新增一个 Agent = 新增 10+ 个 addEventListener + 专用 handler

// Workflow-Skills：8 个标准事件类型，与 Agent 数量无关
// 无论有多少个 Agent，前端只需注册这 8 个事件监听器
eventSource.addEventListener('flow_step', handler);       // 活动级步骤进度
eventSource.addEventListener('flow_plan', handler);        // 流程骨架
eventSource.addEventListener('flow_complete', handler);    // 流程结束

// phase_* 事件由 _handleFlowSSEEvent 统一处理：
//   phase_started    → 记录当前阶段 ID，初始化 phases 数组
//   phase_advanced   → 旧阶段标记 done，新阶段标记 running
//   phase_paused     → 阶段标记 paused（出口 HUMAN）
//   phase_rolled_back → 阶段标记 rolled_back
//   phase_summary    → 存储阶段总结（产出物、耗时）

_handleFlowSSEEvent(data) // 统一处理器，根据 eventData.type 分发
// 新增 Agent 无需修改前端代码，事件类型不变

// 普通管道：手动渲染 timeline，每个 Agent 独立渲染
_handleDbFirstEvent: function(data) {
    var meta = data.meta || host._inferDefaultMeta(dbfirstEventType);
    var stage = data.stage != null ? data.stage : (meta.stage != null ? meta.stage : '');
    host._ensureSubFlowInstance(data.processInstId, 'dbfirst-build', ...);
    // 手动追加 timeline 条目，无阶段分组概念
    host._appendTimelineItem(data.processInstId, data.activityInstId, 'step', {
        title: stage !== '' ? ('Stage' + stage + ': ' + meta.label) : meta.label,
        body: bodyText, status: meta.status || 'running', ...
    });
    // 重复追加：同一个阶段从两个路径推送，渲染两次
}

// Workflow-Skills：通用渲染，按阶段分组，与 Agent 类型无关
_renderActivitiesGroupedByPhase: function(inst, activities, flowKey, nestLevel, focusedAid) {
    // 三种渲染模式，由 phase_* 事件自动填充的 inst.phases 数组驱动：
    //   running/paused → 展开：详细活动卡片 + 阶段时间线
    //   done → 折叠：可展开查看历史活动
    //   pending → 折叠大纲：仅显示阶段名 + 活动数量
    //
    // 此外，_advanceFlowPhase 管理流程级 phase（plan → run → history）
    // 当第一个 flow_step 到达时推进 plan→run，flow_complete 推进 run→history
    //
    // 关键：渲染引擎不关心 Agent 的具体类型，只关心 phases 数组
    // 新增 Agent 只需在 definition.json 中定义阶段，前端自动渲染
}

// ===== definition.json 中的 Agent 活动定义 =====
// 每个活动可以配置不同的执行模式

{
  "activities": [
    {
      "activityId": "rag-knowledge-query",
      "type": "TASK",
      "config": {
        "stage": 0,
        "skillId": "rag-knowledge-query",
        // 本地执行，无需额外配置
      }
    },
    {
      "activityId": "codegen-entity",
      "type": "TASK",
      "config": {
        "stage": 2,
        "skillId": "codegen-entity",
        "agentConfig": {
          "executorType": "remote",        // 远程执行
          "targetService": "aiserver",     // 目标服务
          "timeout": 60000,                // 60s 超时
          "retryCount": 3,                 // 重试 3 次
          "retryBackoff": "EXPONENTIAL",   // 指数退避
          "fallbackMode": "HUMAN_ESCALATE" // 降级到人工
        }
      }
    },
    {
      "activityId": "human_design_confirm",
      "type": "HUMAN",
      "config": {
        "stage": 1,
        "humanMode": "CONFIRM",
        "prompt": "请确认 DDD 设计方案是否满足需求？"
      }
    }
  ]
}

// ===== ActivityDispatcher 运行时路由 =====
// 根据 agentConfig 自动选择执行方式

public class ActivityDispatcher {
    public void executeActivity(ProcessInstance instance,
                                 ActivityInstance actInstance,
                                 ActivityDefinition actDef) {
        Object agentConfig = actDef.getConfigValue("agentConfig");
        if (agentConfig instanceof Map) {
            String executorType = (String) ((Map) agentConfig).get("executorType");
            if ("remote".equals(executorType)) {
                // 远程执行：通过 RemoteWorkflowBridge 桥接
                remoteBridge.execute(instance, actDef, (Map) agentConfig);
                return;
            }
        }
        // 本地执行：通过 skillId 查找 ActivityExecutor
        ActivityExecutor executor = findExecutor(actDef.getSkillId());
        executor.execute(new SkillExecutionContext(instance, actDef));
    }
}

// 编排器手写 stageIndex，与 definition.json 的 config.stage 不一致
pushStep(ctx, 0, "executing", "开始意图识别...");  // stageIndex=0 ✅ 匹配
pushStep(ctx, 1, "executing", "开始连接...");       // stageIndex=1 ✅ 匹配
pushStep(ctx, 2, "executing", "开始用途导向...");    // stageIndex=2 ❌ 实际是 Stage1.5
// 此后所有阶段编号与定义文件相差 1
// 前端显示的 "Stage2" 与设计文档 "Stage1.5" 不一致

// PhaseManager 自动处理 Agent 协作阶段
// 三个关键 hook 点对应三种 Agent 协作场景：

// 1. 流程启动时 — 注入初始阶段，确定哪个 Agent 先执行
PhaseManager.onProcessStarted()
  → phase_started(phase_0, "RAG 知识查询")

// 2. 活动完成时 — 检测是否推进到下一个 Agent 阶段
PhaseManager.onActivityCompleted()
  → 当前阶段所有 Agent 活动完成 → phase_advanced(phase_0→phase_1)
  → 下一个 Agent (DDD 设计) 开始执行

// 3. 活动暂停时 — 检测出口 HUMAN，等待用户确认
PhaseManager.onActivityPaused()
  → 匹配出口 HUMAN 活动 ID → phase_paused
  → 用户确认后 → phase_advanced

// 阶段定义来自 definition.json 单一来源，Agent 编排逻辑一目了然
// "phases": [
//   { "phaseId": "phase_0", "displayName": "知识查询", "agent": "rag" },
//   { "phaseId": "phase_1", "displayName": "领域设计", "agent": "ddd" },
//   { "phaseId": "phase_2", "displayName": "代码生成", "agent": "codegen" },
//   { "phaseId": "phase_3", "displayName": "代码审查", "agent": "review" }
// ]

事件源                         事件类型                    消费端
─────────                     ────────                    ──────
DBFirst 管道 (内含所有 Agent)  flow_dbfirst_* (17+ 种)     → _handleDbFirstEvent
ViewFirst 管道                flow_viewfirst_* (类似数量)  → _handleViewFirstEvent
DesignerFirst 管道            flow_designerfirst_* (类似)  → _handleDesignerFirstEvent
                              === 同时还有 ===
                              flow_step (来自 pushStep)     → _handleFlowSSEEvent + DBFirstFlowListener

问题：3 套专用事件 + 1 套通用事件 = 4 套通道并行
     每个 Agent 的事件类型需要独立注册、独立维护
     同一 Agent 阶段通过 2 条通道推送，前端重复渲染
     新增 Agent 需要：新增 10+ 事件类型 + 新增前端 handler + 新增 SSE 监听器

事件源                         事件类型                    消费端
─────────                     ────────                    ──────
SkillFlowEngine               flow_plan                    → _handleFlowSSEEvent
  + PhaseManager              flow_start                   → _handleFlowSSEEvent
  + ActivityExecutor          flow_step (活动级步骤)        → _handleFlowSSEEvent
  (任意 Agent)                phase_started (阶段开始)       → _handleFlowSSEEvent
                              phase_advanced (阶段推进)      → _handleFlowSSEEvent
                              phase_paused (阶段暂停)        → _handleFlowSSEEvent
                              phase_rolled_back (阶段回滚)   → _handleFlowSSEEvent
                              phase_summary (阶段总结)       → _handleFlowSSEEvent
                              flow_complete (流程结束)       → _handleFlowSSEEvent

优势：8 种标准事件类型，1 个消费者
     所有 Agent 共用同一套事件体系，新增 Agent 无需新增事件类型
     前端无需关心事件来自哪个 Agent，统一处理

维护成本
  ^
  │                         普通管道 (O(N))
  │                         ─────────────
  │                         / 每个新 Agent 增加 100% 成本
  │                        /  (新事件类型 + 新前端 handler + 新编排代码)
  │                       /
  │                      /
  │                     /
  │                    /
  │  Workflow-Skills (O(1))
  │  ──────────────── 新增 Agent 成本趋近于 0
  │  (仅注册执行器 + 定义阶段，前端和事件通道不变)
  │
  └───────────────────────────────────→ Agent 数量
       1        2        3        4