Planning Pattern - ReAct Technique¶
source : https://github.com/neural-maze/agentic_patterns
So, we've seen agents capable of reflecting and using tools to access the outside world. But ... what about planning, i.e. deciding what sequence of steps to follow to accomplish a large task?
That is exactly what the Planning Pattern provides; ways for the LLM to break a task into smaller, more easily accomplished subgoals without losing track of the end goal.
The most paradigmatic example of the planning pattern is the ReAct technique, displayed in the diagram above.
In this notebook, you'll learn how this technique actually works. This is the third lesson of the "Agentic Patterns from Scratch" series. Take a look at the previous lessons if you haven't!
Relevant imports and Groq Client¶
We start by importing all the libraries we'll be using in this tutorial as well as the Groq client.
In [ ]:
Copied!
import os
import re
import math
import json
from dotenv import load_dotenv
from groq import Groq
# Remember to load the environment variables. You should have the Groq API Key in there :)
load_dotenv()
MODEL = "llama-3.1-70b-versatile"
GROQ_CLIENT = Groq()
import os
import re
import math
import json
from dotenv import load_dotenv
from groq import Groq
# Remember to load the environment variables. You should have the Groq API Key in there :)
load_dotenv()
MODEL = "llama-3.1-70b-versatile"
GROQ_CLIENT = Groq()
If you are not familiar with the
tooldecorator, changes are you are missed the previous tutorial about the Tool Pattern. Check the video here.
Utils¶
In [ ]:
Copied!
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()
A System Prompt for the ReAct Loop¶
As we did with the Tool Pattern, we also need a System Prompt for the ReAct technique. This System Prompt is very similar, the difference is that it describes the ReAct loop, so that the LLM is aware of the three operations it's allowed to use:
- Thought: The LLM will think about which action to take
- Action: The LLM will use a Tool to "act on the environment"
- Observation: The LLM will observe the tool output and reflect on the next thing to do.
Another key difference from the Tool Pattern System Prompt is that we are going to enclose all the messages with tags, like these:
Ok! So without further ado, there's the prompt!
In [ ]:
Copied!
# Define the System Prompt as a constant
REACT_SYSTEM_PROMPT = """
You are a function calling AI model. You operate by running a loop with the following steps: Thought, Action, Observation.
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' 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 / actions:
<tools>
%s
</tools>
Example session:
<question>What's the current temperature in Madrid?</question>
<thought>I need to get the current weather in Madrid</thought>
<tool_call>{"name": "get_current_weather","arguments": {"location": "Madrid", "unit": "celsius"}, "id": 0}</tool_call>
You will be called again with this:
<observation>{0: {"temperature": 25, "unit": "celsius"}}</observation>
You then output:
<response>The current temperature in Madrid is 25 degrees Celsius</response>
Additional constraints:
- If the user asks you something unrelated to any of the tools above, answer freely enclosing your answer with <response></response> tags.
"""
# Define the System Prompt as a constant
REACT_SYSTEM_PROMPT = """
You are a function calling AI model. You operate by running a loop with the following steps: Thought, Action, Observation.
You are provided with function signatures within
{"name": ,"arguments": , "id": }
Here are the available tools / actions:
%s
Example session:
What's the current temperature in Madrid?
I need to get the current weather in Madrid
{"name": "get_current_weather","arguments": {"location": "Madrid", "unit": "celsius"}, "id": 0}
You will be called again with this:
{0: {"temperature": 25, "unit": "celsius"}}
You then output:
The current temperature in Madrid is 25 degrees Celsius
Additional constraints:
- If the user asks you something unrelated to any of the tools above, answer freely enclosing your answer with
Example step by step¶
Defining the Tools¶
Let's build an example that involves the use of three tools, like the following ones.
In [ ]:
Copied!
@tool
def sum_two_elements(a: int, b: int) -> int:
"""
Computes the sum of two integers.
Args:
a (int): The first integer to be summed.
b (int): The second integer to be summed.
Returns:
int: The sum of `a` and `b`.
"""
return a + b
@tool
def multiply_two_elements(a: int, b: int) -> int:
"""
Multiplies two integers.
Args:
a (int): The first integer to multiply.
b (int): The second integer to multiply.
Returns:
int: The product of `a` and `b`.
"""
return a * b
@tool
def compute_log(x: int) -> float | str:
"""
Computes the logarithm of an integer `x` with an optional base.
Args:
x (int): The integer value for which the logarithm is computed. Must be greater than 0.
Returns:
float: The logarithm of `x` to the specified `base`.
"""
if x <= 0:
return "Logarithm is undefined for values less than or equal to 0."
return math.log(x)
available_tools = {
"sum_two_elements": sum_two_elements,
"multiply_two_elements": multiply_two_elements,
"compute_log": compute_log
}
@tool
def sum_two_elements(a: int, b: int) -> int:
"""
Computes the sum of two integers.
Args:
a (int): The first integer to be summed.
b (int): The second integer to be summed.
Returns:
int: The sum of `a` and `b`.
"""
return a + b
@tool
def multiply_two_elements(a: int, b: int) -> int:
"""
Multiplies two integers.
Args:
a (int): The first integer to multiply.
b (int): The second integer to multiply.
Returns:
int: The product of `a` and `b`.
"""
return a * b
@tool
def compute_log(x: int) -> float | str:
"""
Computes the logarithm of an integer `x` with an optional base.
Args:
x (int): The integer value for which the logarithm is computed. Must be greater than 0.
Returns:
float: The logarithm of `x` to the specified `base`.
"""
if x <= 0:
return "Logarithm is undefined for values less than or equal to 0."
return math.log(x)
available_tools = {
"sum_two_elements": sum_two_elements,
"multiply_two_elements": multiply_two_elements,
"compute_log": compute_log
}
Remember that the @tool operator allows us to convert a Python function into a Tool automatically. We cana check that very easily with some of the functions above.
In [ ]:
Copied!
print("Tool name: ", sum_two_elements.name)
print("Tool signature: ", sum_two_elements.fn_signature)
print("Tool name: ", sum_two_elements.name)
print("Tool signature: ", sum_two_elements.fn_signature)
Adding the Tools signature to the System Prompt¶
Now, we just concatenate the tools signature and add them to the System Prompt.
In [ ]:
Copied!
tools_signature = sum_two_elements.fn_signature + ",\n" + multiply_two_elements.fn_signature + ",\n" + compute_log.fn_signature
tools_signature = sum_two_elements.fn_signature + ",\n" + multiply_two_elements.fn_signature + ",\n" + compute_log.fn_signature
In [ ]:
Copied!
print(tools_signature)
print(tools_signature)
In [ ]:
Copied!
REACT_SYSTEM_PROMPT = REACT_SYSTEM_PROMPT % tools_signature
REACT_SYSTEM_PROMPT = REACT_SYSTEM_PROMPT % tools_signature
In [ ]:
Copied!
print(REACT_SYSTEM_PROMPT)
print(REACT_SYSTEM_PROMPT)
ReAct Loop Step 1¶
In [ ]:
Copied!
USER_QUESTION = "I want to calculate the sum of 1234 and 5678 and multiply the result by 5. Then, I want to take the logarithm of this result"
chat_history = [
{
"role": "system",
"content": REACT_SYSTEM_PROMPT
},
{
"role": "user",
"content": f"<question>{USER_QUESTION}</question>"
}
]
USER_QUESTION = "I want to calculate the sum of 1234 and 5678 and multiply the result by 5. Then, I want to take the logarithm of this result"
chat_history = [
{
"role": "system",
"content": REACT_SYSTEM_PROMPT
},
{
"role": "user",
"content": f"{USER_QUESTION} "
}
]
In [ ]:
Copied!
output = GROQ_CLIENT.chat.completions.create(
messages=chat_history,
model=MODEL
).choices[0].message.content
print(output)
output = GROQ_CLIENT.chat.completions.create(
messages=chat_history,
model=MODEL
).choices[0].message.content
print(output)
In [ ]:
Copied!
chat_history.append(
{
"role": "assistant",
"content": output
}
)
chat_history.append(
{
"role": "assistant",
"content": output
}
)
ReAct Loop Step 2¶
In [ ]:
Copied!
tool_call = extract_tag_content(output, tag="tool_call")
tool_call = extract_tag_content(output, tag="tool_call")
In [ ]:
Copied!
tool_call
tool_call
In [ ]:
Copied!
tool_call = json.loads(tool_call.content[0])
tool_call = json.loads(tool_call.content[0])
In [ ]:
Copied!
tool_call
tool_call
In [ ]:
Copied!
tool_result = available_tools[tool_call["name"]].run(**tool_call["arguments"])
tool_result = available_tools[tool_call["name"]].run(**tool_call["arguments"])
In [ ]:
Copied!
assert tool_result == 1234 + 5678
assert tool_result == 1234 + 5678
In [ ]:
Copied!
chat_history.append(
{
"role": "user",
"content": f"<observation>{tool_result}</observation>"
}
)
chat_history.append(
{
"role": "user",
"content": f"{tool_result} "
}
)
ReAct Loop Step 3¶
In [ ]:
Copied!
output = GROQ_CLIENT.chat.completions.create(
messages=chat_history,
model=MODEL
).choices[0].message.content
print(output)
output = GROQ_CLIENT.chat.completions.create(
messages=chat_history,
model=MODEL
).choices[0].message.content
print(output)
In [ ]:
Copied!
chat_history.append(
{
"role": "assistant",
"content": output
}
)
chat_history.append(
{
"role": "assistant",
"content": output
}
)
ReAct Loop Step 4¶
In [ ]:
Copied!
tool_call = extract_tag_content(output, tag="tool_call")
tool_call = json.loads(tool_call.content[0])
tool_result = available_tools[tool_call["name"]].run(**tool_call["arguments"])
tool_call = extract_tag_content(output, tag="tool_call")
tool_call = json.loads(tool_call.content[0])
tool_result = available_tools[tool_call["name"]].run(**tool_call["arguments"])
In [ ]:
Copied!
tool_result
tool_result
In [ ]:
Copied!
assert tool_result == (1234 + 5678) * 5
assert tool_result == (1234 + 5678) * 5
In [ ]:
Copied!
chat_history.append(
{
"role": "user",
"content": f"<observation>{tool_result}</observation>"
}
)
chat_history.append(
{
"role": "user",
"content": f"{tool_result} "
}
)
ReAct Loop Step 5¶
In [ ]:
Copied!
output = GROQ_CLIENT.chat.completions.create(
messages=chat_history,
model=MODEL
).choices[0].message.content
print(output)
output = GROQ_CLIENT.chat.completions.create(
messages=chat_history,
model=MODEL
).choices[0].message.content
print(output)
In [ ]:
Copied!
chat_history.append(
{
"role": "assistant",
"content": output
}
)
chat_history.append(
{
"role": "assistant",
"content": output
}
)
ReAct Loop Step 6¶
In [ ]:
Copied!
tool_call = extract_tag_content(output, tag="tool_call")
tool_call = json.loads(tool_call.content[0])
tool_result = available_tools[tool_call["name"]].run(**tool_call["arguments"])
tool_call = extract_tag_content(output, tag="tool_call")
tool_call = json.loads(tool_call.content[0])
tool_result = available_tools[tool_call["name"]].run(**tool_call["arguments"])
In [ ]:
Copied!
tool_result
tool_result
In [ ]:
Copied!
assert tool_result == math.log((1234 + 5678) * 5)
assert tool_result == math.log((1234 + 5678) * 5)
In [ ]:
Copied!
chat_history.append(
{
"role": "user",
"content": f"<observation>{tool_result}</observation>"
}
)
chat_history.append(
{
"role": "user",
"content": f"{tool_result} "
}
)
ReAct Loop Step 7¶
In [ ]:
Copied!
output = GROQ_CLIENT.chat.completions.create(
messages=chat_history,
model=MODEL
).choices[0].message.content
print(output)
output = GROQ_CLIENT.chat.completions.create(
messages=chat_history,
model=MODEL
).choices[0].message.content
print(output)
In [ ]:
Copied!
import json
import re
from colorama import Fore
from dotenv import load_dotenv
from groq import Groq
load_dotenv()
BASE_SYSTEM_PROMPT = ""
REACT_SYSTEM_PROMPT = """
You operate by running a loop with the following steps: Thought, Action, Observation.
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' 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 / actions:
<tools>
%s
</tools>
Example session:
<question>What's the current temperature in Madrid?</question>
<thought>I need to get the current weather in Madrid</thought>
<tool_call>{"name": "get_current_weather","arguments": {"location": "Madrid", "unit": "celsius"}, "id": 0}</tool_call>
You will be called again with this:
<observation>{0: {"temperature": 25, "unit": "celsius"}}</observation>
You then output:
<response>The current temperature in Madrid is 25 degrees Celsius</response>
Additional constraints:
- If the user asks you something unrelated to any of the tools above, answer freely enclosing your answer with <response></response> tags.
"""
class ReactAgent:
"""
A class that represents an agent using the ReAct logic that interacts with tools to process
user inputs, make decisions, and execute tool calls. The agent can run interactive sessions,
collect tool signatures, and process multiple tool calls in a given round of interaction.
Attributes:
client (Groq): The Groq client used to handle model-based completions.
model (str): The name of the model used for generating responses. Default is "llama-3.1-70b-versatile".
tools (list[Tool]): A list of Tool instances available for execution.
tools_dict (dict): A dictionary mapping tool names to their corresponding Tool instances.
"""
def __init__(
self,
tools: Tool | list[Tool],
model: str = "llama-3.1-70b-versatile",
system_prompt: str = BASE_SYSTEM_PROMPT,
) -> None:
self.client = Groq()
self.model = model
self.system_prompt = system_prompt
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,
max_rounds: int = 10,
) -> str:
"""
Executes a user interaction session, where the agent processes user input, generates responses,
handles tool calls, and updates chat history until a final response is ready or the maximum
number of rounds is reached.
Args:
user_msg (str): The user's input message to start the interaction.
max_rounds (int, optional): Maximum number of interaction rounds the agent should perform. Default is 10.
Returns:
str: The final response generated by the agent after processing user input and any tool calls.
"""
user_prompt = build_prompt_structure(
prompt=user_msg, role="user", tag="question"
)
if self.tools:
self.system_prompt += (
"\n" + REACT_SYSTEM_PROMPT % self.add_tool_signatures()
)
chat_history = ChatHistory(
[
build_prompt_structure(
prompt=self.system_prompt,
role="system",
),
user_prompt,
]
)
if self.tools:
# Run the ReAct loop for max_rounds
for _ in range(max_rounds):
completion = completions_create(self.client, chat_history, self.model)
response = extract_tag_content(str(completion), "response")
if response.found:
return response.content[0]
thought = extract_tag_content(str(completion), "thought")
tool_calls = extract_tag_content(str(completion), "tool_call")
update_chat_history(chat_history, completion, "assistant")
print(Fore.MAGENTA + f"\nThought: {thought.content[0]}")
if tool_calls.found:
observations = self.process_tool_calls(tool_calls.content)
print(Fore.BLUE + f"\nObservations: {observations}")
update_chat_history(chat_history, f"{observations}", "user")
return completions_create(self.client, chat_history, self.model)
import json
import re
from colorama import Fore
from dotenv import load_dotenv
from groq import Groq
load_dotenv()
BASE_SYSTEM_PROMPT = ""
REACT_SYSTEM_PROMPT = """
You operate by running a loop with the following steps: Thought, Action, Observation.
You are provided with function signatures within
{"name": ,"arguments": , "id": }
Here are the available tools / actions:
%s
Example session:
What's the current temperature in Madrid?
I need to get the current weather in Madrid
{"name": "get_current_weather","arguments": {"location": "Madrid", "unit": "celsius"}, "id": 0}
You will be called again with this:
{0: {"temperature": 25, "unit": "celsius"}}
You then output:
The current temperature in Madrid is 25 degrees Celsius
Additional constraints:
- If the user asks you something unrelated to any of the tools above, answer freely enclosing your answer with
In [ ]:
Copied!
agent = ReactAgent(tools=[sum_two_elements, multiply_two_elements, compute_log])
agent = ReactAgent(tools=[sum_two_elements, multiply_two_elements, compute_log])
In [ ]:
Copied!
agent.run(user_msg="I want to calculate the sum of 1234 and 5678 and multiply the result by 5. Then, I want to take the logarithm of this result")
agent.run(user_msg="I want to calculate the sum of 1234 and 5678 and multiply the result by 5. Then, I want to take the logarithm of this result")
We did it!! A ReAct Agent working as expected, completely from Scratch! 🚀🚀🚀🚀