Tool Pattern¶

source : https://github.com/neural-maze/agentic_patterns

---

As you may already know, the information stored in LLM weights is (usually) 𝐧𝐨𝐭 𝐞𝐧𝐨𝐮𝐠𝐡 to give accurate and insightful answers to our questions.

That's why we need to provide the LLM with ways to access the outside world. 🌍

In practice, you can build tools for whatever you want (at the end of the day they are just functions the LLM can use), from a tool that let's you access Wikipedia, another to analyse the content of YouTube videos or calculate difficult integrals using Wolfram Alpha.

The second pattern we are going to implement is the tool pattern.

In this notebook, you'll learn how tools actually work. This is the second lesson of the "Agentic Patterns from Scratch" series. Take a look at the first lesson if you haven't!

• First Lesson: The Reflection Pattern

A simple function¶

Take a look at this function 👇

In [ ]:

import json

def get_current_weather(location: str, unit: str):
	"""
	Get the current weather in a given location

	location (str): The city and state, e.g. Madrid, Barcelona
	unit (str): The unit. It can take two values; "celsius", "fahrenheit"
	"""
	if location == "Madrid":
		return json.dumps({"temperature": 25, "unit": unit})

	else:
		return json.dumps({"temperature": 58, "unit": unit})

import json def get_current_weather(location: str, unit: str): """ Get the current weather in a given location location (str): The city and state, e.g. Madrid, Barcelona unit (str): The unit. It can take two values; "celsius", "fahrenheit" """ if location == "Madrid": return json.dumps({"temperature": 25, "unit": unit}) else: return json.dumps({"temperature": 58, "unit": unit})

Very simple, right? You provide a location and a unit and it returns the temperature.

In [ ]:

get_current_weather(location="Madrid", unit="celsius")

get_current_weather(location="Madrid", unit="celsius")

But the question is:

How can you make this function available to an LLM?

An LLM is a type of NLP system, so it expects text as input. But how can we transform this function into text?

A System Prompt that works¶

For the LLM to be aware of this function, we need to provide some relevant information about it in the context. I'm referring to the function name, attributes, description, etc. Take a look at the following System Prompt.

You are a function calling AI model. You are provided with function signatures within <tools></tools> XML tags. 
You may call one or more functions to assist with the user query. Don't make assumptions about what values to plug 
into functions. Pay special attention to the properties 'types'. You should use those types as in a Python dict.
For each function call return a json object with function name and arguments within <tool_call></tool_call> XML tags as follows:

<tool_call>
{"name": <function-name>,"arguments": <args-dict>}
</tool_call>

Here are the available tools:

<tools> {
    "name": "get_current_weather",
    "description": "Get the current weather in a given location location (str): The city and state, e.g. Madrid, Barcelona unit (str): The unit. It can take two values; 'celsius', 'fahrenheit'",
    "parameters": {
        "properties": {
            "location": {
                "type": "string"
            },
            "unit": {
                "type": "string"
            }
        }
    }
}
</tools>

As you can see, the LLM enforces the LLM to behave as a function calling AI model who, given a list of function signatures inside the XML tags will select which one to use. When the model decides a function to use, it will return a json like the following, representing a function call:

<tool_call>
{"name": <function-name>,"arguments": <args-dict>}
</tool_call>

Let's see how it works in practise! 👇

In [ ]:

import os
import re
from groq import Groq
from dotenv import load_dotenv

# Remember to load the environment variables. You should have the Groq API Key in there :)
load_dotenv()

MODEL = "llama3-groq-70b-8192-tool-use-preview"
GROQ_CLIENT = Groq()

# Define the System Prompt as a constant
TOOL_SYSTEM_PROMPT = """
You are a function calling AI model. You are provided with function signatures within <tools></tools> XML tags. 
You may call one or more functions to assist with the user query. Don't make assumptions about what values to plug 
into functions. Pay special attention to the properties 'types'. You should use those types as in a Python dict.
For each function call return a json object with function name and arguments within <tool_call></tool_call> XML tags as follows:

<tool_call>
{"name": <function-name>,"arguments": <args-dict>}
</tool_call>

Here are the available tools:

<tools> {
    "name": "get_current_weather",
    "description": "Get the current weather in a given location location (str): The city and state, e.g. Madrid, Barcelona unit (str): The unit. It can take two values; 'celsius', 'fahrenheit'",
    "parameters": {
        "properties": {
            "location": {
                "type": "str"
            },
            "unit": {
                "type": "str"
            }
        }
    }
}
</tools>
"""

