Upload 17 files

crewai/__init__.py

@@ -0,0 +1,4 @@
+from crewai.agent import Agent
+from crewai.crew import Crew
+from crewai.process import Process
+from crewai.task import Task

crewai/agent.py

@@ -0,0 +1,181 @@
+import uuid
+from typing import Any, List, Optional
+
+from langchain.agents.format_scratchpad import format_log_to_str
+from langchain.chat_models import ChatOpenAI
+from langchain.memory import ConversationSummaryMemory
+from langchain.tools.render import render_text_description
+from langchain_core.runnables.config import RunnableConfig
+from pydantic import (
+    UUID4,
+    BaseModel,
+    ConfigDict,
+    Field,
+    InstanceOf,
+    field_validator,
+    model_validator,
+)
+from pydantic_core import PydanticCustomError
+
+from crewai.agents import (
+    CacheHandler,
+    CrewAgentExecutor,
+    CrewAgentOutputParser,
+    ToolsHandler,
+)
+from crewai.prompts import Prompts
+
+
+class Agent(BaseModel):
+    """Represents an agent in a system.
+
+    Each agent has a role, a goal, a backstory, and an optional language model (llm).
+    The agent can also have memory, can operate in verbose mode, and can delegate tasks to other agents.
+
+    Attributes:
+            agent_executor: An instance of the CrewAgentExecutor class.
+            role: The role of the agent.
+            goal: The objective of the agent.
+            backstory: The backstory of the agent.
+            llm: The language model that will run the agent.
+            memory: Whether the agent should have memory or not.
+            verbose: Whether the agent execution should be in verbose mode.
+            allow_delegation: Whether the agent is allowed to delegate tasks to other agents.
+    """
+
+    __hash__ = object.__hash__
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+    id: UUID4 = Field(
+        default_factory=uuid.uuid4,
+        frozen=True,
+        description="Unique identifier for the object, not set by user.",
+    )
+    role: str = Field(description="Role of the agent")
+    goal: str = Field(description="Objective of the agent")
+    backstory: str = Field(description="Backstory of the agent")
+    llm: Optional[Any] = Field(
+        default_factory=lambda: ChatOpenAI(
+            temperature=0.7,
+            model_name="gpt-4",
+        ),
+        description="Language model that will run the agent.",
+    )
+    memory: bool = Field(
+        default=True, description="Whether the agent should have memory or not"
+    )
+    verbose: bool = Field(
+        default=False, description="Verbose mode for the Agent Execution"
+    )
+    allow_delegation: bool = Field(
+        default=True, description="Allow delegation of tasks to agents"
+    )
+    tools: List[Any] = Field(
+        default_factory=list, description="Tools at agents disposal"
+    )
+    agent_executor: Optional[InstanceOf[CrewAgentExecutor]] = Field(
+        default=None, description="An instance of the CrewAgentExecutor class."
+    )
+    tools_handler: Optional[InstanceOf[ToolsHandler]] = Field(
+        default=None, description="An instance of the ToolsHandler class."
+    )
+    cache_handler: Optional[InstanceOf[CacheHandler]] = Field(
+        default=CacheHandler(), description="An instance of the CacheHandler class."
+    )
+
+    @field_validator("id", mode="before")
+    @classmethod
+    def _deny_user_set_id(cls, v: Optional[UUID4]) -> None:
+        if v:
+            raise PydanticCustomError(
+                "may_not_set_field", "This field is not to be set by the user.", {}
+            )
+
+    @model_validator(mode="after")
+    def check_agent_executor(self) -> "Agent":
+        if not self.agent_executor:
+            self.set_cache_handler(self.cache_handler)
+        return self
+
+    def execute_task(
+        self, task: str, context: str = None, tools: List[Any] = None
+    ) -> str:
+        """Execute a task with the agent.
+
+        Args:
+            task: Task to execute.
+            context: Context to execute the task in.
+            tools: Tools to use for the task.
+
+        Returns:
+            Output of the agent
+        """
+        if context:
+            task = "\n".join(
+                [task, "\nThis is the context you are working with:", context]
+            )
+
+        tools = tools or self.tools
+        self.agent_executor.tools = tools
+
+        return self.agent_executor.invoke(
+            {
+                "input": task,
+                "tool_names": self.__tools_names(tools),
+                "tools": render_text_description(tools),
+            },
+            RunnableConfig(callbacks=[self.tools_handler]),
+        )["output"]
+
+    def set_cache_handler(self, cache_handler) -> None:
+        self.cache_handler = cache_handler
+        self.tools_handler = ToolsHandler(cache=self.cache_handler)
+        self.__create_agent_executor()
+
+    def __create_agent_executor(self) -> CrewAgentExecutor:
+        """Create an agent executor for the agent.
+
+        Returns:
+            An instance of the CrewAgentExecutor class.
+        """
+        agent_args = {
+            "input": lambda x: x["input"],
+            "tools": lambda x: x["tools"],
+            "tool_names": lambda x: x["tool_names"],
+            "agent_scratchpad": lambda x: format_log_to_str(x["intermediate_steps"]),
+        }
+        executor_args = {
+            "tools": self.tools,
+            "verbose": self.verbose,
+            "handle_parsing_errors": True,
+        }
+
+        if self.memory:
+            summary_memory = ConversationSummaryMemory(
+                llm=self.llm, memory_key="chat_history", input_key="input"
+            )
+            executor_args["memory"] = summary_memory
+            agent_args["chat_history"] = lambda x: x["chat_history"]
+            prompt = Prompts.TASK_EXECUTION_WITH_MEMORY_PROMPT
+        else:
+            prompt = Prompts.TASK_EXECUTION_PROMPT
+
+        execution_prompt = prompt.partial(
+            goal=self.goal,
+            role=self.role,
+            backstory=self.backstory,
+        )
+
+        bind = self.llm.bind(stop=["\nObservation"])
+        inner_agent = (
+            agent_args
+            | execution_prompt
+            | bind
+            | CrewAgentOutputParser(
+                tools_handler=self.tools_handler, cache=self.cache_handler
+            )
+        )
+        self.agent_executor = CrewAgentExecutor(agent=inner_agent, **executor_args)
+
+    @staticmethod
+    def __tools_names(tools) -> str:
+        return ", ".join([t.name for t in tools])

