[개념정리] Streamlit👑 소개 및 활용 가이드

지난 시간에 이어, streamlit을 활용하여 RAG 데모 서비스를 구축한다.
해당 실습을 진행하며 해결한 오류 및 알게된 점을 작성한다.

---

💥 1. Streamlit 해결

🟢 progress bar

streamlit을 활용해 답변을 생성하는 스트리밍 과정과 검색된 청크를 보여주어야 한다.
이때, 처리 시간은 progress bar를 이용하고, 청크 출력에는 청크의 파일명과 앞 200자만 사용한다.

• 프로세스 진행 상황을 나타내는 진행률 바(progress bar)

start_time = time.time()  # ⏱ 처리 시간 측정 시작
my_bar = st.progress(0, text="🚀 준비 중...")  # ✅ 진행률 표시 바 생성
st.write(f"💬 질문: {question}")

...

my_bar.progress(10, text="🧠 질문 임베딩 생성 중...")
question_embeddings = get_embeddings([question])

...

my_bar.progress(40, text="🔍 Milvus 검색 중...")
context_texts = search_milvus(search_vector, selected_collection, top_k=search_limit)

...

my_bar.progress(70, text="💬 LLM 답변 생성 시작...")
response_stream = generate_answer_stream(question, full_context)     

...

my_bar.progress(100, text="✅ 답변 완성!")
st.success("✅ 답변이 성공적으로 생성되었습니다!") 

---

🔴 selectbox - index

selectbox를 사용하여 milvus 벡터 컬렉션 중 하나를 선택하고 싶었다.
다만, 사용자가 A 컬렉션을 선택하여도 임의로 컬렉션이 변경되는 듯 고정되지 않는 값이 발생하는 문제가 매번 발생하였다.

이에 대해서 찾아보니, streamlit은 selectbox에서 index 기반으로 선택한 옵션을 기억한다고 하여, 컬렉션의 순서를 보장하기 위해 sorted()를 사용해야 한다는 것이었다.

• Streamlit의 selectbox는 index 기반 상태 관리를 한다
• Milvus의 컬렉션 리스트 순서가 불안정하면, index와 실제 값 매칭이 꼬일 수 있다.
• sorted()로 리스트 순서를 고정하면, index 기반 상태 관리가 안정적으로 작동한다.

# 사이드바 설정
# 1. 벡터 DB 컬렉션 리스트 조회 및 선택
st.sidebar.title("🔧 Settings")
st.sidebar.subheader("📚 1. Milvus Collections")
try:
    # 🧨 문제: 컬렉션 A 선택 시, (다른 컬렉션이 선택되고) 고정되지 않는 문제
    # 🧵 해결: 컬렉션 리스트를 정렬하여 순서를 보장하고, selectbox의 선택된 컬렉션을 유지
    # 🤷‍♀️ 이유: Streamlit이 index 기반으로 selectbox 선택값을 기억하기 때문에, 순서가 바뀌면 선택도 바뀌는 문제 발생.
    collections = sorted(
        client.list_collections()
    )  # 컬렉션 리스트 알파벳순 정렬 → 순서 보장
    if collections:  # Check if the list is not empty
        selected_collection = st.sidebar.selectbox(
            "📍 Select a collection", collections
        )
        st.sidebar.write(f"You Selected: {selected_collection}")
    else:
        st.sidebar.write("❌ No collections found.")
except Exception as e:
    st.sidebar.error(f"❌ Failed to retrieve collections: {e}")
st.write(
    f"📍 선택된 컬렉션: {selected_collection if 'selected_collection' in locals() else 'None'}"
)
st.sidebar.divider()

---

🙋‍♀️ 2. 코드 분해 및 로깅

작년 인턴의 여파..? 덕분인지(?)
코드를 작성하거나 작성된 코드를 이해하려고 할때
사용된 함수나 인자를 더욱 자세히 바라보는 습관이 생겼다.
해당 변수에는 어떤 값이 담겨있고, 이 함수의 용도는 뭐고, 필수 인자는 무엇이고 선택 인자는 무엇인지 등등..
이번 실습을 진행할 때도 궁금증이 생기면 해결하고 지나가야 하는 성격이 되어
코드 하나하나를 뜯어보며 값을 출력하고 확인하며 실습을 이어나갔다.

🔎 milvus client search

# ✅ 첫 번째 쿼리 결과만 사용 (단일 쿼리이므로 results[0] 안에 top_k개 결과 존재)
hits = results[0]

