| 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 | |