Move Generative Agent definition to Experimental (langchain-ai#3245)

Extending @BeautyyuYanli 's langchain-ai#3220 to move from the notebook

---------

Co-authored-by: BeautyyuYanli <[email protected]>

docs/reference.rst

@@ -14,3 +14,4 @@ Full documentation on all methods, classes, and APIs in LangChain.
    ./reference/modules/chains.rst
    ./reference/agents.rst
    ./reference/modules/utilities.rst
+   ./reference/modules/experimental.rst

docs/reference/modules/experimental.rst

@@ -0,0 +1,28 @@
+==========
+Experimental Modules
+==========
+
+This module contains experimental modules and reproductions of existing work using LangChain primitives.
+
+Autonomous Agents
+------------------
+
+Here, we document the BabyAGI and AutoGPT classes from the langchain.experimental module.
+
+.. autoclass:: langchain.experimental.BabyAGI
+   :members:
+
+.. autoclass:: langchain.experimental.AutoGPT
+   :members:
+
+
+Generative Agents
+------------------
+
+Here, we document the GenerativeAgent and GenerativeAgentMemory classes from the langchain.experimental module.
+
+.. autoclass:: langchain.experimental.GenerativeAgent
+   :members:
+
+.. autoclass:: langchain.experimental.GenerativeAgentMemory
+   :members:

langchain/experimental/__init__.py

@@ -1,4 +1,6 @@
 from langchain.experimental.autonomous_agents.autogpt.agent import AutoGPT
 from langchain.experimental.autonomous_agents.baby_agi.baby_agi import BabyAGI
+from langchain.experimental.generative_agents.generative_agent import GenerativeAgent
+from langchain.experimental.generative_agents.memory import GenerativeAgentMemory
 
-__all__ = ["BabyAGI", "AutoGPT"]
+__all__ = ["BabyAGI", "AutoGPT", "GenerativeAgent", "GenerativeAgentMemory"]

langchain/experimental/generative_agents/__init__.py

@@ -0,0 +1,5 @@
+"""Generative Agents primitives."""
+from langchain.experimental.generative_agents.generative_agent import GenerativeAgent
+from langchain.experimental.generative_agents.memory import GenerativeAgentMemory
+
+__all__ = ["GenerativeAgent", "GenerativeAgentMemory"]

langchain/experimental/generative_agents/generative_agent.py

@@ -0,0 +1,230 @@
+import re
+from datetime import datetime
+from typing import Any, Dict, List, Optional, Tuple
+
+from pydantic import BaseModel, Field
+
+from langchain import LLMChain
+from langchain.experimental.generative_agents.memory import GenerativeAgentMemory
+from langchain.prompts import PromptTemplate
+from langchain.schema import BaseLanguageModel
+
+
+class GenerativeAgent(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: GenerativeAgentMemory
+    """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) -> 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()
+        relevant_memories_str = self.summarize_related_memories(observation)
+        current_time_str = datetime.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) -> 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)
+        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}"
+            },
+        )
+        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) -> 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)
+        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}"
+                },
+            )
+            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}"
+                },
+            )
+            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) -> str:
+        """Return a descriptive summary of the agent."""
+        current_time = datetime.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) -> str:
+        """Return a full header of the agent's status, summary, and current time."""
+        summary = self.get_summary(force_refresh=force_refresh)
+        current_time_str = datetime.now().strftime("%B %d, %Y, %I:%M %p")
+        return (
+            f"{summary}\nIt is {current_time_str}.\n{self.name}'s status: {self.status}"
+        )