import os import re from groq import Groq from dotenv import load_dotenv # Remember to load the environment variables. You should have the Groq API Key in there :) load_dotenv() MODEL = "llama3-groq-70b-8192-tool-use-preview" GROQ_CLIENT = Groq() # Define the System Prompt as a constant TOOL_SYSTEM_PROMPT = """ You are a function calling AI model. You are provided with function signatures within XML tags. You may call one or more functions to assist with the user query. Don't make assumptions about what values to plug into functions. Pay special attention to the properties 'types'. You should use those types as in a Python dict. For each function call return a json object with function name and arguments within XML tags as follows: {"name": ,"arguments": } Here are the available tools: { "name": "get_current_weather", "description": "Get the current weather in a given location location (str): The city and state, e.g. Madrid, Barcelona unit (str): The unit. It can take two values; 'celsius', 'fahrenheit'", "parameters": { "properties": { "location": { "type": "str" }, "unit": { "type": "str" } } } } """

Let's ask a very simple question: "What's the current temperature in Madrid, in Celsius?"

In [ ]:

tool_chat_history = [
    {
        "role": "system",
        "content": TOOL_SYSTEM_PROMPT
    }
]
agent_chat_history = []

user_msg = {
    "role": "user",
    "content": "What's the current temperature in Madrid, in Celsius?"
}

tool_chat_history.append(user_msg)
agent_chat_history.append(user_msg)

output = GROQ_CLIENT.chat.completions.create(
    messages=tool_chat_history,
    model=MODEL
).choices[0].message.content

print(output)

tool_chat_history = [ { "role": "system", "content": TOOL_SYSTEM_PROMPT } ] agent_chat_history = [] user_msg = { "role": "user", "content": "What's the current temperature in Madrid, in Celsius?" } tool_chat_history.append(user_msg) agent_chat_history.append(user_msg) output = GROQ_CLIENT.chat.completions.create( messages=tool_chat_history, model=MODEL ).choices[0].message.content print(output)

---

That's an improvement! We may not have the proper answer but, with this information, we can obtain it! How? Well, we just need to:

1. Parse the LLM output. By this I mean deleting the XML tags
2. Load the output as a proper Python dict

The function below does exactly this.

---

In [ ]:

def parse_tool_call_str(tool_call_str: str):
    pattern = r'</?tool_call>'
    clean_tags = re.sub(pattern, '', tool_call_str)
    
    try:
        tool_call_json = json.loads(clean_tags)
        return tool_call_json
    except json.JSONDecodeError:
        return clean_tags
    except Exception as e:
        print(f"Unexpected error: {e}")
        return "There was some error parsing the Tool's output"

def parse_tool_call_str(tool_call_str: str): pattern = r'' clean_tags = re.sub(pattern, '', tool_call_str) try: tool_call_json = json.loads(clean_tags) return tool_call_json except json.JSONDecodeError: return clean_tags except Exception as e: print(f"Unexpected error: {e}") return "There was some error parsing the Tool's output"

In [ ]:

parsed_output = parse_tool_call_str(output)
parsed_output

parsed_output = parse_tool_call_str(output) parsed_output

We can simply run the function now, by passing the arguments like this 👇

In [ ]:

result = get_current_weather(**parsed_output["arguments"])

result = get_current_weather(**parsed_output["arguments"])

In [ ]:

result

result

That's it! A temperature of 25 degrees Celsius.

As you can see, we're dealing with a string, so we can simply add the parsed_output to the chat_history so that the LLM knows the information it has to return to the user.

In [ ]:

agent_chat_history.append({
    "role": "user",
    "content": f"Observation: {result}"
})

agent_chat_history.append({ "role": "user", "content": f"Observation: {result}" })

In [ ]:

GROQ_CLIENT.chat.completions.create(
    messages=agent_chat_history,
    model=MODEL
).choices[0].message.content

GROQ_CLIENT.chat.completions.create( messages=agent_chat_history, model=MODEL ).choices[0].message.content

Implementing everything the good way¶

To recap, we have a way for the LLM to generate tool_calls that we can use later to properly run the functions. But, as you may imagine, there are some pieces missing:

1. We need to automatically transform any function into a description like we saw in the initial system prompt.
2. We need a way to tell the agent that this function is a tool

Let's do it!

The tool decorator¶

We are going to use the tool decorator to transform any Python function into a tool. You can see the implementation here. To test it out, let's make a more complex tool than before. For example, a tool that interacts with Hacker News, getting the current top stories.

Reminder: To automatically generate the function signature for the tool, we need a way to infer the arguments types. For this reason, we need to create the typing annotations.

In [ ]:

import json
from typing import Callable


def get_fn_signature(fn: Callable) -> dict:
    """
    Generates the signature for a given function.

    Args:
        fn (Callable): The function whose signature needs to be extracted.

    Returns:
        dict: A dictionary containing the function's name, description,
              and parameter types.
    """
    fn_signature: dict = {
        "name": fn.__name__,
        "description": fn.__doc__,
        "parameters": {"properties": {}},
    }
    schema = {
        k: {"type": v.__name__} for k, v in fn.__annotations__.items() if k != "return"
    }
    fn_signature["parameters"]["properties"] = schema
    return fn_signature


