Pipeline Orchestration from Python #680
Replies: 14 comments 2 replies
-
|
Do we want to implement a Python Node Director first?
In a later phase, proven patterns are promoted to native engine/UI primitives. |
Beta Was this translation helpful? Give feedback.
-
|
RPC via Host Adapter (callback bridge) execution model Why this model? RocketRide Server already solved this problem in the agent framework. When agent_rocketride (Wave) needs to invoke an LLM or a tool from Python, it doesn't do it directly — it uses a host adapter (AgentHostServices) that routes the call back to the engine through instance.invoke(param, component_id). The agent framework already has a clean separation between generic and Wave-specific layers: AgentBase (generic - reusable) AgentHostServices already exposes channels for LLM, Tools, and Memory — all auto-discovered from connected nodes via getControllerNodeIds(). This is exactly what the Director needs. What we CAN'T reuse as-is The Wave driver (RocketRideDriver._run()) runs a loop where:
This is LLM-driven. The Python Node Director needs the opposite: Python-code-driven, where the user writes the flow explicitly |
Beta Was this translation helpful? Give feedback.
-
Beta Was this translation helpful? Give feedback.
-
|
Real-World Use Case: Multi-Lane Deduplication (Frame Grabber → Gemini) related Result: duplicated API calls, doubled cost and latency, redundant outputs. The user found a clever caching workaround to call Gemini only once, but this approach is fragile — it requires careful state management to determine whether the incoming data is "the same image" or a new one, and it violates the engine's core contract: Engine rule: For every input on a lane, there is either no output or all outputs are produced. There is no cross-lane deduplication. This rule exists because the engine dispatches per-lane — when N input lanes are connected to a node, the node's handler is invoked N times independently. The engine has no concept of "these two lane events are related." This is not an edge case. Any pipeline that connects multiple output lanes from the same source to a single downstream node will trigger multiple invocations. Common scenarios:
|
Beta Was this translation helpful? Give feedback.
-
Beta Was this translation helpful? Give feedback.
-
Real-World Use Case: Multi-Lane DeduplicationA concrete case that motivates this work: a user connected a Frame Grabber to Gemini Object Recognition using both the The engine rule is correct: for every input on a lane, there is either no output or all outputs are produced. There is no cross-lane deduplication. But the user needed a way to consolidate multi-lane input, apply logic, and emit selectively. Today there is no pipeline node that does this. The Simple Solution First: Extend
|
Beta Was this translation helpful? Give feedback.
-
Practical API: How Users Access Lane Data in PythonFollowing up on the lane-aware The Sandbox API (4 globals, that's it)lane # str — which lane triggered this execution ("text", "documents", etc.)
data # the lane object (str, List[Doc], Question, Answer, dict)
emit # function: emit(target_lane, value) — send to an output lane
state # dict — persists across lane calls within the same entryNo imports, no framework, no boilerplate. What
|
| Lane | data type |
Access pattern |
|---|---|---|
text |
str |
data directly |
table |
str (markdown) |
data directly |
documents |
List[Doc] |
data[0].page_content, data[0].score, data[0].metadata.parent |
questions |
Question |
data.questions[0].text, data.role, data.context |
answers |
Answer |
data.getText(), data.getJson(), data.isJson() |
image |
dict |
data["data"] (base64), data["mimeType"] |
Example 1: Simple transform — documents → text
[PDF Ingestion] → lane:documents → [tool_python] → lane:text → [Response]
for doc in data:
emit("text", doc.page_content)Example 2: Filter documents by relevance score
[Vector DB] → lane:documents → [tool_python] → lane:documents → [LLM]
good_docs = [doc for doc in data if doc.score > 0.75]
if good_docs:
emit("documents", good_docs)Example 3: Documents → Questions (generate questions from content)
[PDF Ingestion] → lane:documents → [tool_python] → lane:questions → [LLM]
for doc in data:
source = doc.metadata.parent
emit("questions", f"Based on {source}, summarize: {doc.page_content[:500]}")Example 4: Answers → Documents (persist LLM output for storage)
[LLM] → lane:answers → [tool_python] → lane:documents → [Vector DB Store]
answer_text = data.getText()
if data.isJson():
import json
answer_text = json.dumps(data.getJson(), indent=2)
emit("documents", answer_text)Example 5: Multi-lane consolidation (Frame Grabber → single Gemini call)
[Frame Grabber] → lane:documents → [tool_python] → lane:questions → [Gemini]
→ lane:image →
if lane == "documents":
state["doc"] = data[0]
elif lane == "image":
doc = state.get("doc")
if doc:
frame = doc.metadata.chunkId
timestamp = doc.metadata.time_stamp
emit("questions",
f"Frame {frame} at {timestamp}s. "
f"Context: {doc.page_content[:200]}. "
f"Describe what you see in this image.")
state.clear()Example 6: Smart routing (alternative to conditional node)
[Webhook] → lane:questions → [tool_python] → lane:questions → [GPT-4]
→ lane:text → [Response]
question_text = data.questions[0].text
if len(question_text) < 20:
emit("text", f"Too short: '{question_text}'. Please elaborate.")
else:
emit("questions", data)Example 7: Enrich metadata before storage
[Embedding] → lane:documents → [tool_python] → lane:documents → [Vector DB]
for doc in data:
doc.metadata.signature = "enriched_v2"
doc.type = "EnrichedDocument"
if doc.embedding and len(doc.embedding) > 0:
emit("documents", [doc])Example 8: Merge table + text into a single document
[Data Source] → lane:text → [tool_python] → lane:documents → [LLM]
→ lane:table →
if lane == "text":
state["text"] = data
elif lane == "table":
text = state.get("text", "")
emit("documents", f"## Analysis
{text}
## Data
{data}")
state.clear()Why this works
- Zero learning curve — users already know Python.
lane,data,emit,stateis the entire API. - Pydantic objects pass through — Doc, Question, Answer are standard models, no serialization needed. Dot-notation access works in the sandbox.
- Covers the common patterns: transform, filter, route, merge, consolidate. All without a full orchestrator.
- Builds toward the Director — once users are comfortable with
emit()and lane-aware Python
Beta Was this translation helpful? Give feedback.
-
Beta Was this translation helpful? Give feedback.
-
|
I think this is a really cool idea, are you proposing like a Python code node essentially that gives developers the power to use Python in their pipelines rather than being constrained to just nodes? |
Beta Was this translation helpful? Give feedback.
-
|
yes that is idea behind it |
Beta Was this translation helpful? Give feedback.
-
Beta Was this translation helpful? Give feedback.
-
Beta Was this translation helpful? Give feedback.
-
|
After cross-reading this with #628 and #408 (programmatic pipeline builder, cc @joshuadarron), I think the three tickets form a natural chain and the work should land in this order:
(engine, C++) — setTargetFilter dispatch filter in binder.cpp, wildcard lane validator in pipeline_config.cpp, branches metadata propagated through services.cpp. +41/−5 across 7 files, additive and default-off. The only task that will be still pending after merge is the docs update (ROCKETRIDE_PIPELINE_RULES.md + ROCKETRIDE_COMPONENT_REFERENCE.md) — called out below as the first post-merge step. Also builds on the exploratory work from @nihalnihalani referenced in #628.
flow_for, flow_while (bounded via bounds.py), flow_switch — additive nodes that reuse the wildcard lane contract and the dispatch filter introduced in #690. No engine changes needed if the filter already supports multi-branch metadata. This keeps the invariant stated at the bottom of this ticket — "Lanes and components remain the flow foundation" — intact. Every decision, every invocation, and every loop iteration becomes an engine-level event that's already traced, sandboxed (flow_base/sandbox.py), and bounded (flow_base/bounds.py). The "Mandatory Observability" section is satisfied by the existing trace infrastructure rather than reimplemented on the client. And the TS and MCP clients get the same capabilities as Python on day one, because everything is declarative pipeline metadata. |
Beta Was this translation helpful? Give feedback.
-
Beta Was this translation helpful? Give feedback.
-
Beta Was this translation helpful? Give feedback.
Uh oh!
There was an error while loading. Please reload this page.
-
This discussion directly defines what we want to build: orchestrating RocketRide pipelines from Python.
Objective
We want Python to control pipeline execution in a programmable, secure, and traceable way:
if/else, dynamic routes).for, boundedwhile).embedding,vector db,llm,tools) through a Python API.What Will Be Implemented
1) Python orchestration API
A library-oriented Python API is exposed (more natural for developers):
from rocketride.nodes import routing, transforms, embedding, vector_db, llmrouting.branch(...)for branching/routing.embedding.transform(...),vector_db.search(...),llm.ask(...)to invoke capabilities.transforms.normalize(...),transforms.map(...),transforms.validate(...)for data transformation.routing.emit(...)to publish controlled outputs.state.get()/set()for execution state per task/token.2) Programmable flow control
if/elseover payload, metadata, and intermediate results.3) Bounded loops
forover doc/chunk collections.whileonly with limits (max_iter, timeout, budget).4) Node invocation from Python
Expected usage:
embedding.transform(input=documents, profile="miniLM", config=...)vector_db.qdrant_search(input=questions, collection="docs", config=...)llm.openai_ask(input=questions, profile="openai-5", config=...)Each call validates lane contracts to prevent invalid connections.
5) Data transformers from Python
A transformer layer is added to prepare data before and after node invocation:
json,dict, lists, lane objects).source,score,trace_id, business tags).Target conversions (examples):
document -> question(generate questions from content or context).document -> table(extract tabular structure from text/document).document -> text(summarization, extraction, or text normalization).table -> text(narrative/insights from rows and columns).text -> question(convert free input into structured questions).question -> text(serialization for logging, audit, or prompts).Use Cases to enabled
LLM to Enhance Python (Concrete Approach)
We also want to enable a mode where Python uses an LLM to decide orchestration steps.
Not to execute arbitrary code, but to propose structured plans that Python validates and executes.
Proposed pattern: Planner + Executor
embedding,vector_db,llm,transforms).Example plan actions:
retrieve_docsrerank_resultsask_llmstore_memoryrespond_userWhy this approach
Security and Control Rules
max_steps,max_invokes, budget).python_orchestration_enabled.Mandatory Observability
Every orchestration action must log:
Success Criteria
Alignment with RocketRide
The implementation respects current principles:
chat()vssend()).Beta Was this translation helpful? Give feedback.
All reactions