Xyberos

The cognitive platform for AI systems.

Xyberos is a complete, layered platform for building AI applications — agents, tools, workflows, multi-agent collaboration, streaming, memory, knowledge, planning, plugins, observability, and security. Every subsystem is swappable through stable contracts. The core has zero runtime dependencies.

                 ┌──────────────────────────────┐
                 │          Kernel               │
                 │  Config · Logger · Registry   │
                 │  EventBus · Plugins · Security │
                 └──────────────┬───────────────┘
                                │
    ┌───────────────┐  ┌────────┴────────┐  ┌───────────────┐
    │    Runtime    │  │     Brain        │  │   Contracts   │
    │  sync · async │  │  Pipeline Engine │  │ 15 interfaces │
    └───────┬───────┘  └────────┬─────────┘  └───────────────┘
            │                   │
            └──── Context ──────┘

You bring what the system should do. Xyberos provides how — the pipeline, the memory, the planning, the tools, the agents, the guardrails.

Platform at a Glance

Install

pip install -e .

That's it. No runtime dependencies.

pip install -e ".[dev]"   # pytest + coverage for development

Quick Start

from xyberos import create_app

app = create_app()
print(app.chat("Hello, world!"))  # "Hello, world!"

No API keys. No config. The default EchoLLM echoes your prompt — zero dependencies, zero setup. Swap in a real model when you're ready:

from xyberos.llm import OllamaLLM

app = create_app(llm=OllamaLLM(model="qwen2.5:1.5b"))
print(app.chat("Explain quantum computing in one sentence."))

What You Can Build

AI-Powered IDE or Dev Tool

Multi-agent code review with streaming, guardrails blocking destructive ops, tools for read_file / run_test / git_diff. Each step is a workflow node with human approval.

Robotics Controller

Perception → Plan → Act loop. Hierarchical agents (supervisor → navigation → manipulation). Literal emergency stop via Security.engage_kill_switch() — all motor commands halt immediately.

Customer Support Platform

Intent routing via typed tools, escalation through agent handoffs, refund workflows that pause for human approval, persistent SQLite conversation history, full audit trail.

Autonomous Research Assistant

LLMPlanner decomposes "summarize the state of X" into search → read → synthesize → cite. Every result streams token-by-token.

Anything else

Every subsystem is a plugin surface. The platform is done — the rest is building blocks.

Core Concepts

Security & Kill Switch

app.security.engage_kill_switch("emergency maintenance")
app.chat("hello")  # raises SecurityHaltError

app.security.disengage_kill_switch()
app.chat("hello")  # works again

# Block harmful prompts
from xyberos import Guardrail
app.security.add_guardrail(
    Guardrail("no-hacks", lambda ctx: "hack" not in ctx.prompt)
)

Multi-Agent Collaboration

from xyberos.agents import RoleAgent, handoff, post

def supervisor(context):
    post(context, handoff("worker", sender="supervisor"))
    return context

def worker(context):
    context.response = f"Handled: {context.prompt}"
    return context

app.register_agent(RoleAgent("supervisor", "triage", run=supervisor))
app.register_agent(RoleAgent("worker", "resolver", run=worker))
app.run_agents("escalate this", agent_names=["supervisor", "worker"])

Human-in-the-Loop Workflows

from xyberos.workflows import GraphWorkflow
from xyberos.exceptions import WorkflowPaused

def approve(context):
    if context.metadata.get("approved"):
        context.response = "Approved!"
        return context
    raise WorkflowPaused("Approve this action? yes/no")

graph = GraphWorkflow("approve")
graph.add_node("approve", approve)

run = graph.execute(context)
while run.status == "paused":
    answer = input(run.prompt + " ")   # human decides
    run = graph.resume(run, answer)

Streaming & Async

# Stream tokens as they arrive
app.events.subscribe("brain.token_streamed", lambda e: print(e.data["token"], end=""))
app.chat("Write a haiku about code.")

# Async pipeline
response = await app.achat("Summarize this document.")

Observability

from xyberos.events import EventRecorder

recorder = EventRecorder(limit=10_000).subscribe_to(app.events)
app.chat("hello")
print(recorder.counts())
# {'brain.response_produced': 1, 'brain.memory_stored': 1, ...}

LLM-Driven Planning

app = create_app(
    config={"brain.inject_plan": True},
    planner=LLMPlanner(your_llm),
)
# The model sees: "Plan: 1. research 2. draft 3. review\n\nUser: ..."

Persistent Memory & Knowledge

app = create_app(
    memory=SqliteMemory("chat.db"),       # survives restarts
    knowledge=SqliteKnowledge("facts.db"), # curated domain facts
)
app.knowledge.add("hours", "Support is available 9am-6pm Mon-Fri.")

Production Hardening

Built-in, config-driven, all off by default:

app = create_app(config={
    "brain.max_attempts": 3,       # retry on failure
    "brain.retry_backoff": 0.5,    # exponential backoff
    "brain.rate_limit": 10.0,      # calls per second
    "brain.timeout": 30,           # seconds
})

• Retries with exponential backoff
• Rate limiting with token bucket
• Timeouts on LLM calls
• Checkpointing — paused workflows persist to SQLite across restarts
• Kill switch — emergency halt for all processing

Tests

pip install -e ".[dev]"
pytest

242 tests, 95% coverage. The test suite is the authoritative reference for current behavior.

Documentation

Full documentation at xyberos-docs.pages.dev (or mkdocs serve locally):

• Tutorial
• API Reference
• Roadmap & Vision
• 15 Architecture RFCs

Run locally:

pip install mkdocs mkdocs-material
mkdocs serve

Examples

License

Apache 2.0 — see LICENSE.

Core done. Build anything.

Testing

Run the test suite:

pytest

Run with coverage:

pytest --cov=xyberos

Future Enhancements

The current implementation is a working foundation with a fully automated cognitive pipeline. The enhancement backlog — events and observability, persistent memory and knowledge backends, branching workflows, streaming, multi-agent collaboration, and production hardening — is tracked in the Roadmap.

Notes

• The package requires Python 3.10 or newer.
• The repository uses setuptools packaging.
• The public API is intentionally small and stable at the package root.