def validate_arguments(tool_call: dict, tool_signature: dict) -> dict:
    """
    Validates and converts arguments in the input dictionary to match the expected types.

    Args:
        tool_call (dict): A dictionary containing the arguments passed to the tool.
        tool_signature (dict): The expected function signature and parameter types.

    Returns:
        dict: The tool call dictionary with the arguments converted to the correct types if necessary.
    """
    properties = tool_signature["parameters"]["properties"]

    # TODO: This is overly simplified but enough for simple Tools.
    type_mapping = {
        "int": int,
        "str": str,
        "bool": bool,
        "float": float,
    }

    for arg_name, arg_value in tool_call["arguments"].items():
        expected_type = properties[arg_name].get("type")

        if not isinstance(arg_value, type_mapping[expected_type]):
            tool_call["arguments"][arg_name] = type_mapping[expected_type](arg_value)

    return tool_call


class Tool:
    """
    A class representing a tool that wraps a callable and its signature.

    Attributes:
        name (str): The name of the tool (function).
        fn (Callable): The function that the tool represents.
        fn_signature (str): JSON string representation of the function's signature.
    """

    def __init__(self, name: str, fn: Callable, fn_signature: str):
        self.name = name
        self.fn = fn
        self.fn_signature = fn_signature

    def __str__(self):
        return self.fn_signature

    def run(self, **kwargs):
        """
        Executes the tool (function) with provided arguments.

        Args:
            **kwargs: Keyword arguments passed to the function.

        Returns:
            The result of the function call.
        """
        return self.fn(**kwargs)


def tool(fn: Callable):
    """
    A decorator that wraps a function into a Tool object.

    Args:
        fn (Callable): The function to be wrapped.

    Returns:
        Tool: A Tool object containing the function, its name, and its signature.
    """

    def wrapper():
        fn_signature = get_fn_signature(fn)
        return Tool(
            name=fn_signature.get("name"), fn=fn, fn_signature=json.dumps(fn_signature)
        )

    return wrapper()

import json from typing import Callable def get_fn_signature(fn: Callable) -> dict: """ Generates the signature for a given function. Args: fn (Callable): The function whose signature needs to be extracted. Returns: dict: A dictionary containing the function's name, description, and parameter types. """ fn_signature: dict = { "name": fn.__name__, "description": fn.__doc__, "parameters": {"properties": {}}, } schema = { k: {"type": v.__name__} for k, v in fn.__annotations__.items() if k != "return" } fn_signature["parameters"]["properties"] = schema return fn_signature def validate_arguments(tool_call: dict, tool_signature: dict) -> dict: """ Validates and converts arguments in the input dictionary to match the expected types. Args: tool_call (dict): A dictionary containing the arguments passed to the tool. tool_signature (dict): The expected function signature and parameter types. Returns: dict: The tool call dictionary with the arguments converted to the correct types if necessary. """ properties = tool_signature["parameters"]["properties"] # TODO: This is overly simplified but enough for simple Tools. type_mapping = { "int": int, "str": str, "bool": bool, "float": float, } for arg_name, arg_value in tool_call["arguments"].items(): expected_type = properties[arg_name].get("type") if not isinstance(arg_value, type_mapping[expected_type]): tool_call["arguments"][arg_name] = type_mapping[expected_type](arg_value) return tool_call class Tool: """ A class representing a tool that wraps a callable and its signature. Attributes: name (str): The name of the tool (function). fn (Callable): The function that the tool represents. fn_signature (str): JSON string representation of the function's signature. """ def __init__(self, name: str, fn: Callable, fn_signature: str): self.name = name self.fn = fn self.fn_signature = fn_signature def __str__(self): return self.fn_signature def run(self, **kwargs): """ Executes the tool (function) with provided arguments. Args: **kwargs: Keyword arguments passed to the function. Returns: The result of the function call. """ return self.fn(**kwargs) def tool(fn: Callable): """ A decorator that wraps a function into a Tool object. Args: fn (Callable): The function to be wrapped. Returns: Tool: A Tool object containing the function, its name, and its signature. """ def wrapper(): fn_signature = get_fn_signature(fn) return Tool( name=fn_signature.get("name"), fn=fn, fn_signature=json.dumps(fn_signature) ) return wrapper()

In [ ]:

import re
from dataclasses import dataclass
import time

from colorama import Fore
from colorama import Style

def completions_create(client, messages: list, model: str) -> str:
    """
    Sends a request to the client's `completions.create` method to interact with the language model.

    Args:
        client (Groq): The Groq client object
        messages (list[dict]): A list of message objects containing chat history for the model.
        model (str): The model to use for generating tool calls and responses.

    Returns:
        str: The content of the model's response.
    """
    response = client.chat.completions.create(messages=messages, model=model)
    return str(response.choices[0].message.content)


