Skip to content

Commit

Permalink
including tools
Browse files Browse the repository at this point in the history
  • Loading branch information
@MervinPraison
MervinPraison committed Dec 26, 2024
1 parent 159ef73 commit 5d0f168
Show file tree
Hide file tree
Showing 2 changed files with 211 additions and 83 deletions.
133 changes: 99 additions & 34 deletions agents/example.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,39 @@
from rich.panel import Panel
from rich.text import Text
from praisonaiagents import Agent, Task, PraisonAIAgents, error_logs
import inspect

def generate_tool_definition(func):
"""
Generate a tool definition for a given function.
Args:
func (function): The target function.
Returns:
dict: A dictionary representing the tool definition.
"""
sig = inspect.signature(func)
parameters = {
"type": "object",
"properties": {},
"required": []
}

for name, param in sig.parameters.items():
param_type = "string" # Default type, can be extended for other types
parameters["properties"][name] = {"type": param_type}
if param.default == inspect.Parameter.empty:
parameters["required"].append(name)

return {
"type": "function",
"function": {
"name": func.__name__,
"description": func.__doc__.strip() if func.__doc__ else "No description available",
"parameters": parameters
}
}

# Configure logging
logging.basicConfig(
Expand All @@ -16,8 +49,35 @@ def my_callback(output):
logging.info(f"Callback function called after task execution: {output.description}")
logging.info(f"Task output: {output}")

from duckduckgo_search import DDGS
def tool():
"""
Decorator to automatically register a function as a tool and generate its definition.
"""
def decorator(func):
tool_name = func.__name__
globals()[tool_name + "_definition"] = generate_tool_definition(func)
globals()[tool_name] = func
return func
return decorator

# Example usage of @tool
define_tool = tool()

@define_tool
def my_tool(question: str) -> str:
"""
Clear description for what this tool is useful for, your agent will need this information to use it.
"""
# Function logic here
return "Result from your custom tool"

# Access the generated tool definition and use the tool
# print(my_tool_definition)
# result = my_tool(question="What is AI?")
# print(result)

from duckduckgo_search import DDGS
@define_tool
def internet_search_tool(query):
"""
Perform a search using DuckDuckGo.
Expand All @@ -42,40 +102,45 @@ def internet_search_tool(query):
print(f"Error during DuckDuckGo search: {e}")
return []

# # Access the generated tool definition and use the tool
# print(internet_search_tool_definition)
# result = internet_search_tool(query="What is AI?")
# print(result)

def main():
# Make sure OPENAI_API_KEY is set
if not os.environ.get("OPENAI_API_KEY"):
raise ValueError("Please set OPENAI_API_KEY environment variable")

# Define tools
internet_search_tool = {
"type": "function",
"function": {
"name": "internet_search_tool",
"description": "Use this to perform search queries",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string"},
},
"required": ["query"],
},
},
}
get_weather_tool = {
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the weather for a location",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string"},
},
"required": ["location"],
},
},
}
# # Define tools
# internet_search_tool = {
# "type": "function",
# "function": {
# "name": "internet_search_tool",
# "description": "Use this to perform search queries",
# "parameters": {
# "type": "object",
# "properties": {
# "query": {"type": "string"},
# },
# "required": ["query"],
# },
# },
# }
# get_weather_tool = {
# "type": "function",
# "function": {
# "name": "get_weather",
# "description": "Get the weather for a location",
# "parameters": {
# "type": "object",
# "properties": {
# "location": {"type": "string"},
# },
# "required": ["location"],
# },
# },
# }

# Create agents
researcher = Agent(
Expand All @@ -86,7 +151,7 @@ def main():
skilled in identifying trends and analyzing complex data.""",
verbose=True,
allow_delegation=False,
tools=[internet_search_tool],
tools=['internet_search_tool'],
llm="gpt-4o",
markdown=True
)
Expand All @@ -99,7 +164,7 @@ def main():
verbose=True,
allow_delegation=True,
llm="gpt-4o",
tools=[get_weather_tool],
tools=[],
markdown=True
)

