diff --git a/swarms/agents/agent.py b/swarms/agents/agent.py new file mode 100644 index 00000000..ba0b26fa --- /dev/null +++ b/swarms/agents/agent.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +from typing import List, Optional + +from langchain.chains.llm import LLMChain + +from swarms.agents.memory.base import VectorStoreRetriever +from swarms.agents.memory.base_memory import BaseChatMessageHistory, ChatMessageHistory +from swarms.agents.memory.document import Document +from swarms.agents.models.base import AbstractModel +from swarms.agents.models.prompts.agent_prompt_auto import ( + MessageFormatter, + PromptConstructor, +) +from swarms.agents.models.prompts.agent_prompt_generator import FINISH_NAME +from swarms.agents.models.prompts.base import ( + AIMessage, + HumanMessage, + SystemMessage, +) +from swarms.agents.tools.base import BaseTool +from swarms.agents.utils.Agent import AgentOutputParser +from swarms.agents.utils.human_input import HumanInputRun + + +class Agent: + """Base Agent class""" + def __init__( + self, + ai_name: str, + chain: LLMChain, + memory: VectorStoreRetriever, + output_parser: AgentOutputParser, + tools: List[BaseTool], + feedback_tool: Optional[HumanInputRun] = None, + chat_history_memory: Optional[BaseChatMessageHistory] = None, + ): + self.ai_name = ai_name + self.chain = chain + self.memory = memory + self.next_action_count = 0 + self.output_parser = output_parser + self.tools = tools + self.feedback_tool = feedback_tool + self.chat_history_memory = chat_history_memory or ChatMessageHistory() + + @classmethod + def from_llm_and_tools( + cls, + ai_name: str, + ai_role: str, + memory: VectorStoreRetriever, + tools: List[BaseTool], + llm: AbstractModel, + human_in_the_loop: bool = False, + output_parser: Optional[AgentOutputParser] = None, + chat_history_memory: Optional[BaseChatMessageHistory] = None, + ) -> Agent: + prompt_constructor = PromptConstructor(ai_name=ai_name, + ai_role=ai_role, + tools=tools) + message_formatter = MessageFormatter() + human_feedback_tool = HumanInputRun() if human_in_the_loop else None + chain = LLMChain(llm=llm, prompt_constructor=prompt_constructor, message_formatter=message_formatter) + return cls( + ai_name, + memory, + chain, + output_parser or AgentOutputParser(), + tools, + feedback_tool=human_feedback_tool, + chat_history_memory=chat_history_memory, + ) + + def run(self, goals: List[str]) -> str: + user_input = ( + "Determine which next command to use, and respond using the format specified above:" + ) + loop_count = 0 + while True: + loop_count += 1 + + # Send message to AI, get response + assistant_reply = self.chain.run( + goals=goals, + messages=self.chat_history_memory.messages, + memory=self.memory, + user_input=user_input, + ) + + print(assistant_reply) + self.chat_history_memory.add_message(HumanMessage(content=user_input)) + self.chat_history_memory.add_message(AIMessage(content=assistant_reply)) + + # Get command name and arguments + action = self.output_parser.parse(assistant_reply) + tools = {t.name: t for t in self.tools} + if action.name == FINISH_NAME: + return action.args["response"] + if action.name in tools: + tool = tools[action.name] + try: + observation = tool.run(action.args) + except Exception as error: + observation = ( + f"Validation Error in args: {str(error)}, args: {action.args}" + ) + except Exception as e: + observation = ( + f"Error: {str(e)}, {type(e).__name__}, args: {action.args}" + ) + result = f"Command {tool.name} returned: {observation}" + elif action.name == "ERROR": + result = f"Error: {action.args}. " + else: + result = ( + f"""Unknown command '{action.name}'. + Please refer to the 'COMMANDS' list for available + commands and only respond in the specified JSON format.""" + ) + memory_to_add = ( + f"Assistant Reply: {assistant_reply} " f"\nResult: {result} " + ) + if self.feedback_tool is not None: + feedback = f"\n{self.feedback_tool.run('Input: ')}" + if feedback in {"q", "stop"}: + print("EXITING") + return "EXITING" + memory_to_add += feedback + + self.memory.add_documents([Document(page_content=memory_to_add)]) + self.chat_history_memory.add_message(SystemMessage(content=result)) + + diff --git a/swarms/agents/base.py b/swarms/agents/base.py index ba0b26fa..12c21cbc 100644 --- a/swarms/agents/base.py +++ b/swarms/agents/base.py @@ -1,134 +1,25 @@ -from __future__ import annotations +from abc import ABC, abstractmethod -from typing import List, Optional - -from langchain.chains.llm import LLMChain - -from swarms.agents.memory.base import VectorStoreRetriever -from swarms.agents.memory.base_memory import BaseChatMessageHistory, ChatMessageHistory -from swarms.agents.memory.document import Document -from swarms.agents.models.base import AbstractModel -from swarms.agents.models.prompts.agent_prompt_auto import ( - MessageFormatter, - PromptConstructor, -) -from swarms.agents.models.prompts.agent_prompt_generator import FINISH_NAME -from swarms.agents.models.prompts.base import ( - AIMessage, - HumanMessage, - SystemMessage, -) -from swarms.agents.tools.base import BaseTool -from swarms.agents.utils.Agent import AgentOutputParser -from swarms.agents.utils.human_input import HumanInputRun - - -class Agent: - """Base Agent class""" +class AbstractAgent(ABC): + #absrtact agent class + + @classmethod def __init__( - self, - ai_name: str, - chain: LLMChain, - memory: VectorStoreRetriever, - output_parser: AgentOutputParser, - tools: List[BaseTool], - feedback_tool: Optional[HumanInputRun] = None, - chat_history_memory: Optional[BaseChatMessageHistory] = None, + self, + ai_name: str = None, + ai_role: str = None, + memory = None, + tools = None, + llm = None, + human_in_the_loop=None, + output_parser = None, + chat_history_memory=None, + *args, + **kwargs ): - self.ai_name = ai_name - self.chain = chain - self.memory = memory - self.next_action_count = 0 - self.output_parser = output_parser - self.tools = tools - self.feedback_tool = feedback_tool - self.chat_history_memory = chat_history_memory or ChatMessageHistory() - - @classmethod - def from_llm_and_tools( - cls, - ai_name: str, - ai_role: str, - memory: VectorStoreRetriever, - tools: List[BaseTool], - llm: AbstractModel, - human_in_the_loop: bool = False, - output_parser: Optional[AgentOutputParser] = None, - chat_history_memory: Optional[BaseChatMessageHistory] = None, - ) -> Agent: - prompt_constructor = PromptConstructor(ai_name=ai_name, - ai_role=ai_role, - tools=tools) - message_formatter = MessageFormatter() - human_feedback_tool = HumanInputRun() if human_in_the_loop else None - chain = LLMChain(llm=llm, prompt_constructor=prompt_constructor, message_formatter=message_formatter) - return cls( - ai_name, - memory, - chain, - output_parser or AgentOutputParser(), - tools, - feedback_tool=human_feedback_tool, - chat_history_memory=chat_history_memory, - ) - - def run(self, goals: List[str]) -> str: - user_input = ( - "Determine which next command to use, and respond using the format specified above:" - ) - loop_count = 0 - while True: - loop_count += 1 - - # Send message to AI, get response - assistant_reply = self.chain.run( - goals=goals, - messages=self.chat_history_memory.messages, - memory=self.memory, - user_input=user_input, - ) - - print(assistant_reply) - self.chat_history_memory.add_message(HumanMessage(content=user_input)) - self.chat_history_memory.add_message(AIMessage(content=assistant_reply)) - - # Get command name and arguments - action = self.output_parser.parse(assistant_reply) - tools = {t.name: t for t in self.tools} - if action.name == FINISH_NAME: - return action.args["response"] - if action.name in tools: - tool = tools[action.name] - try: - observation = tool.run(action.args) - except Exception as error: - observation = ( - f"Validation Error in args: {str(error)}, args: {action.args}" - ) - except Exception as e: - observation = ( - f"Error: {str(e)}, {type(e).__name__}, args: {action.args}" - ) - result = f"Command {tool.name} returned: {observation}" - elif action.name == "ERROR": - result = f"Error: {action.args}. " - else: - result = ( - f"""Unknown command '{action.name}'. - Please refer to the 'COMMANDS' list for available - commands and only respond in the specified JSON format.""" - ) - memory_to_add = ( - f"Assistant Reply: {assistant_reply} " f"\nResult: {result} " - ) - if self.feedback_tool is not None: - feedback = f"\n{self.feedback_tool.run('Input: ')}" - if feedback in {"q", "stop"}: - print("EXITING") - return "EXITING" - memory_to_add += feedback - - self.memory.add_documents([Document(page_content=memory_to_add)]) - self.chat_history_memory.add_message(SystemMessage(content=result)) + pass + @abstractmethod + def run(self, goals=None): + pass