diff --git a/src/crewai/agent.py b/src/crewai/agent.py index 303d7236b..9fc10e0a2 100644 --- a/src/crewai/agent.py +++ b/src/crewai/agent.py @@ -267,31 +267,6 @@ class Agent(BaseAgent): def get_output_converter(self, llm, text, model, instructions): return Converter(llm=llm, text=text, model=model, instructions=instructions) - def copy(self): - """Create a deep copy of the Agent.""" - exclude = { - "id", - "_logger", - "_rpm_controller", - "_request_within_rpm_limit", - "_token_process", - "agent_executor", - "tools", - "tools_handler", - "cache_handler", - "llm", - } - - # Copy llm and clear callbacks - existing_llm = shallow_copy(self.llm) - existing_llm.callbacks = [] - copied_data = self.model_dump(exclude=exclude) - copied_data = {k: v for k, v in copied_data.items() if v is not None} - - copied_agent = Agent(**copied_data, llm=existing_llm, tools=self.tools) - - return copied_agent - def _parse_tools(self, tools: List[Any]) -> List[LangChainTool]: """Parse tools to be used for the task.""" tools_list = [] diff --git a/src/crewai/agents/agent_builder/base_agent.py b/src/crewai/agents/agent_builder/base_agent.py index 3d1206e01..a9549e5a8 100644 --- a/src/crewai/agents/agent_builder/base_agent.py +++ b/src/crewai/agents/agent_builder/base_agent.py @@ -1,6 +1,7 @@ import uuid from abc import ABC, abstractmethod -from typing import Any, Dict, List, Optional +from copy import copy as shallow_copy +from typing import Any, Dict, List, Optional, TypeVar from pydantic import ( UUID4, @@ -18,6 +19,8 @@ from crewai.agents import CacheHandler, ToolsHandler from crewai.agents.agent_builder.utilities.base_token_process import TokenProcess from crewai.utilities import I18N, Logger, RPMController +T = TypeVar("T", bound="BaseAgent") + class BaseAgent(ABC, BaseModel): """Abstract Base Class for all third party agents compatible with CrewAI. @@ -187,10 +190,30 @@ class BaseAgent(ABC, BaseModel): """Get the converter class for the agent to create json/pydantic outputs.""" pass - @abstractmethod - def copy(self): - """Create a copy of the agent.""" - pass + def copy(self: T) -> T: + """Create a deep copy of the Agent.""" + exclude = { + "id", + "_logger", + "_rpm_controller", + "_request_within_rpm_limit", + "_token_process", + "agent_executor", + "tools", + "tools_handler", + "cache_handler", + "llm", + } + + # Copy llm and clear callbacks + existing_llm = shallow_copy(self.llm) + existing_llm.callbacks = [] + copied_data = self.model_dump(exclude=exclude) + copied_data = {k: v for k, v in copied_data.items() if v is not None} + + copied_agent = type(self)(**copied_data, llm=existing_llm, tools=self.tools) + + return copied_agent def interpolate_inputs(self, inputs: Dict[str, Any]) -> None: """Interpolate inputs into the agent description and backstory."""