[AI 서비스 개발] Langchain 대화 내용 기록 RunnableWithMessageHistory (1) Message 구조

- ConversationCahin이 최신 langchain 버전에서는 지원하지 않아 대체 코드를 찾기 위해 공식 문서를 찾아보았다.

- 공식 문서의 예제 코드를 이해하려고 하니 class개념이 부족해 개념을 먼저 학습하였다 : https://bravesol.tistory.com/181

 

- langchain 라이브러리 공식 문서 :https://python.langchain.com/v0.2/api_reference/core/runnables/langchain_core.runnables.history.RunnableWithMessageHistory.html

from operator import itemgetter
from typing import List

from langchain_openai.chat_models import ChatOpenAI

from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.documents import Document
from langchain_core.messages import BaseMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import (
    RunnableLambda,
    ConfigurableFieldSpec,
    RunnablePassthrough,
)
from langchain_core.runnables.history import RunnableWithMessageHistory


class InMemoryHistory(BaseChatMessageHistory, BaseModel): #2
    """In memory implementation of chat message history."""

    messages: List[BaseMessage] = Field(default_factory=list)

    def add_messages(self, messages: List[BaseMessage]) -> None:
        """Add a list of messages to the store"""
        self.messages.extend(messages)

    def clear(self) -> None:
        self.messages = []

# Here we use a global variable to store the chat message history.
# This will make it easier to inspect it to see the underlying results.
store = {}

def get_by_session_id(session_id: str) -> BaseChatMessageHistory: #1
    if session_id not in store:
        store[session_id] = InMemoryHistory()
    return store[session_id]


history = get_by_session_id("1")
history.add_message(AIMessage(content="hello")) #3
print(store)  # noqa: T201

 

1. get_by_session_id 함수 이해하기

def get_by_session_id(session_id: str) -> BaseChatMessageHistory: # 1,2
    if session_id not in store:
        store[session_id] = InMemoryHistory() # 3
    return store[session_id]

1) session_id가 필요한 이유

- 하나의 애플리케이션에서 여러 사용자가 동시에 AI와 대화할 수 있다.

- 모든 사용자의 대화 기록을 하나의 전역 저장소(store)에 섞어 관리하면, 특정 사용자의 대화 기록을 추적하기 어렵다.

 

2) (session_id: str) -> BaseChatMessageHistory 의 의미

- 타입 힌팅: 가독성과 안정성을 높이기 위해 변수, 함수의 매개변수, 반환값의 데이터 타입을 명시적으로 표시하는 것

(파이썬 3.5부터 도입되었다)

- get_by_session_id함수의 매개변수인 session_id은 str(문자열)이고, 반환값(->)은 BaseChatMessageHistory 타입

- InMemoryHistory는 BaseChatMessageHistory를 상속하므로 반환 타입이 올바르다.

 

2. InMemoryHistory 클래스 이해하기

class InMemoryHistory(BaseChatMessageHistory, BaseModel): #1)
    """In memory implementation of chat message history."""

    messages: List[BaseMessage] = Field(default_factory=list) #2)

    def add_messages(self, messages: List[BaseMessage]) -> None: # 3)
        """Add a list of messages to the store"""
        self.messages.extend(messages)

    def clear(self) -> None: # 4)
        self.messages = []

1) BaseChatMessageHistory와 BaseModel을 상속받음

- BaseChatMessageHistory와 BaseModel의 변수와 기능을 사용할 수 있다.

 

2) messages: List[BaseMessage] = Field(default_factory=list)

- 타입힌팅: messages의 데이터 타입은 List[BaseMessage]이다

- List: 리스트

- BaseMessage: 리스트 안에 들어갈 요소의 타입

=> messages의 데이터 타입은 BaseMessage 객체들을 요소로 갖는 list이다

- BaseMessage: 메세지의 기본 속성과 구조를 정의, 대화 기록 관리에서 사용되는 기본 데이터 타입