# ✅ 결과에서 'text' 필드만 추출하여 리스트로 반환
return [hit["entity"]["text"] for hit in hits if hit["entity"].get("text")]

client.search()를 호출하면 결과가 2중 리스트 형태로 온다.

📢 예시 (질문 하나를 검색했을 때):

results = [
    [   # ✅ 첫 번째 쿼리에 대한 검색 결과 (top_k 만큼)
        {"id": 1, "distance": 0.12, "entity": {"text": "첫 번째 문장", "filenm": "a.pdf"}},
        {"id": 2, "distance": 0.20, "entity": {"text": "두 번째 문장", "filenm": "b.pdf"}},
        {"id": 3, "distance": 0.30, "entity": {"text": "세 번째 문장", "filenm": "c.pdf"}}
    ]
]

• results 자체가 바깥 리스트 ➡ 여러 개 질문(query)을 동시에 보낼 때를 대비해서 이렇게 구조화됨.

• 우리는 질문 하나만 보내기 때문에 results[0] 만 꺼내면 됨.

• hits = results[0] ➡ “첫 번째(그리고 유일한) 쿼리의 검색 결과(top_k개)”를 hits 에 담는 것.

📢 이제 hits 안에는 이런 딕셔너리들이 들어있다:

[
    {"id": 1, "distance": 0.12, "entity": {"text": "첫 번째 문장", "filenm": "a.pdf"}},
    {"id": 2, "distance": 0.20, "entity": {"text": "두 번째 문장", "filenm": "b.pdf"}}
]

• 각 hit는 검색된 하나의 결과(row)
  • hit["entity"] → Milvus에 저장된 메타데이터(여기선 text, filenm)
  • hit["entity"]["text"] → 검색된 chunk의 실제 텍스트
• 리스트 컴프리헨션으로:
  • hits 안에 있는 각 hit에서
  • hit["entity"]["text"] 값을 꺼내서 새 리스트에 담음
  • if hit["entity"].get("text") 조건 → 혹시 text 값이 None이거나 없는 경우 제외

---

🔎 question embedding

# ✅ [1단계] 질문 임베딩 생성
my_bar.progress(10, text="🧠 질문 임베딩 생성 중...")

• Streamlit의 progress bar 를 10%로 올리면서 현재 단계 표시
• 사용자에게 “지금 질문을 임베딩(벡터화)하는 단계예요” 라는 메시지를 보여줌
• 즉, RAG 첫 번째 단계(질문 → 벡터 변환)임을 알려주는 UI 피드백

question_embeddings = get_embeddings([question])

• OpenAI Embedding API (또는 지정한 임베딩 모델)를 호출해 사용자가 입력한 질문(question)을 벡터화하는 코드
  
• [question] 으로 리스트 형태로 넣는 이유:
  • OpenAI API는 여러 텍스트를 한 번에 임베딩할 수 있기 때문에 input을 리스트로 받음
  • 질문이 1개여도 리스트로 감싸서 넣는 게 표준 사용법
• 결과:
  • question_embeddings → 리스트 안에 벡터가 들어 있음
  • 예: [[0.123, 0.456, 0.789, ...]]

if question_embeddings:  # ✅ 임베딩이 성공적으로 생성됐는지 확인

• 임베딩이 제대로 생성되었는지 확인
• get_embeddings() 함수에서 API 오류가 나면 None을 리턴하게 되어 있음
• 즉, 임베딩이 없으면(if False) → 이후 검색 단계 진행 안 함

search_vector = question_embeddings[0]  # 첫 번째(유일한) 질문 임베딩 사용

• question_embeddings는 리스트이므로 question_embeddings[0]을 꺼내야 함
• 질문이 여러 개였다면 여러 벡터가 들어있겠지만, 여기서는 질문이 1개니까 첫 번째(그리고 유일한) 벡터 사용
  
• 결과:
  • search_vector → Milvus에서 유사도 검색에 사용할 질문 벡터 (FLOAT 리스트)

---

🔎 streaming response generation

OpenAI에서 스트리밍으로 답변을 받아서 화면에 실시간으로 보여주는 핵심 로직
1️⃣ OpenAI API가 보내는 토큰을 하나씩 받음
2️⃣ 받을 때마다 partial_answer에 추가 → 진행률(progress) 업데이트 → UI에 출력
3️⃣ 스트리밍이 끝나면 progress bar 100%, 완료 메시지 표시

