sanguo/sanguo_moziplus_v2

Fork 0

Files

T

cfdaily f613895360 auto-sync: 2026-05-28 08:45:47

2026-05-28 08:45:47 +08:00

30 KiB

Raw Blame History

v2.8 Pipeline 架构设计

版本: v2.0
日期: 2026-05-27
作者: 庞统
状态: 调研完成 → 设计提案 v2.0，待用户确认

v1.0→v2.0 变更：补充执行模式调研（parallel/loop/saga/interactive/event-driven），补充场景×执行模式映射表，明确业务类型 vs 执行模式的分层关系。

1. 设计目标

消灭 46 处 if/_mail 分支（ticker 14 + dispatcher 23 + spawner 9）
不同 task_type 走不同执行路径（代码评审不需要 planning，数据下载不需要路由）
新增 task_type 只加配置不改核心代码（Plugin/Registry）
执行模式是有限的几种，业务类型无限扩展（用户选业务类型，系统映射到执行模式）

2. 两层抽象：业务类型 vs 执行模式

2.1 概念

层	定义	谁选	数量	例子
业务类型（task_type）	用户视角的"做什么"	用户创建任务时选	无限扩展	coding, review, data, deploy, research, discuss, data_collect, risk_patrol, ...
执行模式（execution_pattern）	系统视角的"怎么跑"	YAML Profile 声明	有限（~7种）	single_step, multi_step, parallel, loop, saga, interactive, event_trigger

2.2 映射关系

# config/pipelines/review.yaml
task_type: review                    # 业务类型（用户视角）
execution_pattern: single_step       # 执行模式（系统视角）

业务类型无限扩展，执行模式有限。新增业务类型 = 新增一个 YAML。新增执行模式 = 新增代码。

3. 执行模式调研

每种模式的结论来自外部来源，不自己发明。

3.1 single_step（单步执行）

描述：一个 agent 一次 spawn 就完成。

适用场景：代码评审、数据下载、Mail 投递、异常监控告警。

优秀实践：

Temporal 的 Activity 概念——原子操作，独立执行，失败可重试（来源：Temporal docs）
Hermes Kanban 的 todo → ready → claimed → running → done 生命周期——单步但状态完整（来源：知识库 hermes-agent）
OpenClaw Lobster 单步 command——exec --shell 'cmd --json'（来源：/opt/homebrew/lib/node_modules/openclaw/docs/tools/lobster.md）

关键设计点：

完成 ≠ 成功：需要 verification（如检查产出文件是否存在）
失败可重试，但有上限

3.2 multi_step（多步串行）

描述：多个 stage 依次执行，前一步的输出是后一步的输入。

适用场景：编码（plan→implement→verify）、调研（plan→research→report）、部署（build→test→deploy）。

优秀实践：

Lobster 的 steps + stdin: $step.stdout 管道——声明式步骤链，步骤间 JSON 管道传递（来源：lobster.md）
OpenClaw TaskFlow——持久化多步 flow 管理，跨 gateway 重启（来源：taskflow.md）
Temporal Workflow——步骤自动持久化，不怕进程重启（来源：Temporal docs）

关键设计点：

步骤间数据传递：通过 DB（stages_json）或 prompt 注入前序 stage 的 output
每个 stage 对应一次 agent spawn
中间 stage 失败 → 可从当前 stage 重试（不需要从头开始）

3.3 parallel（并行执行 / Fan-Out / Scatter-Gather）

描述：同一个任务拆成多个子任务并行执行，结果汇聚。

适用场景：批量回测（N 组参数）、并行协作（张飞写代码 + 关羽做风控）、多角度评审（乐观/悲观/务实三路并行）。

优秀实践：