crewai/agents/__init__.py

@@ -0,0 +1,4 @@
+from .cache.cache_handler import CacheHandler
+from .executor import CrewAgentExecutor
+from .output_parser import CrewAgentOutputParser
+from .tools_handler import ToolsHandler

crewai/agents/cache/__init__.py

@@ -0,0 +1,2 @@
+from .cache_handler import CacheHandler
+from .cache_hit import CacheHit

crewai/agents/cache/cache_handler.py

@@ -0,0 +1,20 @@
+from typing import Optional
+
+from pydantic import PrivateAttr
+
+
+class CacheHandler:
+    """Callback handler for tool usage."""
+
+    _cache: PrivateAttr = {}
+
+    def __init__(self):
+        self._cache = {}
+
+    def add(self, tool, input, output):
+        input = input.strip()
+        self._cache[f"{tool}-{input}"] = output
+
+    def read(self, tool, input) -> Optional[str]:
+        input = input.strip()
+        return self._cache.get(f"{tool}-{input}")

crewai/agents/cache/cache_hit.py

@@ -0,0 +1,14 @@
+from langchain_core.agents import AgentAction
+from pydantic.v1 import BaseModel, Field
+
+from .cache_handler import CacheHandler
+
+
+class CacheHit(BaseModel):
+    """Cache Hit Object."""
+
+    class Config:
+        arbitrary_types_allowed = True
+
+    action: AgentAction = Field(description="Action taken")
+    cache: CacheHandler = Field(description="Cache Handler for the tool")

crewai/agents/exceptions.py