• 👉 “OpenAI가 스트리밍으로 보내는 답변을 토큰 단위로 받으면서 진행률을 올리고, 최종적으로 답변 생성을 완료시키는 단계”

# ✅ [4단계] 스트리밍 답변 생성
for chunk in response_stream:

• response_stream은 generate_answer_stream()에서 받아온 OpenAI API 스트리밍 응답 객체
• 이 객체는 LLM이 생성하는 답변을 토큰(token) 단위로 하나씩 보내줌.
• for chunk in response_stream: → 토큰이 도착할 때마다 chunk가 하나씩 반복문에 들어옴.

if chunk.choices[0].delta.content:  # ✅ 토큰이 있으면

chunk 내부
{
  "id": "chatcmpl-abc123",
  "object": "chat.completion.chunk",
  "created": 1730000000,
  "model": "gpt-4o-mini",
  "choices": [
    {
      "index": 0,
      "delta": {
        "role": "assistant",
        "content": "안녕하세요"
      },
      "finish_reason": null
    }
  ]
}
# chunk.choices[0].delta → 새로 도착한 토큰(조각) 이 들어있는 부분
# chunk.choices[0].delta.content → 실제로 화면에 출력할 텍스트 토큰

• chunk 안에는 여러 메타데이터가 들어있는데, 그중 choices[0].delta.content에 실제 텍스트 토큰이 담겨 있다
• 어떤 chunk는 “역할(role) 정보”만 오거나, “대화 종료 신호”만 오기도 하기 때문에, delta.content가 있는 경우만 처리.
  

token = chunk.choices[0].delta.content
partial_answer += token
token_count += 1

• token: 새로 도착한 한 조각의 텍스트.
• partial_answer에 계속 이어 붙임 → 답변이 점점 길어짐.
• token_count를 1씩 증가 → 몇 번째 토큰인지 카운트.

# ✅ 진행률 70→100% 점진적 증가
progress_ratio = 70 + min(
    int((token_count / total_tokens_estimate) * 30),
    30,
)

• Progress bar는 LLM 답변 생성 시작 시 70%부터 시작.
• 토큰이 올 때마다 조금씩 진행률을 늘림.
• total_tokens_estimate (예: 120)과 현재 받은 token_count를 이용해서 30% 구간(70→100)을 채움.
• min(..., 30)을 쓴 이유 → 100% 이상 안 넘어가도록 제한.

my_bar.progress(
    progress_ratio,
    text=f"💬 답변 생성 중... ({progress_ratio}%)",
)

• Progress bar를 새 값으로 업데이트.
• 70%~100% 사이에서 점점 차오르는 효과.

# ✅ 실시간 답변 표시 (placeholder 갱신)
# answer_box.markdown(partial_answer)

• partial_answer를 Streamlit placeholder(answer_box)에 표시해서 실시간으로 화면에 답변 출력.
• 현재는 주석 처리돼 있지만, 이 줄을 살리면 토큰이 도착할 때마다 UI가 업데이트됨.
• ChatGPT처럼 “답변이 줄줄 나오는 느낌”을 구현할 수 있음.

# ✅ 스트리밍 완료 후 상태 업데이트
my_bar.progress(100, text="✅ 답변 완성!")
st.success("✅ 답변이 성공적으로 생성되었습니다!")

• 스트리밍이 끝났을 때(= OpenAI가 답변을 다 보냈을 때) 실행.
• Progress bar를 100%로 바꾸고, 상태 메시지를 “✅ 답변 완성!”으로 업데이트.
• st.success()로 화면에 “답변이 성공적으로 생성되었습니다!”라는 확인 메시지를 띄움.

---

🔎 chunk vs token

🔵 1️⃣ 청크(chunk)

• RAG(Retrieval-Augmented Generation), 데이터 전처리, 검색 시스템에서 주로 쓰는 개념.
• 큰 문서를 잘라서 작은 단위로 나눈 조각을 말함.
• 예: 50페이지 PDF → 여러 문단 단위로 쪼개면 각 문단이 “청크”

📌 왜 청크로 나누나?

• LLM(예: GPT)은 한 번에 처리할 수 있는 입력 길이(토큰 수)가 제한됨.
• 검색할 때도 짧은 단위로 저장하면 더 정확한 검색 가능.
• 그래서 문서를 chunk_size (예: 500자, 1000자) 단위로 쪼개고 벡터 DB(Milvus 등)에 넣음.

📊 예시

• 원문:

  인공지능(AI)은 컴퓨터가 사람처럼 학습하고 추론하는 기술이다.
  최근 LLM은 RAG와 결합해 더 정확한 답변을 생성한다.
  

