Typed Graph-based Pipeline Builder

A pydantically typed, lightweight graph framework for Python that combines features from Langgraph with static type security.

A community collection of nodes will be available here.

Overview

Edgy Graph is a framework for building and executing graph-based pipelines. It supports:

• Pydantic Typing
  Built on Pydantic and Generics for complete static type safety.
• Inheritance and Variance:
  Easily extend and specialize state and node classes.
• Parallel Task Processing:
  Multiple nodes can run simultaneously
• Dual State Management:
  • State with automatic change extraction and conflict detection
  • Shared state accessible by all nodes, protected via explicit locking
• Flexible Routing:
  Define simple node-to-node edges or dynamic routing based on functions.
• Streaming:
  A standardized interface for streaming data between nodes.

Installation

PyPI

pip install edgygraph

Python 3.13+ is required

Example Workflow

Import Classes

from edgygraph import State, Shared, Node, START, END, Graph
import asyncio

Create a State

class MyState(State):

    capslock: bool = False

Create a Node

class MyNode(Node[MyState, Shared]):

    async def run(self, state: MyState, shared: Shared) -> None:

        if state.capslock:
            print("HELLO WORLD!")
        else:
            print("Hello World!")

Create Instances

state = MyState(capslock=True)
shared = Shared()

node = MyNode()

Create a Graph

Graph[MyState, Shared](
    edges=[
        (
            START,
            node
        ),
        (
            node,
            END
        )
    ]
)

Run Graph

asyncio.run(graph(state, shared))

More examples can be found in the examples folder