Memory Using SQLite
Author: Heesun Moon
Proofread : Juni Lee
This is a part of LangChain Open Tutorial
Overview
This tutorial explains the SQLChatMessageHistory
class, which allows storing chat history in any database supported by SQLAlchemy
.
Structured Query Language (SQL)
is a domain-specific language used in programming and designed for managing data held in a Relational Database Management System (RDBMS), or for stream processing in a Relational Data Stream Management System (RDSMS). It is particularly useful for handling structured data, including relationships between entities and variables.
SQLAlchemy
is an open-source SQL toolkit and Object-Relational Mapper (ORM) for the Python programming language, released under the MIT License.
To use a database other than SQLite
, please make sure to install the appropriate database driver first.
Table of Contents
References
Environment Setup
Set up the environment. You may refer to Environment Setup for more details.
[Note]
langchain-opentutorial
is a package that provides a set of easy-to-use environment setup, useful functions and utilities for tutorials.You can checkout the
langchain-opentutorial
for more details.
%%capture --no-stderr
%pip install langchain_opentutorial
# Install required packages
from langchain_opentutorial import package
package.install(
[
"langchain_community",
"langchain_openai",
"langchain_core",
"SQLAlchemy",
],
verbose=False,
upgrade=False,
)
# Set environment variables
from langchain_opentutorial import set_env
set_env(
{
"OPENAI_API_KEY": "",
"LANGCHAIN_API_KEY": "",
"LANGCHAIN_TRACING_V2": "true",
"LANGCHAIN_ENDPOINT": "https://api.smith.langchain.com",
"LANGCHAIN_PROJECT": "MemoryUsingSQLite",
}
)
You can alternatively set OPENAI_API_KEY
in .env
file and load it.
[Note] This is not necessary if you've already set OPENAI_API_KEY
in previous steps.
from dotenv import load_dotenv
load_dotenv()
Usage
To use the storage, you need to provide only the following 2 things:
session_id
- A unique identifier for the session, such as a user name, email, chat ID, etc.connection
- A string that specifies the database connection. This string will be passed to SQLAlchemy'screate_engine
function.
from langchain_community.chat_message_histories import SQLChatMessageHistory
# Initialize chat history with session ID and database connection.
chat_message_history = SQLChatMessageHistory(
session_id="sql_history", connection="sqlite:///sqlite.db"
)
# Add a user message
chat_message_history.add_user_message(
"Hello, nice to meet you! My name is Heesun :) I'm a LangChain developer. I look forward to working with you!"
)
# Add an AI message
chat_message_history.add_ai_message(
"Hi, Heesun! Nice to meet you. I look forward to working with you too!"
)
Now, let's check the stored conversation history.
chat_message_history.messages
[HumanMessage(content="Hello, nice to meet you! My name is Heesun :) I'm a LangChain developer. I look forward to working with you!", additional_kwargs={}, response_metadata={}),
AIMessage(content='Hi, Heesun! Nice to meet you. I look forward to working with you too!', additional_kwargs={}, response_metadata={})]
You can also clear the session memory from db:
# Clear the session memory
chat_message_history.clear()
chat_message_history.messages
[]
Adding Metadata
Metadata can be added by directly creating HumanMessage
and AIMessage
objects. This approach enables flexible data handling and logging.
Parameters:
additional_kwargs
- Stores custom tags or metadata, such as priority or task type.response_metadata
- Captures AI response details, including model, timestamp, and token count.
These fields enhance debugging and task tracking through detailed data storage.
from langchain_core.messages import HumanMessage
# Add a user message with additional metadata.
user_message = HumanMessage(
content="Can you help me summarize this text?",
additional_kwargs={"task": "summarization"},
)
# Add the message to chat history.
chat_message_history.add_message(user_message)
chat_message_history.messages
[HumanMessage(content='Can you help me summarize this text?', additional_kwargs={'task': 'summarization'}, response_metadata={})]
from langchain_core.messages import AIMessage
# Add an AI message with response metadata.
ai_message = AIMessage(
content="Sure! Here's the summary of the provided text.",
response_metadata={"model": "gpt-4", "token_count": 30, "response_time": "150ms"},
)
# Add the message to chat history.
chat_message_history.add_message(ai_message)
chat_message_history.messages
[HumanMessage(content='Can you help me summarize this text?', additional_kwargs={'task': 'summarization'}, response_metadata={}),
AIMessage(content="Sure! Here's the summary of the provided text.", additional_kwargs={}, response_metadata={'model': 'gpt-4', 'token_count': 30, 'response_time': '150ms'})]
Chaining
You can easily integrate this chat history class with LCEL Runnables.
from langchain_core.prompts import (
ChatPromptTemplate,
MessagesPlaceholder,
)
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
prompt = ChatPromptTemplate.from_messages(
[
("system", "You are a helpful assistant."),
# Placeholder for chat history
MessagesPlaceholder(variable_name="chat_history"),
("human", "{question}"),
]
)
# Chaining
chain = prompt | ChatOpenAI(model_name="gpt-4o") | StrOutputParser()
The following shows how to create a function that returns chat history from sqlite.db
.
def get_chat_history(user_id, conversation_id):
return SQLChatMessageHistory(
table_name=user_id,
session_id=conversation_id,
connection="sqlite:///sqlite.db",
)
Set config_fields
to provide reference information when retrieving conversation details.
from langchain_core.runnables.utils import ConfigurableFieldSpec
config_fields = [
ConfigurableFieldSpec(
id="user_id",
annotation=str,
name="User ID",
description="Unique identifier for a user.",
default="",
is_shared=True,
),
ConfigurableFieldSpec(
id="conversation_id",
annotation=str,
name="Conversation ID",
description="Unique identifier for a conversation.",
default="",
is_shared=True,
),
]
chain_with_history = RunnableWithMessageHistory(
chain,
get_chat_history,
input_messages_key="question",
history_messages_key="chat_history",
# Set parameters for retrieving chat history
history_factory_config=config_fields,
)
Set the "user_id"
and "conversation_id"
key-value pairs under the "configurable"
key.
# Config settings
config = {"configurable": {"user_id": "user1", "conversation_id": "conversation1"}}
Let's ask a question about the name. If there is any previously saved conversation history, it will provide the correct response.
Use the
invoke
method of thechain_with_history
object to generate an answer to the question.Pass a question dictionary and
config
settings to theinvoke
method as inputs.
# Execute by passing the question and config
chain_with_history.invoke(
{"question": "Hi, nice to meet you. My name is Heesun."}, config
)
'Hi Heesun! Nice to meet you again. How can I help you today?'
# Execute a follow-up question
chain_with_history.invoke({"question": "What is my name?"}, config)
'Your name is Heesun.'
This time, set the same user_id
but use a different value for conversation_id
.
# Config settings
config = {"configurable": {"user_id": "user1", "conversation_id": "conversation2"}}
# Execute by passing the question and config
chain_with_history.invoke({"question": "What is my name?"}, config)
"I'm sorry, but I don't have access to personal information, so I don't know your name. If you'd like, you can tell me your name, and I can address you by it."
Last updated