def build_prompt_structure(prompt: str, role: str, tag: str = "") -> dict:
    """
    Builds a structured prompt that includes the role and content.

    Args:
        prompt (str): The actual content of the prompt.
        role (str): The role of the speaker (e.g., user, assistant).

    Returns:
        dict: A dictionary representing the structured prompt.
    """
    if tag:
        prompt = f"<{tag}>{prompt}</{tag}>"
    return {"role": role, "content": prompt}


def update_chat_history(history: list, msg: str, role: str):
    """
    Updates the chat history by appending the latest response.

    Args:
        history (list): The list representing the current chat history.
        msg (str): The message to append.
        role (str): The role type (e.g. 'user', 'assistant', 'system')
    """
    history.append(build_prompt_structure(prompt=msg, role=role))


class ChatHistory(list):
    def __init__(self, messages: list | None = None, total_length: int = -1):
        """Initialise the queue with a fixed total length.

        Args:
            messages (list | None): A list of initial messages
            total_length (int): The maximum number of messages the chat history can hold.
        """
        if messages is None:
            messages = []

        super().__init__(messages)
        self.total_length = total_length

    def append(self, msg: str):
        """Add a message to the queue.

        Args:
            msg (str): The message to be added to the queue
        """
        if len(self) == self.total_length:
            self.pop(0)
        super().append(msg)


class FixedFirstChatHistory(ChatHistory):
    def __init__(self, messages: list | None = None, total_length: int = -1):
        """Initialise the queue with a fixed total length.

        Args:
            messages (list | None): A list of initial messages
            total_length (int): The maximum number of messages the chat history can hold.
        """
        super().__init__(messages, total_length)

    def append(self, msg: str):
        """Add a message to the queue. The first messaage will always stay fixed.

        Args:
            msg (str): The message to be added to the queue
        """
        if len(self) == self.total_length:
            self.pop(1)
        super().append(msg)




@dataclass
class TagContentResult:
    """
    A data class to represent the result of extracting tag content.

    Attributes:
        content (List[str]): A list of strings containing the content found between the specified tags.
        found (bool): A flag indicating whether any content was found for the given tag.
    """

    content: list[str]
    found: bool


def extract_tag_content(text: str, tag: str) -> TagContentResult:
    """
    Extracts all content enclosed by specified tags (e.g., <thought>, <response>, etc.).

    Parameters:
        text (str): The input string containing multiple potential tags.
        tag (str): The name of the tag to search for (e.g., 'thought', 'response').

    Returns:
        dict: A dictionary with the following keys:
            - 'content' (list): A list of strings containing the content found between the specified tags.
            - 'found' (bool): A flag indicating whether any content was found for the given tag.
    """
    # Build the regex pattern dynamically to find multiple occurrences of the tag
    tag_pattern = rf"<{tag}>(.*?)</{tag}>"

    # Use findall to capture all content between the specified tag
    matched_contents = re.findall(tag_pattern, text, re.DOTALL)

    # Return the dataclass instance with the result
    return TagContentResult(
        content=[content.strip() for content in matched_contents],
        found=bool(matched_contents),
    )


def fancy_print(message: str) -> None:
    """
    Displays a fancy print message.

    Args:
        message (str): The message to display.
    """
    print(Style.BRIGHT + Fore.CYAN + f"\n{'=' * 50}")
    print(Fore.MAGENTA + f"{message}")
    print(Style.BRIGHT + Fore.CYAN + f"{'=' * 50}\n")
    time.sleep(0.5)


def fancy_step_tracker(step: int, total_steps: int) -> None:
    """
    Displays a fancy step tracker for each iteration of the generation-reflection loop.

    Args:
        step (int): The current step in the loop.
        total_steps (int): The total number of steps in the loop.
    """
    fancy_print(f"STEP {step + 1}/{total_steps}")

import json
from typing import Callable


def get_fn_signature(fn: Callable) -> dict:
    """
    Generates the signature for a given function.

    Args:
        fn (Callable): The function whose signature needs to be extracted.

    Returns:
        dict: A dictionary containing the function's name, description,
              and parameter types.
    """
    fn_signature: dict = {
        "name": fn.__name__,
        "description": fn.__doc__,
        "parameters": {"properties": {}},
    }
    schema = {
        k: {"type": v.__name__} for k, v in fn.__annotations__.items() if k != "return"
    }
    fn_signature["parameters"]["properties"] = schema
    return fn_signature


