# pipline

**Repository Path**: cng1985/pipline

## Basic Information

- **Project Name**: pipline
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: MIT
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 1
- **Forks**: 0
- **Created**: 2026-06-08
- **Last Updated**: 2026-09-07

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# pipeline

轻量级 Java Pipeline 框架，用于将业务逻辑拆分为可组合、可复用的 Stage（阶段），并按顺序、并行或 DAG 依赖关系执行。项目包含核心执行能力和 Spring/SSE 适配能力。

## 特性

- **Stage 抽象**：每个阶段实现 `PipelineStage`，只关注当前阶段的业务逻辑
- **上下文共享**：`PipelineContext` 在阶段间传递数据，支持键值读写、失败中断和事件输出
- **顺序执行**：`DefaultPipelineRuntime` 按列表顺序执行 Stage，遇到 `context.fail()` 或异常后中断
- **组合 Stage**：`SequentialStage` 串行执行子阶段，`ParallelStage` 并行执行子阶段
- **Stage 注册**：`StageRegistry` 按 code 注册和查找 Stage，Spring 模块可自动注册容器中的 Stage Bean
- **DAG 编排**：`PipelineBuilder` / `DagEngine` 支持按依赖关系调度任务，同层任务可并行
- **vue-flow 支持**：`VueFlowParser` 可直接解析前端 vue-flow 导出的 JSON
- **执行控制**：`exec` 引擎支持重试、超时、条件跳过、出错继续、补偿、路由、挂起/恢复
- **Spring 集成**：提供按 Stage code 或 Stage class 执行 Pipeline 的 Spring Handler
- **SSE 流式输出**：基于 Spring MVC `SseEmitter` 将 Pipeline 执行过程中的事件实时推送给前端

## 环境要求

- JDK 21+
- Maven 3.x

> 当前 `pom.xml` 使用 Java 21 编译，并且 `pipeline-spring` 的 SSE Handler 使用了虚拟线程。

## 模块与依赖

项目是 Maven 多模块结构：

| 模块 | 说明 |
|------|------|
| `pipeline-core` | 核心 Pipeline、Stage、DAG、执行引擎、vue-flow 解析 |
| `pipeline-spring` | Spring 容器适配、Spring Handler、SSE 输出 |

根工程 `pipeline` 是聚合工程（`packaging=pom`）。业务项目通常依赖具体模块。

### 仅使用核心能力

```xml
<dependency>
  <groupId>com.nbsaas.boot</groupId>
  <artifactId>pipeline-core</artifactId>
  <version>1.10</version>
</dependency>
```

### 使用 Spring / SSE 能力

```xml
<dependency>
  <groupId>com.nbsaas.boot</groupId>
  <artifactId>pipeline-spring</artifactId>
  <version>1.10</version>
</dependency>
```

`pipeline-spring` 已依赖 `pipeline-core`。Spring 相关依赖在模块中是 `provided` scope，业务应用需要由 Spring Boot 或自身工程提供 `spring-context`、`spring-webmvc` 等运行时依赖。

本地开发可先安装到本地仓库：

```bash
mvn clean install
```

## 快速开始：核心用法

### 1. 定义 Stage

`PipelineStage` 只要求实现 `execute(PipelineContext context)`。Stage code 不在接口方法里定义，注册时通常来自 `@Stage(code = "...")` 或类名规则。

```java
import com.nbsaas.boot.pipeline.PipelineContext;
import com.nbsaas.boot.pipeline.annotation.Stage;
import com.nbsaas.boot.pipeline.api.PipelineStage;

@Stage(code = "validate")
public class ValidateStage implements PipelineStage {

    @Override
    public void execute(PipelineContext context) {
        String name = context.get("name", String.class);
        if (name == null || name.isBlank()) {
            context.fail(400, "name 不能为空");
        }
    }
}
```

```java
import com.nbsaas.boot.pipeline.PipelineContext;
import com.nbsaas.boot.pipeline.annotation.Stage;
import com.nbsaas.boot.pipeline.api.PipelineStage;

@Stage(code = "save")
public class SaveStage implements PipelineStage {

    @Override
    public void execute(PipelineContext context) {
        context.set("saved", true);
    }
}
```

### 2. 组装并执行 Pipeline

