[LangChain] Chatbot Message History

🤓 Message History가 필요한 이유

---

model을 load하여 이름을 알려주고 내 이름이 무엇인지 다시 질문해보자.

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import HumanMessage

model = ChatGoogleGenerativeAI(model="gemini-pro",
                               convert_system_message_to_human=True,
                               )
model.invoke([HumanMessage(content="Hi! I'm Bob")])

model.invoke([HumanMessage(content="What's my name?")])

이름을 알려준 첫번째 대화와 나의 이름을 물어보는 두번째 대화가 같은 맥락으로 이어지지 않는 것을 확인할 수 있다.

대화 1)

대화 2)

이 문제를 해결하기 위해선 전체 대화 기록을 모델에 전달해야한다.

하나의 invoke() 메소드에 리스트 형태로 HumanMessage를 전달해보자.

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import HumanMessage
from langchain_core.messages import AIMessage

model.invoke(
  [
    HumanMessage(content="Hi, I'm Bob"),
    AIMessage(content="Hello Bob! How can i assist you today?"),
    HumanMessage(content="What's my name?")
  ]
)

이전과 달리 이전 대화 기록을 기억하고 답변하는 것을 확인할 수 있다.

📑 Message History 구현

Message History 사용법

Message History 클래스를 사용하여 모델을 래핑하여 대화 상태를 유지하도록 구현할 수 있다.

이 방법을 이용하면 모델의 입력과 출력을 추적해 데이터 저장소에 저장하고 이후의 상호작용에서 데이터 저장소에 저장된 데이터를 활용하여 체인의 일부로 전달한다.

from dotenv import load_dotenv
from langchain_google_genai import ChatGoogleGenerativeAI

from langchain_core.messages import (
  HumanMessage,
  AIMessage 
)

from langchain_core.chat_history import (
  BaseChatMessageHistory,
  InMemoryChatMessageHistory,
)
from langchain_core.runnables.history import RunnableWithMessageHistory

# load settings
load_dotenv()

# define model
model = ChatGoogleGenerativeAI(model="gemini-pro",
                               convert_system_message_to_human=True,
)

# session_id를 저장할 저장소
store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
  if session_id not in store:
    store[session_id] = InMemoryChatMessageHistory()
  return store[session_id]

with_message_history = RunnableWithMessageHistory(model, get_session_history)

config = {"configurable":
    {"session_id": "ABC"}
  }

response = with_message_history.invoke(
  [HumanMessage(content="Hi, I'm Bob")],
  config=config,
)
print(response.content)

response = with_message_history.invoke(
  [HumanMessage(content="What's my name?")],
  config=config,
)
print(response.content)

get_session_history() 함수는 기록된 session_id가 존재하는 경우 해당 session_id를 반환하여 RunnableWithMessageHistory를 통해 챗봇과 대화를 이어나갈 수 있게 해준다.

만약 다른 session_id가 전달되는 경우 새로운 대화로 인식한다.

PromptTemplate 적용

...
prompt = ChatPromptTemplate.from_messages(
  [
    (
      "system",
      "너는 훌륭한 외국어 공부 보조 AI야. 한국어로 먼저 답변해주고 이어서 {language}로 답변해줘."
    ),
    MessagesPlaceholder(variable_name="messages"),
  ]
)

chain = prompt | model

with_message_history = RunnableWithMessageHistory(chain, get_session_history, input_messages_key="messages",)

response = with_message_history.invoke(
    {"messages": [HumanMessage(content="난 한국에서 태어난 남자야! 영어를 공부하고 싶어!")], "language": "english",},
    config=config,
)

response = with_message_history.invoke(
    {"messages": [HumanMessage(content="일본어도 공부하고싶은데 내 성별의 말투로 인사하는 방법을 알려줘!")], "language": "japanese"},
    config=config,
)

대화 1)

대화 2)

🛠 Conversation History 관리

챗봇을 구축할 때 가장 중요한 개념 중 하나는 대화 기록을 관리하는 방법이다.
대화 기록을 관리하지 않으면 질문마다 메세지 목록이 증가하여 LLM의 Context Window를 오버플로우할 가능성이 있다.

LangChain에서 제공하는 Trimmer를 이용해 기존의 대화내역을 모델의 제약조건에 맞추어 전달할 수 있다.

먼저 테스트용 대화내역을 만들고 Trimmer에 적용한다. 이 때 Trimmer의 max_tokens 파라미터를 이용해 최대로 유지하는 대화내역을 조절할 수 있다.

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, trim_messages

trimmer = trim_messages(
    max_tokens=1000,
    strategy="last",
    token_counter=model,
    include_system=True,
    allow_partial=False,
    start_on="human",
)

messages = [
    HumanMessage(content="안녕 난 고길동이야! 곧 유학을 떠날 예정인데 회화좀 도와줄래?"),
    AIMessage(content="괜찮습니다! 제가 도와드릴게요!"),
    HumanMessage(content="외국인 친구 역할로 대답해줘!"),
    AIMessage(content="네 좋아요! 최선을 다해 도와드릴게요."),
]

trimmer.invoke(messages)

from operator import itemgetter

chain = (
    RunnablePassthrough.assign(messages=itemgetter("messages") | trimmer)
    | prompt
    | model
)

with_message_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="messages",
)

with_message_history.invoke(
    {
        "messages": messages + [HumanMessage(content="내 이름이 뭐라고~?")],
        "language": "English",
    },
    config=config,
)

🤖 Streaming

Streaming 방식의 답변을 원하는 경우 다음과 같이 사용할 수 있다.

# Streaming
for r in with_message_history.stream(
    {
        "messages": [HumanMessage(content="내 첫 외국인 친구야! 앞으로 친하게 지내자!")],
        "language": "English",
    },
    config=config,
):
    print(r.content, end="|")

>>> 저도 여러분의 첫 외국인 친구가 될 수 있어서 기|쁩니다! 앞으로 친하게 지내기를 바랍니다!|

Ref : https://python.langchain.com/v0.2/docs/tutorials/chatbot/