def validate_arguments(tool_call: dict, tool_signature: dict) -> dict:
    """
    Validates and converts arguments in the input dictionary to match the expected types.

    Args:
        tool_call (dict): A dictionary containing the arguments passed to the tool.
        tool_signature (dict): The expected function signature and parameter types.

    Returns:
        dict: The tool call dictionary with the arguments converted to the correct types if necessary.
    """
    properties = tool_signature["parameters"]["properties"]

    # TODO: This is overly simplified but enough for simple Tools.
    type_mapping = {
        "int": int,
        "str": str,
        "bool": bool,
        "float": float,
    }

    for arg_name, arg_value in tool_call["arguments"].items():
        expected_type = properties[arg_name].get("type")

        if not isinstance(arg_value, type_mapping[expected_type]):
            tool_call["arguments"][arg_name] = type_mapping[expected_type](arg_value)

    return tool_call


class Tool:
    """
    A class representing a tool that wraps a callable and its signature.

    Attributes:
        name (str): The name of the tool (function).
        fn (Callable): The function that the tool represents.
        fn_signature (str): JSON string representation of the function's signature.
    """

    def __init__(self, name: str, fn: Callable, fn_signature: str):
        self.name = name
        self.fn = fn
        self.fn_signature = fn_signature

    def __str__(self):
        return self.fn_signature

    def run(self, **kwargs):
        """
        Executes the tool (function) with provided arguments.

        Args:
            **kwargs: Keyword arguments passed to the function.

        Returns:
            The result of the function call.
        """
        return self.fn(**kwargs)


def tool(fn: Callable):
    """
    A decorator that wraps a function into a Tool object.

    Args:
        fn (Callable): The function to be wrapped.

    Returns:
        Tool: A Tool object containing the function, its name, and its signature.
    """

    def wrapper():
        fn_signature = get_fn_signature(fn)
        return Tool(
            name=fn_signature.get("name"), fn=fn, fn_signature=json.dumps(fn_signature)
        )

    return wrapper()

import re from dataclasses import dataclass import time from colorama import Fore from colorama import Style def completions_create(client, messages: list, model: str) -> str: """ Sends a request to the client's `completions.create` method to interact with the language model. Args: client (Groq): The Groq client object messages (list[dict]): A list of message objects containing chat history for the model. model (str): The model to use for generating tool calls and responses. Returns: str: The content of the model's response. """ response = client.chat.completions.create(messages=messages, model=model) return str(response.choices[0].message.content) def build_prompt_structure(prompt: str, role: str, tag: str = "") -> dict: """ Builds a structured prompt that includes the role and content. Args: prompt (str): The actual content of the prompt. role (str): The role of the speaker (e.g., user, assistant). Returns: dict: A dictionary representing the structured prompt. """ if tag: prompt = f"<{tag}>{prompt}" return {"role": role, "content": prompt} def update_chat_history(history: list, msg: str, role: str): """ Updates the chat history by appending the latest response. Args: history (list): The list representing the current chat history. msg (str): The message to append. role (str): The role type (e.g. 'user', 'assistant', 'system') """ history.append(build_prompt_structure(prompt=msg, role=role)) class ChatHistory(list): def __init__(self, messages: list | None = None, total_length: int = -1): """Initialise the queue with a fixed total length. Args: messages (list | None): A list of initial messages total_length (int): The maximum number of messages the chat history can hold. """ if messages is None: messages = [] super().__init__(messages) self.total_length = total_length def append(self, msg: str): """Add a message to the queue. Args: msg (str): The message to be added to the queue """ if len(self) == self.total_length: self.pop(0) super().append(msg) class FixedFirstChatHistory(ChatHistory): def __init__(self, messages: list | None = None, total_length: int = -1): """Initialise the queue with a fixed total length. Args: messages (list | None): A list of initial messages total_length (int): The maximum number of messages the chat history can hold. """ super().__init__(messages, total_length) def append(self, msg: str): """Add a message to the queue. The first messaage will always stay fixed. Args: msg (str): The message to be added to the queue """ if len(self) == self.total_length: self.pop(1) super().append(msg) @dataclass class TagContentResult: """ A data class to represent the result of extracting tag content. Attributes: content (List[str]): A list of strings containing the content found between the specified tags. found (bool): A flag indicating whether any content was found for the given tag. """ content: list[str] found: bool def extract_tag_content(text: str, tag: str) -> TagContentResult: """ Extracts all content enclosed by specified tags (e.g., , , etc.). Parameters: text (str): The input string containing multiple potential tags. tag (str): The name of the tag to search for (e.g., 'thought', 'response'). Returns: dict: A dictionary with the following keys: - 'content' (list): A list of strings containing the content found between the specified tags. - 'found' (bool): A flag indicating whether any content was found for the given tag. """ # Build the regex pattern dynamically to find multiple occurrences of the tag tag_pattern = rf"<{tag}>(.*?)" # Use findall to capture all content between the specified tag matched_contents = re.findall(tag_pattern, text, re.DOTALL) # Return the dataclass instance with the result return TagContentResult( content=[content.strip() for content in matched_contents], found=bool(matched_contents), ) def fancy_print(message: str) -> None: """ Displays a fancy print message. Args: message (str): The message to display. """ print(Style.BRIGHT + Fore.CYAN + f"\n{'=' * 50}") print(Fore.MAGENTA + f"{message}") print(Style.BRIGHT + Fore.CYAN + f"{'=' * 50}\n") time.sleep(0.5) def fancy_step_tracker(step: int, total_steps: int) -> None: """ Displays a fancy step tracker for each iteration of the generation-reflection loop. Args: step (int): The current step in the loop. total_steps (int): The total number of steps in the loop. """ fancy_print(f"STEP {step + 1}/{total_steps}") import json from typing import Callable def get_fn_signature(fn: Callable) -> dict: """ Generates the signature for a given function. Args: fn (Callable): The function whose signature needs to be extracted. Returns: dict: A dictionary containing the function's name, description, and parameter types. """ fn_signature: dict = { "name": fn.__name__, "description": fn.__doc__, "parameters": {"properties": {}}, } schema = { k: {"type": v.__name__} for k, v in fn.__annotations__.items() if k != "return" } fn_signature["parameters"]["properties"] = schema return fn_signature def validate_arguments(tool_call: dict, tool_signature: dict) -> dict: """ Validates and converts arguments in the input dictionary to match the expected types. Args: tool_call (dict): A dictionary containing the arguments passed to the tool. tool_signature (dict): The expected function signature and parameter types. Returns: dict: The tool call dictionary with the arguments converted to the correct types if necessary. """ properties = tool_signature["parameters"]["properties"] # TODO: This is overly simplified but enough for simple Tools. type_mapping = { "int": int, "str": str, "bool": bool, "float": float, } for arg_name, arg_value in tool_call["arguments"].items(): expected_type = properties[arg_name].get("type") if not isinstance(arg_value, type_mapping[expected_type]): tool_call["arguments"][arg_name] = type_mapping[expected_type](arg_value) return tool_call class Tool: """ A class representing a tool that wraps a callable and its signature. Attributes: name (str): The name of the tool (function). fn (Callable): The function that the tool represents. fn_signature (str): JSON string representation of the function's signature. """ def __init__(self, name: str, fn: Callable, fn_signature: str): self.name = name self.fn = fn self.fn_signature = fn_signature def __str__(self): return self.fn_signature def run(self, **kwargs): """ Executes the tool (function) with provided arguments. Args: **kwargs: Keyword arguments passed to the function. Returns: The result of the function call. """ return self.fn(**kwargs) def tool(fn: Callable): """ A decorator that wraps a function into a Tool object. Args: fn (Callable): The function to be wrapped. Returns: Tool: A Tool object containing the function, its name, and its signature. """ def wrapper(): fn_signature = get_fn_signature(fn) return Tool( name=fn_signature.get("name"), fn=fn, fn_signature=json.dumps(fn_signature) ) return wrapper()