```java
import com.nbsaas.boot.pipeline.DefaultPipelineContext;
import com.nbsaas.boot.pipeline.api.impl.DefaultPipelineRuntime;

import java.util.List;

DefaultPipelineContext context = new DefaultPipelineContext();
context.set("name", "demo");

DefaultPipelineRuntime runtime = new DefaultPipelineRuntime(
        List.of(new ValidateStage(), new SaveStage())
);
runtime.execute(context);

if (context.isStop()) {
    return context.toResponseObject();
}
```

### 3. 注册并按 code 获取 Stage

```java
import com.nbsaas.boot.pipeline.api.StageRegistry;
import com.nbsaas.boot.pipeline.api.PipelineStage;
import com.nbsaas.boot.pipeline.api.impl.DefaultStageRegistry;

StageRegistry registry = new DefaultStageRegistry();
registry.register(new ValidateStage());
registry.register(new SaveStage());

PipelineStage stage = registry.get("validate");
stage.execute(context);
```

建议显式写 `@Stage(code = "...")`。未标注时，默认注册表会按类名首字母小写生成 code，例如 `ValidateStage` 对应 `validateStage`。

### 4. 使用组合 Stage

顺序执行子阶段：

```java
SequentialStage sequential = new SequentialStage();
sequential.add(new ValidateStage());
sequential.add(new SaveStage());
sequential.execute(context);
```

并行执行子阶段：

```java
ParallelStage parallel = new ParallelStage(Executors.newFixedThreadPool(4));
parallel.add(new LoadUserStage());
parallel.add(new LoadConfigStage());
parallel.execute(context);
```

`ParallelStage` 会并发读写同一个 `PipelineContext`。`DefaultPipelineContext` 的底层数据容器是 `ConcurrentHashMap`，但复合对象的内部状态仍需业务方自行保证线程安全。

## Spring 用法

`pipeline-spring` 会把 Spring 容器中的 `PipelineStage` Bean 注册到 `SpringStageRegistry` 和 `SpringStageClassRegistry`，然后通过 Handler 执行。

### 1. 定义 Spring Stage

```java
import com.nbsaas.boot.pipeline.PipelineContext;
import com.nbsaas.boot.pipeline.annotation.Stage;
import com.nbsaas.boot.pipeline.api.PipelineStage;
import org.springframework.stereotype.Component;

@Component
@Stage(code = "validatePrompt", title = "校验 Prompt")
public class ValidatePromptStage implements PipelineStage {

    @Override
    public void execute(PipelineContext context) {
        String prompt = context.get("prompt", String.class);
        if (prompt == null || prompt.isBlank()) {
            context.fail(400, "prompt 不能为空");
        }
    }
}
```

Spring 注册规则：

- 优先使用 `@Stage(code = "...")`
- 未配置有效 code 时，使用 AOP 目标类简单类名首字母小写，例如 `ValidatePromptStage` -> `validatePromptStage`
- 相同 code 不允许重复注册，启动时会抛出异常

### 2. 按 Stage code 执行

适合由配置、数据库、前端编排结果动态指定阶段顺序。

```java
import com.nbsaas.boot.pipeline.DefaultPipelineContext;
import com.nbsaas.boot.pipeline.handler.PipelineCodeHandler;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class PromptController {

    private final PipelineCodeHandler pipelineCodeHandler;

    public PromptController(PipelineCodeHandler pipelineCodeHandler) {
        this.pipelineCodeHandler = pipelineCodeHandler;
    }

    @PostMapping("/api/prompt/run")
    public Object run(@RequestBody PromptRequest request) {
        DefaultPipelineContext context = pipelineCodeHandler.handle(
                ctx -> ctx.set("prompt", request.prompt()),
                "validatePrompt",
                "callAi",
                "saveResult"
        );

        if (context.isStop()) {
            return context.toResponseObject();
        }

        return context.data();
    }
}
```

### 3. 按 Stage class 执行

适合阶段顺序固定、希望编译期能引用具体 Stage 类型的场景。

```java
import com.nbsaas.boot.pipeline.DefaultPipelineContext;
import com.nbsaas.boot.pipeline.handler.PipelineClassHandler;
import org.springframework.stereotype.Service;

@Service
public class PromptService {

    private final PipelineClassHandler pipelineClassHandler;

    public PromptService(PipelineClassHandler pipelineClassHandler) {
        this.pipelineClassHandler = pipelineClassHandler;
    }

    public DefaultPipelineContext run(String prompt) {
        return pipelineClassHandler.handle(
                ctx -> ctx.set("prompt", prompt),
                ValidatePromptStage.class,
                CallAiStage.class,
                SaveResultStage.class
        );
    }
}
```

