mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-08 15:48:29 +00:00
123 lines
3.9 KiB
Python
123 lines
3.9 KiB
Python
import json
|
|
from typing import Any, Dict, List, Optional, Union
|
|
|
|
from pydantic import (
|
|
BaseModel,
|
|
Field,
|
|
InstanceOf,
|
|
Json,
|
|
field_validator,
|
|
model_validator,
|
|
)
|
|
from pydantic_core import PydanticCustomError
|
|
|
|
from crewai.agent import Agent
|
|
from crewai.agents 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."""
|
|
|
|
class Config:
|
|
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."
|
|
)
|
|
|
|
@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)
|