call one tool at a time

agent.py

@@ -4,6 +4,7 @@ import os
 import logging
 from groq import Groq
 import functions
+from utils import python_type, raise_error
 from tools import tools
 
 # Set up logging
@@ -31,14 +32,13 @@ def create_message(prompt, message_type):
         )
     elif message_type == "function_call":
         system_message = (
-            "You are a movie search assistant bot who uses TMDB to help users "
-            "find movies. Think step by step and identify the sequence of "
-            "function calls that will help to answer the user's query. Use the "
-            "available functions to gather the necessary data. "
-            "Do not call multiple functions when they need to be executed in sequence. "
-            "Only call multiple functions when they can be executed in parallel. "
-            "Stop with a discover_movie function call that returns a list of movie ids. "
-            "Ensure the discover_movie function call includes all the necessary parameters to filter the movies accurately."
+            "You are a movie search assistant bot that utilizes TMDB to help users find movies. "
+            "Approach each query step by step, determining the sequence of function calls needed to gather the necessary information. "
+            "Execute functions sequentially, using the output from one function to inform the next function call when required. "
+            "Only call multiple functions simultaneously when they can run independently of each other. "
+            "Once you have identified all the required parameters from previous calls, "
+            "finalize your process with a discover_movie function call that returns a list of movie IDs. "
+            "Ensure that this call includes all necessary parameters to accurately filter the movies."
         )
     else:
         raise ValueError(
@@ -57,8 +57,8 @@ def create_message(prompt, message_type):
 
 
 def get_response(client, model, messages, tool_choice="auto"):
-    logging.debug(
-        f"Getting response with model: {model}, messages: {messages}, tool_choice: {tool_choice}")
+    logging.info(
+        f"Getting response with model: {model}, \nmessages: {json.dumps(messages, indent=2)}, \ntool_choice: {tool_choice}")
     response = client.chat.completions.create(
         model=model,
         messages=messages,
@@ -79,27 +79,135 @@ def generate_reasoning_chain(user_prompt):
     if cot_response.choices[0].finish_reason == "stop":
         return cot_response.choices[0]
     else:
-        logging.error(
-            "Failed to generate reasoning chain. Got response: ", cot_response)
-        raise Exception("Failed to generate reasoning chain")
+        raise_error("Failed to generate reasoning chain. Got response: " +
+                    str(cot_response), Exception)
 
 
-def gather_movie_data(messages, iteration=0, max_iterations=2):
+def validate_params(tool_params, param_name, param_value):
+    """
+    Checks if the parameter value matches with the one defined in tools.py
+    """
     logging.debug(
-        f"Gathering movie data with messages: {messages}, iteration: {iteration}")
-    response = get_response(client, MODEL, messages, tool_choice="required")
+        f"Validating parameter: {param_name} with value: {param_value}")
+    param_def = tool_params.get(param_name, None)
+    if param_def is None:
+        logging.error(
+            f"Parameter {param_name} not found in tools. Dropping this tool call.")
+        return False
+    try:
+        param_value = python_type(param_def["type"])(param_value)
+    except ValueError:
+        logging.error(
+            f"Parameter {param_name} value cannot be cast to {param_def['type']}. Dropping this tool call.")
+        return False
+    return True
+
+
+def extract_leaf_values(json_obj):
+    """Recursively extract leaf values from a JSON object or string."""
+    # Check if the input is a string and try to parse it
+    if isinstance(json_obj, str):
+        try:
+            json_obj = json.loads(json_obj)
+        except json.JSONDecodeError:
+            return [json_obj]  # Return the string if it's not valid JSON
+
+    if isinstance(json_obj, dict):
+        values = []
+        for value in json_obj.values():
+            values.extend(extract_leaf_values(value))
+        return values
+    elif isinstance(json_obj, list):
+        values = []
+        for item in json_obj:
+            values.extend(extract_leaf_values(item))
+        return values
+    else:
+        return [json_obj]
+
+
+def is_tool_valid(tool_name):
+    """Check if the tool name is valid and return its definition."""
+    return next((tool for tool in tools if tool["function"]["name"] == tool_name), None)
+
+
+def validate_tool_parameters(tool_def, tool_args):
+    """Validate the parameters of the tool against its definition."""
+    tool_params = tool_def["function"]["parameters"]["properties"]
+    for param_name, param_value in tool_args.items():
+        if not validate_params(tool_params, param_name, param_value):
+            logging.error(
+                f"Invalid parameter {param_name} for tool {tool_def['function']['name']}. Dropping this tool call.")
+            return False
+    return True
+
+
+def are_arguments_valid(tool_args, user_query_values, previous_values):
+    """Check if all argument values are valid."""
+    arg_values = tool_args.values()
+    return all(str(value) in user_query_values or value in previous_values for value in arg_values)
+
+
+def verify_tool_calls(tool_calls, messages):
+    """
+    Verify tool calls based on user query and previous tool outputs.
+
+    :param tool_calls: List of tool calls with arguments.
+    :param messages: List containing user query and previous tool outputs.
+    :return: List of valid tool calls.
+    """
+    # Extract user query from the first message with role 'user'
+    user_query_values = next((msg["content"]
+                              for msg in messages if msg["role"] == "user"), None)
+
+    # Extract previous tool outputs from messages with role 'tool'
+    previous_tool_outputs = [msg["content"]
+                             for msg in messages if msg["role"] == "tool"]
+
+    previous_values = [
+        value for output in previous_tool_outputs for value in extract_leaf_values(output)]
+
+    valid_tool_calls = []
+
+    for tool_call in tool_calls:
+        tool_name = tool_call.function.name
+        tool_args = json.loads(tool_call.function.arguments)
+
+        tool_def = is_tool_valid(tool_name)
+        if tool_def:
+            if validate_tool_parameters(tool_def, tool_args):
+                valid_tool_calls.append(tool_call)
+        else:
+            logging.error(
+                f"Tool {tool_name} not found in tools. Dropping this tool call.")
+
+    tool_calls_str = [json.dumps(tool_call.__dict__, default=str)
+                      for tool_call in valid_tool_calls]
     logging.info(
-        f"Gathering movie data response: {response}")
+        'Tool calls validated successfully. Valid tool calls are: %s', tool_calls_str)
+    return valid_tool_calls
+
+
+def gather_movie_data(messages):
+    logging.debug(f"Gathering movie data with messages: {messages}")
+    response = get_response(client, MODEL, messages, tool_choice="required")
+    logging.debug(f"Calling tools based on the response: {response}")
     if response.choices[0].finish_reason == "tool_calls":
         tool_calls = response.choices[0].message.tool_calls
