RunnableWithMessageHistory

정의

LangChain의 RunnableWithMessageHistory체인(또는 러너블)에 자동 메모리 관리 기능을 추가하는 래퍼. ChatMessageHistory와 다르게, invoke 호출 시 메시지 저장/로드를 자동으로 처리하고, session_id를 통한 다중 사용자 세션 관리를 지원한다.

핵심 특징

특징설명
자동 메모리invoke 시 자동 저장/로드 (수동 호출 불필요)
Session ID사용자별 독립적인 대화 컨텍스트 유지
상태 보존같은 session_id = 같은 히스토리 (다중 요청에서 일관성)
래퍼 패턴기존 chain을 감싸서 기능 확장

아키텍처

사용자 입력 + Session ID
        ↓
RunnableWithMessageHistory
├─ Session 히스토리 로드
├─ 프롬프트 구성 (chat_history + input)
├─ 체인 실행
├─ 응답 생성
└─ 히스토리에 자동 저장
        ↓
AI 응답

구현 패턴

기본 설정

from langchain.memory import ChatMessageHistory
from langchain.runnables.history import RunnableWithMessageHistory
from langchain.prompts import ChatPromptTemplate
from langchain.chat_models import ChatOpenAI
 
# 1. 세션 저장소 (in-memory)
session_history = {}
 
def get_session_history(session_id: str) -> ChatMessageHistory:
    """세션 ID로 해당 히스토리 조회 또는 생성"""
    if session_id not in session_history:
        session_history[session_id] = ChatMessageHistory()
    return session_history[session_id]
 
# 2. 기본 체인 구성
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 도움을 주는 어시스턴트입니다."),
    ("placeholder", "{chat_history}"),
    ("human", "{input}")
])
 
model = ChatOpenAI(model="gpt-4o")
chain = prompt | model
 
# 3. RunnableWithMessageHistory로 감싸기
runnable_with_history = RunnableWithMessageHistory(
    runnable=chain,
    get_session_history=get_session_history,
    input_messages_key="input",              # 프롬프트에서 사용자 입력 키
    history_messages_key="chat_history"      # 프롬프트에서 히스토리 키
)
 
# 4. 호출 (session_id와 함께)
response = runnable_with_history.invoke(
    {"input": "안녕하세요!"},
    config={"configurable": {"session_id": "user123"}}
)

매개변수 상세

매개변수설명필수
runnable실행할 체인 또는 러너블
get_session_historysession_id → ChatMessageHistory 반환 함수
input_messages_key프롬프트에서 사용자 입력을 받을 키 이름
history_messages_key프롬프트에서 히스토리를 받을 키 이름

세션 관리 예제

# User A와 User B의 독립 대화
 
# User A
response1 = runnable_with_history.invoke(
    {"input": "나는 말차를 좋아해"},
    config={"configurable": {"session_id": "userA"}}
)
 
# User B (독립 세션)
response2 = runnable_with_history.invoke(
    {"input": "나는 커피를 좋아해"},
    config={"configurable": {"session_id": "userB"}}
)
 
# User A 재방문 (기존 세션 유지)
response3 = runnable_with_history.invoke(
    {"input": "디저트 추천해줘"},
    config={"configurable": {"session_id": "userA"}}
)
# AI가 "당신이 말차를 좋아한다는 걸 알고 있습니다..."라고 응답

vs. ChatMessageHistory

항목ChatMessageHistoryRunnableWithMessageHistory
메모리 저장수동 (invoke + add_ai_message)자동 (invoke만 호출)
세션 관리없음✅ session_id 기반
다중 사용자별도 구현 필요내장 지원
사용 간편도낮음높음
설정 복잡도낮음중간 (get_session_history 함수)

실제 예제 1: 음악 추천 챗봇

from langchain.memory import ChatMessageHistory
from langchain.runnables.history import RunnableWithMessageHistory
from langchain.prompts import ChatPromptTemplate
from langchain.chat_models import ChatOpenAI
 
# 세션 저장소
session_history = {}
 
def get_session_history(session_id: str):
    if session_id not in session_history:
        session_history[session_id] = ChatMessageHistory()
    return session_history[session_id]
 
# 기본 체인
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 음악 추천 전문가입니다."),
    ("placeholder", "{chat_history}"),
    ("human", "{input}")
])
 
model = ChatOpenAI(model="gpt-4o")
chain = prompt | model
 
# 래퍼 추가
runnable_with_history = RunnableWithMessageHistory(
    runnable=chain,
    get_session_history=get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)
 
# 사용
response = runnable_with_history.invoke(
    {"input": "안녕"},
    config={"configurable": {"session_id": "music_user1"}}
)
print(response)  # "안녕하세요! 어떤 음악을 좋아하세요?"

실제 예제 2: 끝말잇기 게임

# 게임 프롬프트
game_prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 끝말잇기 게임을 진행합니다. 규칙: 이전 단어 마지막 글자로 시작."),
    ("placeholder", "{chat_history}"),
    ("human", "{input}")
])
 
game_chain = game_prompt | model
 
game_runnable = RunnableWithMessageHistory(
    runnable=game_chain,
    get_session_history=get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)
 
# 게임 시작
response1 = game_runnable.invoke(
    {"input": "사과"},
    config={"configurable": {"session_id": "game1"}}
)
# AI: "과자"
 
response2 = game_runnable.invoke(
    {"input": "자동차"},
    config={"configurable": {"session_id": "game1"}}
)
# AI: "차멀미" (이전 "자동차"의 마지막 글자로 시작)

토큰 최적화: Message Summarization 적용

def summarize_messages(chat_history: ChatMessageHistory) -> str:
    """대화 내용을 요약하여 토큰 절감"""
    if len(chat_history.messages) == 0:
        return ""
    
    summary = (summarization_prompt | model).invoke({
        "chat_history": chat_history.messages
    })
    
    # 원본 히스토리 삭제 후 요약으로 교체
    chat_history.clear()
    chat_history.add_ai_message(summary.content)
    return ""
 
# Summarization을 체인에 추가
full_chain = (
    RunnableLambda(summarize_messages(get_session_history(session_id))) |
    runnable_with_history
)

라이프사이클

invoke() 호출
    ↓
config에서 session_id 추출
    ↓
get_session_history(session_id) 호출
    ↓
기존 ChatMessageHistory 로드 (또는 새로 생성)
    ↓
프롬프트에 chat_history + input 삽입
    ↓
체인 실행 (모델 호출)
    ↓
AI 응답 생성
    ↓
ChatMessageHistory에 자동 저장
    ↓
응답 반환

주요 사용 사례

사용 사례예시
챗봇고객 지원, 상담, 추천
게임끝말잇기, 20 질문 게임
교육튜터링, 대화형 강의
개인화사용자별 선호도 기반 응답

통합 개념


관련 소스: word-chain-chatbot (Method 3, 4 예제)