andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-01-10 08:38:30 +00:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Run pre-commit hooks

Browse Source 
In the title !
This commit is contained in:
Greyson Lalonde
2023-12-27 13:23:45 -05:00
parent 542a794e64
commit a4e93cea75
11 changed files with 582 additions and 562 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
crewai/tools/agent_tools.py
View File

@@ -1,61 +1,74 @@
from typing import List, Any
from pydantic.v1 import BaseModel, Field
from textwrap import dedent
from typing import List
from langchain.tools import Tool
from pydantic.v1 import BaseModel, Field
from ..agent import Agent
class AgentTools(BaseModel):
"""Tools for generic agent."""
agents: List[Agent] = Field(description="List of agents in this crew.")
def tools(self):
return [
Tool.from_function(
func=self.delegate_work,
name="Delegate Work to Co-Worker",
description=dedent(f"""Useful to delegate a specific task to one of the
class AgentTools(BaseModel):
"""Tools for generic agent."""
agents: List[Agent] = Field(description="List of agents in this crew.")
def tools(self):
return [
Tool.from_function(
func=self.delegate_work,
name="Delegate Work to Co-Worker",
description=dedent(
f"""Useful to delegate a specific task to one of the
following co-workers: [{', '.join([agent.role for agent in self.agents])}].
The input to this tool should be a pipe (|) separated text of length
three, representing the role you want to delegate it to, the task and
information necessary. For example, `coworker|task|information`.
""")
),
Tool.from_function(
func=self.ask_question,
name="Ask Question to Co-Worker",
description=dedent(f"""Useful to ask a question, opinion or take from on
"""
),
),
Tool.from_function(
func=self.ask_question,
name="Ask Question to Co-Worker",
description=dedent(
f"""Useful to ask a question, opinion or take from on
of the following co-workers: [{', '.join([agent.role for agent in self.agents])}].
The input to this tool should be a pipe (|) separated text of length
three, representing the role you want to ask it to, the question and
information necessary. For example, `coworker|question|information`.
""")
),
]
"""
),
),
]
def delegate_work(self, command):
"""Useful to delegate a specific task to a coworker."""
return self.__execute(command)
def delegate_work(self, command):
"""Useful to delegate a specific task to a coworker."""
return self.__execute(command)
def ask_question(self, command):
"""Useful to ask a question, opinion or take from a coworker."""
return self.__execute(command)
def ask_question(self, command):
"""Useful to ask a question, opinion or take from a coworker."""
return self.__execute(command)
def __execute(self, command):
"""Execute the command."""
try:
agent, task, information = command.split("|")
except ValueError:
return "Error executing tool. Missing exact 3 pipe (|) separated values. For example, `coworker|task|information`."
def __execute(self, command):
"""Execute the command."""
try:
agent, task, information = command.split("|")
except ValueError:
return "Error executing tool. Missing exact 3 pipe (|) separated values. For example, `coworker|task|information`."
if not agent or not task or not information:
return "Error executing tool. Missing exact 3 pipe (|) separated values. For example, `coworker|question|information`."
if not agent or not task or not information:
return "Error executing tool. Missing exact 3 pipe (|) separated values. For example, `coworker|question|information`."
agent = [available_agent for available_agent in self.agents if available_agent.role == agent]
agent = [
available_agent
for available_agent in self.agents
if available_agent.role == agent
]
if len(agent) == 0:
return "Error executing tool. Co-worker not found, double check the co-worker."
if len(agent) == 0:
return (
"Error executing tool. Co-worker not found, double check the co-worker."
)
agent = agent[0]
result = agent.execute_task(task, information)
return result
agent = agent[0]
result = agent.execute_task(task, information)
return result