Langchain instance

AuConversationSummaryBufferMemory

Bases: ConversationSummaryBufferMemory, SummarizerMixin

Long term memory to store conversation memory.

If memories exceeds the max tokens limit, memories will be compressed by the AuConversationSummaryBufferMemory.predict_new_summary method.

Attributes:

Name	Type	Description
messages	List[BaseMessage]	List of BaseMessage to save.

Source code in agentuniverse/agent/memory/langchain_instance.py

Python

class AuConversationSummaryBufferMemory(ConversationSummaryBufferMemory, SummarizerMixin):
    """Long term memory to store conversation memory.

    If memories exceeds the max tokens limit,
    memories will be compressed by the AuConversationSummaryBufferMemory.predict_new_summary method.

    Attributes:
        messages (List[BaseMessage]): List of BaseMessage to save.
    """

    messages: Optional[List[Message]] = None
    prompt_version: Optional[str] = None

    def __init__(self, **kwargs):
        """The __init__ method.

        The initialization method saves the memory context from messages in `messages` attribute to the memory buffer.

        Args:
            **kwargs: Arbitrary keyword arguments.
        """

        super().__init__(**kwargs)
        self.build_memory()

    @property
    def load_memory(self) -> List[BaseMessage]:
        """ General method: load the memory context from the memory buffer."""
        buffer = self.buffer
        if self.moving_summary_buffer != "":
            moving_messages: List[BaseMessage] = [
                self.summary_message_cls(content=self.moving_summary_buffer)
            ]
            buffer = moving_messages + buffer
        return buffer

    @property
    def load_memory_str(self) -> str:
        """ General method: load the memory context from the memory buffer as string format."""
        buffer = get_buffer_string(self.buffer)
        if self.moving_summary_buffer is not None:
            buffer = self.moving_summary_buffer + buffer
        return buffer

    def build_memory(self):
        """Save the memory context from messages in `messages` attribute to the memory buffer.

        Note:
            If conversation messages contains the system message, it will be saved in the `moving_summary_buffer` attribute.
        """
        if self.messages is None:
            return
        # save system messages
        for message in self.messages:
            if message.type.lower() == ChatMessageEnum.SYSTEM.value:
                self.moving_summary_buffer += message.content
                self.messages.remove(message)
        # generate pairs of Human and AI conversation messages and save to the memory buffer
        for i in range(0, len(self.messages), 2):
            inputs, outputs = self.generate_chat_messages(self.messages[i], self.messages[i + 1])
            self.save_context(inputs, outputs)

    def generate_chat_messages(self, *pairs: Message):
        """generate pairs of Human and AI conversation messages"""
        human_dict = {}
        ai_dict = {}
        for pair in pairs:
            if pair.type.lower() == ChatMessageEnum.HUMAN.value:
                human_dict = {self.input_key: pair.content}
            elif pair.type.lower() == ChatMessageEnum.AI.value:
                ai_dict = {self.output_key: pair.content}

        return human_dict, ai_dict

    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
        """Save context from the conversation to buffer and prune memories"""

        def message_to_dict(message):
            return {
                "content": message.content,
                "type": message.type
            }

        super().save_context(inputs, outputs)
        message_list = self.load_memory
        messages_dicts = [message_to_dict(message) for message in message_list]
        inputs[self.memory_key] = messages_dicts

    def predict_new_summary(
            self, messages: List[BaseMessage], existing_summary: str
    ) -> str:
        """Predict new summary, summarize memories in multiple rounds of conversations."""
        new_lines = get_buffer_string(
            messages,
            human_prefix=self.human_prefix,
            ai_prefix=self.ai_prefix,
        )
        prompt_version = self.prompt_version if self.prompt_version else 'chat_memory.summarizer_cn'
        prompt: Prompt = PromptManager().get_instance_obj(prompt_version)
        chain = LLMChain(llm=self.llm, prompt=prompt.as_langchain())
        return chain.predict(summary=existing_summary, new_lines=new_lines)

load_memory: List[BaseMessage] property

General method: load the memory context from the memory buffer.

load_memory_str: str property

General method: load the memory context from the memory buffer as string format.

__init__(**kwargs)

The init method.

The initialization method saves the memory context from messages in messages attribute to the memory buffer.

Parameters:

