NEXOS GRAPH STATE MACHINE ENGINE
Cyclic State Graphs with Durable Checkpointing
The definitive foundation for cyclical multi-agent workflows. Move beyond linear chains into stateful graphs with human-in-the-loop approval, time-travel rollbacks, and CRDT state convergence.
Cyclic Feedback Loops
Construct recursive reasoning cycles with guaranteed convergence bounds. Agents critique, iterate, and verify until state predicates evaluate to true.
Durable Checkpointing
Every graph node transition automatically saves immutable state checkpoints to Postgres, Redis, or SQLite with sub-millisecond write latency.
Time-Travel & Human Interruption
Pause execution before critical mutations for human sign-off (`interrupt_before`). Rewind graph state to any historical step and re-fork alternative paths.
100% Python & TypeScript Parity
from nexos.runtime import StateGraph, MemorySaver
# 1. Initialize cyclic state graph with durable saver
checkpointer = MemorySaver()
graph = StateGraph(checkpointer=checkpointer)
@graph.node("researcher")
def retrieve_hypotheses(state: dict) -> dict:
return {"hypotheses": ["Vector state verified"]}
@graph.node("auditor")
def audit_consensus(state: dict) -> dict:
return {"status": "VERIFIED", "risk_score": 0.001}
# 2. Add edges with conditional branch
graph.add_edge("researcher", "auditor")
graph.add_conditional_edges(
"auditor",
lambda state: "human_review" if state["risk_score"] > 0.05 else "__END__"
)
graph.set_entry_point("researcher")
app = graph.compile(interrupt_before=["human_review"])
run = app.execute({"task": "Sub-millisecond verification"})