Agents
XYBEROS provides a multi-agent runtime with communication, shared memory, and task delegation.
Built-in Agents
| Agent | Role |
|---|---|
PlannerAgent |
Decomposes goals into sub-tasks and delegates |
ResearcherAgent |
Queries knowledge, memory, and web sources |
ReviewerAgent |
Quality, safety, and consistency checks |
ExecutorAgent |
Runs actions via ActionExecutor |
Using Agents
from xyberos.agents import Agent, AgentManager
# Create a manager
manager = AgentManager()
# Register agents
manager.register("planner", PlannerAgent())
manager.register("researcher", ResearcherAgent())
# Get an agent
agent = manager.get("planner")
Creating a Custom Agent
from xyberos.agents import Agent
from xyberos.agents.context import AgentContext
from xyberos.agents.message_bus import Message
class MyAgent(Agent):
@property
def name(self) -> str:
return "my_agent"
async def execute(self, task: str, context: AgentContext) -> str:
# Process the task
result = f"Processed: {task}"
# Send a message to another agent
await self.send_message(
Message(
sender=self.name,
recipient="reviewer",
content=result,
)
)
return result
Agent Communication
Agents communicate via a MessageBus:
from xyberos.agents import Message, MessageBus
bus = MessageBus()
# Send a message
await bus.publish(Message(
sender="planner",
recipient="researcher",
content="Research the latest AI trends",
metadata={"priority": "high"},
))
# Subscribe to messages
async def handler(msg: Message):
print(f"Received: {msg.content}")
await bus.subscribe("researcher", handler)
Task Delegation
from xyberos.agents import Task, TaskManager
manager = TaskManager()
# Create a delegatable task
task = Task(
id="task-001",
description="Summarize the document",
assigned_to="researcher",
priority=1,
)
# Dispatch
await manager.dispatch(task)