Source code for langchain.memory.vectorstore_token_buffer_memory
"""
Class for a conversation memory buffer with older messages stored in a vectorstore .
This implements a conversation memory in which the messages are stored in a memory
buffer up to a specified token limit. When the limit is exceeded, older messages are
saved to a vectorstore backing database. The vectorstore can be made persistent across
sessions.
"""
import warnings
from datetime import datetime
from typing import Any, Dict, List
from langchain_core.messages import BaseMessage
from langchain_core.prompts.chat import SystemMessagePromptTemplate
from langchain_core.vectorstores import VectorStoreRetriever
from pydantic import Field, PrivateAttr
from langchain.memory import ConversationTokenBufferMemory, VectorStoreRetrieverMemory
from langchain.memory.chat_memory import BaseChatMemory
from langchain.text_splitter import RecursiveCharacterTextSplitter
DEFAULT_HISTORY_TEMPLATE = """
Current date and time: {current_time}.
Potentially relevant timestamped excerpts of previous conversations (you
do not need to use these if irrelevant):
{previous_history}
"""
TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S %Z"
[docs]
class ConversationVectorStoreTokenBufferMemory(ConversationTokenBufferMemory):
"""Conversation chat memory with token limit and vectordb backing.
load_memory_variables() will return a dict with the key "history".
It contains background information retrieved from the vector store
plus recent lines of the current conversation.
To help the LLM understand the part of the conversation stored in the
vectorstore, each interaction is timestamped and the current date and
time is also provided in the history. A side effect of this is that the
LLM will have access to the current date and time.
Initialization arguments:
This class accepts all the initialization arguments of
ConversationTokenBufferMemory, such as `llm`. In addition, it
accepts the following additional arguments
retriever: (required) A VectorStoreRetriever object to use
as the vector backing store
split_chunk_size: (optional, 1000) Token chunk split size
for long messages generated by the AI
previous_history_template: (optional) Template used to format
the contents of the prompt history
Example using ChromaDB:
.. code-block:: python
from langchain.memory.token_buffer_vectorstore_memory import (
ConversationVectorStoreTokenBufferMemory
)
from langchain_chroma import Chroma
from langchain_community.embeddings import HuggingFaceInstructEmbeddings
from langchain_openai import OpenAI
embedder = HuggingFaceInstructEmbeddings(
query_instruction="Represent the query for retrieval: "
)
chroma = Chroma(collection_name="demo",
embedding_function=embedder,
collection_metadata={"hnsw:space": "cosine"},
)
retriever = chroma.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={
'k': 5,
'score_threshold': 0.75,
},
)
conversation_memory = ConversationVectorStoreTokenBufferMemory(
return_messages=True,
llm=OpenAI(),
retriever=retriever,
max_token_limit = 1000,
)
conversation_memory.save_context({"Human": "Hi there"},
{"AI": "Nice to meet you!"}
)
conversation_memory.save_context({"Human": "Nice day isn't it?"},
{"AI": "I love Wednesdays."}
)
conversation_memory.load_memory_variables({"input": "What time is it?"})
"""
retriever: VectorStoreRetriever = Field(exclude=True)
memory_key: str = "history"
previous_history_template: str = DEFAULT_HISTORY_TEMPLATE
split_chunk_size: int = 1000
_memory_retriever: VectorStoreRetrieverMemory = PrivateAttr(default=None)
_timestamps: List[datetime] = PrivateAttr(default_factory=list)
@property
def memory_retriever(self) -> VectorStoreRetrieverMemory:
"""Return a memory retriever from the passed retriever object."""
if self._memory_retriever is not None:
return self._memory_retriever
self._memory_retriever = VectorStoreRetrieverMemory(retriever=self.retriever)
return self._memory_retriever
[docs]
def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
"""Return history and memory buffer."""
try:
with warnings.catch_warnings():
warnings.simplefilter("ignore")
memory_variables = self.memory_retriever.load_memory_variables(inputs)
previous_history = memory_variables[self.memory_retriever.memory_key]
except AssertionError: # happens when db is empty
previous_history = ""
current_history = super().load_memory_variables(inputs)
template = SystemMessagePromptTemplate.from_template(
self.previous_history_template
)
messages = [
template.format(
previous_history=previous_history,
current_time=datetime.now().astimezone().strftime(TIMESTAMP_FORMAT),
)
]
messages.extend(current_history[self.memory_key])
return {self.memory_key: messages}
[docs]
def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
"""Save context from this conversation to buffer. Pruned."""
BaseChatMemory.save_context(self, inputs, outputs)
self._timestamps.append(datetime.now().astimezone())
# Prune buffer if it exceeds max token limit
buffer = self.chat_memory.messages
curr_buffer_length = self.llm.get_num_tokens_from_messages(buffer)
if curr_buffer_length > self.max_token_limit:
while curr_buffer_length > self.max_token_limit:
self._pop_and_store_interaction(buffer)
curr_buffer_length = self.llm.get_num_tokens_from_messages(buffer)
[docs]
def save_remainder(self) -> None:
"""
Save the remainder of the conversation buffer to the vector store.
This is useful if you have made the vectorstore persistent, in which
case this can be called before the end of the session to store the
remainder of the conversation.
"""
buffer = self.chat_memory.messages
while len(buffer) > 0:
self._pop_and_store_interaction(buffer)
def _pop_and_store_interaction(self, buffer: List[BaseMessage]) -> None:
input = buffer.pop(0)
output = buffer.pop(0)
timestamp = self._timestamps.pop(0).strftime(TIMESTAMP_FORMAT)
# Split AI output into smaller chunks to avoid creating documents
# that will overflow the context window
ai_chunks = self._split_long_ai_text(str(output.content))
for index, chunk in enumerate(ai_chunks):
self.memory_retriever.save_context(
{"Human": f"<{timestamp}/00> {str(input.content)}"},
{"AI": f"<{timestamp}/{index:02}> {chunk}"},
)
def _split_long_ai_text(self, text: str) -> List[str]:
splitter = RecursiveCharacterTextSplitter(chunk_size=self.split_chunk_size)
return [chunk.page_content for chunk in splitter.create_documents([text])]