In [ ]:

import json
import requests
import json
import re

from colorama import Fore
from dotenv import load_dotenv
from groq import Groq



load_dotenv()


TOOL_SYSTEM_PROMPT = """
You are a function calling AI model. You are provided with function signatures within <tools></tools> XML tags.
You may call one or more functions to assist with the user query. Don't make assumptions about what values to plug
into functions. Pay special attention to the properties 'types'. You should use those types as in a Python dict.
For each function call return a json object with function name and arguments within <tool_call></tool_call>
XML tags as follows:

<tool_call>
{"name": <function-name>,"arguments": <args-dict>,  "id": <monotonically-increasing-id>}
</tool_call>

Here are the available tools:

<tools>
%s
</tools>
"""


class ToolAgent:
    """
    The ToolAgent class represents an agent that can interact with a language model and use tools
    to assist with user queries. It generates function calls based on user input, validates arguments,
    and runs the respective tools.

    Attributes:
        tools (Tool | list[Tool]): A list of tools available to the agent.
        model (str): The model to be used for generating tool calls and responses.
        client (Groq): The Groq client used to interact with the language model.
        tools_dict (dict): A dictionary mapping tool names to their corresponding Tool objects.
    """

    def __init__(
        self,
        tools: Tool | list[Tool],
        model: str = "llama3-groq-70b-8192-tool-use-preview",
    ) -> None:
        self.client = Groq()
        self.model = model
        self.tools = tools if isinstance(tools, list) else [tools]
        self.tools_dict = {tool.name: tool for tool in self.tools}

    def add_tool_signatures(self) -> str:
        """
        Collects the function signatures of all available tools.

        Returns:
            str: A concatenated string of all tool function signatures in JSON format.
        """
        return "".join([tool.fn_signature for tool in self.tools])

    def process_tool_calls(self, tool_calls_content: list) -> dict:
        """
        Processes each tool call, validates arguments, executes the tools, and collects results.

        Args:
            tool_calls_content (list): List of strings, each representing a tool call in JSON format.

        Returns:
            dict: A dictionary where the keys are tool call IDs and values are the results from the tools.
        """
        observations = {}
        for tool_call_str in tool_calls_content:
            tool_call = json.loads(tool_call_str)
            tool_name = tool_call["name"]
            tool = self.tools_dict[tool_name]

            print(Fore.GREEN + f"\nUsing Tool: {tool_name}")

            # Validate and execute the tool call
            validated_tool_call = validate_arguments(
                tool_call, json.loads(tool.fn_signature)
            )
            print(Fore.GREEN + f"\nTool call dict: \n{validated_tool_call}")

            result = tool.run(**validated_tool_call["arguments"])
            print(Fore.GREEN + f"\nTool result: \n{result}")

            # Store the result using the tool call ID
            observations[validated_tool_call["id"]] = result

        return observations

    def run(
        self,
        user_msg: str,
    ) -> str:
        """
        Handles the full process of interacting with the language model and executing a tool based on user input.

        Args:
            user_msg (str): The user's message that prompts the tool agent to act.

        Returns:
            str: The final output after executing the tool and generating a response from the model.
        """
        user_prompt = build_prompt_structure(prompt=user_msg, role="user")

        tool_chat_history = ChatHistory(
            [
                build_prompt_structure(
                    prompt=TOOL_SYSTEM_PROMPT % self.add_tool_signatures(),
                    role="system",
                ),
                user_prompt,
            ]
        )
        agent_chat_history = ChatHistory([user_prompt])

        tool_call_response = completions_create(
            self.client, messages=tool_chat_history, model=self.model
        )
        tool_calls = extract_tag_content(str(tool_call_response), "tool_call")

        if tool_calls.found:
            observations = self.process_tool_calls(tool_calls.content)
            update_chat_history(
                agent_chat_history, f'f"Observation: {observations}"', "user"
            )

        return completions_create(self.client, agent_chat_history, self.model)