来源	做法	关键点
AWS Prescriptive Guidance	scatter-gather：分发→并行执行→汇聚	"sends tasks to multiple services in parallel, waits for responses, aggregates results"（web_search 摘要）
Azure Architecture Center	concurrent orchestration / fan-out-fan-in / map-reduce	"also known as parallel, fan-out/fan-in, scatter-gather, map-reduce"（web_search 摘要）
Temporal	`asyncio.gather(*futures, return_exceptions=True)`	`return_exceptions=True` 意味着部分失败不阻塞整体，结果里包含异常（来源：dev.to Temporal AI agents）
open-multi-agent	`AgentPool.runParallel()` + fan-out-aggregate pattern	多 agent 并行，结果由 synthesizer 聚合（来源：知识库 `examples/patterns/fan-out-aggregate.ts`）
GSD (get-shit-done)	Wave Execution——按依赖分组，wave 内并行，wave 间串行	"Plans are grouped into waves. Within each wave, plans run in parallel. Waves run sequentially."（来源：知识库 `get-shit-done/README.md`）

关键设计点：

结果汇聚：需要一个 aggregator（可以是固定逻辑，也可以是一个 agent）
部分失败处理：
- fail_fast：一个失败全部取消（适合 saga 类场景）
- best_effort：收集所有结果，失败标记，汇总时报告（适合批量回测）
并发控制：受 counter 全局上限约束（max_global_agents）
GSD 的 Wave 启发：parallel 可以和 multi_step 组合——一个多步 pipeline 的某个 stage 是并行的

3.4 loop（循环 / 审议 / Evaluator-Reflect-Refine）

描述：执行→评审→不通过→修改→再提交，循环直到通过或达到上限。

适用场景：审议循环（门下省封驳重做）、自愈（检测→诊断→修复→验证循环）、质量迭代（写→评审→改→再审）。

优秀实践：

来源	做法	关键点
AWS Prescriptive Guidance	Evaluator-Reflect-Refine Loop	"The loop repeats until the result meets criteria, is approved, or reaches a retry limit"（web_search 摘要）
三省六部 Edict	门下省封驳→返回中书省→修改→重新审议，最多 3 轮	制度化循环，有硬上限（来源：知识库 `edict/docs/task-dispatch-architecture.md`）
Cloudflare Agents	Human-in-the-loop + schedule escalation	"Implement timeouts — use schedule() to escalate or auto-reject after reasonable periods"（web_search 摘要）
Reddit AI_Agents 实战	"Set a hard rule: if no CRITICAL items, output APPROVED — loop terminates. Max 3 cycles, then escalate to human"（web_search 摘要）

关键设计点：

终止条件：通过 / 达到 max_rounds（硬上限）/ 人工介入
收敛保障：max_rounds 防止无限循环
上下文传递：每次循环需要把上一次的评审意见传给执行者
和 multi_step 的关系：loop 是 multi_step 的某个阶段加上循环条件。可以建模为 multi_step + loop_guard

3.5 saga（补偿事务）

描述：多步串行执行，任何一步失败，反向执行之前步骤的补偿动作。

适用场景：数据 ETL（下载→清洗→入库，入库失败要清掉已洗的数据）、实盘交易（下单→成交确认→风控检查，失败需要撤单）、策略上线（回测→模拟盘→实盘，失败需要回退）。

优秀实践：

来源	做法	关键点
SagaLLM (arXiv:2503.11951)	AI Agent 场景的 saga——保留 context 用于 recovery，按 dependency graph 反向执行补偿	"Upon failure, system invokes compensatory actions using logs and rollback specifications stored in... traverses dependency graph to orchestrate reverse execution paths"（web_search 摘要）
SparkCo	每步定义 forward + compensate 动作，orchestrator 管理补偿链	"Each step treated as independent transaction, with compensating actions defined for rollback scenarios"（web_search 摘要）
Temporal	内置 Saga 支持：`await workflow.execute_activity(compensate_A)`	每个 Activity 注册对应的补偿函数（来源：Temporal docs）

关键设计点：

补偿动作定义：每个 stage 必须声明 compensate 动作（可以是 agent spawn 或固定脚本）
补偿链：失败后从当前 stage 反向执行之前所有 stage 的 compensate
补偿也可能失败：需要记录补偿状态，人工介入
和 multi_step 的关系：saga 是 multi_step + compensate 定义。可以建模为 multi_step + per-stage-compensate

3.6 interactive（人机交互 / Human-in-the-Loop）

描述：执行到关键点暂停等人确认，确认后继续。

适用场景：策略上线审批、实盘交易确认、多级审批链。

优秀实践：

