diff --git a/CHANGELOG.md b/CHANGELOG.md index a510d6d..d9329ca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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, diff --git a/docs/en/BUSINESS_CALLBACKS.md b/docs/en/BUSINESS_CALLBACKS.md index f0869d6..d98bac2 100644 --- a/docs/en/BUSINESS_CALLBACKS.md +++ b/docs/en/BUSINESS_CALLBACKS.md @@ -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 diff --git a/docs/zh/BUSINESS_CALLBACKS.md b/docs/zh/BUSINESS_CALLBACKS.md index 0dae445..79f6045 100644 --- a/docs/zh/BUSINESS_CALLBACKS.md +++ b/docs/zh/BUSINESS_CALLBACKS.md @@ -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 或业务操作,宿主负责清理其资源。 同步回调入口应迅速返回,阻塞任务应交给异步执行器。