08. Add memory to LCEL Chain

LCEL (Remember Conversation): Add Memory

Shows how to add memory to any chain. You can use the current memory class, but you have to connect it manually.

Copy

from dotenv import load_dotenv

load_dotenv()

Copy

True

Copy

from operator import itemgetter
from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_openai import ChatOpenAI


# ChatOpenAI Initialize the model.
model = ChatOpenAI()

# Creates an interactive prompt. This prompt includes system messages, previous conversation history, and user input.
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful chatbot"),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "{input}"),
    ]
)

Memory to store conversation ConversationBufferMemory Create and return_messages parameter True By setting it to, the generated instance returns the message.

• memory_key Settings: Chain later prompt It is the key to be brought inside. You can change it and use it.

Copy

# Creates a conversation buffer memory and enables message return functionality.
memory = ConversationBufferMemory(return_messages=True, memory_key="chat_history")

Check the saved minutes. Since we haven't saved it yet, the dialog is empty.

Copy

memory.load_memory_variables({})  # Initializes a memory variable to an empty dictionary.

Copy

{'chat_history': []} 

RunnablePassthrough.assign Using chat_history To variable memory.load_memory_variables Assign the result of the function, and from this result chat_history Extract the value corresponding to the key.

Copy

runnable = RunnablePassthrough.assign(
    chat_history=RunnableLambda(memory.load_memory_variables)
    | itemgetter("chat_history")  # memory_key Enter the same as.
)

Copy

runnable.invoke({"input": "hi"})

Copy

 {'input':'hi','chat_history': {'chat_history': []}} 

Copy

runnable.invoke({"input": "hi"})

Copy

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful chatbot"),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "{input}"),
    ]
)

runnable Start the first conversation on.

• input : User input conversation is passed.

• chat_history : Conversation records are passed.

Copy

runnable.invoke({"input": "hi!"})

Copy

{'input':'hi!','chat_history': []} 

Copy

chain = runnable | prompt | model

Proceed with the first conversation.

Copy

# chain of the object invoke Generate a response to input using a method.
response = chain.invoke({"input": "Nice to meet you. My name is Teddy."})
print(response.content)  # Prints the generated response.

Copy

Nice to meet you, Teddy! What can you help?

Copy

memory.load_memory_variables({})

Copy

 {'chat_history': []} 

Copy

memory.save_context Functions are input data inputs ) And response content ( response.content ) To save to memory. It can be used to record the current state in the learning process of the AI model, or to track user requests and system responses.

Copy

# Stores the entered data and response content in memory.
memory.save_context(
    {"human": "Nice to meet you. My name is Teddy."}, {"ai": response.content}
)

# Prints saved conversation history.
memory.load_memory_variables({})

Copy

 {'chat_history': [HumanMessage (content=' Nice to meet you. My name is Teddy.'), AIMessage (content=' Nice to meet you, Teddy! What can I help you with?')]} 

Ask if you remember your name.

Copy

# I'll ask you further if you remember the name.
response = chain.invoke({"input": "Remember what my name was?"})
# Prints the answer.
print(response.content)

Copy

Yes, I remember Teddy's name. Need some help?

Custom ConversationChain implementation example

Copy

from operator import itemgetter
from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableLambda, RunnablePassthrough, Runnable
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# ChatOpenAI Initialize the model.
llm = ChatOpenAI(model_name="gpt-4o", temperature=0)

# Creates an interactive prompt, which includes system messages, previous conversation history, and user input.
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful chatbot"),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "{input}"),
    ]
)

# Creates a conversation buffer memory and enables message return functionality.
memory = ConversationBufferMemory(return_messages=True, memory_key="chat_history")

Copy

class MyConversationChain(Runnable):

    def __init__(self, llm, prompt, memory, input_key="input"):

        self.prompt = prompt
        self.memory = memory
        self.input_key = input_key

        self.chain = (
            RunnablePassthrough.assign(
                chat_history=RunnableLambda(self.memory.load_memory_variables)
                | itemgetter(memory.memory_key)  # memory_key Enter the same as.
            )
            | prompt
            | llm
            | StrOutputParser()
        )

    def invoke(self, query, configs=None, **kwargs):
        answer = self.chain.invoke({self.input_key: query})
        self.memory.save_context(inputs={"human": query}, outputs={"ai": answer})
        return answer

Copy

# ChatOpenAI Initialize the model.
llm = ChatOpenAI(model_name="gpt-4o", temperature=0)

# Creates an interactive prompt, which includes system messages, previous conversation history, and user input.
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful chatbot"),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "{input}"),
    ]
)

# Creates a conversation buffer memory and enables message return functionality.
memory = ConversationBufferMemory(return_messages=True, memory_key="chat_history")

# If you replace it with summary memory
# memory = ConversationSummaryMemory(
#     llm=llm, return_messages=True, memory_key="chat_history"
# )

conversation_chain = MyConversationChain(llm, prompt, memory)

Copy

conversation_chain.invoke("Hello, nice to meet you. My name is Teddy.")

Copy

'Hello, Teddy! Nice to meet you. How can I help you?'

Copy

conversation_chain.invoke("What is my name?")

Copy

 'You said Teddy? Is that correct?' 

Copy

conversation_chain.invoke("Please answer in English only from now on.Okay.?")

Copy

 'Sure, I understand. How can I assist you today?' 

Copy

conversation_chain.invoke("Please tell me my name again")

Copy

'Your name is Teddy.' 

Copy

conversation_chain.memory.load_memory_variables({})["chat_history"]

Copy

 [HumanMessage (content=' Hello? Nice to meet you. My name is Teddy.'), AIMessage (content='Hello, Teddy! Nice to meet you. How can I help you?'), HumanMessage (content='What is my name?'), AIMessage (content='? Is that correct?'), HumanMessage (content=' Please answer English only in the future, okay?'), AIMessage (content='Sure, I understand. How can I assist you today?'), HumanMessage (content='Please tell me my name again'), AIMessage (content='Your name is Teddy.')]