### 4. 手动构建 PipelineRuntime 后执行

如果需要先自行组装 `PipelineRuntime`，可以使用底层 `PipelineHandler`。

```java
import com.nbsaas.boot.pipeline.DefaultPipelineContext;
import com.nbsaas.boot.pipeline.api.PipelineRuntime;
import com.nbsaas.boot.pipeline.handler.PipelineHandler;
import com.nbsaas.boot.pipeline.registry.SpringStageRegistry;
import org.springframework.stereotype.Service;

@Service
public class ManualPipelineService {

    private final SpringStageRegistry stageRegistry;
    private final PipelineHandler pipelineHandler;

    public ManualPipelineService(
            SpringStageRegistry stageRegistry,
            PipelineHandler pipelineHandler) {
        this.stageRegistry = stageRegistry;
        this.pipelineHandler = pipelineHandler;
    }

    public DefaultPipelineContext run(String name) {
        PipelineRuntime runtime = stageRegistry.build(
                "validate",
                "save"
        );

        return pipelineHandler.handle(
                runtime,
                ctx -> ctx.set("name", name)
        );
    }
}
```

## SSE 用法

SSE 适合 AI 生成、长任务进度、阶段状态等需要服务端持续推送给浏览器的场景。`pipeline-spring` 提供：

| 类 | 用途 |
|----|------|
| `SsePipelineHandler` | 按 Stage code 构建 Pipeline，并返回 `SseEmitter` |
| `SseClassPipelineHandler` | 按 Stage class 构建 Pipeline，并返回 `SseEmitter` |
| `SsePipelineOutput` | `PipelineOutput` 的 SSE 实现，内部使用 Spring MVC `SseEmitter` |

SSE Handler 会自动：

1. 创建 `SseEmitter`
2. 创建 `DefaultPipelineContext`
3. 将 `SsePipelineOutput` 注入到 `context.setOutput(...)`
4. 使用虚拟线程异步执行 Pipeline
5. Pipeline 成功后关闭 SSE
6. Stage 抛异常或 `context.fail(...)` 后发送 `pipeline-error` 事件并关闭 SSE

### 1. Controller 返回 SSE

`SseEmitter` 对应的接口应声明 `produces = MediaType.TEXT_EVENT_STREAM_VALUE`。

```java
import com.nbsaas.boot.pipeline.handler.SsePipelineHandler;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

@RestController
public class PromptStreamController {

    private final SsePipelineHandler ssePipelineHandler;

    public PromptStreamController(SsePipelineHandler ssePipelineHandler) {
        this.ssePipelineHandler = ssePipelineHandler;
    }

    @GetMapping(
            value = "/api/prompt/stream",
            produces = MediaType.TEXT_EVENT_STREAM_VALUE
    )
    public SseEmitter stream(@RequestParam String prompt) {
        return ssePipelineHandler.handle(
                ctx -> ctx.set("prompt", prompt),
                "validatePrompt",
                "callAi",
                "saveResult"
        );
    }
}
```

按 class 执行的写法：

```java
@GetMapping(
        value = "/api/prompt/stream-by-class",
        produces = MediaType.TEXT_EVENT_STREAM_VALUE
)
public SseEmitter streamByClass(@RequestParam String prompt) {
    return sseClassPipelineHandler.handle(
            ctx -> ctx.set("prompt", prompt),
            ValidatePromptStage.class,
            CallAiStage.class,
            SaveResultStage.class
    );
}
```

### 2. Stage 内输出 SSE 事件

Stage 通过 `PipelineContext.emit(event, data)` 输出事件。事件名由业务方约定，数据对象会按 JSON 发送。