(content: 메세지의 텍스트 내용, role: 메세지를 보낸 역할(user, assistant),추가 메타데이터(타임스태프, ID 등)

- Field: message에 추가적인 설정을 제공(검증 조건, 기본값, 설명 등)

- Field(default_factory=list): message 필드의 기본 값을 빈 리스트로 설정, 새로운 리스트를 생성하므로 각 인스턴스가 독립적인 리스트를 갖게 된다

=> message는 기본값으로 빈 리스트( [ ] ) 를 가진다

* Pydantic : 데이터 검증 및 설정 관리를 위한 python 라이브러리, 타입 힌딩과 자연스럽게 통합되어 가독성이 좋다

* typing: 타입 힌팅을 위해 사용되는 다양한 타입을 제공(예: List[int] = [1,2,3]), 파이썬 3.8이하버전은 :list를 지원하지 않아 호환성을 위해 사용

 

3) 위에서 messages의 타입을 지정했는데 왜 또 messages: List[BaseMessage]를 언급할까?

- 위에 있는 messages: List[BaseMessage] = Field(default_factory=list)는 Field를 사용했기 때문에 self가 없어도인스턴스 변수(클래스변수 아님)

- add_messages(self, messages: List[BaseMessage]) -> None에서 message는 위에 정의한 message가 아니라 외부에서 받아오는 새로운 메세지 리스트이다. 그래서 타입을 다시 명시해 주었다.

- -> None: 따로 반환값이 없는 함수

- extend와 append의 차이 : extend는 요소를 전달, append는 값 자체가 전달

# extend 예제
a = [1, 2, 3]
b = [4, 5, 6]

a.extend(b)
print(a)  # [1, 2, 3, 4, 5, 6]

# append 예제
a = [1, 2, 3]
b = [4, 5, 6]

a.append(b)
print(a)  # [1, 2, 3, [4, 5, 6]]

 

4) 메세지 기록 초기화

- 사용 이유: 사용자가 새로운 대화를 시작하려는 경우, 기존 대화기록을 지워하 하므로

- 새로운 객체를 사용해 기존 참조와 완전리 분리하여 다른 참조가 영향을 받지 않게 하기 위해 clear() 대신 [ ] 새로 부여

- list.clear()는 python 내장함수로, 기존 리스트 객체는 그대로 남아 있고, 내부 요소만 지워줌

a = [1,2,3]
a.clear()
a # [ ]

 

3.AIMessage(content="hello") 이해하기

1) AIMessage는 Langchain 에서 메시지 클래스 간의 일관된 인터페이스를 제공

* 인터페이스(interface)란?

- 시스템의 구성 요소간에 상호작용하는 방법
- 시스템이 제공해야 할 동작(기능)을 명세

- 내부 구현 세부사항(어떻게 하는지)은 감추고, 무엇을 제공하는지 명시

- 객체지향 프로그래밍(OOP)에서 인터페이스: 클래스는 인터페이스로, 메서드를 제공, 세부사항은 서브클래스에서 정의

2) langchain 메시지 구조

클래스	설명
BaseMessage	모든 메시지 클래스의 기본 클래스
AIMessage	AI 모델이 "생성한" 메세지
HumanMessage	사람이 "생성한" 메세지
SystemMessage	AI의 역할, 응답 스타일, 행동 방식을 제어

- 사용예시

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

# 대화 초기화
conversation = [
    SystemMessage(content="You are a helpful assistant."),
    HumanMessage(content="What's the capital of Japan?"),
    AIMessage(content="The capital of Japan is Tokyo.")
]

# 대화 출력
for message in conversation:
    print(f"{message.__class__.__name__}: {message.content}")


# 출력
# SystemMessage: You are a helpful assistant.
# HumanMessage: What's the capital of Japan?
# AIMessage: The capital of Japan is Tokyo.

* BaseMessage는 인터페이스인가? 추상클래스인가?

- Python 에서 인터페이스: 구현 세부사항 없이 메서드 시그니처만 정의, 모든 구현은 서브클래스에서 이루어짐

- Python 에서 추상클래스: 인터페이스와 유사하지만, 일부 기본 구현을 포함할 수 있다.

- BaseMessage는 인터페이스처럼 동작하는 추상클래스