+        # validate tool calls
+        valid_tool_calls = verify_tool_calls(tool_calls, messages)
+        # valid_tool_calls = tool_calls
         updated_messages = messages.copy()
-        for tool_call in tool_calls:
+        tool_messages_count = len(
+            [msg for msg in messages if msg["role"] == "tool"])
+        if tool_messages_count <= 3 and valid_tool_calls:
+            tool_call = valid_tool_calls[0]  # Run one tool call at a time
             logging.info(
                 f"Tool call: {tool_call.function.name}, Tool call parameters: {tool_call.function.arguments}")
             tool_output = execute_tool(tool_call)
-            logging.info(
+            logging.debug(
                 f"Tool call output: {json.dumps(tool_output, indent=2)}")
-            if tool_call.function.name == "discover_movie":
+            if tool_call.function.name == "discover_movie" or tool_messages_count > 3:
                 return tool_output["results"]  # A list of movies
             else:
                 updated_messages.append(
@@ -110,15 +218,16 @@ def gather_movie_data(messages, iteration=0, max_iterations=2):
                         "content": str(tool_output),
                     }
                 )
-        if iteration < max_iterations:
-            return gather_movie_data(updated_messages, iteration + 1)
+            return gather_movie_data(updated_messages)
+        else:
+            return "No results found"
     else:
         raise Exception(
             "Failed to gather movie data. Got response: ", response)
 
 
 def execute_tool(tool_call):
-    logging.debug(f"Executing tool: {tool_call.function.name}")
+    logging.info(f"Executing tool: {tool_call.function.name}")
     function_to_call = names_to_functions[tool_call.function.name]
     function_args = json.loads(tool_call.function.arguments)
     return function_to_call(**function_args)
@@ -135,4 +244,5 @@ def chatbot(user_prompt):
 
 
 if __name__ == "__main__":
-    print(json.dumps(chatbot("List comedy movies with tom cruise in it"), indent=2))
+    print(json.dumps(chatbot("Movies of Tom Hanks that were released in 1993"), indent=2))
+    # print(json.dumps(chatbot("List some movies of Tom Cruise"), indent=2))

utils.py

@@ -0,0 +1,23 @@
+import logging
+
+
+def raise_error(error_message, error_type=Exception):
+    logging.error(error_message)
+    raise error_type(error_message)
+
+
+def python_type(type_str):
+    if type_str == "string":
+        return str
+    elif type_str == "integer":
+        return int
+    elif type_str == "number":
+        return float
+    elif type_str == "boolean":
+        return bool
+    elif type_str == "array":
+        return list
+    elif type_str == "object":
+        return dict
+    else:
+        raise ValueError(f"Unknown type: {type_str}")