【免费下载链接】rocketride-server
High-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.
本文以 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 node's 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，能力为invoke（Agent 通过 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对象，输入作为其中的 question |
documents | 单个Doc，输入作为page_content |
table | 原始输入字符串 |
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_content=data)]) 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_schema）：
data：string类型，必填，语义为“要发送给管道的输入”；- 缺失或空字符串会抛出错误（
ValueError）。
返回契约（output_schema）：返回一个对象，包含唯一的result字符串字段。result的描述随return_type变化，源码中维护了一张映射表（IInstance.py 第 48-53 行）：
return_type | result 描述 |
|---|---|
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_type | string | Return Type：将哪个响应车道的值返回给 Agent | "text" |
tool_pipe.tool_description | string | Tool Description：Agent 用来决定何时调用此工具的自然语言描述 | "" |
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(f'tool_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_value）：
text：作为普通字符串直接返回；answers：响应是一个列表，取第一个answer 转为字符串返回；documents/table：列表或字典值会被 JSON 序列化（ensure_ascii=False）为字符串交给 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 | 第二个 start（chat_1）同时喂给子管道头节点sub_head | Control node "Pipeline Tool" (pipe_tool_1) reaches node "Prompt" (sub_head) that the main flow owns; a control node's sub-pipeline must not be shared with the main pipeline or another start |
| incorrect-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_pipe（pipe_tool_1、pipe_tool_2）的子管道都包含shared_node | Pipeline 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 owner |
| incorrect-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 钻石合并子管道（diamond）
examples/tool-pipe-diamond.pipe 展示了tool_pipe最有代表性的用法——两条分支汇入一个 join：
pipe_tool_1（tool_pipe）被agent_1（CrewAI）以control方式绑定，配置为return_type: "answers"；- 输入通过
text车道同时扇出到branch_a_1与branch_b_1两个 Prompt 分支； - 两条分支的
questions汇入join_1（Prompt，指令要求确认两条分支的内容都出现后再回答）； join_1→sub_llm_1（LLM）→sub_response_1（response_answers）收尾；- Agent 指令明确要求“对每条用户消息恰好调用一次
run_pipe，把用户文本作为data参数传入，并把工具结果作为回答返回”。
由于子管道按依赖顺序冲刷，join 会等两条分支都完成后再产出，所以 Agent 拿到的result是两分支合并后的答案，而不是某一支的局部结果——这正是“钻石管道返回合并输出”的落地体现。
8.2 嵌套 tool_pipe
examples/tool-pipe-nested.pipe 演示了合法嵌套：外层tp1（tool_pipe）的子管道里包含一个内层 Agentagent_2，而agent_2又绑定了内层tp2（tool_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.py：
run_pipe工具实现、车道扇出、返回值提取； - IGlobal.py：配置读取与
return_type合法性校验； - services.json：协议、车道、字段与默认配置声明；
- tool-pipe-diamond.pipe：钻石合并合法示例；
- tool-pipe-nested.pipe：嵌套
tool_pipe合法示例； - incorrect 示例说明：三种被拒绝接线的完整对照表。
【免费下载链接】rocketride-server
High-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.
相关推荐
RocketRide tool_drive 节点：将 Google Drive API v3 封装为可安全治理的 Agent 工具
RocketRide tool_drive 节点：将 Google Drive API v3 封装为可安全治理的 Agent 工具 RocketRide 的 t
RocketRide agent_langchain 节点实战：在管道中接入 LangChain 单 Agent 工具调用循环
RocketRide agent_langchain 节点实战：在管道中接入 LangChain 单 Agent 工具调用循环 RocketRide 是一个以
dxwrapper 老游戏兼容修复实战：给二十年前的 DirectX 游戏配一台翻译机
dxwrapper 老游戏兼容修复实战：给二十年前的 DirectX 游戏配一台翻译机 周六深夜，你翻出积灰的光盘，双击 exe，屏幕黑了一秒，随后弹出一行冰冷
游戏开发图形学
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
