Moving copy functionality from Agent to BaseAgent

This commit is contained in:
Brandon Hancock
2024-07-01 16:06:02 -04:00
parent 1f9166f61b
commit f36f73e035
2 changed files with 28 additions and 30 deletions

View File

@@ -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 = []

View File

@@ -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."""