feat(tool_runner): allow empty args_schema + logging

chore: pre-commit

Author: boczekbartek

examples/husarion_poc_example_ros_native.py

@@ -4,19 +4,17 @@
 import rclpy
 from langchain_core.messages import HumanMessage, SystemMessage
 from langchain_openai import ChatOpenAI
+from rclpy.node import Node
 
 from rai.scenario_engine.messages import AgentLoop
 from rai.scenario_engine.scenario_engine import ScenarioPartType, ScenarioRunner
 from rai.tools.ros.cat_demo_tools import FinishTool
 from rai.tools.ros.native import (
     Ros2GetOneMsgFromTopicTool,
+    Ros2GetTopicsNamesAndTypesTool,
     Ros2PubMessageTool,
-    Ros2GetTopicsNamesAndTypesTool
 )
 
-from rclpy.node import Node
-from ros2cli.node.strategy import NodeStrategy
-
 
 def main():
     scenario: List[ScenarioPartType] = [
@@ -34,25 +32,25 @@ def main():
 
     rclpy.init()
 
-    rai_node = Node("rai") # type: ignore
+    rai_node = Node("rai")  # type: ignore
 
+    tools = [
+        Ros2GetTopicsNamesAndTypesTool(),
+        Ros2GetOneMsgFromTopicTool(node=rai_node),
+        Ros2PubMessageTool(node=rai_node),
+        FinishTool(),
+    ]
 
-    runner = ScenarioRunner( 
+    runner = ScenarioRunner(
         scenario,
         llm,
+        tools=tools,
         llm_type="openai",
         scenario_name="Husarion example",
         log_usage=log_usage,
-        logging_level="DEBUG"
+        logging_level="DEBUG",
     )
 
-    tools = [
-        Ros2GetTopicsNamesAndTypesTool(),
-        Ros2GetOneMsgFromTopicTool(node=rai_node),
-        Ros2PubMessageTool(node=rai_node),
-        FinishTool(),
-    ]
-
     runner.bind_tools(tools)
 
     runner.run()

src/rai/scenario_engine/scenario_engine.py

@@ -3,7 +3,6 @@
 import os
 import pickle
 from typing import Callable, Dict, List, Literal, Sequence, Union, cast
-from rclpy.node import Node
 
 import coloredlogs
 from langchain_core.language_models.chat_models import BaseChatModel
@@ -17,6 +16,7 @@
 from langchain_core.runnables import RunnableConfig
 from langchain_core.tools import BaseTool
 from langfuse.callback import CallbackHandler
+from rclpy.node import Node
 
 from rai.history_saver import HistorySaver
 from rai.scenario_engine.messages import AgentLoop, FutureAiMessage
@@ -81,10 +81,10 @@ def __init__(
         log_usage: bool = True,
         use_cache: bool = False,
     ):
-        super().__init__(node_name="rai") # type: ignore
+        super().__init__(node_name="rai")  # type: ignore
         self.scenario = scenario
         self.tools = tools
-        for t in self.tools: # TODO(@boczekbartek): refactor to the method 
+        for t in self.tools:  # TODO(@boczekbartek): refactor to the method
             if isinstance(t, BaseRos2NativeTool):
                 t.set_node(self)
         self.log_usage = log_usage

src/rai/scenario_engine/tool_runner.py

@@ -1,3 +1,4 @@
+import logging
 from typing import Any, Dict, List, Literal, Sequence
 
 from langchain.tools import BaseTool
@@ -22,21 +23,31 @@ def images_to_vendor_format(images: List[str], vendor: str) -> List[Dict[str, An
 
 
 def run_tool_call(
-    tool_call: ToolCall, tools: Sequence[BaseTool]
+    tool_call: ToolCall,
+    tools: Sequence[BaseTool],
 ) -> Dict[str, Any] | Any:
     selected_tool = {k.name: k for k in tools}[tool_call["name"]]
+
     try:
-        args = selected_tool.args_schema(**tool_call["args"])  # type: ignore
+        if selected_tool.args_schema is not None:
+            args = selected_tool.args_schema(**tool_call["args"]).dict()
+        else:
+            args = dict()
     except Exception as e:
-        return f"Error in preparing arguments for {selected_tool.name}: {e}"
+        err_msg = f"Error in preparing arguments for {selected_tool.name}: {e}"
+        logging.error(err_msg)
+        return err_msg
 
-    print(f"Running tool: {selected_tool.name} with args: {args.dict()}")
+    logging.info(f"Running tool: {selected_tool.name} with args: {args}")
 
     try:
-        tool_output = selected_tool.run(args.dict())
+        tool_output = selected_tool.run(args)
     except Exception as e:
-        return f"Error running tool {selected_tool.name}: {e}"
+        err_msg = f"Error in running tool {selected_tool.name}: {e}"
+        logging.error(err_msg)
+        return err_msg
 
+    logging.info(f"Successfully ran tool: {selected_tool.name}. Output: {tool_output}")
     return tool_output
 
 

src/rai/tools/ros/native.py

@@ -12,8 +12,10 @@
 
 from .utils import import_message_from_str
 
+
 class Ros2BaseInput(BaseModel):
-    """ Empty input for ros2 tool """ 
+    """Empty input for ros2 tool"""
+
 
 class Ros2BaseTool(BaseTool):
     node: Node = Field(..., exclude=True, include=False, required=True)
@@ -24,10 +26,11 @@ class Ros2BaseTool(BaseTool):
     def logger(self) -> RcutilsLogger:
         return self.node.get_logger()
 
+
 class Ros2GetTopicsNamesAndTypesTool(BaseTool):
     name: str = "Ros2GetTopicsNamesAndTypes"
     description: str = "A tool for getting all ros2 topics names and types"
-    
+
     def _run(self):
         with NodeStrategy(dict()) as node:
             return [
@@ -61,7 +64,9 @@ def _run(self, topic: str, msg_type: str, timeout_sec: int):
         """Gets the current position from the specified topic."""
         msg_cls: Type = import_message_from_str(msg_type)
 
-        qos_profile = rclpy.qos.qos_profile_sensor_data # TODO(@boczekbartek): infer QoS from topic 
+        qos_profile = (
+            rclpy.qos.qos_profile_sensor_data
+        )  # TODO(@boczekbartek): infer QoS from topic
         success, msg = wait_for_message(
             msg_cls,
             self.node,