来源	做法	关键点
OpenClaw Lobster	`approval: required` + `resumeToken`——暂停等审批，resume 不重跑已完成步骤	"Halted workflows return a token; approve and resume without re-running everything"（来源：`lobster.md`）
Redis AI HITL	工作流暂停在 decision point，保存 state，等 human approve/reject/modify 后 resume	"Workflow pauses at decision point, saves state, waits for human to approve/reject/modify before resuming"（web_search 摘要）
Cloudflare Agents	审批超时自动 escalate 或 auto-reject	"Implement timeouts — use schedule() to escalate or auto-reject"（web_search 摘要）
MindStudio	Iterative Kanban Pattern——agent 产出 → human 评审 → agent 修改 → human 确认	"Task gets handed to agent, agent produces output, human reviews, sends it back, agent revises, human approves"（web_search 摘要）

关键设计点：

resume 机制：暂停时保存状态 + 产出 token，恢复时从 checkpoint 继续
超时处理：审批等待超时 → escalate / auto-reject / auto-approve（可配置）
Lobster 启发：moziplus 可以直接用 Lobster 的 approval: required 模式，不需要自己实现暂停/恢复
和 multi_step 的关系：interactive 是 multi_step 的某个 stage 加了 approval 门控。可以建模为 multi_step + approval_stage

3.7 event_trigger（事件触发 / 统一触发入口）

描述：不是一种执行模式，而是一种触发方式。和执行模式正交——任何执行模式都可以被事件触发。

适用场景：cron 定时触发、webhook 触发、依赖完成触发、条件触发。

优秀实践：

