ConversationEntityMemory Class — langchain Architecture

Architecture documentation for the ConversationEntityMemory class in entity.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  f6985512_c22e_0ae9_c629_acb233df5408["ConversationEntityMemory"]
  48fb025a_9570_1ddd_d527_ec72c1ae6a6f["BaseChatMemory"]
  f6985512_c22e_0ae9_c629_acb233df5408 -->|extends| 48fb025a_9570_1ddd_d527_ec72c1ae6a6f
  1ed59830_902e_f3ce_f9f6_76963a6fd122["entity.py"]
  f6985512_c22e_0ae9_c629_acb233df5408 -->|defined in| 1ed59830_902e_f3ce_f9f6_76963a6fd122
  c2008dca_85be_90b1_0522_48f351f41a7f["buffer()"]
  f6985512_c22e_0ae9_c629_acb233df5408 -->|method| c2008dca_85be_90b1_0522_48f351f41a7f
  f293cc5a_77f9_484e_3075_cfe3083449e8["memory_variables()"]
  f6985512_c22e_0ae9_c629_acb233df5408 -->|method| f293cc5a_77f9_484e_3075_cfe3083449e8
  b961d203_0a7e_862d_a270_668983457309["load_memory_variables()"]
  f6985512_c22e_0ae9_c629_acb233df5408 -->|method| b961d203_0a7e_862d_a270_668983457309
  f048cf72_7d11_089f_89e3_f73dbf79fd42["save_context()"]
  f6985512_c22e_0ae9_c629_acb233df5408 -->|method| f048cf72_7d11_089f_89e3_f73dbf79fd42
  5beabb13_7b04_2420_3161_861f277a73d4["clear()"]
  f6985512_c22e_0ae9_c629_acb233df5408 -->|method| 5beabb13_7b04_2420_3161_861f277a73d4

Relationship Graph

Source Code

libs/langchain/langchain_classic/memory/entity.py lines 465–611

class ConversationEntityMemory(BaseChatMemory):
    """Entity extractor & summarizer memory.

    Extracts named entities from the recent chat history and generates summaries.
    With a swappable entity store, persisting entities across conversations.
    Defaults to an in-memory entity store, and can be swapped out for a Redis,
    SQLite, or other entity store.
    """

    human_prefix: str = "Human"
    ai_prefix: str = "AI"
    llm: BaseLanguageModel
    entity_extraction_prompt: BasePromptTemplate = ENTITY_EXTRACTION_PROMPT
    entity_summarization_prompt: BasePromptTemplate = ENTITY_SUMMARIZATION_PROMPT

    # Cache of recently detected entity names, if any
    # It is updated when load_memory_variables is called:
    entity_cache: list[str] = []

    # Number of recent message pairs to consider when updating entities:
    k: int = 3

    chat_history_key: str = "history"

    # Store to manage entity-related data:
    entity_store: BaseEntityStore = Field(default_factory=InMemoryEntityStore)

    @property
    def buffer(self) -> list[BaseMessage]:
        """Access chat memory messages."""
        return self.chat_memory.messages

    @property
    def memory_variables(self) -> list[str]:
        """Will always return list of memory variables."""
        return ["entities", self.chat_history_key]

    def load_memory_variables(self, inputs: dict[str, Any]) -> dict[str, Any]:
        """Load memory variables.

        Returns chat history and all generated entities with summaries if available,
        and updates or clears the recent entity cache.

        New entity name can be found when calling this method, before the entity
        summaries are generated, so the entity cache values may be empty if no entity
        descriptions are generated yet.
        """
        # Create an LLMChain for predicting entity names from the recent chat history:
        chain = LLMChain(llm=self.llm, prompt=self.entity_extraction_prompt)

        if self.input_key is None:
            prompt_input_key = get_prompt_input_key(inputs, self.memory_variables)
        else:
            prompt_input_key = self.input_key

        # Extract an arbitrary window of the last message pairs from
        # the chat history, where the hyperparameter k is the
        # number of message pairs:
        buffer_string = get_buffer_string(
            self.buffer[-self.k * 2 :],
            human_prefix=self.human_prefix,
            ai_prefix=self.ai_prefix,
        )

        # Generates a comma-separated list of named entities,
        # e.g. "Jane, White House, UFO"
        # or "NONE" if no named entities are extracted:
        output = chain.predict(
            history=buffer_string,
            input=inputs[prompt_input_key],
        )

        # If no named entities are extracted, assigns an empty list.
        if output.strip() == "NONE":
            entities = []
        else:
            # Make a list of the extracted entities:
            entities = [w.strip() for w in output.split(",")]

        # Make a dictionary of entities with summary if exists:
        entity_summaries = {}