@@ -0,0 +1,20 @@
+from langchain_core.exceptions import OutputParserException
+
+
+class TaskRepeatedUsageException(OutputParserException):
+    """Exception raised when a task is used twice in a roll."""
+
+    error: str = "TaskRepeatedUsageException"
+    message: str = "\nI just used the {tool} tool with input {tool_input}. So I already know the result of that.\n"
+
+    def __init__(self, tool: str, tool_input: str):
+        self.tool = tool
+        self.tool_input = tool_input
+        self.message = self.message.format(tool=tool, tool_input=tool_input)
+
+        super().__init__(
+            error=self.error, observation=self.message, send_to_llm=True, llm_output=""
+        )
+
+    def __str__(self):
+        return self.message

crewai/agents/executor.py

@@ -0,0 +1,130 @@
+from typing import Dict, Iterator, List, Optional, Tuple, Union
+
+from langchain.agents import AgentExecutor
+from langchain.agents.agent import ExceptionTool
+from langchain.agents.tools import InvalidTool
+from langchain.callbacks.manager import CallbackManagerForChainRun
+from langchain_core.agents import AgentAction, AgentFinish, AgentStep
+from langchain_core.exceptions import OutputParserException
+from langchain_core.tools import BaseTool
+
+from ..tools.cache_tools import CacheTools
+from .cache.cache_hit import CacheHit
+
+
+class CrewAgentExecutor(AgentExecutor):
+    def _iter_next_step(
+        self,
+        name_to_tool_map: Dict[str, BaseTool],
+        color_mapping: Dict[str, str],
+        inputs: Dict[str, str],
+        intermediate_steps: List[Tuple[AgentAction, str]],
+        run_manager: Optional[CallbackManagerForChainRun] = None,
+    ) -> Iterator[Union[AgentFinish, AgentAction, AgentStep]]:
+        """Take a single step in the thought-action-observation loop.
+
+        Override this to take control of how the agent makes and acts on choices.
+        """
+        try:
+            intermediate_steps = self._prepare_intermediate_steps(intermediate_steps)
+
+            # Call the LLM to see what to do.
+            output = self.agent.plan(
+                intermediate_steps,
+                callbacks=run_manager.get_child() if run_manager else None,
+                **inputs,
+            )
+        except OutputParserException as e:
+            if isinstance(self.handle_parsing_errors, bool):
+                raise_error = not self.handle_parsing_errors
+            else:
+                raise_error = False
+            if raise_error:
+                raise ValueError(
+                    "An output parsing error occurred. "
+                    "In order to pass this error back to the agent and have it try "
+                    "again, pass `handle_parsing_errors=True` to the AgentExecutor. "
+                    f"This is the error: {str(e)}"
+                )
+            text = str(e)
+            if isinstance(self.handle_parsing_errors, bool):
+                if e.send_to_llm:
+                    observation = str(e.observation)
+                    text = str(e.llm_output)
+                else:
+                    observation = "Invalid or incomplete response"
+            elif isinstance(self.handle_parsing_errors, str):
+                observation = self.handle_parsing_errors
+            elif callable(self.handle_parsing_errors):
+                observation = self.handle_parsing_errors(e)
+            else:
+                raise ValueError("Got unexpected type of `handle_parsing_errors`")
+            output = AgentAction("_Exception", observation, text)
+            if run_manager:
+                run_manager.on_agent_action(output, color="green")
+            tool_run_kwargs = self.agent.tool_run_logging_kwargs()
+            observation = ExceptionTool().run(
+                output.tool_input,
+                verbose=self.verbose,
+                color=None,
+                callbacks=run_manager.get_child() if run_manager else None,
+                **tool_run_kwargs,
+            )
+            yield AgentStep(action=output, observation=observation)
+            return
+
+        # If the tool chosen is the finishing tool, then we end and return.
+        if isinstance(output, AgentFinish):
+            yield output
+            return
+
+        # Override tool usage to use CacheTools
+        if isinstance(output, CacheHit):
+            cache = output.cache
+            action = output.action
+            tool = CacheTools(cache_handler=cache).tool()
+            output = action.copy()
+            output.tool_input = f"tool:{action.tool}|input:{action.tool_input}"
+            output.tool = tool.name
+            name_to_tool_map[tool.name] = tool
+            color_mapping[tool.name] = color_mapping[action.tool]
+
+        actions: List[AgentAction]
+        if isinstance(output, AgentAction):
+            actions = [output]
+        else:
+            actions = output
+        for agent_action in actions:
+            yield agent_action
+        for agent_action in actions:
+            if run_manager:
+                run_manager.on_agent_action(agent_action, color="green")
+            # Otherwise we lookup the tool
+            if agent_action.tool in name_to_tool_map:
+                tool = name_to_tool_map[agent_action.tool]
+                return_direct = tool.return_direct
+                color = color_mapping[agent_action.tool]
+                tool_run_kwargs = self.agent.tool_run_logging_kwargs()
+                if return_direct:
+                    tool_run_kwargs["llm_prefix"] = ""
+                # We then call the tool on the tool input to get an observation
+                observation = tool.run(
+                    agent_action.tool_input,
+                    verbose=self.verbose,
+                    color=color,
+                    callbacks=run_manager.get_child() if run_manager else None,
+                    **tool_run_kwargs,
+                )
+            else:
+                tool_run_kwargs = self.agent.tool_run_logging_kwargs()
+                observation = InvalidTool().run(
+                    {
+                        "requested_tool_name": agent_action.tool,
+                        "available_tool_names": list(name_to_tool_map.keys()),
+                    },
+                    verbose=self.verbose,
+                    color=None,
+                    callbacks=run_manager.get_child() if run_manager else None,
+                    **tool_run_kwargs,
+                )
+            yield AgentStep(action=agent_action, observation=observation)