```java
import com.nbsaas.boot.pipeline.PipelineContext;
import com.nbsaas.boot.pipeline.annotation.Stage;
import com.nbsaas.boot.pipeline.api.PipelineStage;
import com.nbsaas.boot.pipeline.event.AiCompleteEvent;
import com.nbsaas.boot.pipeline.event.AiDeltaEvent;
import com.nbsaas.boot.pipeline.event.StageEvent;
import org.springframework.stereotype.Component;

@Component
@Stage(code = "callAi", title = "调用 AI")
public class CallAiStage implements PipelineStage {

    @Override
    public void execute(PipelineContext context) {
        String prompt = context.get("prompt", String.class);

        context.emit(
                "stage",
                new StageEvent("callAi", "开始生成")
        );

        StringBuilder answer = new StringBuilder();
        long index = 0;

        for (String chunk : callModelAsChunks(prompt)) {
            answer.append(chunk);
            context.emit(
                    "ai-delta",
                    new AiDeltaEvent(chunk, index++)
            );
        }

        context.set("answer", answer.toString());
        context.emit(
                "ai-complete",
                new AiCompleteEvent(answer.toString(), index)
        );
    }
}
```

内置事件 DTO：

| 类 | 常用事件名 | 字段 |
|----|------------|------|
| `StageEvent` | `stage` | `stage`, `message` |
| `AiDeltaEvent` | `ai-delta` | `content`, `text`, `index` |
| `AiCompleteEvent` | `ai-complete` | `content`, `chunkCount` |
| `AiErrorEvent` | `pipeline-error` | `message` |

`pipeline-error` 是 Handler 的错误事件名。正常完成时 Handler 只关闭连接，不会自动发送 `ai-complete`；如果前端需要最终内容，业务 Stage 应主动 `emit("ai-complete", ...)`。

### 3. 前端 EventSource 消费

```javascript
const source = new EventSource(
  `/api/prompt/stream?prompt=${encodeURIComponent(prompt)}`
);

source.addEventListener("stage", (event) => {
  const data = JSON.parse(event.data);
  console.log(`[${data.stage}] ${data.message}`);
});

source.addEventListener("ai-delta", (event) => {
  const data = JSON.parse(event.data);
  appendText(data.content);
});

source.addEventListener("ai-complete", (event) => {
  const data = JSON.parse(event.data);
  setFinalAnswer(data.content);
  source.close();
});

source.addEventListener("pipeline-error", (event) => {
  const data = JSON.parse(event.data);
  showError(data.message);
  source.close();
});

source.onerror = () => {
  source.close();
};
```

命令行调试：

```bash
curl -N -H "Accept: text/event-stream" "http://localhost:8080/api/prompt/stream?prompt=hello"
```

### 4. SSE 使用注意

- 浏览器原生 `EventSource` 只支持 GET；如果需要 POST 请求体，需要改用 fetch 流式读取或先创建任务再用 GET 订阅
- Controller 必须返回 `SseEmitter`，并设置 `produces = MediaType.TEXT_EVENT_STREAM_VALUE`
- `context.emit(...)` 在非 SSE 场景下如果没有配置 `PipelineOutput`，会直接忽略，不影响普通 Pipeline 执行
- SSE 连接默认超时时间为 15 分钟
- 客户端断开、超时或发送失败后，`SsePipelineOutput` 会停止继续输出

## DAG 编排

先注册各业务 Stage，再用 `PipelineBuilder` 声明任务依赖关系：

```java
StageRegistry stageRegistry = new DefaultStageRegistry();
stageRegistry.register(new ValidateStage());
stageRegistry.register(new LoadUserStage());
stageRegistry.register(new LoadProductStage());
stageRegistry.register(new SaveStage());

PipelineDefinition definition = PipelineBuilder.named("order-create")
        .displayName("创建订单")
        .executor(Executors.newFixedThreadPool(4))
        .task("validate").stage("validate").end()
        .task("loadUser").stage("loadUser").dependsOn("validate").end()
        .task("loadProduct").stage("loadProduct").dependsOn("validate").end()
        .task("save").stage("save").dependsOn("loadUser", "loadProduct").end()
        .build();

PipelineCatalog catalog = new DefaultPipelineCatalog();
catalog.register(definition);

PipelineOrchestrator orchestrator = new DefaultPipelineOrchestrator(
        stageRegistry,
        catalog
);
orchestrator.execute("order-create", context);
```

编排器会按 DAG 拓扑分层执行：无依赖的任务先跑，同一层级的任务并行执行，上一层全部完成后再执行下一层。

## vue-flow 可视化编排

