|
|
from typing import Any, Dict, List, Optional |
|
|
|
|
|
from langchain_core._api import deprecated |
|
|
from langchain_core.messages import BaseMessage, get_buffer_string |
|
|
from langchain_core.utils import pre_init |
|
|
|
|
|
from langchain.memory.chat_memory import BaseChatMemory, BaseMemory |
|
|
from langchain.memory.utils import get_prompt_input_key |
|
|
|
|
|
|
|
|
@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 ConversationBufferMemory(BaseChatMemory): |
|
|
"""A basic memory implementation that simply stores the conversation history. |
|
|
|
|
|
This stores the entire conversation history in memory without any |
|
|
additional processing. |
|
|
|
|
|
Note that additional processing may be required in some situations when the |
|
|
conversation history is too large to fit in the context window of the model. |
|
|
""" |
|
|
|
|
|
human_prefix: str = "Human" |
|
|
ai_prefix: str = "AI" |
|
|
memory_key: str = "history" |
|
|
|
|
|
@property |
|
|
def buffer(self) -> Any: |
|
|
"""String buffer of memory.""" |
|
|
return self.buffer_as_messages if self.return_messages else self.buffer_as_str |
|
|
|
|
|
async def abuffer(self) -> Any: |
|
|
"""String buffer of memory.""" |
|
|
return ( |
|
|
await self.abuffer_as_messages() |
|
|
if self.return_messages |
|
|
else await self.abuffer_as_str() |
|
|
) |
|
|
|
|
|
def _buffer_as_str(self, messages: List[BaseMessage]) -> str: |
|
|
return get_buffer_string( |
|
|
messages, |
|
|
human_prefix=self.human_prefix, |
|
|
ai_prefix=self.ai_prefix, |
|
|
) |
|
|
|
|
|
@property |
|
|
def buffer_as_str(self) -> str: |
|
|
"""Exposes the buffer as a string in case return_messages is True.""" |
|
|
return self._buffer_as_str(self.chat_memory.messages) |
|
|
|
|
|
async def abuffer_as_str(self) -> str: |
|
|
"""Exposes the buffer as a string in case return_messages is True.""" |
|
|
messages = await self.chat_memory.aget_messages() |
|
|
return self._buffer_as_str(messages) |
|
|
|
|
|
@property |
|
|
def buffer_as_messages(self) -> List[BaseMessage]: |
|
|
"""Exposes the buffer as a list of messages in case return_messages is False.""" |
|
|
return self.chat_memory.messages |
|
|
|
|
|
async def abuffer_as_messages(self) -> List[BaseMessage]: |
|
|
"""Exposes the buffer as a list of messages in case return_messages is False.""" |
|
|
return await self.chat_memory.aget_messages() |
|
|
|
|
|
@property |
|
|
def memory_variables(self) -> List[str]: |
|
|
"""Will always return list of memory variables. |
|
|
|
|
|
:meta private: |
|
|
""" |
|
|
return [self.memory_key] |
|
|
|
|
|
def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]: |
|
|
"""Return history buffer.""" |
|
|
return {self.memory_key: self.buffer} |
|
|
|
|
|
async def aload_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]: |
|
|
"""Return key-value pairs given the text input to the chain.""" |
|
|
buffer = await self.abuffer() |
|
|
return {self.memory_key: buffer} |
|
|
|
|
|
|
|
|
@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 ConversationStringBufferMemory(BaseMemory): |
|
|
"""A basic memory implementation that simply stores the conversation history. |
|
|
|
|
|
This stores the entire conversation history in memory without any |
|
|
additional processing. |
|
|
|
|
|
Equivalent to ConversationBufferMemory but tailored more specifically |
|
|
for string-based conversations rather than chat models. |
|
|
|
|
|
Note that additional processing may be required in some situations when the |
|
|
conversation history is too large to fit in the context window of the model. |
|
|
""" |
|
|
|
|
|
human_prefix: str = "Human" |
|
|
ai_prefix: str = "AI" |
|
|
"""Prefix to use for AI generated responses.""" |
|
|
buffer: str = "" |
|
|
output_key: Optional[str] = None |
|
|
input_key: Optional[str] = None |
|
|
memory_key: str = "history" |
|
|
|
|
|
@pre_init |
|
|
def validate_chains(cls, values: Dict) -> Dict: |
|
|
"""Validate that return messages is not True.""" |
|
|
if values.get("return_messages", False): |
|
|
raise ValueError( |
|
|
"return_messages must be False for ConversationStringBufferMemory" |
|
|
) |
|
|
return values |
|
|
|
|
|
@property |
|
|
def memory_variables(self) -> List[str]: |
|
|
"""Will always return list of memory variables. |
|
|
:meta private: |
|
|
""" |
|
|
return [self.memory_key] |
|
|
|
|
|
def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, str]: |
|
|
"""Return history buffer.""" |
|
|
return {self.memory_key: self.buffer} |
|
|
|
|
|
async def aload_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, str]: |
|
|
"""Return history buffer.""" |
|
|
return self.load_memory_variables(inputs) |
|
|
|
|
|
def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None: |
|
|
"""Save context from this conversation to buffer.""" |
|
|
if self.input_key is None: |
|
|
prompt_input_key = get_prompt_input_key(inputs, self.memory_variables) |
|
|
else: |
|
|
prompt_input_key = self.input_key |
|
|
if self.output_key is None: |
|
|
if len(outputs) != 1: |
|
|
raise ValueError(f"One output key expected, got {outputs.keys()}") |
|
|
output_key = list(outputs.keys())[0] |
|
|
else: |
|
|
output_key = self.output_key |
|
|
human = f"{self.human_prefix}: " + inputs[prompt_input_key] |
|
|
ai = f"{self.ai_prefix}: " + outputs[output_key] |
|
|
self.buffer += "\n" + "\n".join([human, ai]) |
|
|
|
|
|
async def asave_context( |
|
|
self, inputs: Dict[str, Any], outputs: Dict[str, str] |
|
|
) -> None: |
|
|
"""Save context from this conversation to buffer.""" |
|
|
return self.save_context(inputs, outputs) |
|
|
|
|
|
def clear(self) -> None: |
|
|
"""Clear memory contents.""" |
|
|
self.buffer = "" |
|
|
|
|
|
async def aclear(self) -> None: |
|
|
self.clear() |
|
|
|