crewai/agents/output_parser.py

@@ -0,0 +1,76 @@
+import re
+from typing import Union
+
+from langchain.agents.output_parsers import ReActSingleInputOutputParser
+from langchain_core.agents import AgentAction, AgentFinish
+
+from .cache import CacheHandler, CacheHit
+from .exceptions import TaskRepeatedUsageException
+from .tools_handler import ToolsHandler
+
+FINAL_ANSWER_ACTION = "Final Answer:"
+FINAL_ANSWER_AND_PARSABLE_ACTION_ERROR_MESSAGE = (
+    "Parsing LLM output produced both a final answer and a parse-able action:"
+)
+
+
+class CrewAgentOutputParser(ReActSingleInputOutputParser):
+    """Parses ReAct-style LLM calls that have a single tool input.
+
+    Expects output to be in one of two formats.
+
+    If the output signals that an action should be taken,
+    should be in the below format. This will result in an AgentAction
+    being returned.
+
+    ```
+    Thought: agent thought here
+    Action: search
+    Action Input: what is the temperature in SF?
+    ```
+
+    If the output signals that a final answer should be given,
+    should be in the below format. This will result in an AgentFinish
+    being returned.
+
+    ```
+    Thought: agent thought here
+    Final Answer: The temperature is 100 degrees
+    ```
+
+    It also prevents tools from being reused in a roll.
+    """
+
+    class Config:
+        arbitrary_types_allowed = True
+
+    tools_handler: ToolsHandler
+    cache: CacheHandler
+
+    def parse(self, text: str) -> Union[AgentAction, AgentFinish, CacheHit]:
+        FINAL_ANSWER_ACTION in text
+        regex = (
+            r"Action\s*\d*\s*:[\s]*(.*?)[\s]*Action\s*\d*\s*Input\s*\d*\s*:[\s]*(.*)"
+        )
+        action_match = re.search(regex, text, re.DOTALL)
+        if action_match:
+            action = action_match.group(1).strip()
+            action_input = action_match.group(2)
+            tool_input = action_input.strip(" ")
+            tool_input = tool_input.strip('"')
+
+            last_tool_usage = self.tools_handler.last_used_tool
+            if last_tool_usage:
+                usage = {
+                    "tool": action,
+                    "input": tool_input,
+                }
+                if usage == last_tool_usage:
+                    raise TaskRepeatedUsageException(tool=action, tool_input=tool_input)
+
+            result = self.cache.read(action, tool_input)
+            if result:
+                action = AgentAction(action, tool_input, text)
+                return CacheHit(action=action, cache=self.cache)
+
+        return super().parse(text)

