RocketRide tool_pipe 节点实战：将整条内联管道封装为 Agent 可调用的 `run_pipe` 工具

【免费下载链接】rocketride-serverHigh-performance AI pipeline engine with a C core and 50 Python-extensible nodes. Build, debug, and scale LLM workflows with 13 model providers, 8 vector databases, and agent orchestration, all from your IDE. Includes VS Code extension, TypeScript/Python SDKs, and Docker deployment.项目地址https://gitcode.com/gh_mirrors/ro/rocketride-server点击查看免费下载本文以 tool_pipe 节点文档 为核心骨架深入剖析 RocketRide 中这一特殊tool_*节点的设计、生命周期与接线约束它能把画布上一段内联子管道整体封装成一个 AI Agent 可调用的工具函数run_pipe让 Agent 通过invoke能力驱动任意复杂的多分支流程并取回结果。读完本文你将掌握tool_pipe的五个输出车道语义、配置项tool_description/return_type、返回值提取规则以及三条会被引擎在打开管道时直接拒绝的错误接线方式并能读懂仓库中的钻石合并与嵌套示例。一、核心概念把管道当作工具RocketRide 是一个用于连接数据源、处理器、模型与响应的管道系统。通常 Agent 通过工具节点tool_*调用外部能力而tool_pipe的独特之处在于它把管道系统自身的一段子管道封装为工具。也就是说你不必为每个内部流程单独写工具只要在画布上把节点连好tool_pipe就会把这条子管道暴露给 Agent 调用。从 services.json 可以看到它的服务声明{ title: Pipeline Tool, protocol: tool_pipe://, classType: [tool], capabilities: [invoke], register: filter, node: python, path: nodes.tool_pipe, prefix: tool_pipe, description: [ Exposes an inline pipeline as an agent tool., Connect this nodes output lanes to any pipeline nodes on the same canvas., When an agent calls the tool, the input is routed to every connected output lane., End each connected branch with a response node to return results. ], tile: [run_pipe], lanes: { _source: [text, questions, documents, table, answers] } }关键信息一目了然协议为tool_pipe://归类为tool能力为invokeAgent 通过 invoke 绑定工具节点只有一个通道_source但它同时具备五个输出车道text、questions、documents、table、answers这使它成为所有tool_*节点中唯一拥有真实输出车道的节点——其他工具节点通常只有控制流而tool_pipe可以把输入以五种不同载荷形式写入子管道。二、一次调用背后的完整生命周期当 Agent 调用run_pipe时IInstance.py 中run_pipe第 86-121 行按以下顺序执行归一化入参并校验通过normalize_tool_input解析参数取出data若data为空或缺失直接抛出ValueError(tool_pipe: tool requires a non-empty data parameter)。打开子管道self.instance.open(entry)打开子管道实例。扇出输入_send_to_connected_lane(data)把输入字符串写入每一个有监听者的输出车道无监听者的车道自动跳过。冲刷并关闭在finally中依次调用self.instance.closing()冲刷缓冲与self.instance.close()且按依赖顺序关闭——一个 join 节点只有在它所有上游分支都被冲刷之后才会被冲刷因此钻石形diamond子管道返回的是两条分支合并后的输出而不是仅第一条分支的结果。冲刷动作在读取响应值之前完成。读取响应把entry.response通过IJson.toDict转成 Python 字典再按配置的return_type提取返回值。这些响应条目会被快照后移除不会泄漏到父管道中。返回最终返回{result: 字符串}。子管道在每次调用时都经历 open → 输入 → flush(closing) → close 的完整周期因此完全隔离每次调用都是全新的一次运行父管道不会保留子管道的中间状态。三、五个输出车道与负载格式原文档明确指出Agent 传入的输入字符串会写入每一个已连接的输出车道。各车道的载荷格式如下车道发送到下游的载荷text原始输入字符串questions一个Question对象输入作为其中的 questiondocuments单个Doc输入作为page_contenttable原始输入字符串answers一个Answer对象输入作为 answer这一行为在源码中一一对应IInstance.py 的_send_to_connected_lane逐车道检查self.instance.hasListener(...)有监听者才写入if self.instance.hasListener(questions): q Question() q.addQuestion(data) self.instance.writeQuestions(q) if self.instance.hasListener(documents): self.instance.writeDocuments([Doc(page_contentdata)]) if self.instance.hasListener(table): self.instance.writeTable(data) if self.instance.hasListener(text): self.instance.writeText(data) if self.instance.hasListener(answers): answer Answer() answer.setAnswer(data) self.instance.writeAnswers(answer)实用含义如果你想在下游既做文档检索又做问答生成可以把documents车道接检索节点、questions车道接 LLM同一份输入会自动以合适的载荷类型进入两条分支。四、run_pipe工具契约作为工具暴露的函数在 IInstance.py 的装饰器定义 中声明函数说明tool_pipe.run_pipe运行已连接的内联管道必填非空data返回result参数契约input_schemadatastring类型必填语义为“要发送给管道的输入”缺失或空字符串会抛出错误ValueError。返回契约output_schema返回一个对象包含唯一的result字符串字段。result的描述随return_type变化源码中维护了一张映射表IInstance.py 第 48-53 行return_typeresult 描述text管道的纯文本结果answers管道生成的 LLM 回答字符串documents文档对象数组的 JSON 序列化table表格数据的 JSON 序列化tool_pipe是默认的 server-name 前缀。响应是来自配置返回车道的字符串输入缺失会报错而管道未产生所选响应时返回空结果见下文“常见陷阱”。五、配置详解tool_pipe 节点文档 说明节点使用单一默认配置档案default profile初始为空描述、返回text车道。挂载到 Agent 前应先配置描述并让返回类型与子管道实际产出的响应车道匹配。5.1 Tool Description工具描述用自然语言描述所连接子管道的用途帮助 Agent 判断“何时该调用这个工具”。默认值为空字符串会让 Agent 难以选择该工具。例如一个摘要分支可以配置为Summarize a supplied support ticket总结一份传入的支持工单。5.2 Return Type返回类型常规文本响应使用text仅当所连接的响应节点确实写入answers、documents或table车道时才选择对应类型类型不匹配会返回空结果——看起来像是成功但毫无帮助的调用。5.3 完整 Schema原文档末尾给出了由nodes:docs-generate自动生成的参数 Schema生成内容勿手工编辑完整继承如下字段类型描述默认值tool_pipe.return_typestringReturn Type将哪个响应车道的值返回给 Agenttexttool_pipe.tool_descriptionstringTool DescriptionAgent 用来决定何时调用此工具的自然语言描述5.4 源码中的配置校验IGlobal.py 是节点的全局共享状态它在启动时读取工具配置VALID_RETURN_TYPES {text, answers, documents, table} def beginGlobal(self) - None: ... cfg Config.getNodeConfig(self.glb.logicalType, self.glb.connConfig) self.tool_description str(cfg.get(tool_description) or ).strip() self.return_type str(cfg.get(return_type) or ).strip() or text if self.return_type not in VALID_RETURN_TYPES: raise Exception(ftool_pipe: return_type must be one of {sorted(VALID_RETURN_TYPES)})要点return_type只接受{text, answers, documents, table}四个取值非法值在beginGlobal时直接抛异常validateConfig阶段则以 warning 形式提示tool_description读取后.strip()空串会回退为工具函数装饰器中定义的默认描述Accepts a text input and returns the pipeline result.配置默认值与 services.json 的 fields 定义 一致tool_pipe.tool_description默认tool_pipe.return_type默认text且return_type带有[text, answers, documents, table]的枚举约束。六、返回值提取规则子管道完成后节点按return_type读取响应值_extract_return_valuetext作为普通字符串直接返回answers响应是一个列表取第一个answer 转为字符串返回documents/table列表或字典值会被 JSON 序列化ensure_asciiFalse为字符串交给 Agent缺失值返回空字符串。陷阱Gotcha如果子管道完全没有返回数据工具会返回空的result。务必确保每个已连接分支都以响应节点response node收尾且喂给return_type所指车道的那个分支确实产生了响应——否则 Agent 会收到一次“成功但无内容”的调用结果。七、子管道所有权三条会被拒绝的接线由于tool_pipe每次调用都会 open、flush、close 自己的子管道它所触达的每个节点都不能有第二个生命周期所有者。以下三种接线会在管道打开时被直接拒绝fail fast而不是静默产出残缺结果与主流程或第二个 start 共享节点若子管道节点同时从 source 可达主流程喂给它或子管道回流进主流程节点则该节点归主流程所有会在对象结束时才被冲刷导致工具读到不完整结果。应保持每个分支自包含并让每个分支以自身响应节点收尾。在两个tool_pipe之间共享节点一个节点被两个 invoke 节点触达时所有权歧义应各自拥有独立子管道。在 invoke 节点上直接接数据输入tool_pipe没有输入车道只能通过工具控制缝control seam驱动——从 services.json 可见它只声明了_source输出车道没有输入车道因此数据喂入会在车道校验阶段就被拒绝报错形如Component pipe_tool_1 input lane questions not found in service definition。两个容易混淆的“合法”情形两个 Agent 调用同一个tool_pipe合法——那是一个子管道、一个所有者每次调用运行一次嵌套合法——子管道内可以包含一个再调用另一个tool_pipe的 Agent每一层按顺序冲刷自己的子管道。可运行的反面示例仓库 examples/incorrect 目录收录了每种拒绝情形的可运行示例全部故意无效引擎在client.use(...)打开管道时立即抛出RuntimeError属于验证期报错、早于任何数据流动文件错误模式引擎报错原文incorrect-second-start-feeds-subpipe.pipe第二个 startchat_1同时喂给子管道头节点sub_headControl node Pipeline Tool (pipe_tool_1) reaches node Prompt (sub_head) that the main flow owns; a control nodes sub-pipeline must not be shared with the main pipeline or another startincorrect-second-start-feeds-subpipe-node.pipe第二个 start 喂给子管道中间节点sub_mid同上 “…must not be shared with the main pipeline or another start”incorrect-subpipe-merges-into-main.pipe子管道回流进主流程节点main_node工具管道与主管道相交同上incorrect-shared-subpipe-node.pipe两个tool_pipepipe_tool_1、pipe_tool_2的子管道都包含shared_nodePipeline node Prompt (shared_node) is reachable from two control roots ( Pipeline Tool (pipe_tool_1) and Pipeline Tool (pipe_tool_2) ) - a node has exactly one lifecycle ownerincorrect-data-fed-tool-pipe.pipe给tool_pipe同时驱动子管道接数据输入Component pipe_tool_1 input lane questions not found in service definition如何读懂报错引擎用服务标题画布上的标签如Pipeline Tool加组件 id如pipe_tool_1来点名节点所以报错信息直接指向需要修复的接线。例如examples/incorrect/README.md中提到的复现方式启动本分支构建的引擎后运行配套脚本加载对应.pipe文件client.use()即抛出携带引擎消息的RuntimeError管道不会运行。八、实战示例8.1 钻石合并子管道diamondexamples/tool-pipe-diamond.pipe 展示了tool_pipe最有代表性的用法——两条分支汇入一个 joinpipe_tool_1tool_pipe被agent_1CrewAI以control方式绑定配置为return_type: answers输入通过text车道同时扇出到branch_a_1与branch_b_1两个 Prompt 分支两条分支的questions汇入join_1Prompt指令要求确认两条分支的内容都出现后再回答join_1→sub_llm_1LLM→sub_response_1response_answers收尾Agent 指令明确要求“对每条用户消息恰好调用一次run_pipe把用户文本作为data参数传入并把工具结果作为回答返回”。由于子管道按依赖顺序冲刷join 会等两条分支都完成后再产出所以 Agent 拿到的result是两分支合并后的答案而不是某一支的局部结果——这正是“钻石管道返回合并输出”的落地体现。8.2 嵌套 tool_pipeexamples/tool-pipe-nested.pipe 演示了合法嵌套外层tp1tool_pipe的子管道里包含一个内层 Agentagent_2而agent_2又绑定了内层tp2tool_pipe驱动自己的钻石子管道。每一层tool_pipe都拥有并独立冲刷自己的子管道最终结果逐层回流内层钻石 →agent_2→sub_resp1→ 外层tp1→agent_1→response_1。这印证了文档所述“嵌套没问题每一层按顺序冲刷自己的子管道”。九、依赖与运行前提tool_pipe 的 requirements.txt 是空的——节点没有外部 Python 依赖运行时仅依赖 RocketRide 引擎自身的rocketlib与ai.common模块。节点目录结构非常精简init.py 只负责按需加载requirements.txt并导出IGlobal、IInstance两个类分别承载全局配置与每次调用的实例逻辑。从源码结构看节点的工程形态是标准 RocketRide Python 节点services.json声明协议/车道/字段IGlobal管理启动期配置校验IInstance实现工具调用与子管道生命周期。若要把它接入你自己的项目只需要在画布上放置tool_pipe节点 → 配置tool_description与return_type→ 从_source的输出车道连出你自己的子管道分支 → 每个分支以响应节点收尾 → 把tool_pipe以control方式绑定到 Agent 即可。关键文件速查tool_pipe 节点文档本文核心依据IInstance.pyrun_pipe工具实现、车道扇出、返回值提取IGlobal.py配置读取与return_type合法性校验services.json协议、车道、字段与默认配置声明tool-pipe-diamond.pipe钻石合并合法示例tool-pipe-nested.pipe嵌套tool_pipe合法示例incorrect 示例说明三种被拒绝接线的完整对照表。赞分享【免费下载链接】rocketride-serverHigh-performance AI pipeline engine with a C core and 50 Python-extensible nodes. Build, debug, and scale LLM workflows with 13 model providers, 8 vector databases, and agent orchestration, all from your IDE. Includes VS Code extension, TypeScript/Python SDKs, and Docker deployment.项目地址https://gitcode.com/gh_mirrors/ro/rocketride-server点击查看免费下载相关推荐RocketRide tool_drive 节点将 Google Drive API v3 封装为可安全治理的 Agent 工具RocketRide tool_drive 节点将 Google Drive API v3 封装为可安全治理的 Agent 工具 RocketRide 的 tRocketRide agent_langchain 节点实战在管道中接入 LangChain 单 Agent 工具调用循环RocketRide agent_langchain 节点实战在管道中接入 LangChain 单 Agent 工具调用循环 RocketRide 是一个以dxwrapper 老游戏兼容修复实战给二十年前的 DirectX 游戏配一台翻译机dxwrapper 老游戏兼容修复实战给二十年前的 DirectX 游戏配一台翻译机 周六深夜你翻出积灰的光盘双击 exe屏幕黑了一秒随后弹出一行冰冷游戏开发图形学上一篇C 快速参考备忘清单语法基础、智能指针与多线程并发实战下一篇knowledge-work-plugins 产品管理插件连接器指南~~category 占位符机制与十大工具类别全解析创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考