RAG 파이프라인 세부 탐구3

wldbs._.·2025년 8월 4일

AI LLM Milvus python rag 스터디 실습 정리 프로젝트

RAG

목록 보기

10/22

[개념정리] 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))