【免费下载链接】rp-engineYAML-native agent workflow execution engine, written in Rust项目地址https://gitcode.com/gh_mirrors/rp/rp-engine点击查看免费下载 在rp-engineRust 编写的 YAML 原生 Agent 工作流执行引擎中循环工作流最大的隐患是节点每跑一轮上一轮的结果就被覆盖历史彻底丢失。rp-engine 用一套append-only只追加执行状态 short ID 引用的设计化解了这个问题——每次执行都留一条带短 ID 的记录配合max_iterations循环保护和可序列化的 CheckpointState让循环既安全又可追溯。本文将带你完整读懂这套机制。一、循环工作流为什么需要特殊的状态设计先想象一个典型的思考-自我修正循环think节点每轮产出thought、needs_more条件边think → think在needs_more true时回跳直到得出最终答案。如果状态是节点 → 一份输出的字典那么问题后果覆盖历史第 2 轮的输出会顶掉第 1 轮模板里想引用之前所有思考就无从谈起无法定位某一轮调试时无法回答第 1 轮到底输出了什么死循环风险没有边级别的计数与上限条件永远为 true 就会烧穿 tokenrp-engine 的 tests/fixtures/loop.yaml 就是一个标准案例think节点的variables里同时用了history: think[*].thought # 所有轮次数组 task: start.task # 工作流输入而输出里all_thoughts: think[*].thought要把每一轮的思考原样带回。这要求状态必须记住每一轮这正是 append-only 的由来。二、ExecutionState一张只增不改的执行台账核心实现在 src/state/execution_state.rs。可以把ExecutionState理解为一本台账关键字段各司其职node_executions节点ID → VecExecutionRecord每个节点的执行历史按时间顺序追加latest_execution节点ID → 最新 short_id取最新值的快捷索引edge_counts边(from, to) → 已走次数循环保护的计数器dead_nodes/dead_edges被判定为不可达的节点与条件为 false 的边用于 dead path 检测short_id_counter全局递增的短 ID 发号器。每一条ExecutionRecordL18-L42记录了 4 样东西short_id本次执行的短 ID如a、b、zexecution_index该节点的第几次执行0、1、2…outputs本次执行的全部输出字段started_at/timestamp开始与结束墙钟时间戳毫秒二者相减就是该次执行的权威耗时直接供前端时间线/甘特图渲染。写入统一走write/write_with_startL203-L249生成短 ID → 构造记录 →push 进历史数组→ 更新latest_execution。注意它只有 push没有任何修改已有记录的接口——这就是 append-only 语义的落点。️ 还有一个防内存失控的细节MAX_RECORDS_PER_NODE 500当单节点记录数超过上限时会drain掉最旧的部分。长循环下状态体积被硬性封顶不会 OOM。三、short IDbase36 发号 跨断点不重号短 ID 由全局计数器按base36生成L626-L6420 →a1 →b… 25 →z26 →0… 36 →ba三个设计意图短路径里写counter[b]比一串 UUID 友好得多全局唯一计数器是引擎级的不同节点、不同轮次共享同一序列天然不会撞号断点续跑不重号CheckpointState里专门保存了short_id_counterL53-L71from_checkpoint恢复时沿用旧计数器L156-L180。即使工作流暂停后数小时才恢复新记录也不会和历史 short ID 混淆。四、路径表达式五句话就能读完全部历史parse_pathL573-L611把字符串路径解析成段序列支持的引用格式与 docs/FLOW_SPEC.md 的 Path Expressions 一节完全一致路径写法含义node.field该节点最新一次执行的field最常用node[a].field指定short_id的那次执行node[0].field第一次执行node[-1].field最后一次执行等价于不带括号node[*].field所有执行返回数组input.field工作流原始输入特殊命名空间对应的解析逻辑在resolve_segmentsL401-L458与resolve_all_executionsL496-L518。源码内的单元测试test_multiple_executionsL683-L710展示了完整行为// counter 执行了两次count1, count2 state.read(counter.count) // → 2 默认取最新 state.read(counter[a].count) // → 1 按 short_id state.read(counter[0].count) // → 1 按序号 state.read(counter[-1].count) // → 2 负数序号 state.read(counter[*].count) // → [1, 2] 全部这套语法让 YAML 作者在不写任何脚本的情况下就能在循环里精确选择最新一轮第一轮全部轮次的数据。五、edge_counts 与 max_iterations循环安全的最后一道闸append-only 解决的是数据不丢循环安全还需要次数可控。机制在 src/engine/executor.rs每次过边先查计数evaluate_edges在决定是否走某条边前先state.get_edge_count(from, to)与max_iterations比较L1471-L1481超限即跳过过边时计数 1state.increment_edge_count(...)L1516-L1518三级上限边级max_iterationsYAML 中配置→ 引擎级默认ExecutionConfig.max_iterations默认 100见 L264schema 里也允许按边单独声明src/schema/edge.rs。回到loop.yaml的例子think → think的边声明了max_iterations: 3即使模型一直回答needs_more: true引擎最多也只放它循环 3 次随后走is_default: true的兜底边退出——循环永远有出口。此外条件边求值为 false 时会被mark_edge_dead记入dead_edges并经过propagate_dead_pathsL1546-L1588固定点传播所有入边都不活跃的节点被标记 dead从而既避免无谓等待也让引用 dead 节点且没有default的变量快速报错详见 docs/FLOW_SPEC.md 的 Dead Path Detection 一节。六、CheckpointState把整本台账序列化暂停/恢复零损耗长流程尤其含 LLM 调用、等待人工确认的checkpoint节点需要存盘续跑。to_checkpointL134-L153把整份状态快照成可序列化的CheckpointStateinput原始工作流输入executions全部节点的执行记录append-only 的台账本体edge_counts每条边的已走次数short_id_counter短 ID 发号器当前值dead_nodes/dead_edges已判定的死路径。恢复时from_checkpoint会用记录尾部自动重建latest_execution索引rebuild_latestL183-L190并把新输入合并进input。执行侧在遇到 checkpoint 节点时会调用to_checkpoint并返回ExecutionStatus::Pausedsrc/engine/executor.rscheckpoint节点执行器见 src/nodes/checkpoint.rs。由于edge_counts、short_id_counter、历史记录三件套都随快照走恢复后的执行与从未暂停在数据层面完全一致——第 N1 轮循环引用的node[*].field依然包含暂停前的全部轮次。七、执行结果长什么样run_yaml返回的ExecutionResult中executions字段就是台账的 JSON 投影docs/FLOW_SPEC.md Result Schema{ status: Success, executions: { think: [ { short_id: a, execution_index: 0, outputs: { ...: ... }, timestamp: 1776308363051 } // ……循环节点会有一条记录对应一次执行 ] } }也就是说前端只要拿到这一个 JSON 就能渲染完整的执行时间线——这正是ExecutionRecord里保留started_at/timestamp的用武之地。八、新手落地清单写循环工作流时的 5 个要点默认引用取最新node.field永远指向最新一轮多数场景无需写选择器要历史上下文用[*]如history: think[*].thought让模型看到完整轨迹给回跳边加max_iterations不信任任何一定会收敛的条件兜底边用is_default: true确保计数用尽后仍能离开循环调试时按 short_id 回溯node[a].field可精确钉住某一次执行。写在最后rp-engine 的ExecutionState用三个极简决策撑起了循环数据流的全部安全append-only 让历史永不丢失base36 短 ID 让任意一轮可寻址边级计数 max_iterations让循环必然终止——再加上可完整序列化的 CheckpointState暂停/恢复只是台账的一次落盘。读懂 src/state/ 这一个模块你就掌握了这套引擎状态语义的钥匙。赞分享【免费下载链接】rp-engineYAML-native agent workflow execution engine, written in Rust项目地址https://gitcode.com/gh_mirrors/rp/rp-engine点击查看免费下载相关推荐Requests 库 Cookie Jar 深度解析RequestsCookieJar 如何让 Session 记住登录状态Requests 库 Cookie Jar 深度解析RequestsCookieJar 如何让 Session 记住登录状态 本篇技术指南以 Requests人工智能AI 应用AI AgentRuView Evidence Engine 深度解析为物理感知而生的 Append-Only 精度账本ADR-304 与 ruview-evidence 实现RuView Evidence Engine 深度解析为物理感知而生的 Append Only 精度账本ADR 304 与 ruview evidence人工智能计算机视觉物联网智能家居后端嵌入式OpenEMS ESS 循环控制器状态机Controller Ess Cycle深度解析充放电循环与容量测试的完整状态流转OpenEMS ESS 循环控制器状态机Controller Ess Cycle深度解析充放电循环与容量测试的完整状态流转 导读 本文以 OpenEMS后端前端物联网边缘计算创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考