Expand All @@ -110,7 +175,7 @@ def main():
Find major trends, new technologies, and their effects.""",
expected_output="""A detailed report on 2024 AI advancements""",
agent=researcher,
tools=[internet_search_tool]
tools=['internet_search_tool']
)

task2 = Task(
Expand All @@ -123,7 +188,7 @@ def main():
agent=writer,
context=[task1],
callback=my_callback,
tools=[get_weather_tool]
tools=[]
)

task3 = Task(
Expand Down
161 changes: 112 additions & 49 deletions agents/praisonaiagents/agent/agent.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,85 @@
)

class Agent:
def _generate_tool_definition(self, function_name):
"""
Generate a tool definition from a function name by inspecting the function.
"""
# First try to get the tool definition if it exists
tool_def_name = f"{function_name}_definition"
tool_def = globals().get(tool_def_name)
if not tool_def:
import __main__
tool_def = getattr(__main__, tool_def_name, None)

if tool_def:
return tool_def

# If no definition exists, try to generate one from the function
func = globals().get(function_name)
if not func:
import __main__
func = getattr(__main__, function_name, None)

if not func or not callable(func):
return None

import inspect
sig = inspect.signature(func)
parameters = {
"type": "object",
"properties": {},
"required": []
}

# Parse docstring for parameter descriptions
docstring = inspect.getdoc(func)
param_descriptions = {}
if docstring:
import re
param_section = re.split(r'\s*Args:\s*', docstring)
if len(param_section) > 1:
param_lines = param_section[1].split('\n')
for line in param_lines:
line = line.strip()
if line and ':' in line:
param_name, param_desc = line.split(':', 1)
param_descriptions[param_name.strip()] = param_desc.strip()

for name, param in sig.parameters.items():
param_type = "string" # Default type
if param.annotation != inspect.Parameter.empty:
if param.annotation == int:
param_type = "integer"
elif param.annotation == float:
param_type = "number"
elif param.annotation == bool:
param_type = "boolean"
elif param.annotation == list:
param_type = "array"
elif param.annotation == dict:
param_type = "object"

param_info = {"type": param_type}
if name in param_descriptions:
param_info["description"] = param_descriptions[name]

parameters["properties"][name] = param_info
if param.default == inspect.Parameter.empty:
parameters["required"].append(name)

# Extract description from docstring
description = docstring.split('\n')[0] if docstring else f"Function {function_name}"

return {
"type": "function",
"function": {
"name": function_name,
"description": description,
"parameters": parameters
}
}

def __init__(
self,
name: str,
Expand Down Expand Up @@ -53,7 +132,7 @@ def __init__(
self.goal = goal
self.backstory = backstory
self.llm = llm
self.tools = tools if tools else []
self.tools = tools if tools else [] # Store original tools
self.function_calling_llm = function_calling_llm
self.max_iter = max_iter
self.max_rpm = max_rpm
Expand All @@ -79,18 +158,25 @@ def __init__(
self.max_reflection_iter = max_reflection_iter

def execute_tool(self, function_name, arguments):
"""
Execute a tool dynamically based on the function name and arguments.
"""
logging.debug(f"{self.name} executing tool {function_name} with arguments: {arguments}")
if function_name == "get_weather":
location = arguments.get("location", "Unknown Location")
return {"temperature": "25C", "condition": "Sunny", "location": location}
elif function_name == "search_tool":
query = arguments.get("query", "AI trends in 2024")
return {"results": [
{"title": "AI advancements in 2024", "link": "url1", "summary": "Lots of advancements"},
{"title": "New trends in AI", "link": "url2", "summary": "New trends being found"}
]}
else:
return f"Tool '{function_name}' is not recognized"

# Try to get the function from globals first
func = globals().get(function_name)
if not func:
# Then try to get from the main module
import __main__
func = getattr(__main__, function_name, None)

if func and callable(func):
try:
return func(**arguments)
except Exception as e:
return {"error": str(e)}

return {"error": f"Tool '{function_name}' is not callable"}

def clear_history(self):
self.chat_history = []
Expand All @@ -104,26 +190,25 @@ def _chat_completion(self, messages, temperature=0.2, tools=None, stream=True):
logging.debug(f"{self.name} sending messages to LLM: {messages}")

formatted_tools = []
if tools is None:
tools = self.tools
if tools:
for tool in tools:
if isinstance(tool, dict):
if isinstance(tool, str):
# Generate tool definition for string tool names
tool_def = self._generate_tool_definition(tool)
if tool_def:
formatted_tools.append(tool_def)
else:
logging.warning(f"Could not generate definition for tool: {tool}")
elif isinstance(tool, dict):
formatted_tools.append(tool)
elif hasattr(tool, "to_openai_tool"):
formatted_tools.append(tool.to_openai_tool())
elif isinstance(tool, str):
formatted_tools.append({
"type": "function",
"function": {
"name": tool,
"description": f"This is a tool called {tool}",
"parameters": {
"type": "object",
"properties": {},
},
}
})
elif callable(tool):
formatted_tools.append(self._generate_tool_definition(tool.__name__))
else:
display_error(f"Warning: Tool {tool} not recognized")
logging.warning(f"Tool {tool} not recognized")

try:
initial_response = client.chat.completions.create(
Expand Down Expand Up @@ -223,29 +308,7 @@ def chat(self, prompt, temperature=0.2, tools=None, output_json=None):
if self.verbose:
display_instruction(f"Agent {self.name} is processing prompt: {prompt}")

formatted_tools = []
if tools:
for tool in tools:
if isinstance(tool, dict):
formatted_tools.append(tool)
elif hasattr(tool, "to_openai_tool"):
formatted_tools.append(tool.to_openai_tool())
elif isinstance(tool, str):
formatted_tools.append({
"type": "function",
"function": {
"name": tool,
"description": f"This is a tool called {tool}",
"parameters": {
"type": "object",
"properties": {},
},
}
})
else:
display_error(f"Warning: Tool {tool} not recognized")

response = self._chat_completion(messages, temperature=temperature, tools=formatted_tools if formatted_tools else None)
response = self._chat_completion(messages, temperature=temperature, tools=tools if tools else None)
if not response:
return None

Expand Down

0 comments on commit 5d0f168

Please sign in to comment.