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_openai import ChatOpenAI | |
| # from langchain_google_genai import ChatGoogleGenerativeAI | |
| 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]) | |