graphrefly 0.5.0

Reactive graph protocol for human + LLM co-operation. Composable nodes, glitch-free diamond resolution, two-phase push, durable streaming. Zero dependencies.

GraphReFly

Reactive graph protocol for human + LLM co-operation.

One primitive. Zero dependencies. Composable nodes with glitch-free diamond resolution, two-phase push propagation, durable streaming, and async runners for asyncio and trio.

Docs | Spec | TypeScript | API Reference

---

Quick start

pip install graphrefly

from graphrefly import state, derived, effect

count = state(0)
doubled = derived([count], lambda deps, _: deps[0] * 2)

effect([doubled], lambda deps, _: print("doubled:", deps[0]))
# → doubled: 0

count.push(3)
# → doubled: 6

Why GraphReFly?

Most state libraries solve one problem well. GraphReFly solves the space between them:

	Redux / Zustand	RxPY	Pydantic AI	TC39 Signals	GraphReFly
Simple store API	yes	no	no	yes	yes
Streaming operators	no	yes	no	no	yes
Diamond resolution	no	n/a	n/a	partial	glitch-free
Graph introspection	no	no	no	no	describe / observe / diagram
Durable checkpoints	no	no	no	no	file / SQLite / IndexedDB
LLM orchestration	no	no	partial	no	agent_loop / chat_stream / tool_registry
Async runners	n/a	asyncio	asyncio	n/a	asyncio / trio
Dependencies	varies	0	many	n/a	0

One primitive

Everything is a node. Sugar constructors give you the right shape:

from graphrefly import state, derived, producer, effect
from graphrefly.core.messages import DATA

# Writable state
name = state("world")

# Computed (re-runs when deps change)
greeting = derived([name], lambda deps, _: f"Hello, {deps[0]}!")

# Push source (timers, events, async streams)
clock = producer(lambda emit, _: emit([(DATA, time.time())]))

# Side effect
effect([greeting], lambda deps, _: print(deps[0]))

Streaming & operators

70+ operators — transform, combine, buffer, window, rate-limit, retry, circuit-break:

from graphrefly.extra.tier1 import map_op, filter_op, scan
from graphrefly.extra.tier2 import switch_map, debounce_time
from graphrefly.extra.resilience import retry
from graphrefly import pipe

search = pipe(
    user_input,
    debounce_time(0.3),
    switch_map(lambda q: from_promise(fetch(f"/api?q={q}"))),
    retry(strategy="exponential", max_attempts=3),
)

Graph container

Register nodes in a Graph for introspection, snapshot, and persistence:

from graphrefly import Graph, state, derived

g = Graph("pricing")
price = g.register("price", state(100))
tax   = g.register("tax", derived([price], lambda d, _: d[0] * 0.1))
total = g.register("total", derived([price, tax], lambda d, _: d[0] + d[1]))

g.describe()   # → full graph topology as dict
g.diagram()    # → Mermaid diagram string
g.observe(lambda e: print(e))  # → live change stream

AI & orchestration

First-class patterns for LLM streaming, agent loops, and human-in-the-loop workflows:

from graphrefly.patterns.ai import chat_stream, agent_loop, tool_registry
from graphrefly.patterns.memory import collection, decay

# Streaming chat with tool use
chat = chat_stream("assistant", model="claude-sonnet-4-20250514",
                   tools=tool_registry("tools", search=search_fn))

# Full agent loop: observe → think → act → memory
agent = agent_loop("researcher", llm=chat,
                   memory=agent_memory(decay="openviking"))

Async runners

Native asyncio and trio support for async sources and long-running graphs:

from graphrefly.compat.asyncio_runner import AsyncioRunner
from graphrefly.extra.sources import from_async_iter

# Wrap an async generator as a reactive node
async def sse_events():
    async for event in httpx_client.stream("GET", "/events"):
        yield event.data

events = from_async_iter(sse_events())

# Run the graph in an asyncio event loop
runner = AsyncioRunner(graph)
await runner.run()

FastAPI integration

Drop-in integration for reactive backends:

from graphrefly.integrations.fastapi import GraphReflyRouter

router = GraphReflyRouter(graph)
app.include_router(router, prefix="/graph")
# GET /graph/describe  → graph topology
# GET /graph/snapshot  → current state
# WS  /graph/observe   → live change stream

Resilience & checkpoints

Built-in retry, circuit breakers, rate limiters, and persistent checkpoints:

from graphrefly.extra.resilience import retry, circuit_breaker, rate_limiter
from graphrefly.extra.checkpoint import FileCheckpointAdapter, save_graph_checkpoint

# Retry with exponential backoff
resilient = pipe(source, retry(strategy="exponential"))

# Circuit breaker
breaker = circuit_breaker(threshold=5, reset_timeout=30.0)

# Checkpoint to file system
adapter = FileCheckpointAdapter("./checkpoints")
save_graph_checkpoint(graph, adapter)

Project layout

Path	Contents
src/graphrefly/core/	Message protocol, node primitive, batch, sugar constructors
src/graphrefly/extra/	Operators, sources, data structures, resilience, checkpoints
src/graphrefly/graph/	Graph container, describe/observe, snapshot, persistence
src/graphrefly/patterns/	Orchestration, messaging, memory, AI, CQRS, reactive layout
src/graphrefly/compat/	Async runners (asyncio, trio)
src/graphrefly/integrations/	Framework integrations (FastAPI)
docs/	Roadmap, guidance, benchmarks
website/	Astro + Starlight docs site (py.graphrefly.dev)

Scripts

uv run pytest              # run tests
uv run ruff check .        # lint
uv run mypy src/           # type check
uv run pytest --benchmark  # benchmarks

Requirements

Python 3.12 or later. Zero runtime dependencies.

Acknowledgments

GraphReFly builds on ideas from many projects and papers:

Protocol & predecessor:

• Callbag (Andre Staltz) — the original reactive protocol spec. GraphReFly's message-based node communication descends from callbag's function-calling-function model.
• callbag-recharge & callbag-recharge-py — GraphReFly's direct predecessors. The Python port (6 primitives, 18 operators, 100+ tests) established cross-language parity patterns carried forward.

Reactive design patterns:

• SolidJS — two-phase execution (DIRTY propagation + value flow), automatic caching, and effect batching. Closest philosophical neighbor.
• Preact Signals — fine-grained reactivity and cached-flag optimization patterns that informed RESOLVED signal design.
• TC39 Signals Proposal — the .get()/.set() contract and the push toward language-level reactivity.
• RxJS / RxPY — operator naming conventions and the DevTools observability philosophy that inspired the Inspector pattern.

AI & memory:

• OpenViking (Volcengine) — the memory decay formula (sigmoid(log1p(count)) * exp_decay(age, 7d)) and L0/L1/L2 progressive loading strategy used in agent_memory().
• FadeMem (Wei et al., ICASSP 2026) — biologically-inspired dual-layer memory with adaptive exponential decay.
• MAGMA (Jiang et al., 2026) — four-parallel-graph model (semantic/temporal/causal/entity) that informed knowledge_graph() design.
• Letta/MemGPT, Mem0, Zep/Graphiti, Cognee — production memory architectures surveyed during agent_memory() design.

Layout & other:

• Pretext (Cheng Lou) — inspired the reactive layout engine's DOM-free text measurement pipeline.
• CASL — declarative allow()/deny() policy builder DX that inspired policy().
• Nanostores — tiny framework-agnostic API with .get()/.set()/.subscribe() mapping that validated the store ergonomics.

License

MIT