Spaces:
Sleeping
Sleeping
import uuid | |
from typing import Any, List, Optional | |
''' | |
from langchain.prompts.chat import ( | |
ChatPromptTemplate, | |
HumanMessagePromptTemplate, | |
SystemMessagePromptTemplate, | |
) | |
from langchain.schema import HumanMessage, SystemMessage | |
''' | |
from langchain_community.chat_models import ChatOpenAI | |
from langchain.agents.format_scratchpad import format_log_to_str | |
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." | |
) | |
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.", {} | |
) | |
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) | |
def __tools_names(tools) -> str: | |
return ", ".join([t.name for t in tools]) | |