• 30자씩 나눈 chunk: 1️⃣ “인공지능(AI)은 컴퓨터가 사람처럼 학습하고…” 2️⃣ “추론하는 기술이다. 최근 LLM은 RAG와…” 3️⃣ “결합해 더 정확한 답변을 생성한다.”
  

🟢 2️⃣ 토큰(token)

• *LLM(언어 모델)**이 텍스트를 처리하는 가장 작은 단위.
• 단어(word)보다는 더 잘게 쪼갠 문자/어절 단위.
• OpenAI, Claude, LLaMA 같은 모델은 텍스트를 “토큰” 단위로 읽고 생성.

📌 왜 토큰 단위가 중요한가?

• LLM은 입력(prompt)과 출력(answer)의 총 토큰 수에 따라 비용과 속도가 달라짐.
• “한글 1자 = 1토큰” 이런 건 아니고, 보통 단어/어절 기준으로 나뉘지만 언어마다 다름.
• 예: OpenAI의 GPT 모델에서 “사과를 먹었다” → [ "사", "과", "를", " 먹", "었다" ] → 5 토큰
  

📊 예시

• 문장: “I love pizza”
• GPT 토크나이저로 나누면:
  • “I” → 1토큰
  • “ love” → 1토큰
  • “ pizza” → 1토큰 총 3 토큰
    

🔄 비유로 구분하기

• 청크는 검색(RAG)을 위한 문서 단위 (RAG, 검색, 데이터 전처리에서 사용)
• 토큰은 LLM이 텍스트를 이해하고 출력할 때 쓰는 언어 단위 (LLM의 입력/출력 단위)

👉 즉, RAG에서는 “청크”를 나눠 Milvus에 저장하고, LLM은 그 “청크”를 불러와 “토큰” 단위로 읽고 답변을 만들어냄.

---

🔎 chunk

chunk의 내부를 살펴보자.

# ✅ [4단계] 스트리밍 답변 생성
for (chunk in response_stream):  
# chunk = OpenAI가 보내는 스트리밍 응답 조각
# 토큰이 도착할 때마다 chunk가 하나씩 반복문에 들어옴.
         **print(chunk)**
         if chunk.choices[0].delta.content:  # → 토큰이 있으면
                   # choices[0].delta.content*에 실제 텍스트 토큰이 담겨 있음
                   token = chunk.choices[0].delta.content  # token: 새로 도착한 한 조각의 텍스트.
                   partial_answer += token  # partial_answer에 계속 이어 붙임 → 답변이 점점 길어짐.
                   token_count += 1  # token_count를 1씩 증가 → 몇 번째 토큰인지 카운트.

ChatCompletionChunk(
id='chatcmpl-By------...', 
choices=
	[Choice(delta=ChoiceDelta(
		content='', function_call=None, 
		refusal=None, role='assistant', 
		tool_calls=None), 
		finish_reason=None, index=0, logprobs=None)], 
created=1753------, 
model='gpt-4.1-mini-2025-04-14', 
object='chat.completion.chunk', 
service_tier='default', 
system_fingerprint='fp_0e---...', usage=None)

---

🔎 search results

벡터 서치 과정을 알아보자.

results = client.search(
        collection_name=collection_name,  # 검색 대상 컬렉션 이름
        data=[query_vec],  # 입력 쿼리 벡터 (리스트 형태로 전달)
        anns_field="dense_vector",  # 임베딩한 벡터가 저장된 필드명
        search_params=search_params,  # 추가 검색 파라미터 필요 시 사용
        limit=top_k,  # 검색할 유사 문서 개수
        output_fields=["text", "filenm"],  # 결과로 가져올 메타데이터 필드
    )

    print(results)
    
    # ✅ 첫 번째 쿼리 결과만 사용 (단일 쿼리이므로 results[0] 안에 top_k개 결과 존재)
    hits = results[0]  # results[0] → “질문 하나에 대한 top_k 검색 결과 전체”

    # ✅ text가 없는 결과는 건너뜀 & 결과에서 'text' 필드만 추출하여 리스트로 반환
    return [hit["entity"]["text"] for hit in hits if hit["entity"].get("text")]

