summary_buffer.py — langchain Source File

Source Code

from typing import Any

from langchain_core._api import deprecated
from langchain_core.messages import BaseMessage, get_buffer_string
from langchain_core.utils import pre_init
from typing_extensions import override

from langchain_classic.memory.chat_memory import BaseChatMemory
from langchain_classic.memory.summary import SummarizerMixin


@deprecated(
    since="0.3.1",
    removal="1.0.0",
    message=(
        "Please see the migration guide at: "
        "https://python.langchain.com/docs/versions/migrating_memory/"
    ),
)
class ConversationSummaryBufferMemory(BaseChatMemory, SummarizerMixin):
    """Buffer with summarizer for storing conversation memory.

    Provides a running summary of the conversation together with the most recent
    messages in the conversation under the constraint that the total number of
    tokens in the conversation does not exceed a certain limit.
    """

    max_token_limit: int = 2000
    moving_summary_buffer: str = ""
    memory_key: str = "history"

    @property
    def buffer(self) -> str | list[BaseMessage]:
        """String buffer of memory."""
        return self.load_memory_variables({})[self.memory_key]

    async def abuffer(self) -> str | list[BaseMessage]:
        """Async memory buffer."""
        memory_variables = await self.aload_memory_variables({})
        return memory_variables[self.memory_key]

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

    @override
    def load_memory_variables(self, inputs: dict[str, Any]) -> dict[str, Any]:
        """Return history buffer."""
        buffer = self.chat_memory.messages
        if self.moving_summary_buffer != "":
            first_messages: list[BaseMessage] = [
                self.summary_message_cls(content=self.moving_summary_buffer),
            ]
            buffer = first_messages + buffer
        if self.return_messages:
            final_buffer: Any = buffer
        else:
            final_buffer = get_buffer_string(
                buffer,
// ... (89 more lines)