Name	Type	Description	Default
**kwargs		Arbitrary keyword arguments.	{}

Source code in agentuniverse/agent/memory/langchain_instance.py

Python

def __init__(self, **kwargs):
    """The __init__ method.

    The initialization method saves the memory context from messages in `messages` attribute to the memory buffer.

    Args:
        **kwargs: Arbitrary keyword arguments.
    """

    super().__init__(**kwargs)
    self.build_memory()

build_memory()

Save the memory context from messages in messages attribute to the memory buffer.

Note

If conversation messages contains the system message, it will be saved in the moving_summary_buffer attribute.

Source code in agentuniverse/agent/memory/langchain_instance.py

Python

def build_memory(self):
    """Save the memory context from messages in `messages` attribute to the memory buffer.

    Note:
        If conversation messages contains the system message, it will be saved in the `moving_summary_buffer` attribute.
    """
    if self.messages is None:
        return
    # save system messages
    for message in self.messages:
        if message.type.lower() == ChatMessageEnum.SYSTEM.value:
            self.moving_summary_buffer += message.content
            self.messages.remove(message)
    # generate pairs of Human and AI conversation messages and save to the memory buffer
    for i in range(0, len(self.messages), 2):
        inputs, outputs = self.generate_chat_messages(self.messages[i], self.messages[i + 1])
        self.save_context(inputs, outputs)

generate_chat_messages(*pairs)

generate pairs of Human and AI conversation messages

Source code in agentuniverse/agent/memory/langchain_instance.py

Python

def generate_chat_messages(self, *pairs: Message):
    """generate pairs of Human and AI conversation messages"""
    human_dict = {}
    ai_dict = {}
    for pair in pairs:
        if pair.type.lower() == ChatMessageEnum.HUMAN.value:
            human_dict = {self.input_key: pair.content}
        elif pair.type.lower() == ChatMessageEnum.AI.value:
            ai_dict = {self.output_key: pair.content}

    return human_dict, ai_dict

predict_new_summary(messages, existing_summary)

Predict new summary, summarize memories in multiple rounds of conversations.

Source code in agentuniverse/agent/memory/langchain_instance.py

Python

def predict_new_summary(
        self, messages: List[BaseMessage], existing_summary: str
) -> str:
    """Predict new summary, summarize memories in multiple rounds of conversations."""
    new_lines = get_buffer_string(
        messages,
        human_prefix=self.human_prefix,
        ai_prefix=self.ai_prefix,
    )
    prompt_version = self.prompt_version if self.prompt_version else 'chat_memory.summarizer_cn'
    prompt: Prompt = PromptManager().get_instance_obj(prompt_version)
    chain = LLMChain(llm=self.llm, prompt=prompt.as_langchain())
    return chain.predict(summary=existing_summary, new_lines=new_lines)

save_context(inputs, outputs)

Save context from the conversation to buffer and prune memories

Source code in agentuniverse/agent/memory/langchain_instance.py

Python

def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
    """Save context from the conversation to buffer and prune memories"""

    def message_to_dict(message):
        return {
            "content": message.content,
            "type": message.type
        }

    super().save_context(inputs, outputs)
    message_list = self.load_memory
    messages_dicts = [message_to_dict(message) for message in message_list]
    inputs[self.memory_key] = messages_dicts

AuConversationTokenBufferMemory

Bases: ConversationTokenBufferMemory

Short term memory to store conversation memory.

If memories exceeds the max tokens limit, the requirement is met by truncating part of memories

Attributes:

Name	Type	Description
messages	List[BaseMessage]	List of BaseMessage to save.

Source code in agentuniverse/agent/memory/langchain_instance.py

Python