import json import requests import json import re from colorama import Fore from dotenv import load_dotenv from groq import Groq load_dotenv() TOOL_SYSTEM_PROMPT = """ You are a function calling AI model. You are provided with function signatures within XML tags. You may call one or more functions to assist with the user query. Don't make assumptions about what values to plug into functions. Pay special attention to the properties 'types'. You should use those types as in a Python dict. For each function call return a json object with function name and arguments within XML tags as follows: {"name": ,"arguments": , "id": } Here are the available tools: %s """ class ToolAgent: """ The ToolAgent class represents an agent that can interact with a language model and use tools to assist with user queries. It generates function calls based on user input, validates arguments, and runs the respective tools. Attributes: tools (Tool | list[Tool]): A list of tools available to the agent. model (str): The model to be used for generating tool calls and responses. client (Groq): The Groq client used to interact with the language model. tools_dict (dict): A dictionary mapping tool names to their corresponding Tool objects. """ def __init__( self, tools: Tool | list[Tool], model: str = "llama3-groq-70b-8192-tool-use-preview", ) -> None: self.client = Groq() self.model = model self.tools = tools if isinstance(tools, list) else [tools] self.tools_dict = {tool.name: tool for tool in self.tools} def add_tool_signatures(self) -> str: """ Collects the function signatures of all available tools. Returns: str: A concatenated string of all tool function signatures in JSON format. """ return "".join([tool.fn_signature for tool in self.tools]) def process_tool_calls(self, tool_calls_content: list) -> dict: """ Processes each tool call, validates arguments, executes the tools, and collects results. Args: tool_calls_content (list): List of strings, each representing a tool call in JSON format. Returns: dict: A dictionary where the keys are tool call IDs and values are the results from the tools. """ observations = {} for tool_call_str in tool_calls_content: tool_call = json.loads(tool_call_str) tool_name = tool_call["name"] tool = self.tools_dict[tool_name] print(Fore.GREEN + f"\nUsing Tool: {tool_name}") # Validate and execute the tool call validated_tool_call = validate_arguments( tool_call, json.loads(tool.fn_signature) ) print(Fore.GREEN + f"\nTool call dict: \n{validated_tool_call}") result = tool.run(**validated_tool_call["arguments"]) print(Fore.GREEN + f"\nTool result: \n{result}") # Store the result using the tool call ID observations[validated_tool_call["id"]] = result return observations def run( self, user_msg: str, ) -> str: """ Handles the full process of interacting with the language model and executing a tool based on user input. Args: user_msg (str): The user's message that prompts the tool agent to act. Returns: str: The final output after executing the tool and generating a response from the model. """ user_prompt = build_prompt_structure(prompt=user_msg, role="user") tool_chat_history = ChatHistory( [ build_prompt_structure( prompt=TOOL_SYSTEM_PROMPT % self.add_tool_signatures(), role="system", ), user_prompt, ] ) agent_chat_history = ChatHistory([user_prompt]) tool_call_response = completions_create( self.client, messages=tool_chat_history, model=self.model ) tool_calls = extract_tag_content(str(tool_call_response), "tool_call") if tool_calls.found: observations = self.process_tool_calls(tool_calls.content) update_chat_history( agent_chat_history, f'f"Observation: {observations}"', "user" ) return completions_create(self.client, agent_chat_history, self.model)

