Source code for langchain.schema.memory

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Any, Dict, List

from langchain.load.serializable import Serializable
from langchain.schema.messages import AIMessage, BaseMessage, HumanMessage


[docs]class BaseMemory(Serializable, ABC): """Abstract base class for memory in Chains. Memory refers to state in Chains. Memory can be used to store information about past executions of a Chain and inject that information into the inputs of future executions of the Chain. For example, for conversational Chains Memory can be used to store conversations and automatically add them to future model prompts so that the model has the necessary context to respond coherently to the latest input. Example: .. code-block:: python class SimpleMemory(BaseMemory): memories: Dict[str, Any] = dict() @property def memory_variables(self) -> List[str]: return list(self.memories.keys()) def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, str]: return self.memories def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None: pass def clear(self) -> None: pass """ # noqa: E501
[docs] class Config: """Configuration for this pydantic object.""" arbitrary_types_allowed = True
@property @abstractmethod def memory_variables(self) -> List[str]: """The string keys this memory class will add to chain inputs."""
[docs] @abstractmethod def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]: """Return key-value pairs given the text input to the chain."""
[docs] @abstractmethod def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None: """Save the context of this chain run to memory."""
[docs] @abstractmethod def clear(self) -> None: """Clear memory contents."""
[docs]class BaseChatMessageHistory(ABC): """Abstract base class for storing chat message history. See `ChatMessageHistory` for default implementation. Example: .. code-block:: python class FileChatMessageHistory(BaseChatMessageHistory): storage_path: str session_id: str @property def messages(self): with open(os.path.join(storage_path, session_id), 'r:utf-8') as f: messages = json.loads(f.read()) return messages_from_dict(messages) def add_message(self, message: BaseMessage) -> None: messages = self.messages.append(_message_to_dict(message)) with open(os.path.join(storage_path, session_id), 'w') as f: json.dump(f, messages) def clear(self): with open(os.path.join(storage_path, session_id), 'w') as f: f.write("[]") """ messages: List[BaseMessage] """A list of Messages stored in-memory."""
[docs] def add_user_message(self, message: str) -> None: """Convenience method for adding a human message string to the store. Args: message: The string contents of a human message. """ self.add_message(HumanMessage(content=message))
[docs] def add_ai_message(self, message: str) -> None: """Convenience method for adding an AI message string to the store. Args: message: The string contents of an AI message. """ self.add_message(AIMessage(content=message))
# TODO: Make this an abstractmethod.
[docs] def add_message(self, message: BaseMessage) -> None: """Add a Message object to the store. Args: message: A BaseMessage object to store. """ raise NotImplementedError
[docs] @abstractmethod def clear(self) -> None: """Remove all messages from the store"""