실습 3 — 스트리밍 답과 키 읽기

from pathlib import Path

APP_CODE = r'''
import os

import streamlit as st
from dotenv import find_dotenv, load_dotenv
from langchain_core.language_models.fake_chat_models import FakeListChatModel
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder


def get_api_key():
    """클라우드에서는 Secrets 칸, 내 PC 에서는 .env 에서 키를 읽는다. (키, 읽은 곳) 을 돌려준다"""
    try:
        return st.secrets["OPENAI_API_KEY"], "Secrets"
    except (FileNotFoundError, KeyError):
        # secrets.toml 이 없거나 그 안에 키가 없으면 2차시처럼 .env 를 읽는다
        load_dotenv(find_dotenv(usecwd=True))
        return os.getenv("OPENAI_API_KEY"), ".env"


@st.cache_resource
def make_stream_chain():
    """스트리밍을 보여 줄 대화 체인(가짜 모델)"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "너는 구름다리여행의 친절한 여행 비서다."),
        MessagesPlaceholder("history"),
        ("human", "{question}"),
    ])
    model = FakeListChatModel(responses=["제주도 2박 3일이면 첫날은 동쪽 해안, 둘째 날은 오름 산책을 추천합니다."])
    return prompt | model | StrOutputParser()


st.title("구름다리여행 여행 비서")
key, where = get_api_key()
with st.sidebar:
    st.header("설정")
    # 키 자체는 찍지 않는다. 어디서 읽었는지와 찾았는지만 보인다
    st.caption(f"키를 읽은 곳: {where}")
    st.caption(f"키를 찾았나: {bool(key)}")

if "messages" not in st.session_state:
    st.session_state.messages = []
for m in st.session_state.messages:
    with st.chat_message(m["role"]):
        st.markdown(m["content"])

if question := st.chat_input("여행에 대해 물어보세요"):
    with st.chat_message("user"):
        st.markdown(question)
    chain = make_stream_chain()
    with st.chat_message("assistant"):
        # chain.stream 이 내놓는 조각을 write_stream 이 말풍선에 이어 쓰고, 다 쓴 글을 돌려준다
        reply = st.write_stream(chain.stream({"history": st.session_state.messages, "question": question}))
    st.session_state.messages.extend([
        {"role": "user", "content": question},
        {"role": "assistant", "content": reply},
    ])
'''

Path("out_ch07_stream.py").write_text(APP_CODE.lstrip(), encoding="utf-8")
print("저장: out_ch07_stream.py,", len(APP_CODE.strip().splitlines()), "줄")

import streamlit as st
from streamlit.testing.v1 import AppTest

st.cache_resource.clear()

# secrets.toml 이 없으니 get_api_key 는 .env 를 읽는다
at = AppTest.from_file("out_ch07_stream.py").run()
for c in at.sidebar.caption:
    print(c.value)

cloud = AppTest.from_file("out_ch07_stream.py")
# Secrets 칸에 시험용 값을 넣는다. 진짜 키가 아니고, 이 실습은 모델에 키를 쓰지 않는다
cloud.secrets["OPENAI_API_KEY"] = "시험용-가짜-값"
cloud.run()
for c in cloud.sidebar.caption:
    print(c.value)

at.chat_input[0].set_value("제주도 2박 3일 일정 짜 주세요").run()

reply = at.chat_message[-1].markdown[0].value
print("말풍선:", reply)
saved = at.session_state["messages"][-1]
print("기록의 마지막:", saved["role"], "/", type(saved["content"]).__name__, "/ 말풍선과 같은가:", saved["content"] == reply)