data: [[
{'document_id': 'NABO_경제동향_(제45호).pdf_35_0', 
'distance': 0.5186111927032471, 
'entity': 
{'text': '2024년 10월호\n2 고용\n9월 취업자 수는 전년동월대비 14.4만명 증가하였고 실업률은 전년동월대비 0.2%p 하락\n■ 9월 \u200c 취업자 수는 2,884.2만명으로 전년동월대비 14.4만명 증가, 계절조정 취업자 수는 전월대비 5.5만명 증가\n• 취업자 수(전년동월대비, 만명): (’24.7월)17.2 → (8월)12.3 → (9월)14.4\n• 계절조정 취업자 수(전월대비, 만명): (’24.7월)1.9 → (8월)1.4 → (9월)5.5\n| 그림 14\u2009| 전체 취업자 수 추이\n(만명) ■\u2002 취업자수\u2003 \u2002증감(우축) (전년동월대비, 만명)\n3,000 120\n2,884.2 100\n2,900 2,884.2 100\n2,900\n80\n2,800\n60\n2,700 40\n2,600 14.4 20\n20\n2,500 0\n2022.1 4 7 10 2023.1 4 7 10 2024.1 4 7 9\n자료: 통계청, 「경제활동인구조사」\n■ 경제활동참가율(15세 \u200c 이상 인구 중 경제활동인구 비중)은 64.6%로 전년동월과 같은 수준\n• 경제활동참가율(%): (’24.7월)64.9 → (8', 
'filenm': 'NABO_경제동향_(제45호).pdf'}},
 
 {'document_id': 'NABO_경제동향_(제45호).pdf_36_1', 
 'distance': 0.47488752007484436, 
 'entity': 
 {'text': '� 취업자 수(전년동월대비, 만명): (’24.7월)-8.1 → (8월)-8.4 → (9월)-10.0\n9월 「연령별 취업자 수」는 30대, 50대 및 60세 이상의 증가세 유지되는 가운데 15~29세\n및 40대는 감소세 지속\n■ 고령층(60세 \u200c 이상) 취업자 수는 전년동월대비 27.2만명 증가하여 증가세 유지\n• 고령층 취업자 수(전년동월대비, 만명): (’24.7월)27.8 → (8월)23.1 → (9월)27.2\n■ 15~29세 \u200c 취업자 수는 전년동월대비 16.8만명 감소하여 감소세 지속\n• 15~29세 취업자 수(전년동월대비, 만명): (’24.7월)-14.9 → (8월)-14.2 → (9월)-16.8\n■ 30대와 \u200c 50대 취업자 수는 증가, 40대 취업자 수는 하락\n• 30대 취업자 수(전년동월대비, 만명): (’24.7월)11.0 → (8월)9.9 → (9월)7.7\n• 40대 취업자 수(전년동월대비, 만명): (’24.7월)-9.1 → (8월)-6.8 → (9월)-6.2\n• 50대 취업자 수(전년동월대비, 만명): (’24.7월)2.3 → (', 
 'filenm': 'NABO_경제동향_(제45호).pdf'}}, 
 
 {'document_id': 'NABO_경제동향_(제45호).pdf_38_2', 
 'distance': 0.46165961027145386, 
 'entity': 
 {'text': '4 7 10 ’22.1 4 7 10 ’23.1 4 7 10 ’24.1 4 7 9\n자료: 통계청, 「경제활동인구조사」\n| 그림 22\u2009| 빈 일자리 수 변화율 추이\n(전년동월대비, %) \u2002 300인 이상\u2003 \u2002 300인 미만\u2003 \u2002 전체\n30\n20\n10\n0\n0\n-1.4\n-10\n-18.7\n-20\n-20 -19.4\n-19.4\n-30\n-40\n-50\n-60\n’21.1 4 7 10 ’24.1 4 7 8\n자료: 고용노동통계, 「사업체노동력조사」\n36', 
 'filenm': 'NABO_경제동향_(제45호).pdf'}}]]

---

🔎 embedding response

모델이 텍스트를 임베딩하는 과정을 살펴보자.

response = client_openai.embeddings.create(model=model, input=texts)

# ✅ API 응답(response.data) 안에 각 텍스트의 embedding 벡터가 들어 있음
#   - item.embedding: 한 텍스트의 임베딩 (float 값들의 리스트)
print(response)
return [item.embedding for item in response.data]

except Exception as e:
    print(f"❌ 임베딩 생성 오류: {e}")
    return None

CreateEmbeddingResponse(data=[Embedding(embedding=[......,0.002619395265355706], 
index=0, 
object='embedding')], 
model='text-embedding-3-small', 
object='list', 
usage=Usage(prompt_tokens=11, total_tokens=11))