File size: 2,757 Bytes
e5c3819 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 |
from textwrap import dedent
from typing import List
from langchain.tools import Tool
from pydantic import BaseModel, Field
from crewai.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
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
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 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 "\nError executing tool. Missing exact 3 pipe (|) separated values. For example, `coworker|task|information`.\n"
if not agent or not task or not information:
return "\nError executing tool. Missing exact 3 pipe (|) separated values. For example, `coworker|question|information`.\n"
agent = [
available_agent
for available_agent in self.agents
if available_agent.role == agent
]
if len(agent) == 0:
return f"\nError executing tool. Co-worker mentioned on the Action Input not found, it must to be one of the following options: {', '.join([agent.role for agent in self.agents])}.\n"
agent = agent[0]
result = agent.execute_task(task, information)
return result
|