실습 3 — 가짜 모델로 세션별 기록 확인하기

import warnings

from langchain_core.chat_history import InMemoryChatMessageHistory

# InMemoryChatMessageHistory 와 RunnableWithMessageHistory 는 "사용 중단 예정" 경고를 띄운다
# ("손님마다 기록 상자 나누기" 절의 주의할 점). 경고를 읽었으면 이 줄로 숨긴다
warnings.filterwarnings("ignore", category=DeprecationWarning)

store = {}   # {세션 id: 기록 상자}


def get_session_history(session_id):
    """세션 id 로 기록 상자를 찾는다. 처음 보는 손님이면 빈 상자를 새로 만든다"""
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]


# 같은 id 로 두 번 부르면 같은 상자가 돌아온다
print(get_session_history("kim") is get_session_history("kim"))
print("만든 상자:", list(store))

from langchain_core.language_models.fake_chat_models import FakeListChatModel
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableLambda

prompt = ChatPromptTemplate.from_messages([
    ("system", "너는 구름다리여행의 여행 비서다."),
    MessagesPlaceholder("history"),     # 세션의 기록이 들어갈 자리
    ("human", "{question}"),
])


def peek(prompt_value):
    """프롬프트와 모델 사이에서, 모델이 받을 메시지의 수와 역할을 찍고 그대로 넘긴다"""
    msgs = prompt_value.to_messages()
    print(f"  모델이 받는 메시지 {len(msgs)}개:", [m.type for m in msgs])
    return prompt_value


# 가짜 모델은 세션과 상관없이 목록의 답을 차례로 돌려준다
model = FakeListChatModel(responses=[
    "좋습니다. 며칠 일정인가요?",
    "어디로 가고 싶으신가요?",
    "2박 3일이면 동쪽 해안 일정을 추천합니다.",
    "부산은 1박 2일도 충분합니다.",
])
# RunnableLambda: 보통 함수를 체인의 부품으로 만든다. lambda 는 "이름 없는 작은 함수"를 부르는 말
chain = prompt | RunnableLambda(peek) | model
print(type(chain).__name__)

from langchain_core.runnables.history import RunnableWithMessageHistory

bot = RunnableWithMessageHistory(
    chain,                                # 감쌀 체인
    get_session_history,                  # 세션 id → 기록 상자
    input_messages_key="question",        # 입력 사전에서 새 질문이 든 열쇠
    history_messages_key="history",       # 프롬프트의 기록 자리 이름
)

talks = [("kim", "제주도에 가요."), ("lee", "여행 가고 싶어요."), ("kim", "2박 3일이요."), ("lee", "부산이요.")]
for session_id, question in talks:
    print(f"[{session_id}] {question}")
    # configurable: 부를 때마다 바꿀 수 있는 설정. 여기에 세션 id 를 넣는다
    reply = bot.invoke({"question": question}, config={"configurable": {"session_id": session_id}})
    print("  답:", reply.content)

import json
from pathlib import Path

for session_id, history in store.items():
    print(session_id, "기록", len(history.messages), "개:", [m.content for m in history.messages if m.type == "human"])

# 세션마다 {"role", "content"} 사전 목록으로 바꿔 파일로 남긴다
saved = {sid: [{"role": m.type, "content": m.content} for m in h.messages] for sid, h in store.items()}
Path("out_ch04_sessions.json").write_text(json.dumps(saved, ensure_ascii=False, indent=2), encoding="utf-8")
print("저장: out_ch04_sessions.json")