orbitflow 0.1.0

Lightweight graph-based AI workflow engine.

OrbitFlow

OrbitFlow is a lightweight, graph-based Python workflow engine for composing AI, HTTP, RAG, and custom Python steps. A workflow is a set of typed nodes connected by directed edges. Nodes share a State object, so one step can pass results to the next.

Status: pre-release. The public import package is orbitflow.

Install

Install the released package after it is published:

pip install orbitflow

For local development from a clone:

python -m venv venv
# Windows PowerShell
.\venv\Scripts\Activate.ps1
python -m pip install -e .

OrbitFlow requires Python 3.10 or newer. Ollama is only needed for the OllamaLLM and OllamaEmbeddings adapters.

Your first workflow

Create a workflow with a start node, one or more processing nodes, and an end node. The engine begins at the start node and follows the edges.

from orbitflow import Edge, Engine, Node, Workflow, registry

workflow = Workflow(
    nodes=[
        Node("start", "start"),
        Node("message", "variable", {"name": "greeting", "value": "Hello"}),
        Node("end", "end"),
    ],
    edges=[
        Edge("start", "message"),
        Edge("message", "end"),
    ],
)

result = Engine(registry=registry).run(workflow, input="")
print(result.variables["greeting"])

Engine.run() returns a State with these useful fields:

• input: the input supplied to run()
• output: the latest node output, usually from an LLM node
• variables: named values created by variable nodes
• context: structured data shared between nodes, including RAG and HTTP results
• metadata: execution metadata, including condition results

Use an LLM

Provide an LLM implementation when creating the engine. OrbitFlow includes an Ollama adapter; select the model and host appropriate for your environment.

from orbitflow import Edge, Engine, Node, Workflow, registry
from orbitflow.llm import OllamaLLM

llm = OllamaLLM(model=your_model_name, host=your_ollama_host)
workflow = Workflow(
    nodes=[
        Node("start", "start"),
        Node(
            "answer",
            "llm",
            {
                "system_prompt": your_system_prompt,
                "prompt": your_prompt,
                "temperature": your_temperature,
            },
        ),
        Node("end", "end"),
    ],
    edges=[Edge("start", "answer"), Edge("answer", "end")],
)

result = Engine(registry=registry, llm=llm).run(workflow, input=your_input)
print(result.output)

Use RAG

The RAG node contains no bundled documents, model, query, or result limit. Supply these values from your application. Documents can be strings, Document instances, or dictionaries containing content plus optional id and metadata.

from orbitflow import Edge, Engine, Node, Workflow, registry
from orbitflow.rag import OllamaEmbeddings

workflow = Workflow(
    nodes=[
        Node("start", "start"),
        Node(
            "retrieve",
            "rag",
            {
                "documents": your_documents,
                "embeddings": OllamaEmbeddings(
                    model=your_embedding_model,
                    host=your_ollama_host,
                ),
                "top_k": requested_result_count,
                "output_key": "retrieved_documents",
            },
        ),
        Node("end", "end"),
    ],
    edges=[Edge("start", "retrieve"), Edge("retrieve", "end")],
)

result = Engine(registry=registry).run(workflow, input=your_query)
for match in result.context["retrieved_documents"]:
    print(match["score"], match["content"])

Set query in the RAG node configuration when the query should differ from the workflow input. For repeated runs, build an InMemoryVectorStore, add your documents once, wrap it in a Retriever, and pass that retriever in the node configuration.

Built-in nodes

Type	Purpose
start, end	Mark the workflow boundaries.
variable	Store config["value"] at state.variables[config["name"]].
condition	Compare a variable and route outgoing true / false edges.
llm	Generate a response through the engine's configured LLM.
rag	Retrieve relevant documents into a configured context key.
http	Make a configured HTTP request and store its JSON response in context.
python	Run configured Python with state available. Use only with trusted code.

Create a custom node

Extend BaseNode, register it, then reference its registration name in a workflow.

from orbitflow.nodes.base import BaseNode
from orbitflow.state import State

class UppercaseNode(BaseNode):
    def execute(self, state: State) -> State:
        state.output = str(state.input).upper()
        return state

registry.register("uppercase", UppercaseNode)

Development checks

python -m compileall -q orbitflow
python -m pip check

License

OrbitFlow is released under the MIT License.