NEXOS AGENTS

NEXOS AGENTS is an enterprise-grade, model-agnostic multi-agent operating system and runtime.

• Cyclic Directed State Graphs (NexosGraph): State machine execution with zero-overhead durable checkpointing and time-travel debugging.
• Decentralized Peer-to-Peer Swarms (NexosSwarm): Powered by Conflict-Free Replicated Data Types (CRDTs) and work-stealing gossip.
• Hierarchical Hive-Mind Clusters (NexosHive): Queen-governed task decomposition with quadratic consensus voting and unified vector memory.
• Sub-Millisecond Distributed Telemetry (NexosTrace): Trace waterfalls and token accounting.
• Repo-Harness Self-Trainer (nexos train --repo): AST call-graph extraction and DSPy MIPRO-style test optimization.

Installation

pip install nexos-agents

Quickstart

from nexos.runtime import StateGraph

graph = StateGraph()

def research_step(state: dict) -> dict:
    state["findings"] = ["Vector consensus confirmed"]
    return state

def synthesize_step(state: dict) -> dict:
    state["output"] = "Pipeline executed deterministically."
    return state

graph.add_node("researcher", research_step)
graph.add_node("synthesizer", synthesize_step)
graph.set_entry_point("researcher")
graph.add_edge("researcher", "synthesizer")
graph.add_edge("synthesizer", "__END__")

result = graph.execute({"task": "Analyze telemetry anomalies"})
print("Execution Result:", result)

Decentralized Swarm (CRDT)

from nexos.swarm import SwarmCluster, SwarmNode, SwarmTask

cluster = SwarmCluster()
node_a = SwarmNode("node-a", "analyst", ["compute"])
node_b = SwarmNode("node-b", "synthesizer", ["compute"])
cluster.add_node(node_a)
cluster.add_node(node_b)

task = SwarmTask(title="Process data stream", payload={"items": [1, 2, 3]})
cluster.broadcast_task(task)

summary = cluster.run_until_idle()
print("Swarm converged:", summary)

Links

• GitHub Repository: https://github.com/socrates6969/nexos-agents
• Production Web Studio: https://nexos-agents.vercel.app
• License: Apache 2.0