crewai/agents/tools_handler.py

@@ -0,0 +1,44 @@
+from typing import Any, Dict
+
+from langchain.callbacks.base import BaseCallbackHandler
+
+from ..tools.cache_tools import CacheTools
+from .cache.cache_handler import CacheHandler
+
+
+class ToolsHandler(BaseCallbackHandler):
+    """Callback handler for tool usage."""
+
+    last_used_tool: Dict[str, Any] = {}
+    cache: CacheHandler = None
+
+    def __init__(self, cache: CacheHandler = None, **kwargs: Any):
+        """Initialize the callback handler."""
+        self.cache = cache
+        super().__init__(**kwargs)
+
+    def on_tool_start(
+        self, serialized: Dict[str, Any], input_str: str, **kwargs: Any
+    ) -> Any:
+        """Run when tool starts running."""
+        name = serialized.get("name")
+        if name not in ["invalid_tool", "_Exception"]:
+            tools_usage = {
+                "tool": name,
+                "input": input_str,
+            }
+            self.last_used_tool = tools_usage
+
+    def on_tool_end(self, output: str, **kwargs: Any) -> Any:
+        """Run when tool ends running."""
+        if (
+            "is not a valid tool" not in output
+            and "Invalid or incomplete response" not in output
+            and "Invalid Format" not in output
+        ):
+            if self.last_used_tool["tool"] != CacheTools().name:
+                self.cache.add(
+                    tool=self.last_used_tool["tool"],
+                    input=self.last_used_tool["input"],
+                    output=output,
+                )

crewai/crew.py

