Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

## [Unreleased]

## [0.0.5] — 2026-09-09

### Breaking: per-edge conditional routing

- `onRoute` now evaluates one conditional edge at a time: `RouteRequest` describes a single edge (step, condition,
Expand Down
47 changes: 47 additions & 0 deletions docs/en/BUSINESS_CALLBACKS.md
Original file line number Diff line number Diff line change
Expand Up @@ -154,6 +154,53 @@ The engine waits for every route decision before dispatching any successor; a mi
decision fails the workflow and cannot cause partial successor dispatch. Route failure messages identify the source and
target edge while preserving the original cause.

### 6.1 Declaring conditional edges (workflow side)

Each `next` entry is a `JumpCondition(step, condition)`; an empty condition makes the edge unconditional:

```java
WorkflowStep diagnosis = WorkflowStep.builder()
.name("diagnosis")
.subtasks(List.of(Task.builder().agent("SPN Domain Agent City1").description("diagnose").build()))
.next(List.of(
new JumpCondition("notify", ""), // unconditional: always active, never reaches onRoute
new JumpCondition("hardware_recovery", "fault.hardware"),
new JumpCondition("escalate", "fault.unresolved")))
.build();
```

### 6.2 Implementing the per-edge decision (host side)

onRoute receives one conditional edge per call and reads this step's fresh results from `currentResults()`:

```java
.onRoute(request -> {
// Different edges may call back concurrently: read only the request, share no mutable state
String diagnosis = request.currentResults().stream()
.flatMap(r -> r.outputs().stream().map(String::valueOf))
.findFirst()
.orElse("");
boolean hardwareFault = diagnosis.contains("hardware fault");

return switch (request.condition()) {
case "fault.hardware" -> CompletableFuture.completedFuture(
hardwareFault
? RouteDecision.allow("hardware fault confirmed")
: RouteDecision.deny("no hardware fault in: " + diagnosis));
case "fault.unresolved" -> CompletableFuture.completedFuture(
diagnosis.isBlank()
? RouteDecision.allow("diagnosis inconclusive")
: RouteDecision.deny("diagnosis conclusive: " + diagnosis));
default -> CompletableFuture.completedFuture(
RouteDecision.deny("unknown condition: " + request.condition()));
};
})
```

In this example the unconditional `notify` edge always activates; the two conditional edges are judged independently
and may fan out together with it. Replace the business check with real parsing of `currentResults()` outputs (structured
output comes as JSON in outputs; natural-language output calls for text matching or the host's own judgement).

Callbacks for different conditional edges may run concurrently; do not hold shared mutable current-task state. Each
workflow task activation
(preparation and dispatch combined) is bounded by the client timeout (default sendTimeoutSeconds=600). Routing has a
Expand Down
46 changes: 46 additions & 0 deletions docs/zh/BUSINESS_CALLBACKS.md
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,52 @@ RouteDecision.allow (reason) 激活该边,或返回 RouteDecision.deny (reason
引擎等待全部路由判断完成后才下发任何后继节点;回调缺失、返回 null、异常或超时都会使工作流失败,不会发生部分后继任务已下发。
路由失败信息会标明源节点和目标节点,并保留原始异常原因。

### 6.1 声明条件边(工作流侧)

`next` 里每项是一条 `JumpCondition(step, condition)`;condition 留空即为无条件边:

```java
WorkflowStep diagnosis = WorkflowStep.builder()
.name("diagnosis")
.subtasks(List.of(Task.builder().agent("SPN Domain Agent City1").description("diagnose").build()))
.next(List.of(
new JumpCondition("notify", ""), // 无条件:始终激活,不经 onRoute
new JumpCondition("hardware_recovery", "fault.hardware"),
new JumpCondition("escalate", "fault.unresolved")))
.build();
```

### 6.2 实现逐边判断(宿主侧)

onRoute 每次收到一条条件边,从 `currentResults()` 读取本步骤刚产生的结果,按业务规则放行或拒绝:

```java
.onRoute(request -> {
// 不同条件边可能并发回调:只读 request,不共享可变状态
String diagnosis = request.currentResults().stream()
.flatMap(r -> r.outputs().stream().map(String::valueOf))
.findFirst()
.orElse("");
boolean hardwareFault = diagnosis.contains("hardware fault");

return switch (request.condition()) {
case "fault.hardware" -> CompletableFuture.completedFuture(
hardwareFault
? RouteDecision.allow("hardware fault confirmed")
: RouteDecision.deny("no hardware fault in: " + diagnosis));
case "fault.unresolved" -> CompletableFuture.completedFuture(
diagnosis.isBlank()
? RouteDecision.allow("diagnosis inconclusive")
: RouteDecision.deny("diagnosis conclusive: " + diagnosis));
default -> CompletableFuture.completedFuture(
RouteDecision.deny("unknown condition: " + request.condition()));
};
})
```

上面的示例中:`notify` 无条件边总是激活;两条条件边独立判断后可与它同时扇出。业务判断应替换为对
`currentResults()` 输出的真实解析(结构化输出用 outputs 里的 JSON,自然语言输出用文本包含或宿主自己的判定逻辑)。

不同条件边的回调可能并发,不要共享可变的“当前任务”状态。每个工作流任务从内容准备到传输完成受客户端 timeout 总体限制,默认
sendTimeoutSeconds=600; 路由单独限制回调等待时间,dispatch/协商另有总等待截止时间。取消/超时后晚到结果不发送,
但不等于自动取消宿主正在运行的 LLM 或业务操作,宿主负责清理其资源。 同步回调入口应迅速返回,阻塞任务应交给异步执行器。
Expand Down