rp-engine条件边与表达式语言：内置6大函数实现分支、循环与错误路由-平芜编程栈

【免费下载链接】rp-engine

YAML-native agent workflow execution engine, written in Rust

项目地址：https://gitcode.com/gh_mirrors/rp/rp-engine

rp-engine 是一款用 Rust 编写的 YAML 原生 agent 工作流执行引擎：你用 YAML 描述节点和边，引擎负责解析依赖并执行整个图。本文带你读懂它最有代表性的控制流能力——条件边（conditional edges）与表达式语言（expression language）：仅靠边上的condition字段加 6 大内置函数，就能实现工作流分支、循环与错误路由，无需写一行代码。

条件边是什么：一条边上的 4 种"开关"

在 rp-engine 中，edges定义执行图。一条边本质上是一个路由器，引擎在源节点完成时评估它（核心逻辑见 src/engine/executor.rs）：

边类型	写法	何时走这条边
无条件边	只写`from`/`to`	总是执行
条件边	加`condition: "表达式"`	表达式求值为 true 时执行
默认边	加`is_default: true`	无条件执行，作为兜底路径
错误边	加`on_error: true`	仅在源节点失败（`_error: true`）时激活

这些字段的完整定义在 src/schema/edge.rs 的YamlEdge结构中。一个关键细节：条件表达式求值出错时会被当作 false 处理（引擎记录 warning 而不是崩溃），这保证了容错性；而如果同一个源节点存在on_error边且节点失败了，普通边会被自动抑制——这正是"降级回退"模式的基础。

表达式语言速览：操作数、运算符与路径

条件写在condition字符串里，由独立的表达式引擎（分词 → 解析 → 求值，源码在 src/expression/）执行。

运算符：

比较：==、!=、>、<、>=、<=
逻辑：&&、||、!
分组：(a == 1 || b == 2) && c == 3

路径（Path）是表达式读取工作流状态的方式：

node_id.field # 节点最近一次执行的输出 node_id[0].field # 第一次执行的输出 node_id[*].field # 全部执行（数组） input.field # 工作流入参

字面量："math"、3.14、true、false、null。

内置 6 大函数：分支判断的主力

官方规范中的完整语法见 docs/FLOW_SPEC.md（"Condition Expression Language" 一节）。函数实现位于 src/expression/evaluator.rs。

函数	参数	行为	典型用法
`empty(x)`	1	为`null`、`""`、`[]`、`{}`时返回 true	`empty(node.rows)`
`exists(x)`	1	值不为 null 时返回 true	`exists(node.user_id)`
`len(x)`	1	返回长度（唯一可作为数字参与比较的函数）	`len(node.items) > 3`
`contains(h, n)`	2	字符串查子串；数组查元素	`contains(node.text, "error")`
`not_contains(h, n)`	2	`contains`的反义	`not_contains(node.status, "draft")`
`matches(s, r)`	2	正则匹配（基于 regex_lite）	`matches(node.email, "^[^@]+@.+$")`

几个实用小贴士：

len()特殊：它返回真实长度数值，所以len(node.messages) >= 5这种数值比较完全合法；其余函数在布尔上下文中使用
contains类型不匹配时返回 false 而非报错，方便在||链中组合
裸路径本身是"真值判断"：非空字符串、非零数字、非空数组/对象均为 true

场景一：用条件边实现工作流分支

官方测试用例 tests/fixtures/conditional.yaml 演示了最经典的"分类器分流"：先让 LLM 判断问题类型，再走不同处理路径。

edges: - from: start to: classifier - from: classifier to: math_solver condition: "classifier.category == 'math'" - from: classifier to: general_answer condition: "classifier.category != 'math'"

两个分支最后汇聚回response节点。注意分支未执行的节点会被引擎标记为"死路径"（dead path）——如果response用default兜底引用另一分支的输出，就不会卡住：

- id: response config: output: result: from: math_solver.answer default: ""

死路径检测机制的详细规则见 docs/FLOW_SPEC.md 的 "Dead Path Detection" 章节，这是 rp-engine 无需"跳过节点"技巧就能干净收敛分支的关键设计。

场景二：用 max_iterations 实现受保护的循环

想让一个节点"思考→再思考"多轮迭代？把边连回自身，加上max_iterations即可。官方循环用例 tests/fixtures/loop.yaml：

edges: - from: start to: think - from: think to: think condition: "think.needs_more == true" max_iterations: 3 # 最多循环 3 轮，防止死循环 - from: think to: response is_default: true # 兜底：其余情况直接出结果

每一轮循环中，think节点通过think[*].thought读取所有历史轮次的输出作为上下文——路径表达式里的[*]选择器让"循环携带记忆"变得非常简单。max_iterations达到上限后该边自动失效，引擎会记录 debug 日志并跳过，工作流不会失控。

场景三：on_error 错误路由与降级回退

当某个节点（比如调用外部 API）不可靠时，组合error_handling与on_error边即可实现优雅降级：

nodes: - id: primary type: generate config: provider: openai model: gpt-4o template: my_prompt error_handling: retry: 2 timeout_ms: 15000 on_exhausted: continue # 重试耗尽不炸流程，输出 _error: true edges: - from: primary to: success_path # 成功时走这里 - from: primary to: fallback # 失败时自动改道 on_error: true

规则很直白：on_error: true的边只在源节点输出_error: true时激活，同时抑制同一源节点的普通边。这样降级节点（fallback）可以引用primary._error_message拿到失败原因，用备用模型或缓存策略继续服务，工作流无需终止。

另外，supervisor（多智能体规划节点）还有一类专属的exit 边：通过exit_reason（如todo_empty、timeout、max_iterations、default）按退出原因路由到不同收尾节点，适合"任务完成/超时/超限"走不同下游的复杂编排。

上手建议与延伸阅读

分支尽量成对写condition并保留一条is_default兜底边，避免"条件全 false 导致流程卡死"（引擎会以Partial状态报告卡住的节点）
判断"有没有数据"优先用empty()/exists()而不是== null，语义更清晰且覆盖空串/空数组
表达式写错不会让引擎崩溃（按 false 处理），但请留意 warning 日志，排查条件为何未生效

想深入细节，推荐按顺序阅读：