In [ ]:

def fetch_top_hacker_news_stories(top_n: int):
    """
    Fetch the top stories from Hacker News.

    This function retrieves the top `top_n` stories from Hacker News using the Hacker News API. 
    Each story contains the title, URL, score, author, and time of submission. The data is fetched 
    from the official Firebase Hacker News API, which returns story details in JSON format.

    Args:
        top_n (int): The number of top stories to retrieve.
    """
    top_stories_url = 'https://hacker-news.firebaseio.com/v0/topstories.json'
    
    try:
        response = requests.get(top_stories_url)
        response.raise_for_status()  # Check for HTTP errors
        
        # Get the top story IDs
        top_story_ids = response.json()[:top_n]
        
        top_stories = []
        
        # For each story ID, fetch the story details
        for story_id in top_story_ids:
            story_url = f'https://hacker-news.firebaseio.com/v0/item/{story_id}.json'
            story_response = requests.get(story_url)
            story_response.raise_for_status()  # Check for HTTP errors
            story_data = story_response.json()
            
            # Append the story title and URL (or other relevant info) to the list
            top_stories.append({
                'title': story_data.get('title', 'No title'),
                'url': story_data.get('url', 'No URL available'),
            })
        
        return json.dumps(top_stories)

    except requests.exceptions.RequestException as e:
        print(f"An error occurred: {e}")
        return []

def fetch_top_hacker_news_stories(top_n: int): """ Fetch the top stories from Hacker News. This function retrieves the top `top_n` stories from Hacker News using the Hacker News API. Each story contains the title, URL, score, author, and time of submission. The data is fetched from the official Firebase Hacker News API, which returns story details in JSON format. Args: top_n (int): The number of top stories to retrieve. """ top_stories_url = 'https://hacker-news.firebaseio.com/v0/topstories.json' try: response = requests.get(top_stories_url) response.raise_for_status() # Check for HTTP errors # Get the top story IDs top_story_ids = response.json()[:top_n] top_stories = [] # For each story ID, fetch the story details for story_id in top_story_ids: story_url = f'https://hacker-news.firebaseio.com/v0/item/{story_id}.json' story_response = requests.get(story_url) story_response.raise_for_status() # Check for HTTP errors story_data = story_response.json() # Append the story title and URL (or other relevant info) to the list top_stories.append({ 'title': story_data.get('title', 'No title'), 'url': story_data.get('url', 'No URL available'), }) return json.dumps(top_stories) except requests.exceptions.RequestException as e: print(f"An error occurred: {e}") return []

If we run this Python function, we'll obtain the top HN stories, as you can see below (the top 5 in this case).

In [ ]:

json.loads(fetch_top_hacker_news_stories(top_n=5))

json.loads(fetch_top_hacker_news_stories(top_n=5))

To transform the fetch_top_hacker_news_stories function into a Tool, we can use the tool decorator.

In [ ]:

hn_tool = tool(fetch_top_hacker_news_stories)

hn_tool = tool(fetch_top_hacker_news_stories)

The Tool has the following parameters: a name, a fn_signature and the fn (this is the function we are going to call, this case fetch_top_hacker_news_stories)

In [ ]:

hn_tool.name

hn_tool.name

By default, the tool gets its name from the function name.

In [ ]:

json.loads(hn_tool.fn_signature)

json.loads(hn_tool.fn_signature)

As you can see, the function signature has been automatically generated. It contains the name, a description (taken from the docstrings) and the parameters, whose types come from the tying annotations. Now that we have a tool, let's run the agent.

The ToolAgent¶

To create the agent, we just need to pass a list of tools (in this case, just one).

In [ ]:

tool_agent = ToolAgent(tools=[hn_tool])

tool_agent = ToolAgent(tools=[hn_tool])

A quick check to see that everything works fine. If we ask the agent something unrelated to Hacker News, it shouldn't use the tool.

In [ ]:

output = tool_agent.run(user_msg="Tell me your name")

output = tool_agent.run(user_msg="Tell me your name")

In [ ]:

print(output)

print(output)

Now, let's ask for specific information about Hacker News.

In [ ]:

output = tool_agent.run(user_msg="Tell me the top 5 Hacker News stories right now")

output = tool_agent.run(user_msg="Tell me the top 5 Hacker News stories right now")

In [ ]:

print(output)

print(output)

---

There you have it!! A fully functional Tool!! 🛠️