来源	做法	关键点
Gobii (hermes-agent #491)	统一 scheduled 和 event-driven 到一个 durable processing loop——cron / IMAP idle / webhook / agent-to-agent message 全部统一	"unifies scheduled and event-driven execution into a single durable processing loop"（web_search 摘要）
AgentC2	Agent 同时支持 cron + webhook，hybrid pattern	"Agent can have both cron schedule and event triggers — daily summary at 6 AM + webhook for critical alerts"（web_search 摘要）
Edict	Event Bus + 主题路由（`task.created`, `task.completed` 等）	所有状态变更发布事件，订阅者按需触发（来源：知识库 `edict/edict_agent_architecture.md`）
OpenClaw Cron	`cron add --cron "0 7 * * 1-5" --session session:xxx --message "..."`	定时触发 agent session，声明式配置（来源：OpenClaw cron 工具）

关键设计点：

触发方式 vs 执行模式正交：trigger 是"什么时候开始"，execution_pattern 是"开始后怎么跑"
四种触发：manual（人工）/ cron（定时）/ webhook（外部事件）/ dependency（前序完成）
Gobii 启发：统一到一个 loop，不需要为每种触发建独立系统
OpenClaw Cron 已有能力：当前 moziplus 可以直接用 OpenClaw cron 做定时触发，不需要自己实现

4. 执行模式关系图

                    ┌─────────────────────────────┐
                    │      Trigger（正交层）        │
                    │  manual / cron / webhook /   │
                    │  dependency                  │
                    └──────────────┬──────────────┘
                                   │ 触发任务
                                   ▼
┌──────────┐  ┌──────────────┐  ┌──────────┐  ┌──────────┐
│ single   │  │ multi_step   │  │ parallel │  │   mail   │
│ _step    │  │              │  │          │  │          │
└──────────┘  └──────┬───────┘  └──────────┘  └──────────┘
                     │
            ┌────────┼────────┐
            ▼        ▼        ▼
     ┌──────────┐ ┌──────┐ ┌──────┐
     │interactive│ │ loop │ │ saga │
     │ (审批门控)│ │(循环)│ │(补偿)│
     └──────────┘ └──────┘ └──────┘

关系说明：

single_step 和 multi_step 是两种基础执行模式
parallel 是独立的执行模式（并行子任务）
mail 是特殊的 single_step（独立 pipeline，不走通用路径）
interactive、loop、saga 是 multi_step 的增强——它们在 multi_step 骨架上加了额外能力
trigger 和执行模式正交

建模方式

增强类型	本质	建模
interactive	multi_step + approval stage	`multi_step` + profile 中某些 stage 声明 `approval: required`
loop	multi_step + 循环 guard	`multi_step` + profile 中声明 `loop: { max_rounds: 3, convergence_check: true }`
saga	multi_step + compensate	`multi_step` + profile 中每个 stage 声明 `compensate: { agent: "...", message: "..." }`

不需要 3 个独立的 Pipeline 子类。只需要 MultiStepPipeline 识别 profile 中的增强声明，执行时加入对应逻辑。

5. 场景 × 执行模式映射

5.1 已有业务类型

业务类型	执行模式	触发方式	说明
coding	multi_step	manual	plan→implement→verify→review
review	single_step	manual / dependency	固定 agent，一杆到底
data	single_step	manual / cron	固定 agent（赵云），下载+验证
deploy	multi_step (interactive)	manual	build→test→approve→deploy
research	multi_step	manual	plan→research→report
discuss	single_step	manual	讨论，一杆到底
mail (request)	mail	API 创建	点对点投递
mail (inform)	mail	API 创建	点对点，自动 done

5.2 调研的 28 个场景映射

#	场景	业务类型	执行模式	触发
A1	Mail 投递	mail	mail	API
A2	Task 派发	coding/review/...	各自映射	manual
B1	定时数据采集	data_collect	single_step	cron
B2	早朝简报	morning_brief	multi_step	cron
B3	批量回测	batch_backtest	parallel	manual
B4	风控巡查	risk_patrol	single_step	cron / event
B5	策略上线审批	strategy_launch	multi_step (interactive)	manual
B6	实盘交易信号	trade_signal	multi_step (saga)	event
C1	Webhook 触发	（任意）	由 task_type 决定	webhook
C2	Saga 链式任务	data_etl	multi_step (saga)	manual / dependency
C3	并行协作	parallel_collab	parallel	manual
C4	人机交互	（任意 multi_step）	multi_step (interactive)	manual
C5	执行工具替换	（架构层）	不映射	N/A
D1	隔夜编码	overnight_coding	multi_step	cron
D2	持仓再平衡	portfolio_rebalance	multi_step (interactive)	cron
D3	交易异常监控	fraud_detect	single_step	event
D4	数据 ETL 管道	data_etl	multi_step (saga)	cron / dependency
D5	合规审计追踪	audit_trail	multi_step	event
D6	策略实验跟踪	experiment_track	multi_step	manual
E1	三省六部分权	edict_dispatch	multi_step (loop)	manual
E2	事件驱动响应	（任意）	由 task_type 决定	event
E3	动态计划调整	dynamic_plan	multi_step	manual
E4	自愈/故障恢复	self_heal	multi_step (loop)	event
E5	审批链	approval_chain	multi_step (interactive)	manual
E6	成本/配额控制	quota_guard	single_step	event
E7	经验蒸馏	experience_distill	single_step	dependency
F1	条件触发	（任意）	由 task_type 决定	event (condition)
F2	依赖触发	（任意）	由 task_type 决定	dependency
F3	人工触发	（任意）	由 task_type 决定	manual

6. 设计方案

6.1 设计模式选择

Registry + Template Method + Strategy

Registry：Pipeline 注册中心，按 task_type 查找
Template Method：Pipeline 基类定义 tick() 骨架，子类覆写差异步骤
Strategy：触发方式、路由策略、验证策略作为可注入的策略对象

来源：LangGraph Routing Pattern（来源：web_search "LangGraph Routing"）、Argo WorkflowTemplate（来源：Argo docs）、Temporal Worker Framework（来源：Temporal docs）。

6.2 Pipeline 类体系

TaskPipeline (ABC)                # Template Method 骨架
├── MailPipeline                  # Mail 专属（硬编码，不走 YAML）
├── SingleStepPipeline            # 单步执行
├── MultiStepPipeline             # 多步执行（含增强）
│   ├── 识别 interactive 声明 → approval gate
│   ├── 识别 loop 声明 → 循环 guard
│   └── 识别 saga 声明 → compensate chain
└── ParallelPipeline              # 并行执行（fan-out + aggregate）

PipelineRouter                    # Registry，按 task_type 路由

为什么只有 4 个子类：interactive/loop/saga 不是独立子类，而是 MultiStepPipeline 通过 profile 声明启用的增强能力。

6.3 Pipeline 基类

class TaskPipeline(ABC):
    """Pipeline 基类（Template Method）"""
    
    def __init__(self, spawner, counter, router, config, profile=None):
        self.spawner = spawner
        self.counter = counter
        self.router = router
        self.config = config
        self.profile = profile or {}
    
    async def tick(self, db_path: Path, project_config: dict) -> TickResult:
        """每个 ticker 周期的入口"""
        await self.pre_tick(db_path)
        tasks = self.load_pending(db_path)
        dispatched = 0
        for task in tasks:
            if not self.can_dispatch(task, db_path):
                continue
            result = await self.dispatch_one(task, db_path, project_config)
            if result:
                dispatched += 1
        return TickResult(dispatched=dispatched, total=len(tasks))
    
    # ── 钩子方法 ──
    async def pre_tick(self, db_path): 
        self.check_timeouts(db_path)
    
    @abstractmethod
    def load_pending(self, db_path) -> list: ...
    
    def can_dispatch(self, task, db_path) -> bool:
        return True
    
    @abstractmethod
    async def dispatch_one(self, task, db_path, project_config) -> bool: ...
    
    def check_timeouts(self, db_path): ...

6.4 MultiStepPipeline 增强

class MultiStepPipeline(TaskPipeline):
    """多步执行（含 interactive/loop/saga 增强）"""
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 从 profile 读取增强配置
        self.loop_config = self.profile.get("loop")          # { max_rounds, convergence_check }
        self.saga_config = self.profile.get("saga")          # { stages: [{compensate: ...}] }
        self.interactive_stages = self._extract_approval_stages()
    
    def _extract_approval_stages(self):
        """从 profile stages 中提取需要 approval 的 stage"""
        return [s for s in self.profile.get("stages", []) 
                if s.get("approval") == "required"]
    
    async def dispatch_one(self, task, db_path, project_config):
        stage = self._get_current_stage(task)
        
        # interactive: approval gate
        if stage.get("approval") == "required" and not self._is_approved(task, stage):
            return False  # 等审批，不 spawn
        
        # spawn agent
        agent_id = self._resolve_agent(task, stage)
        message = self._build_stage_prompt(task, stage)
        return await self._spawn_and_track(agent_id, message, task, stage, db_path)
    
    def _on_stage_complete(self, task, stage, outcome, db_path):
        # saga: 记录 compensate 动作
        if self.saga_config and stage.get("compensate"):
            self._record_compensate(task, stage, db_path)
        
        # loop: 检查收敛
        if self.loop_config:
            round_num = self._get_current_round(task)
            if outcome == "rejected" and round_num < self.loop_config["max_rounds"]:
                self._advance_to_next_round(task, db_path)
                return
            elif round_num >= self.loop_config["max_rounds"]:
                self._escalate(task, db_path, reason="loop_max_rounds")
                return
        
        # 正常推进到下一个 stage
        self._advance_stage(task, db_path)

6.5 ParallelPipeline

class ParallelPipeline(TaskPipeline):
    """并行执行（fan-out + aggregate）"""
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.failure_policy = self.profile.get("failure_policy", "best_effort")
        self.aggregator = self.profile.get("aggregator", {})
    
    async def dispatch_one(self, task, db_path, project_config):
        # 1. 拆分子任务
        sub_tasks = self._fan_out(task)
        
        # 2. 并行 spawn（受 counter 约束）
        futures = []
        for sub in sub_tasks:
            agent_id = self._resolve_agent(sub)
            if not self.counter.can_acquire(agent_id, sub.id):
                continue
            self.counter.acquire(agent_id, sub.id)
            futures.append(self._spawn_subtask(agent_id, sub, db_path))
        
        # 3. 等待所有完成
        results = await asyncio.gather(*futures, return_exceptions=True)
        
        # 4. 处理部分失败
        if self.failure_policy == "fail_fast":
            if any(isinstance(r, Exception) for r in results):
                self._cancel_remaining(futures)
                self._mark_failed(task, db_path)
                return True
        # best_effort: 继续聚合
        
        # 5. 聚合结果
        if self.aggregator.get("type") == "agent":
            await self._aggregate_with_agent(task, results, db_path)
        else:
            self._aggregate_simple(task, results, db_path)
        
        return True

6.6 PipelineRouter

class PipelineRouter:
    def __init__(self):
        self._pipelines: Dict[str, TaskPipeline] = {}
        self._default: TaskPipeline = None
    
    def register(self, task_type: str, pipeline: TaskPipeline):
        self._pipelines[task_type] = pipeline
    
    def route(self, task_type: str = None, is_mail: bool = False) -> TaskPipeline:
        if is_mail:
            return self._pipelines["_mail"]
        return self._pipelines.get(task_type, self._default)

6.7 Pipeline Profile（YAML 声明）

# config/pipelines/coding.yaml
task_type: coding
execution_pattern: multi_step
stages:
  - id: plan
    description: "分析需求，制定方案"
    agent_selection: route_by_capability
    capabilities_required: ["coding", "design"]
    timeout_minutes: 10
  - id: implement
    description: "编码实现"
    reuse_plan_agent: true
    timeout_minutes: 30
    depends_on: [plan]
  - id: verify
    description: "自动验证"
    specific_agent: simayi-challenger
    timeout_minutes: 15
    depends_on: [implement]
completion:
  status: done
  require_verification: true

# config/pipelines/review.yaml
task_type: review
execution_pattern: single_step
agent_selection: specific_agent
specific_agent: simayi-challenger
timeout_minutes: 20
completion:
  status: done

# config/pipelines/strategy_launch.yaml
task_type: strategy_launch
execution_pattern: multi_step
stages:
  - id: backtest
    description: "回测验证"
    specific_agent: zhangfei-dev
    timeout_minutes: 60
  - id: simulate
    description: "模拟盘运行"
    specific_agent: zhangfei-dev
    timeout_minutes: 120
    depends_on: [backtest]
  - id: approve
    description: "上线审批"
    approval: required                  # ← interactive 增强
    timeout_hours: 24                   # 审批超时
    timeout_action: escalate            # auto_reject / escalate / auto_approve
    depends_on: [simulate]
  - id: live
    description: "实盘部署"
    specific_agent: jiangwei-infra
    timeout_minutes: 30
    depends_on: [approve]

# config/pipelines/data_etl.yaml
task_type: data_etl
execution_pattern: multi_step
stages:
  - id: download
    description: "数据下载"
    specific_agent: zhaoyun-data
    timeout_minutes: 60
    compensate:                         # ← saga 增强
      action: "删除已下载文件"
      command: "rm -rf /tmp/download/{task_id}"
  - id: clean
    description: "数据清洗"
    specific_agent: zhaoyun-data
    timeout_minutes: 30
    depends_on: [download]
    compensate:
      action: "删除已清洗文件"
      command: "rm -rf /tmp/clean/{task_id}"
  - id: validate
    description: "数据验证"
    specific_agent: zhaoyun-data
    timeout_minutes: 15
    depends_on: [clean]

# config/pipelines/edict_dispatch.yaml
task_type: edict_dispatch
execution_pattern: multi_step
loop:                                   # ← loop 增强
  max_rounds: 3
  convergence_check: "review_passed"
stages:
  - id: plan
    description: "中书省规划"
    specific_agent: zhongshu-agent
    timeout_minutes: 15
  - id: review
    description: "门下省审议"
    specific_agent: menxia-agent
    timeout_minutes: 10
    depends_on: [plan]
    # review 不通过 → loop 回到 plan（最多3轮）
  - id: dispatch
    description: "尚书省派发"
    depends_on: [review]

# config/pipelines/batch_backtest.yaml
task_type: batch_backtest
execution_pattern: parallel
failure_policy: best_effort             # 部分失败不阻塞
aggregator:
  type: agent                           # 用 agent 聚合结果
  specific_agent: pangtong-fujunshi
fan_out:
  source: task.params.param_sets        # 从任务参数获取参数组列表
  per_item:
    specific_agent: zhangfei-dev
    timeout_minutes: 30

7. Ticker 改造

class Ticker:
    def __init__(self, pipeline_router: PipelineRouter, ...):
        self.pipeline_router = pipeline_router
    
    async def _tick_project(self, project_id, project_config):
        is_mail = project_id == "_mail"
        db_path = self._get_db_path(project_id)
        
        if is_mail:
            pipeline = self.pipeline_router.route(is_mail=True)
            return await pipeline.tick(db_path, project_config)
        
        # 查所有活跃 task_type，每种走对应 pipeline
        task_types = self._get_active_task_types(db_path)
        results = []
        for tt in task_types:
            p = self.pipeline_router.route(tt)
            r = await p.tick(db_path, project_config)
            results.append(r)
        return merge_results(results)

8. 实施范围

一次性全做

#	事项	预估行数
1	新建 `pipeline/` 目录：base.py, mail.py, single_step.py, multi_step.py, parallel.py, router.py	~500 行
2	Pipeline Profile YAML（coding, review, data, deploy, research, discuss）	~200 行
3	Profile 加载器（扫描 config/pipelines/）	~50 行
4	Ticker 改造（删除 _mail 分支，改调 pipeline_router）	-150 行（标注废弃）
5	Dispatcher 标注废弃（逻辑迁移到各 Pipeline）	标注
6	Spawner 瘦身（mail prompt 搬到 MailPipeline）	标注
7	前端：task_type 选择 + stage 进度显示	待评估

MultiStepPipeline 增强实现优先级

增强	优先级	理由
基础 multi_step	P0	当前主流程
interactive (approval gate)	P1	有 Lobster 参考，实现简单
loop (循环 guard)	P2	有 edict 参考，逻辑清晰
saga (compensate)	P3	场景需求相对远期
parallel (fan-out)	P3	需要 sub-task 并行调度，改动大

9. 来源引用

结论	来源
parallel = fan-out/fan-in/scatter-gather/map-reduce	Azure Architecture Center（web_search）；AWS Prescriptive Guidance（web_search）；open-multi-agent `fan-out-aggregate.ts`（知识库）
Wave Execution（并行分组）	get-shit-done README（知识库）
loop = evaluator-reflect-refine	AWS Prescriptive Guidance（web_search）
门下省封驳最多3轮	edict `task-dispatch-architecture.md`（知识库）
saga + compensate for AI	SagaLLM arXiv:2503.11951（web_search）；SparkCo（web_search）
interactive = approval + resumeToken	OpenClaw Lobster docs（`lobster.md`）
interactive 超时处理	Cloudflare Agents docs（web_search）
event + cron 统一 loop	Gobii / hermes-agent #491（web_search）
Event Bus + topic routing	Edict architecture（知识库）
TaskFlow 持久化多步	OpenClaw TaskFlow docs（`taskflow.md`）
AgentPool.runParallel()	open-multi-agent `fan-out-aggregate.ts`（知识库）

10. 待确认

4 个 Pipeline 子类是否够？还是 interactive/loop/saga 需要独立子类？
parallel 是否现在实现？还是只预留接口？
discuss 业务类型：走 single_step 还是 multi_step？
Trigger 层是否纳入本次实现？还是继续用 OpenClaw cron 做触发？
前端 task_type 选择：创建任务时的 UI 交互怎么设计？

30 KiB Raw Blame History Unescape Escape

v2.8 Pipeline 架构设计

1. 设计目标

2. 两层抽象：业务类型 vs 执行模式

2.1 概念

2.2 映射关系

3. 执行模式调研

3.1 single_step（单步执行）

3.2 multi_step（多步串行）

3.3 parallel（并行执行 / Fan-Out / Scatter-Gather）

3.4 loop（循环 / 审议 / Evaluator-Reflect-Refine）

3.5 saga（补偿事务）

3.6 interactive（人机交互 / Human-in-the-Loop）

3.7 event_trigger（事件触发 / 统一触发入口）

4. 执行模式关系图

建模方式

5. 场景 × 执行模式映射

5.1 已有业务类型

5.2 调研的 28 个场景映射

6. 设计方案

6.1 设计模式选择

6.2 Pipeline 类体系

6.3 Pipeline 基类

6.4 MultiStepPipeline 增强

6.5 ParallelPipeline

6.6 PipelineRouter

6.7 Pipeline Profile（YAML 声明）

7. Ticker 改造

8. 实施范围

一次性全做

MultiStepPipeline 增强实现优先级

9. 来源引用

10. 待确认

30 KiB

Raw Blame History