Modular AI for Robotics with OpenMind/OM1

OpenMind/OM1

OpenMind/OM1 is a modular AI runtime designed specifically for robotics. Think of it as a flexible framework that helps you build and manage intelligent robotic systems. It's built on top of ROS 2 (Robot Operating System 2), which is the standard for robotics development.

The key benefit here is modularity. Instead of creating one giant, monolithic program for your robot, OM1 allows you to break down the AI logic into smaller, reusable components or "agents." Each agent can have a specific task, like navigating a room, identifying an object, or making a decision. This approach makes your code cleaner, easier to debug, and more scalable.

For a software engineer, OM1's value lies in its structured approach to complexity. Here's how

Simplifies Complex Systems
Building a multi-agent system from scratch is hard. OM1 provides the foundational architecture, so you can focus on the specific AI tasks rather than the low-level communication and coordination between agents. It's like having a well-organized city plan before you start building houses.

Encourages Reusability
Each agent is a self-contained module. This means you can easily reuse an agent you built for one project (e.g., a "pathfinding agent") in another, saving you a ton of time.

Improved Maintainability
When a bug appears, you often know which agent is responsible. You can debug or update that single module without affecting the entire system. This is a massive improvement over traditional, tightly-coupled codebases.

Scalability
Need to add a new feature? Just create a new agent. The modular design makes it easy to scale your robot's capabilities without a complete rewrite.

Before you begin, make sure you have ROS 2 (Humble or later is recommended) and Python 3 installed.

First, you'll need to clone the repository and install the necessary dependencies. Open your terminal and run these commands

# Clone the repository
git clone https://github.com/OpenMind-robotics/OM1.git
cd OM1

# Install dependencies
pip install -r requirements.txt

Let's create a simple agent that just prints a message. We'll call it HelloAgent.

Create a new file, say my_agent.py, and add the following code

import rclpy
from rclpy.node import Node
from om1_agent.agent import Agent

class HelloAgent(Agent):
    def __init__(self):
        super().__init__('hello_agent')
        # This is where you can initialize your agent's state or resources.
        self.timer = self.create_timer(1.0, self.say_hello)

    def say_hello(self):
        self.get_logger().info('Hello from the HelloAgent!')

def main(args=None):
    rclpy.init(args=args)
    agent = HelloAgent()
    rclpy.spin(agent)
    agent.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

Let's imagine a more realistic scenario
a system with a "Monitor" agent and a "Worker" agent. The Monitor agent will send a command to the Worker agent, which will then perform a task.

This agent will listen for a command and then "work" for a few seconds.

File
worker_agent.py

import rclpy
from rclpy.node import Node
from om1_agent.agent import Agent
from std_msgs.msg import String # We'll use a simple string message for communication

class WorkerAgent(Agent):
    def __init__(self):
        super().__init__('worker_agent')
        # Subscribe to a topic where it expects commands
        self.subscription = self.create_subscription(
            String,
            'worker_commands', # Topic name
            self.command_callback,
            10
        )
        self.is_working = False

    def command_callback(self, msg):
        command = msg.data
        if command == "start_work" and not self.is_working:
            self.get_logger().info('Received command: "start_work". Starting task...')
            self.is_working = True
            self.work_timer = self.create_timer(3.0, self.stop_working) # Work for 3 seconds
        elif command == "stop_work":
            self.stop_working()

    def stop_working(self):
        if self.is_working:
            self.get_logger().info('Work complete. Ready for new commands.')
            self.is_working = False
            self.work_timer.cancel() # Stop the timer

def main(args=None):
    rclpy.init(args=args)
    agent = WorkerAgent()
    rclpy.spin(agent)
    agent.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

This agent will send the start_work command to the Worker.

File
monitor_agent.py

import rclpy
from rclpy.node import Node
from om1_agent.agent import Agent
from std_msgs.msg import String

class MonitorAgent(Agent):
    def __init__(self):
        super().__init__('monitor_agent')
        # Create a publisher to send commands to the worker
        self.publisher = self.create_publisher(String, 'worker_commands', 10)
        self.timer = self.create_timer(5.0, self.send_command) # Send a command every 5 seconds
        self.sent_command = False

    def send_command(self):
        if not self.sent_command:
            msg = String()
            msg.data = "start_work"
            self.publisher.publish(msg)
            self.get_logger().info(f'Published command: "{msg.data}" to the worker.')
            self.sent_command = True

def main(args=None):
    rclpy.init(args=args)
    agent = MonitorAgent()
    rclpy.spin(agent)
    agent.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

Now you can run both agents in separate terminals.

Terminal 1

# Run the worker agent
python3 worker_agent.py

Terminal 2

# Run the monitor agent
python3 monitor_agent.py