Xyberos

The cognitive platform for AI systems.

Xyberos is a continuously evolving, layered platform for building AI applications. Its architecture is designed to support capabilities including agents, tools, workflows, multi-agent collaboration, streaming, memory, knowledge, planning, trainable intent and learning engines, plugins, observability, and security as the platform develops. Every subsystem is built around stable contracts, allowing components to be independently extended, replaced, or improved over time. The core is designed with 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 xyberos

That's it. Zero runtime dependencies — the standard library is all it needs.

pip install xyberos[dev]     # pytest + coverage for development

Or install from source:

git clone https://github.com/xyberos/xyberos.git
cd xyberos
pip install -e .

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."))

Fully-local, LLM-free-in-practice

One local Ollama server can provide both chat and real semantic embeddings (OllamaEmbeddingLLM calls /api/embed over stdlib HTTP — no SDK). Plug both into the hybrid router and common requests are answered by the LLM-free tiers (template → tool → knowledge → memory → cache), with the LLM reserved for the novel tail and teaching the cache:

from xyberos import create_semantic_app
from xyberos.llm import OllamaLLM, OllamaEmbeddingLLM

app = create_semantic_app(
    llm=OllamaLLM(model="qwen2.5:1.5b"),
    embedder=OllamaEmbeddingLLM(model="nomic-embed-text"),  # ollama pull nomic-embed-text
    router="hybrid",
)

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 xyberos[dev]
pytest

533 tests, 89% coverage. The test suite is the authoritative reference for current behavior.

Documentation

The full docs are hosted at docs.xyberos.com and live in docs/. Start here:

• Tutorial — build your first app
• Training Tutorial — capture, feedback, learn, evaluate, and distill
• Configuring Services — explicit, factory, and plugin wiring
• Extension Surfaces — contracts, plugins, and customization
• Lifecycle & Services — start/stop and service behavior
• API Reference
• Roadmap & Vision

Architecture RFCs — the reasoning behind every layer, all in docs/RFCs/:

• RFC-0001 Architecture through RFC-0018 Smarter Learning

Run the docs locally:

pip install mkdocs mkdocs-material
mkdocs serve

Public API Map

Import the main facade from the package root:

from xyberos import Xyberos, achat, chat, create_app

Useful supporting modules:

• xyberos.kernel — configuration, logging, registry, lifecycle, event bus
• xyberos.runtime — cognitive context and runtime execution (sync + async)
• xyberos.brain — automated cognitive pipeline
• xyberos.agents — multi-agent runtime, roles, messaging, and handoffs
• xyberos.workflows — sequential workflows, state graphs, and checkpoints
• xyberos.plugins — plugin loading and auto-discovery (entry points + convention scan)
• xyberos.llm — model providers (incl. local OllamaLLM + OllamaEmbeddingLLM), streaming/async, structured output, and adapters
• xyberos.memory / xyberos.knowledge — in-memory, SQLite, vector, and consolidating providers
• xyberos.planner — fixed, LLM, adaptive, reflective planners, and plan execution
• xyberos.intent — heuristic, LLM, embedding, and cascade intent engines
• xyberos.vector — vector store contract and providers (cosine, chroma, pgvector)
• xyberos.experience / xyberos.learning — episode store, promote/demote, example promotion
• xyberos.trainer — offline training/distillation and artifact registry
• xyberos.tools — registries, runners, and typed function tools
• xyberos.events — event bus, tracing, and exporters
• xyberos.utils — resilience helpers (retry, rate limiting, timeouts) + evaluation metrics
• xyberos.contracts — extension contracts
• xyberos.exceptions — typed domain exceptions

Reading Order

New to the project? Read the docs in this order:

1. This README
2. docs/extensions.md
3. docs/tutorial.md
4. docs/api-reference.md
5. docs/lifecycle.md
6. docs/RFCs/RFC-0001-architecture.md, then the remaining RFCs in docs/RFCs/

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.