@@ -0,0 +1,137 @@
+import json
+import uuid
+from typing import Any, Dict, List, Optional, Union
+
+from pydantic import (
+    UUID4,
+    BaseModel,
+    ConfigDict,
+    Field,
+    InstanceOf,
+    Json,
+    field_validator,
+    model_validator,
+)
+from pydantic_core import PydanticCustomError
+
+from crewai.agent import Agent
+from crewai.agents.cache import CacheHandler
+from crewai.process import Process
+from crewai.task import Task
+from crewai.tools.agent_tools import AgentTools
+
+
+class Crew(BaseModel):
+    """Class that represents a group of agents, how they should work together and their tasks."""
+
+    __hash__ = object.__hash__
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+    tasks: List[Task] = Field(description="List of tasks", default_factory=list)
+    agents: List[Agent] = Field(
+        description="List of agents in this crew.", default_factory=list
+    )
+    process: Process = Field(
+        description="Process that the crew will follow.", default=Process.sequential
+    )
+    verbose: Union[int, bool] = Field(
+        description="Verbose mode for the Agent Execution", default=0
+    )
+    config: Optional[Union[Json, Dict[str, Any]]] = Field(
+        description="Configuration of the crew.", default=None
+    )
+    cache_handler: Optional[InstanceOf[CacheHandler]] = Field(
+        default=CacheHandler(), description="An instance of the CacheHandler class."
+    )
+    id: UUID4 = Field(
+        default_factory=uuid.uuid4,
+        frozen=True,
+        description="Unique identifier for the object, not set by user.",
+    )
+
+    @field_validator("id", mode="before")
+    @classmethod
+    def _deny_user_set_id(cls, v: Optional[UUID4]) -> None:
+        if v:
+            raise PydanticCustomError(
+                "may_not_set_field", "This field is not to be set by the user.", {}
+            )
+
+    @classmethod
+    @field_validator("config", mode="before")
+    def check_config_type(cls, v: Union[Json, Dict[str, Any]]):
+        if isinstance(v, Json):
+            return json.loads(v)
+        return v
+
+    @model_validator(mode="after")
+    def check_config(self):
+        if not self.config and not self.tasks and not self.agents:
+            raise PydanticCustomError(
+                "missing_keys", "Either agents and task need to be set or config.", {}
+            )
+
+        if self.config:
+            if not self.config.get("agents") or not self.config.get("tasks"):
+                raise PydanticCustomError(
+                    "missing_keys_in_config", "Config should have agents and tasks", {}
+                )
+
+            self.agents = [Agent(**agent) for agent in self.config["agents"]]
+
+            tasks = []
+            for task in self.config["tasks"]:
+                task_agent = [agt for agt in self.agents if agt.role == task["agent"]][
+                    0
+                ]
+                del task["agent"]
+                tasks.append(Task(**task, agent=task_agent))
+
+            self.tasks = tasks
+
+        if self.agents:
+            for agent in self.agents:
+                agent.set_cache_handler(self.cache_handler)
+        return self
+
+    def kickoff(self) -> str:
+        """Kickoff the crew to work on its tasks.
+
+        Returns:
+            Output of the crew for each task.
+        """
+        for agent in self.agents:
+            agent.cache_handler = self.cache_handler
+
+        if self.process == Process.sequential:
+            return self.__sequential_loop()
+
+    def __sequential_loop(self) -> str:
+        """Loop that executes the sequential process.
+
+        Returns:
+            Output of the crew.
+        """
+        task_outcome = None
+        for task in self.tasks:
+            # Add delegation tools to the task if the agent allows it
+            if task.agent.allow_delegation:
+                tools = AgentTools(agents=self.agents).tools()
+                task.tools += tools
+
+            self.__log("debug", f"Working Agent: {task.agent.role}")
+            self.__log("info", f"Starting Task: {task.description} ...")
+
+            task_outcome = task.execute(task_outcome)
+
+            self.__log("debug", f"Task output: {task_outcome}")
+
+        return task_outcome
+
+    def __log(self, level, message):
+        """Log a message"""
+        level_map = {"debug": 1, "info": 2}
+        verbose_level = (
+            2 if isinstance(self.verbose, bool) and self.verbose else self.verbose
+        )
+        if verbose_level and level_map[level] <= verbose_level:
+            print(message)

crewai/process.py

@@ -0,0 +1,11 @@
+from enum import Enum
+
+
+class Process(str, Enum):
+    """
+    Class representing the different processes that can be used to tackle tasks
+    """
+
+    sequential = "sequential"
+    # TODO: consensual = 'consensual'
+    # TODO: hierarchical = 'hierarchical'

crewai/prompts.py

@@ -0,0 +1,84 @@
+"""Prompts for generic agent."""
+
+from textwrap import dedent
+from typing import ClassVar
+
+from langchain.prompts import PromptTemplate
+from pydantic import BaseModel
+
+
+class Prompts(BaseModel):
+    """Prompts for generic agent."""
+
+    TASK_SLICE: ClassVar[str] = dedent(
+        """\
+	Begin! This is VERY important to you, your job depends on it!
+
+	Current Task: {input}"""
+    )
+
+    SCRATCHPAD_SLICE: ClassVar[str] = "\n{agent_scratchpad}"
+
+    MEMORY_SLICE: ClassVar[str] = dedent(
+        """\
+	This is the summary of your work so far:
+	{chat_history}"""
+    )
+
+    ROLE_PLAYING_SLICE: ClassVar[str] = dedent(
+        """\
+	You are {role}.
+	{backstory}
+
+	Your personal goal is: {goal}"""
+    )
+
+    TOOLS_SLICE: ClassVar[str] = dedent(
+        """\
+
+
+	TOOLS:
+	------
+	You have access to the following tools:
+
+	{tools}
+
+	To use a tool, please use the exact following format:
+
+	```
+	Thought: Do I need to use a tool? Yes
+	Action: the action to take, should be one of [{tool_names}], just the name.
+	Action Input: the input to the action
+	Observation: the result of the action
+	```
+
+	When you have a response for your task, or if you do not need to use a tool, you MUST use the format:
+
+	```
+	Thought: Do I need to use a tool? No
+	Final Answer: [your response here]
+	```"""
+    )
+
+    VOTING_SLICE: ClassVar[str] = dedent(
+        """\
+	You are working on a crew with your co-workers and need to decide who will execute the task.
+
+	These are your format instructions:
+	{format_instructions}
+
+	These are your co-workers and their roles:
+	{coworkers}"""
+    )
+
+    TASK_EXECUTION_WITH_MEMORY_PROMPT: ClassVar[str] = PromptTemplate.from_template(
+        ROLE_PLAYING_SLICE + TOOLS_SLICE + MEMORY_SLICE + TASK_SLICE + SCRATCHPAD_SLICE
+    )
+
+    TASK_EXECUTION_PROMPT: ClassVar[str] = PromptTemplate.from_template(
+        ROLE_PLAYING_SLICE + TOOLS_SLICE + TASK_SLICE + SCRATCHPAD_SLICE
+    )
+
+    CONSENSUNS_VOTING_PROMPT: ClassVar[str] = PromptTemplate.from_template(
+        ROLE_PLAYING_SLICE + VOTING_SLICE + TASK_SLICE + SCRATCHPAD_SLICE
+    )