class AuConversationTokenBufferMemory(ConversationTokenBufferMemory):
    """Short term memory to store conversation memory.

    If memories exceeds the max tokens limit, the requirement is met by truncating part of memories

    Attributes:
        messages (List[BaseMessage]): List of BaseMessage to save.
    """

    messages: Optional[List[Message]] = None

    def __init__(self, **kwargs):
        """The __init__ method.

        The initialization method saves the memory context from messages in `messages` attribute to the memory buffer.

        Args:
            **kwargs: Arbitrary keyword arguments.
        """
        super().__init__(**kwargs)
        self.build_memory()

    @property
    def load_memory(self) -> List[BaseMessage]:
        """ General method: load the memory context from the memory buffer."""
        return self.chat_memory.messages

    @property
    def load_memory_str(self) -> str:
        """ General method: load the memory context from the memory buffer as string format."""
        return get_buffer_string(self.chat_memory.messages)

    def build_memory(self):
        """Save the memory context from messages in `messages` attribute to the memory buffer.

        Note:
            If conversation messages contains the system message, it will be discarded from the memory.
        """
        if self.messages is None:
            return
        # discard system messages
        for message in self.messages:
            if message.type.lower() == ChatMessageEnum.SYSTEM.value:
                self.messages.remove(message)
        # generate pairs of Human and AI conversation messages and save to the memory buffer
        for i in range(0, len(self.messages), 2):
            inputs, outputs = self.generate_chat_messages(self.messages[i], self.messages[i + 1])
            self.save_context(inputs, outputs)

    def generate_chat_messages(self, *pairs: Message):
        """generate pairs of Human and AI conversation messages"""
        human_dict = {}
        ai_dict = {}
        for pair in pairs:
            if pair.type == ChatMessageEnum.HUMAN.value:
                human_dict = {self.input_key: pair.content}
            elif pair.type == ChatMessageEnum.AI.value:
                ai_dict = {self.output_key: pair.content}

        return human_dict, ai_dict

    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
        """Save context from the conversation to buffer and truncate part of memories."""

        def message_to_dict(message):
            return {
                "content": message.content,
                "type": message.type
            }

        super().save_context(inputs, outputs)
        message_list = self.load_memory
        messages_dicts = [message_to_dict(message) for message in message_list]
        inputs[self.memory_key] = messages_dicts

load_memory: List[BaseMessage] property

General method: load the memory context from the memory buffer.

load_memory_str: str property

General method: load the memory context from the memory buffer as string format.

__init__(**kwargs)

The init method.

The initialization method saves the memory context from messages in messages attribute to the memory buffer.

Parameters:

Name	Type	Description	Default
**kwargs		Arbitrary keyword arguments.	{}

Source code in agentuniverse/agent/memory/langchain_instance.py

Python

def __init__(self, **kwargs):
    """The __init__ method.

    The initialization method saves the memory context from messages in `messages` attribute to the memory buffer.

    Args:
        **kwargs: Arbitrary keyword arguments.
    """
    super().__init__(**kwargs)
    self.build_memory()

build_memory()

Save the memory context from messages in messages attribute to the memory buffer.

Note

If conversation messages contains the system message, it will be discarded from the memory.

Source code in agentuniverse/agent/memory/langchain_instance.py

Python

def build_memory(self):
    """Save the memory context from messages in `messages` attribute to the memory buffer.

    Note:
        If conversation messages contains the system message, it will be discarded from the memory.
    """
    if self.messages is None:
        return
    # discard system messages
    for message in self.messages:
        if message.type.lower() == ChatMessageEnum.SYSTEM.value:
            self.messages.remove(message)
    # generate pairs of Human and AI conversation messages and save to the memory buffer
    for i in range(0, len(self.messages), 2):
        inputs, outputs = self.generate_chat_messages(self.messages[i], self.messages[i + 1])
        self.save_context(inputs, outputs)

generate_chat_messages(*pairs)

generate pairs of Human and AI conversation messages

Source code in agentuniverse/agent/memory/langchain_instance.py

Python

def generate_chat_messages(self, *pairs: Message):
    """generate pairs of Human and AI conversation messages"""
    human_dict = {}
    ai_dict = {}
    for pair in pairs:
        if pair.type == ChatMessageEnum.HUMAN.value:
            human_dict = {self.input_key: pair.content}
        elif pair.type == ChatMessageEnum.AI.value:
            ai_dict = {self.output_key: pair.content}

    return human_dict, ai_dict

save_context(inputs, outputs)

Save context from the conversation to buffer and truncate part of memories.

Source code in agentuniverse/agent/memory/langchain_instance.py

Python

def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
    """Save context from the conversation to buffer and truncate part of memories."""

    def message_to_dict(message):
        return {
            "content": message.content,
            "type": message.type
        }

    super().save_context(inputs, outputs)
    message_list = self.load_memory
    messages_dicts = [message_to_dict(message) for message in message_list]
    inputs[self.memory_key] = messages_dicts