From 8eba7aab897fcde2e524996b9b828e9e2fabdca7 Mon Sep 17 00:00:00 2001 From: yanzz <84555416+paipeline@users.noreply.github.com> Date: Wed, 10 Jan 2024 03:29:50 +0100 Subject: [PATCH] improved readability (#90) --- crewai/crew.py | 131 ++++++++++++++++++++-------------------------- crewai/prompts.py | 53 +++++++++---------- 2 files changed, 83 insertions(+), 101 deletions(-) diff --git a/crewai/crew.py b/crewai/crew.py index 00db60918..b131aa8bd 100644 --- a/crewai/crew.py +++ b/crewai/crew.py @@ -3,14 +3,7 @@ import uuid from typing import Any, Dict, List, Optional, Union from pydantic import ( - UUID4, - BaseModel, - ConfigDict, - Field, - InstanceOf, - Json, - field_validator, - model_validator, + BaseModel, ConfigDict, Field, InstanceOf, Json, UUID4 ) from pydantic_core import PydanticCustomError @@ -20,118 +13,110 @@ 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.""" + """ + Represents a group of agents, defining how they should collaborate and the tasks they should perform. + + Attributes: + tasks: List of tasks assigned to the crew. + agents: List of agents part of this crew. + process: The process flow that the crew will follow (e.g., sequential). + verbose: Indicates the verbosity level for logging during execution. + config: Configuration settings for the crew. + cache_handler: Handles caching for the crew's operations. + id: A unique identifier for the crew instance. + """ __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 - ) + tasks: List[Task] = Field(default_factory=list) + agents: List[Agent] = Field(default_factory=list) + process: Process = Field(default=Process.sequential) + verbose: Union[int, bool] = Field(default=0) + config: Optional[Union[Json, Dict[str, Any]]] = Field(default=None) cache_handler: Optional[InstanceOf[CacheHandler]] = Field( - default=CacheHandler(), description="An instance of the CacheHandler class." + default=CacheHandler() ) id: UUID4 = Field( - default_factory=uuid.uuid4, - frozen=True, - description="Unique identifier for the object, not set by user.", + default_factory=uuid.uuid4, frozen=True ) @field_validator("id", mode="before") @classmethod def _deny_user_set_id(cls, v: Optional[UUID4]) -> None: + """Prevent manual setting of the 'id' field by users.""" if v: raise PydanticCustomError( - "may_not_set_field", "This field is not to be set by the user.", {} + "may_not_set_field", "The 'id' field cannot be set by the user.", {} ) @classmethod @field_validator("config", mode="before") def check_config_type(cls, v: Union[Json, Dict[str, Any]]): + """Ensures the 'config' field is a valid JSON or dictionary.""" if isinstance(v, Json): return json.loads(v) return v @model_validator(mode="after") def check_config(self): + """Validates that the crew is properly configured with agents and tasks.""" 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.", {} + "missing_keys", "Either 'agents' and 'tasks' 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 + self._setup_from_config() 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. + def _setup_from_config(self): + """Initializes agents and tasks from the provided 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'.", {} + ) - Returns: - Output of the crew for each task. - """ + self.agents = [Agent(**agent) for agent in self.config["agents"]] + self.tasks = [self._create_task(task) for task in self.config["tasks"]] + + def _create_task(self, task_config): + """Creates a task instance from its configuration.""" + task_agent = next(agt for agt in self.agents if agt.role == task_config["agent"]) + del task_config["agent"] + return Task(**task_config, agent=task_agent) + + def kickoff(self) -> str: + """Starts the crew to work on its assigned tasks.""" for agent in self.agents: agent.cache_handler = self.cache_handler if self.process == Process.sequential: - return self.__sequential_loop() + return self._sequential_loop() - def __sequential_loop(self) -> str: - """Loop that executes the sequential process. - - Returns: - Output of the crew. - """ + def _sequential_loop(self) -> str: + """Executes tasks sequentially and returns the final output.""" task_output = None for task in self.tasks: - # Add delegation tools to the task if the agent allows it - if task.agent.allow_delegation: - agent_tools = AgentTools(agents=self.agents).tools() - task.tools += agent_tools - - self.__log("debug", f"Working Agent: {task.agent.role}") - self.__log("info", f"Starting Task: {task.description}") - + self._prepare_and_execute_task(task) task_output = task.execute(task_output) - self.__log( - "debug", f"\n\n[{task.agent.role}] Task output: {task_output}\n\n" - ) return task_output - def __log(self, level, message): - """Log a message""" + def _prepare_and_execute_task(self, task): + """Prepares and logs information about the task being executed.""" + if task.agent.allow_delegation: + task.tools += AgentTools(agents=self.agents).tools() + + self._log("debug", f"Working Agent: {task.agent.role}") + self._log("info", f"Starting Task: {task.description}") + + def _log(self, level, message): + """Logs a message at the specified verbosity level.""" level_map = {"debug": 1, "info": 2} - verbose_level = ( - 2 if isinstance(self.verbose, bool) and self.verbose else self.verbose - ) + 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) diff --git a/crewai/prompts.py b/crewai/prompts.py index 037efcd70..c805ebb65 100644 --- a/crewai/prompts.py +++ b/crewai/prompts.py @@ -1,53 +1,50 @@ -"""Prompts for generic agent.""" import json import os from typing import ClassVar, Dict, Optional from langchain.prompts import PromptTemplate -from pydantic import BaseModel, Field, PrivateAttr, model_validator - +from pydantic import BaseModel, Field, PrivateAttr, model_validator, ValidationError class Prompts(BaseModel): - """Prompts for generic agent.""" + """Manages and generates prompts for a generic agent with support for different languages.""" _prompts: Optional[Dict[str, str]] = PrivateAttr() language: Optional[str] = Field( default="en", - description="Language of crewai prompts.", + description="Language of the prompts.", ) @model_validator(mode="after") def load_prompts(self) -> "Prompts": - """Load prompts from file.""" - dir_path = os.path.dirname(os.path.realpath(__file__)) - prompts_path = os.path.join(dir_path, f"prompts/{self.language}.json") + """Load prompts from a JSON file based on the specified language.""" + try: + dir_path = os.path.dirname(os.path.realpath(__file__)) + prompts_path = os.path.join(dir_path, f"prompts/{self.language}.json") - with open(prompts_path, "r") as f: - self._prompts = json.load(f)["slices"] + with open(prompts_path, "r") as f: + self._prompts = json.load(f)["slices"] + except FileNotFoundError: + raise ValidationError(f"Prompt file for language '{self.language}' not found.") + except json.JSONDecodeError: + raise ValidationError(f"Error decoding JSON from the prompts file.") return self SCRATCHPAD_SLICE: ClassVar[str] = "\n{agent_scratchpad}" def task_execution_with_memory(self) -> str: - return PromptTemplate.from_template( - self._prompts["role_playing"] - + self._prompts["tools"] - + self._prompts["memory"] - + self._prompts["task"] - + self.SCRATCHPAD_SLICE - ) + """Generate a prompt for task execution with memory components.""" + return self._build_prompt(["role_playing", "tools", "memory", "task"]) def task_execution_without_tools(self) -> str: - return PromptTemplate.from_template( - self._prompts["role_playing"] - + self._prompts["task"] - + self.SCRATCHPAD_SLICE - ) + """Generate a prompt for task execution without tools components.""" + return self._build_prompt(["role_playing", "task"]) def task_execution(self) -> str: - return PromptTemplate.from_template( - self._prompts["role_playing"] - + self._prompts["tools"] - + self._prompts["task"] - + self.SCRATCHPAD_SLICE - ) + """Generate a standard prompt for task execution.""" + return self._build_prompt(["role_playing", "tools", "task"]) + + def _build_prompt(self, components: [str]) -> str: + """Constructs a prompt string from specified components.""" + prompt_parts = [self._prompts[component] for component in components if component in self._prompts] + prompt_parts.append(self.SCRATCHPAD_SLICE) + return PromptTemplate.from_template("".join(prompt_parts))