前端用 [vue-flow](https://vueflow.dev/) 画布编排流程后，`toObject()` 导出的 JSON 可直接交给 `VueFlowParser` 解析为 `DagDefinition`，再由 `DagEngine` 执行：

```java
String json = """
        {
          "nodes": [
            {"id": "1", "position": {"x": 250, "y": 5},
             "data": {"label": "校验订单", "stage": "validate"}},
            {"id": "2", "position": {"x": 100, "y": 100},
             "data": {"label": "加载用户", "stage": "loadUser"}},
            {"id": "3", "position": {"x": 400, "y": 100},
             "data": {"label": "加载商品", "stage": "loadProduct", "retry": 2}},
            {"id": "4", "position": {"x": 250, "y": 200},
             "data": {"label": "保存订单", "stage": "save"}}
          ],
          "edges": [
            {"id": "e1-2", "source": "1", "target": "2"},
            {"id": "e1-3", "source": "1", "target": "3"},
            {"id": "e2-4", "source": "2", "target": "4"},
            {"id": "e3-4", "source": "3", "target": "4"}
          ]
        }
        """;

DagDefinition dag = VueFlowParser.parse(json, "order-create");
DagResult result = new DagEngine(stageRegistry).execute(dag, context);
```

解析规则：

- `nodes[].id` 作为 DAG 节点 id
- `edges[].source -> target` 表示 target 依赖 source
- 画布元数据（`position`、`zoom`、`viewport` 等）自动忽略
- 解析时校验节点 id 重复、边指向未知节点、循环依赖
- JSON 解析由内置极简解析器完成，无需引入第三方 JSON 依赖

`nodes[].data` 支持的执行策略字段：

| 字段 | 类型 | 说明 |
|------|------|------|
| `stage` / `stageCode` | string | 关联的 Stage code，缺省依次回退到 `label`、节点 id |
| `label` | string | 显示名称，可兼作 stage code |
| `retry` | number | 最大重试次数，不含首次执行 |
| `timeoutMs` | number | 单次执行超时毫秒数 |
| `failurePolicy` | string | `FAIL_FAST` / `SKIP_DOWNSTREAM` / `IGNORE`，不区分大小写 |
| `disabled` | boolean | `true` 时节点被跳过，状态为 `SKIPPED` |

## 执行引擎（exec）

`PipelineEngine` 提供比 `DefaultPipelineRuntime` 更丰富的执行控制：

| 能力 | 配置/调用方式 |
|------|---------------|
| 重试 | `StageDefinition.stage("x").retry(2)`，异常或 Stage 内 `control.retry()` 触发 |
| 超时 | `.timeout(Duration.ofSeconds(3))`，超时状态为 `TIMEOUT` |
| 条件跳过 | `.when(ctx -> ...)`，条件不满足时状态为 `SKIPPED` |
| 出错继续 | `.continueOnError()` 或 Stage 内 `control.continueOnError()` |
| 失败补偿 | `.compensation("undoX")`，失败时逆序补偿已成功 Stage |
| 路由跳转 | Stage 内 `control.route("stageCode")` |
| 主动停止 | Stage 内 `control.stop()`，整体状态 `CANCELLED` |
| 挂起/恢复 | Stage 内 `control.suspend()`，凭 `result.suspendedAt()` 恢复 |

```java
StageRegistry registry = new DefaultStageRegistry();
registry.register(new ValidateStage());
registry.register(new SaveStage());
registry.register(new UndoSaveStage());
registry.register(new NotifyStage());

com.nbsaas.boot.pipeline.exec.PipelineDefinition definition =
        com.nbsaas.boot.pipeline.exec.PipelineDefinition.named("order-create")
                .stage("validate")
                .stage(StageDefinition.stage("save")
                        .retry(2)
                        .timeout(Duration.ofSeconds(3))
                        .compensation("undoSave")
                        .build())
                .stage(StageDefinition.stage("notify").continueOnError().build())
                .build();

PipelineEngine engine = new PipelineEngine(registry);
PipelineResult result = engine.execute(definition, context);

if (result.status() == PipelineStatus.SUSPENDED) {
    engine.execute(definition, context, result.suspendedAt());
}
```

需要在 Stage 内部干预执行流程时，实现 `ControllableStage`：

```java
public class AuditStage implements ControllableStage {

    @Override
    public void execute(StageExecution<? extends PipelineContext> execution) {
        if (needManualReview(execution.context())) {
            execution.control().suspend();
        }
    }
}
```

说明：`exec.PipelineDefinition`（顺序执行 + 执行控制）与 `orchestration.PipelineDefinition`（DAG 依赖编排）是两套互补模型，按场景选用。

## 核心概念

### PipelineStage

所有阶段的统一接口：

| 方法 | 说明 |
|------|------|
| `execute(PipelineContext context)` | 执行阶段逻辑，可读写上下文、输出事件或标记失败 |

### PipelineContext

阶段间共享的上下文：

| 方法 | 说明 |
|------|------|
| `get(key)` / `get(key, type)` | 读取数据 |
| `set(key, value)` | 写入数据；value 为 `null` 时移除 key |
| `contains(key)` | 判断是否包含键 |
| `data()` | 获取底层数据 Map |
| `isStop()` | 是否已停止 |
| `fail(code, msg)` | 标记失败并停止后续阶段 |
| `emit(event, data)` | 输出 Pipeline 事件；未配置输出实现时忽略 |

`DefaultPipelineContext` 额外提供 `stop()`、`toResponseObject()`、`setOutput(...)` 等方法。

### PipelineRuntime

`DefaultPipelineRuntime` 按 Stage 列表顺序执行，行为如下：

1. 每个 Stage 执行前检查 `context.isStop()`，为 `true` 则中断
2. Stage 抛出异常时，调用 `context.fail(500, e.getMessage())` 并中断
3. Stage 内部调用 `context.fail()` 后，后续 Stage 不再执行

### 装饰器 Stage

位于 `stages/decorator` 包，可包装任意 Stage 织入横切能力：

| 类 | 行为 |
|----|------|
| `RetryStage` | 异常时自动重试，超过次数后抛出最后一次异常；`context.fail()` 标记的业务失败不触发重试 |
| `LoggingStage` | 执行前后输出开始/结束日志 |
| `MetricsStage` | 统计并输出执行耗时 |

```java
PipelineStage stage = new RetryStage(
        new MetricsStage(new SaveStage()),
        3
);
```

## 项目结构

```text
pipeline
├── pom.xml
├── pipeline-core
│   └── src/main/java/com/nbsaas/boot/pipeline
│       ├── PipelineContext.java
│       ├── DefaultPipelineContext.java
│       ├── annotation
│       │   ├── Stage.java
│       │   └── Pipeline.java
│       ├── api
│       │   ├── PipelineStage.java
│       │   ├── PipelineRuntime.java
│       │   ├── StageRegistry.java
│       │   └── impl
│       ├── dag
│       │   ├── DagEngine.java
│       │   ├── DagDefinition.java
│       │   └── vueflow
│       ├── exec
│       │   ├── PipelineEngine.java
│       │   ├── StageDefinition.java
│       │   └── impl
│       ├── orchestration
│       │   ├── PipelineBuilder.java
│       │   ├── PipelineCatalog.java
│       │   └── impl
│       ├── output
│       │   └── PipelineOutput.java
│       └── stages
│           ├── SequentialStage.java
│           ├── ParallelStage.java
│           ├── ConditionalStage.java
│           └── decorator
└── pipeline-spring
    └── src/main/java/com/nbsaas/boot/pipeline
        ├── registry
        │   ├── SpringStageRegistry.java
        │   └── SpringStageClassRegistry.java
        ├── handler
        │   ├── PipelineHandler.java
        │   ├── PipelineCodeHandler.java
        │   ├── PipelineClassHandler.java
        │   ├── SsePipelineHandler.java
        │   └── SseClassPipelineHandler.java
        ├── listenter
        │   └── SsePipelineOutput.java
        └── event
            ├── StageEvent.java
            ├── AiDeltaEvent.java
            ├── AiCompleteEvent.java
            └── AiErrorEvent.java
```

## 运行测试

```bash
mvn test
```

## 使用建议

- 每个 Stage 只做一件事，通过 `PipelineContext` 传递输入输出
- 可预期的业务错误使用 `context.fail(code, msg)`，不要依赖抛异常表达业务分支
- Spring 场景建议显式使用 `@Stage(code = "...")`，避免类名变化影响调用方
- SSE 场景中，业务 Stage 应明确约定事件名和事件数据结构
- 长任务或 AI 流式输出建议在关键阶段发送 `stage` 事件，便于前端展示进度和排查问题
- 并行执行时避免多个 Stage 修改同一个非线程安全对象

## License

内部项目，按需使用。