pathway-engine 0.1.0

Pathway VM - declarative workflow execution engine for AI agents

Pathway Engine

Declarative workflow execution engine for AI agents.

Installation

pip install pathway-engine

With LLM providers:

pip install pathway-engine[openai]
pip install pathway-engine[anthropic]
pip install pathway-engine[all-llm]

Quick Start

from pathway_engine import Pathway, LLMNode, TransformNode, Connection, PathwayVM, Context

# Define a pathway
pathway = Pathway(
    id="greeting",
    name="Greeting Pipeline",
    nodes={
        "greet": LLMNode(
            id="greet",
            prompt="Write a friendly greeting for {{name}}",
            model="gpt-4o",
        ),
        "format": TransformNode(
            id="format",
            expr="greet.response.upper()",
        ),
    },
    connections=[
        Connection(from_node="greet", to_node="format"),
    ],
)

# Execute
ctx = Context(tools={})  # Add tools as needed
vm = PathwayVM(ctx)
result = await vm.execute(pathway, inputs={"name": "World"})
print(result.outputs)

Core Concepts

Pathway

A directed graph of nodes that defines a workflow.

Nodes

Compute units that process data:

• LLMNode - Call language models
• ToolNode - Execute tools/functions
• TransformNode - Transform data with expressions
• RouterNode - Route to different paths based on conditions
• GateNode - Binary if/else routing
• MemoryReadNode / MemoryWriteNode - Persist state
• EventSourceNode - Stream events (timers, webhooks)
• AgentLoopNode - Agentic tool-use loops

Connection

Data flow between nodes.

Context

Runtime services injected into the VM:

ctx = Context(
    tools={
        "search.web": my_search_handler,
        "workspace.read": my_file_handler,
    },
    memory=my_memory_store,
    extras={"custom": "services"},
)

PathwayVM

Executes pathways:

vm = PathwayVM(ctx)
result = await vm.execute(pathway, inputs={"key": "value"})

Streaming

Built-in event sources for reactive workflows:

from pathway_engine import EventSourceNode, WEBHOOK_BUS

# Timer-based
timer = EventSourceNode(
    id="tick",
    source="timer.interval",
    config={"seconds": 60},
)

# Webhook-based
webhook = EventSourceNode(
    id="events", 
    source="webhook.listen",
    config={"topic": "my-events"},
)

# Publish to webhook
WEBHOOK_BUS.publish("my-events", {"data": "hello"})

Packs & Triggers

Organize pathways into deployable packs:

from pathway_engine import Pack, Trigger, pack_builder

@pack_builder
def my_pack():
    return Pack(
        id="my-pack",
        name="My Pack",
        pathways={"main": my_pathway},
        triggers=[
            Trigger(
                id="on-webhook",
                source="webhook.my-topic",
                pathway="main",
            ),
        ],
    )

License

MIT