From ea3c96c2020c7bd4d099418b35aa151ef87446aa Mon Sep 17 00:00:00 2001 From: Kye Date: Thu, 27 Jul 2023 14:34:10 -0400 Subject: [PATCH] clean up --- swarms/workers/generative_worker.py | 248 ---------------------------- 1 file changed, 248 deletions(-) diff --git a/swarms/workers/generative_worker.py b/swarms/workers/generative_worker.py index af1d9bd1..45f4d7ad 100644 --- a/swarms/workers/generative_worker.py +++ b/swarms/workers/generative_worker.py @@ -21,8 +21,6 @@ logger = logging.getLogger(__name__) from pydantic import BaseModel, Field - - class WorkerSims(BaseMemory): llm: BaseLanguageModel """The core language model.""" @@ -307,249 +305,3 @@ class WorkerSims(BaseMemory): def clear(self) -> None: """Clear memory contents.""" # TODO - - - -####################### MAIN CLASS - - -class WorkerSimsAgent(BaseModel): - """A character with memory and innate characteristics.""" - - name: str - """The character's name.""" - - age: Optional[int] = None - """The optional age of the character.""" - traits: str = "N/A" - """Permanent traits to ascribe to the character.""" - status: str - """The traits of the character you wish not to change.""" - memory: WorkerSims - """The memory object that combines relevance, recency, and 'importance'.""" - llm: BaseLanguageModel - """The underlying language model.""" - verbose: bool = False - summary: str = "" #: :meta private: - """Stateful self-summary generated via reflection on the character's memory.""" - - summary_refresh_seconds: int = 3600 #: :meta private: - """How frequently to re-generate the summary.""" - - last_refreshed: datetime = Field(default_factory=datetime.now) # : :meta private: - """The last time the character's summary was regenerated.""" - - daily_summaries: List[str] = Field(default_factory=list) # : :meta private: - """Summary of the events in the plan that the agent took.""" - - class Config: - """Configuration for this pydantic object.""" - - arbitrary_types_allowed = True - - # LLM-related methods - @staticmethod - def _parse_list(text: str) -> List[str]: - """Parse a newline-separated string into a list of strings.""" - lines = re.split(r"\n", text.strip()) - return [re.sub(r"^\s*\d+\.\s*", "", line).strip() for line in lines] - - def chain(self, prompt: PromptTemplate) -> LLMChain: - return LLMChain( - llm=self.llm, prompt=prompt, verbose=self.verbose, memory=self.memory - ) - - def _get_entity_from_observation(self, observation: str) -> str: - prompt = PromptTemplate.from_template( - "What is the observed entity in the following observation? {observation}" - + "\nEntity=" - ) - return self.chain(prompt).run(observation=observation).strip() - - def _get_entity_action(self, observation: str, entity_name: str) -> str: - prompt = PromptTemplate.from_template( - "What is the {entity} doing in the following observation? {observation}" - + "\nThe {entity} is" - ) - return ( - self.chain(prompt).run(entity=entity_name, observation=observation).strip() - ) - - def summarize_related_memories(self, observation: str) -> str: - """Summarize memories that are most relevant to an observation.""" - prompt = PromptTemplate.from_template( - """ -{q1}? -Context from memory: -{relevant_memories} -Relevant context: -""" - ) - entity_name = self._get_entity_from_observation(observation) - entity_action = self._get_entity_action(observation, entity_name) - q1 = f"What is the relationship between {self.name} and {entity_name}" - q2 = f"{entity_name} is {entity_action}" - return self.chain(prompt=prompt).run(q1=q1, queries=[q1, q2]).strip() - - def _generate_reaction( - self, observation: str, suffix: str, now: Optional[datetime] = None - ) -> str: - """React to a given observation or dialogue act.""" - prompt = PromptTemplate.from_template( - "{agent_summary_description}" - + "\nIt is {current_time}." - + "\n{agent_name}'s status: {agent_status}" - + "\nSummary of relevant context from {agent_name}'s memory:" - + "\n{relevant_memories}" - + "\nMost recent observations: {most_recent_memories}" - + "\nObservation: {observation}" - + "\n\n" - + suffix - ) - agent_summary_description = self.get_summary(now=now) - relevant_memories_str = self.summarize_related_memories(observation) - current_time_str = ( - datetime.now().strftime("%B %d, %Y, %I:%M %p") - if now is None - else now.strftime("%B %d, %Y, %I:%M %p") - ) - kwargs: Dict[str, Any] = dict( - agent_summary_description=agent_summary_description, - current_time=current_time_str, - relevant_memories=relevant_memories_str, - agent_name=self.name, - observation=observation, - agent_status=self.status, - ) - consumed_tokens = self.llm.get_num_tokens( - prompt.format(most_recent_memories="", **kwargs) - ) - kwargs[self.memory.most_recent_memories_token_key] = consumed_tokens - return self.chain(prompt=prompt).run(**kwargs).strip() - - def _clean_response(self, text: str) -> str: - return re.sub(f"^{self.name} ", "", text.strip()).strip() - - def generate_reaction( - self, observation: str, now: Optional[datetime] = None - ) -> Tuple[bool, str]: - """React to a given observation.""" - call_to_action_template = ( - "Should {agent_name} react to the observation, and if so," - + " what would be an appropriate reaction? Respond in one line." - + ' If the action is to engage in dialogue, write:\nSAY: "what to say"' - + "\notherwise, write:\nREACT: {agent_name}'s reaction (if anything)." - + "\nEither do nothing, react, or say something but not both.\n\n" - ) - full_result = self._generate_reaction( - observation, call_to_action_template, now=now - ) - result = full_result.strip().split("\n")[0] - # AAA - self.memory.save_context( - {}, - { - self.memory.add_memory_key: f"{self.name} observed " - f"{observation} and reacted by {result}", - self.memory.now_key: now, - }, - ) - if "REACT:" in result: - reaction = self._clean_response(result.split("REACT:")[-1]) - return False, f"{self.name} {reaction}" - if "SAY:" in result: - said_value = self._clean_response(result.split("SAY:")[-1]) - return True, f"{self.name} said {said_value}" - else: - return False, result - - def generate_dialogue_response( - self, observation: str, now: Optional[datetime] = None - ) -> Tuple[bool, str]: - """React to a given observation.""" - call_to_action_template = ( - "What would {agent_name} say? To end the conversation, write:" - ' GOODBYE: "what to say". Otherwise to continue the conversation,' - ' write: SAY: "what to say next"\n\n' - ) - full_result = self._generate_reaction( - observation, call_to_action_template, now=now - ) - result = full_result.strip().split("\n")[0] - if "GOODBYE:" in result: - farewell = self._clean_response(result.split("GOODBYE:")[-1]) - self.memory.save_context( - {}, - { - self.memory.add_memory_key: f"{self.name} observed " - f"{observation} and said {farewell}", - self.memory.now_key: now, - }, - ) - return False, f"{self.name} said {farewell}" - if "SAY:" in result: - response_text = self._clean_response(result.split("SAY:")[-1]) - self.memory.save_context( - {}, - { - self.memory.add_memory_key: f"{self.name} observed " - f"{observation} and said {response_text}", - self.memory.now_key: now, - }, - ) - return True, f"{self.name} said {response_text}" - else: - return False, result - - ###################################################### - # Agent stateful' summary methods. # - # Each dialog or response prompt includes a header # - # summarizing the agent's self-description. This is # - # updated periodically through probing its memories # - ###################################################### - def _compute_agent_summary(self) -> str: - """""" - prompt = PromptTemplate.from_template( - "How would you summarize {name}'s core characteristics given the" - + " following statements:\n" - + "{relevant_memories}" - + "Do not embellish." - + "\n\nSummary: " - ) - # The agent seeks to think about their core characteristics. - return ( - self.chain(prompt) - .run(name=self.name, queries=[f"{self.name}'s core characteristics"]) - .strip() - ) - - def get_summary( - self, force_refresh: bool = False, now: Optional[datetime] = None - ) -> str: - """Return a descriptive summary of the agent.""" - current_time = datetime.now() if now is None else now - since_refresh = (current_time - self.last_refreshed).seconds - if ( - not self.summary - or since_refresh >= self.summary_refresh_seconds - or force_refresh - ): - self.summary = self._compute_agent_summary() - self.last_refreshed = current_time - age = self.age if self.age is not None else "N/A" - return ( - f"Name: {self.name} (age: {age})" - + f"\nInnate traits: {self.traits}" - + f"\n{self.summary}" - ) - - def get_full_header( - self, force_refresh: bool = False, now: Optional[datetime] = None - ) -> str: - """Return a full header of the agent's status, summary, and current time.""" - now = datetime.now() if now is None else now - summary = self.get_summary(force_refresh=force_refresh, now=now) - current_time_str = now.strftime("%B %d, %Y, %I:%M %p") - return ( - f"{summary}\nIt is {current_time_str}.\n{self.name}'s status: {self.status}" - ) \ No newline at end of file