codekingpro/portable-devtools

1import warnings2from abc import ABC3from typing import Any4 5from langchain_core._api import deprecated6from langchain_core.chat_history import (7    BaseChatMessageHistory,8    InMemoryChatMessageHistory,9)10from langchain_core.messages import AIMessage, HumanMessage11from pydantic import Field12 13from langchain_classic.base_memory import BaseMemory14from langchain_classic.memory.utils import get_prompt_input_key15 16 17@deprecated(18    since="0.3.1",19    removal="2.0.0",20    alternative="langchain.agents.create_agent",21    addendum=(22        "For agents that need to remember prior interactions, use "23        "`create_agent` with checkpointing or the `Store` API. See "24        "https://docs.langchain.com/oss/python/langchain/short-term-memory and "25        "https://docs.langchain.com/oss/python/langchain/long-term-memory"26    ),27)28class BaseChatMemory(BaseMemory, ABC):29    """Abstract base class for chat memory.30 31    **ATTENTION** This abstraction was created prior to when chat models had32        native tool calling capabilities.33        It does **NOT** support native tool calling capabilities for chat models and34        will fail SILENTLY if used with a chat model that has native tool calling.35 36    DO NOT USE THIS ABSTRACTION FOR NEW CODE.37    """38 39    chat_memory: BaseChatMessageHistory = Field(40        default_factory=InMemoryChatMessageHistory,41    )42    output_key: str | None = None43    input_key: str | None = None44    return_messages: bool = False45 46    def _get_input_output(47        self,48        inputs: dict[str, Any],49        outputs: dict[str, str],50    ) -> tuple[str, str]:51        if self.input_key is None:52            prompt_input_key = get_prompt_input_key(inputs, self.memory_variables)53        else:54            prompt_input_key = self.input_key55        if self.output_key is None:56            if len(outputs) == 1:57                output_key = next(iter(outputs.keys()))58            elif "output" in outputs:59                output_key = "output"60                warnings.warn(61                    f"'{self.__class__.__name__}' got multiple output keys:"62                    f" {outputs.keys()}. The default 'output' key is being used."63                    f" If this is not desired, please manually set 'output_key'.",64                    stacklevel=3,65                )66            else:67                msg = (68                    f"Got multiple output keys: {outputs.keys()}, cannot "69                    f"determine which to store in memory. Please set the "70                    f"'output_key' explicitly."71                )72                raise ValueError(msg)73        else:74            output_key = self.output_key75        return inputs[prompt_input_key], outputs[output_key]76 77    def save_context(self, inputs: dict[str, Any], outputs: dict[str, str]) -> None:78        """Save context from this conversation to buffer."""79        input_str, output_str = self._get_input_output(inputs, outputs)80        self.chat_memory.add_messages(81            [82                HumanMessage(content=input_str),83                AIMessage(content=output_str),84            ],85        )86 87    async def asave_context(88        self,89        inputs: dict[str, Any],90        outputs: dict[str, str],91    ) -> None:92        """Save context from this conversation to buffer."""93        input_str, output_str = self._get_input_output(inputs, outputs)94        await self.chat_memory.aadd_messages(95            [96                HumanMessage(content=input_str),97                AIMessage(content=output_str),98            ],99        )100 101    def clear(self) -> None:102        """Clear memory contents."""103        self.chat_memory.clear()104 105    async def aclear(self) -> None:106        """Clear memory contents."""107        await self.chat_memory.aclear()108