Kwangmin Kim - SQL (SQLAlchemy)

Structured Query Language (SQL)은 프로그래밍에 사용되는 도메인 특화 언어로, 관계형 데이터베이스 관리 시스템(RDBMS)에서 데이터를 관리하거나 관계형 데이터 스트림 관리 시스템(RDSMS)에서 스트림 처리를 위해 설계되었습니다. 특히 엔티티와 변수 간의 관계를 포함하는 구조화된 데이터를 다루는 데 유용합니다.

SQLAlchemy는 MIT 라이선스에 따라 배포되는 Python 프로그래밍 언어용 오픈 소스 SQL 툴킷이자 객체 관계 매퍼(ORM)입니다.

이 노트북에서는 SQLAlchemy가 지원하는 모든 데이터베이스에 채팅 기록을 저장할 수 있는 SQLChatMessageHistory 클래스에 대해 설명합니다.

SQLite 이외의 데이터베이스와 함께 사용하려면 해당 데이터베이스 드라이버를 설치해야 합니다.

# API KEY를 환경변수로 관리하기 위한 설정 파일
from dotenv import load_dotenv

# API KEY 정보로드
load_dotenv()

1 사용방법

storage를 사용하려면 다음 2가지만 제공하면 됩니다:

session_id - 사용자 이름, 이메일, 채팅 ID 등과 같은 세션의 고유 식별자입니다.
connection - 데이터베이스 연결을 지정하는 문자열입니다. 이 문자열은 SQLAlchemy의 create_engine 함수에 전달됩니다.

from langchain_community.chat_message_histories import SQLChatMessageHistory

# SQLChatMessageHistory 객체를 생성하고 세션 ID와 데이터베이스 연결 파일을 설정
chat_message_history = SQLChatMessageHistory(
    session_id="sql_history", connection="sqlite:///sqlite.db"
)

# 사용자 메시지를 추가합니다.
chat_message_history.add_user_message(
    "안녕? 만나서 반가워. 내 이름은 테디야. 나는 랭체인 개발자야. 앞으로 잘 부탁해!"
)
# AI 메시지를 추가합니다.
chat_message_history.add_ai_message("안녕 테디, 만나서 반가워. 나도 잘 부탁해!")

저장된 대화내용을 확인합니다. chat_message_history.messages

# 채팅 메시지 기록의 메시지들
chat_message_history.messages

2 Chain 에 적용

우리는 이 메시지 기록 클래스를 LCEL Runnables 와 쉽게 결합할 수 있습니다.

from langchain_core.prompts import (
    ChatPromptTemplate,
    MessagesPlaceholder,
)
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages(
    [
        # 시스템 메시지
        ("system", "You are a helpful assistant."),
        # 대화 기록을 위한 Placeholder
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "{question}"),  # 질문
    ]
)

# chain 을 생성합니다.
chain = prompt | ChatOpenAI(model_name="gpt-4.1-mini") | StrOutputParser()

sqlite.db 에서 대화내용을 가져오는 함수를 만듭니다.

def get_chat_history(user_id, conversation_id):
    return SQLChatMessageHistory(
        table_name=user_id,
        session_id=conversation_id,
        connection="sqlite:///sqlite.db",
    )

config_fields 를 설정합니다. 이는 대화정보를 조회할 때 참고 정보로 활용합니다.

user_id: 사용자 ID
conversation_id: 대화 ID

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",  # 입력 메시지의 키를 "question"으로 설정
    history_messages_key="chat_history",  # 대화 기록 메시지의 키를 "history"로 설정
    history_factory_config=config_fields,  # 대화 기록 조회시 참고할 파라미터를 설정합니다.
)

"configurable" 키 아래에 "user_id", "conversation_id" key-value 쌍을 설정합니다.

# config 설정
config = {"configurable": {"user_id": "user1", "conversation_id": "conversation1"}}

질문에 이름을 물어보는 질문을 해보겠습니다. 이전에 저장한 대화가 있다면, 올바르게 답할 것입니다.

chain_with_history 객체의 invoke 메서드를 호출하여 질문에 대한 답변을 생성합니다.
invoke 메서드에는 질문 딕셔너리와 config 설정이 전달됩니다.

# 질문과 config 를 전달하여 실행합니다.
chain_with_history.invoke({"question": "안녕 반가워, 내 이름은 테디야"}, config)

# 후속 질문을 실해합니다.
chain_with_history.invoke({"question": "내 이름이 뭐라고?"}, config)

이번에는 같은 user_id 를 가지지만 conversion_id 가 다른 값을 가지도록 설정합니다.

# config 설정
config = {"configurable": {"user_id": "user1", "conversation_id": "conversation2"}}

# 질문과 config 를 전달하여 실행합니다.
chain_with_history.invoke({"question": "내 이름이 뭐라고?"}, config)