refactor: allow passing list of arguments (loose restrictions)

feat: better system prompt

Author: maciejmajek

src/rai/rai/tools/debugging_assistant.py

@@ -21,6 +21,7 @@
     ros2_action,
     ros2_interface,
     ros2_node,
+    ros2_param,
     ros2_service,
     ros2_topic,
 )
@@ -32,8 +33,10 @@ def initialize_graph():
     llm = get_llm_model(model_type="complex_model", streaming=True)
     agent = create_conversational_agent(
         llm,
-        [ros2_topic, ros2_interface, ros2_node, ros2_service, ros2_action],
-        system_prompt="You are a helpful assistant that can answer questions about ROS 2.",
+        [ros2_topic, ros2_interface, ros2_node, ros2_service, ros2_action, ros2_param],
+        system_prompt="""You are a ROS 2 expert helping a user with their ROS 2 questions. You have access to various tools that allow you to query the ROS 2 system.
+                Be proactive and use the tools to answer questions. Retrieve as much information from the ROS 2 system as possible.
+                """,
     )
     return agent
 

src/rai/rai/tools/ros/debugging.py

@@ -14,7 +14,7 @@
 
 from subprocess import PIPE, Popen
 from threading import Timer
-from typing import List, Literal
+from typing import List, Literal, Optional
 
 from langchain_core.tools import tool
 
@@ -55,114 +55,90 @@ def run_command(cmd: List[str], timeout: int = 5):
 @tool
 def ros2_action(
     command: Literal["info", "list", "type"],
-    action_name: str = "",
+    arguments: Optional[List[str]] = None,
     timeout: int = 5,
 ):
     """Run a ROS2 action command
     Args:
         command: The action command to run (info/list/type)
-        action_name: Name of the action (required for info and type)
+        arguments: Additional arguments for the command as a list of strings
         timeout: Command timeout in seconds
     """
-    if command in ["info", "type"]:
-        if not action_name:
-            raise ValueError("Action name is required for info and type commands")
-
     cmd = ["ros2", "action", command]
-    if action_name:
-        cmd.append(action_name)
+    if arguments:
+        cmd.extend(arguments)
     return run_command(cmd, timeout)
 
 
 @tool
 def ros2_service(
     command: Literal["call", "find", "info", "list", "type"],
-    service_name: str = "",
+    arguments: Optional[List[str]] = None,
     timeout: int = 5,
 ):
     """Run a ROS2 service command
     Args:
         command: The service command to run
-        service_name: Name of the service (required for call, info, and type)
+        arguments: Additional arguments for the command as a list of strings
         timeout: Command timeout in seconds
     """
-    if command in ["call", "info", "type"]:
-        if not service_name:
-            raise ValueError(
-                "Service name is required for call, info, and type commands"
-            )
-
     cmd = ["ros2", "service", command]
-    if service_name:
-        cmd.append(service_name)
+    if arguments:
+        cmd.extend(arguments)
     return run_command(cmd, timeout)
 
 
 @tool
-def ros2_node(command: Literal["info", "list"], node_name: str = "", timeout: int = 5):
+def ros2_node(
+    command: Literal["info", "list"],
+    arguments: Optional[List[str]] = None,
+    timeout: int = 5,
+):
     """Run a ROS2 node command
     Args:
         command: The node command to run
-        node_name: Name of the node (required for info)
+        arguments: Additional arguments for the command as a list of strings
         timeout: Command timeout in seconds
     """
-    if command == "info":
-        if not node_name:
-            raise ValueError("Node name is required for info command")
-
     cmd = ["ros2", "node", command]
-    if node_name:
-        cmd.append(node_name)
+    if arguments:
+        cmd.extend(arguments)
     return run_command(cmd, timeout)
 
 
 @tool
 def ros2_param(
     command: Literal["delete", "describe", "dump", "get", "list", "set"],
-    node_name: str = "",
-    param_name: str = "",
+    arguments: Optional[List[str]] = None,
     timeout: int = 5,
 ):
     """Run a ROS2 parameter command
     Args:
         command: The parameter command to run
-        node_name: Name of the node
-        param_name: Name of the parameter (required for get, set, delete)
+        arguments: Additional arguments for the command as a list of strings
         timeout: Command timeout in seconds
     """
-    if command in ["get", "set", "delete"]:
-        if not param_name:
-            raise ValueError(
-                "Parameter name is required for get, set, and delete commands"
-            )
-
     cmd = ["ros2", "param", command]
-    if node_name:
-        cmd.append(node_name)
-    if param_name:
-        cmd.append(param_name)
+    if arguments:
+        cmd.extend(arguments)
     return run_command(cmd, timeout)
 
 
 @tool
 def ros2_interface(
     command: Literal["list", "package", "packages", "proto", "show"],
-    interface_name: str = "",
+    arguments: Optional[List[str]] = None,
     timeout: int = 5,
 ):
     """Run a ROS2 interface command
     Args:
         command: The interface command to run
-        interface_name: Name of the interface (required for show, proto)
+        arguments: Additional arguments for the command as a list of strings
         timeout: Command timeout in seconds
     """
-    if command in ["show", "proto"]:
-        if not interface_name:
-            raise ValueError("Interface name is required for show and proto commands")
-
     cmd = ["ros2", "interface", command]
-    if interface_name:
-        cmd.append(interface_name)
+    if arguments:
+        cmd.extend(arguments)
     return run_command(cmd, timeout)
 
 
@@ -171,7 +147,7 @@ def ros2_topic(
     command: Literal[
         "bw", "delay", "echo", "find", "hz", "info", "list", "pub", "type"
     ],
-    topic_name: str = "",
+    arguments: Optional[List[str]] = None,
     timeout: int = 5,
 ):
     """Run a ROS2 topic command
@@ -186,14 +162,10 @@ def ros2_topic(
             - list: Output a list of available topics
             - pub: Publish a message to a topic
             - type: Print a topic's type
-        topic_name: Name of the topic (required for all commands except list)
+        arguments: Additional arguments for the command as a list of strings
         timeout: Command timeout in seconds
     """
-    if command in ["bw", "delay", "echo", "hz", "info", "pub", "type"]:
-        if not topic_name:
-            raise ValueError("Topic name is required for all commands except list")
-
     cmd = ["ros2", "topic", command]
-    if topic_name:
-        cmd.append(topic_name)
+    if arguments:
+        cmd.extend(arguments)
     return run_command(cmd, timeout)