crewai/task.py

@@ -0,0 +1,55 @@
+import uuid
+from typing import Any, List, Optional
+
+from pydantic import UUID4, BaseModel, Field, field_validator, model_validator
+from pydantic_core import PydanticCustomError
+
+from crewai.agent import Agent
+
+
+class Task(BaseModel):
+    """Class that represent a task to be executed."""
+
+    __hash__ = object.__hash__
+    description: str = Field(description="Description of the actual task.")
+    agent: Optional[Agent] = Field(
+        description="Agent responsible for the task.", default=None
+    )
+    tools: List[Any] = Field(
+        default_factory=list,
+        description="Tools the agent are limited to use for this task.",
+    )
+    id: UUID4 = Field(
+        default_factory=uuid.uuid4,
+        frozen=True,
+        description="Unique identifier for the object, not set by user.",
+    )
+
+    @field_validator("id", mode="before")
+    @classmethod
+    def _deny_user_set_id(cls, v: Optional[UUID4]) -> None:
+        if v:
+            raise PydanticCustomError(
+                "may_not_set_field", "This field is not to be set by the user.", {}
+            )
+
+    @model_validator(mode="after")
+    def check_tools(self):
+        if not self.tools and (self.agent and self.agent.tools):
+            self.tools.extend(self.agent.tools)
+        return self
+
+    def execute(self, context: str = None) -> str:
+        """Execute the task.
+
+        Returns:
+            Output of the task.
+        """
+        if self.agent:
+            return self.agent.execute_task(
+                task=self.description, context=context, tools=self.tools
+            )
+        else:
+            raise Exception(
+                f"The task '{self.description}' has no agent assigned, therefore it can't be executed directly and should be executed in a Crew using a specific process that support that, either consensual or hierarchical."
+            )

crewai/tools/agent_tools.py

@@ -0,0 +1,72 @@
+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

crewai/tools/cache_tools.py

@@ -0,0 +1,28 @@
+from langchain.tools import Tool
+from pydantic import BaseModel, ConfigDict, Field
+
+from crewai.agents.cache import CacheHandler
+
+
+class CacheTools(BaseModel):
+    """Default tools to hit the cache."""
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+    name: str = "Hit Cache"
+    cache_handler: CacheHandler = Field(
+        description="Cache Handler for the crew",
+        default=CacheHandler(),
+    )
+
+    def tool(self):
+        return Tool.from_function(
+            func=self.hit_cache,
+            name=self.name,
+            description="Reads directly from the cache",
+        )
+
+    def hit_cache(self, key):
+        split = key.split("tool:")
+        tool = split[1].split("|input:")[0].strip()
+        tool_input = split[1].split("|input:")[1].strip()
+        return self.cache_handler.read(tool, tool_input)