data_driven_characters/memory/retrieval.py

from typing import Any, List, Dict
from langchain.memory import VectorStoreRetrieverMemory

from langchain.schema import Document


class ConversationVectorStoreRetrieverMemory(VectorStoreRetrieverMemory):
    input_prefix = "Human"
    output_prefix = "AI"
    blacklist = []  # keys to ignore

    def _form_documents(
        self, inputs: Dict[str, Any], outputs: Dict[str, str]
    ) -> List[Document]:
        """Format context from this conversation to buffer."""
        # Each document should only include the current turn, not the chat history
        filtered_inputs = {
            k: v
            for k, v in inputs.items()
            if k != self.memory_key and k not in self.blacklist
        }
        texts = []
        for k, v in list(filtered_inputs.items()) + list(outputs.items()):
            if k == "input":
                k = self.input_prefix
            elif k == "response":
                k = self.output_prefix
            texts.append(f"{k}: {v}")
        page_content = "\n".join(texts)
        return [Document(page_content=page_content)]