Source code for langchain_core.memory
"""**Memory** maintains Chain state, incorporating context from past runs.
**Class hierarchy for Memory:**
.. code-block::
BaseMemory --> <name>Memory --> <name>Memory # Examples: BaseChatMemory -> MotorheadMemory
""" # noqa: E501
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any, Dict, List
from langchain_core.load.serializable import Serializable
from langchain_core.runnables import run_in_executor
[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
class Config:
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.
Args:
inputs: The inputs to the chain.
Returns:
A dictionary of key-value pairs.
"""
[docs] async def aload_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
"""Async return key-value pairs given the text input to the chain.
Args:
inputs: The inputs to the chain.
Returns:
A dictionary of key-value pairs.
"""
return await run_in_executor(None, self.load_memory_variables, inputs)
[docs] @abstractmethod
def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
"""Save the context of this chain run to memory.
Args:
inputs: The inputs to the chain.
outputs: The outputs of the chain.
"""
[docs] async def asave_context(
self, inputs: Dict[str, Any], outputs: Dict[str, str]
) -> None:
"""Async save the context of this chain run to memory.
Args:
inputs: The inputs to the chain.
outputs: The outputs of the chain.
"""
await run_in_executor(None, self.save_context, inputs, outputs)
[docs] @abstractmethod
def clear(self) -> None:
"""Clear memory contents."""
[docs] async def aclear(self) -> None:
"""Async clear memory contents."""
await run_in_executor(None, self.clear)