Lab 05: from-scratch 미니 RAG

• 문서를 청크로 나누고, Titan 임베딩으로 각 청크를 벡터로 변환하여 인덱스를 직접 구축합니다
• 코사인 유사도를 코드로 구현하여 질문과 가장 비슷한 청크를 찾는 벡터 검색을 만듭니다
• 검색된 청크를 ‘참고 문서’로 묶어 Converse API로 답을 생성하는 RAG 전체 흐름을 완성합니다
• “문서에 근거해서만 답하라”는 시스템 프롬프트로 환각(hallucination)을 억제하는 방법을 익힙니다
• top_k와 청킹 방식을 바꿔 보며 RAG가 어디서 무너지는지 직접 관찰합니다

RAG(Retrieval-Augmented Generation)는 거창해 보이지만, 실제로는 네 개의 단순한 단계로 이루어집니다. 콘솔 KB가 자동으로 처리하던 이 네 단계를, 이번 Lab에서는 각각 함수 하나로 만듭니다.

graph TD
  A["1. 청킹<br/>문서를 조각으로"] --> B["2. 임베딩<br/>각 청크를 벡터로"]
  B --> C["3. 인덱스 구축<br/>(청크, 벡터) 저장"]
  C --> D["4. 벡터 검색<br/>질문과 가까운 청크 찾기"]
  D --> E["5. 생성<br/>찾은 청크를 근거로 답변"]

핵심 한 문장으로 정리하면 이렇습니다. RAG = 질문과 비슷한 내 문서 조각을 찾아(retrieve), 그걸 근거로 모델이 답을 쓰게(generate) 하는 것. 검색이 절반, 생성이 절반입니다. 검색이 틀리면 답도 틀린다(garbage in, garbage out)는 점을 기억하며 진행하세요.

사용할 데이터 — 주주서한

이 Lab은 아마존 주주서한에서 발췌·정리한 문서(shareholder_letters.txt)를 지식 소스로 씁니다. 빈 줄(\n\n)로 구분된 10개의 문단으로 되어 있고, 각 문단이 하나의 주제를 담고 있습니다. 예를 들면 이런 내용들입니다.

• 아마존 ‘Day 1’ 철학 — 창업 첫날처럼 고객에게 집착하고 빠르게 실험하는 상태
• AWS의 탄생 — 내부 인프라 문제를 해결하다 시작된, 큰 씨앗에서 자란 사업
• 의사결정의 속도 — 되돌릴 수 없는 Type 1 결정과 되돌릴 수 있는 Type 2 결정

문단 하나하나가 의미적으로 잘 분리되어 있어서, “빈 줄 기준으로 자른다”는 가장 단순한 청킹 전략이 잘 먹힙니다. 이건 우연이 아니라 의도된 것입니다 — 데이터가 깔끔할수록 RAG가 잘 동작한다는 사실을 먼저 체감한 뒤, 뒤에서 일부러 무너뜨려 볼 것입니다.

공통 설정 — config.py (넥스트클라우드 제공 블랙박스)

config.py는 넥스트클라우드가 제공하는 모듈입니다. 학생은 수정하지 않고 import만 합니다. 이 Lab에서 쓰는 값은 두 가지입니다.

from config import get_runtime, EMBED_MODEL, MODEL_HAIKU

모델을 부를 때는 bedrock-runtime 클라이언트를 씁니다. 이전 Lab에서 관리형 KB를 호출할 때 쓰던 bedrock-agent-runtime과는 다른 클라이언트라는 점에 주의하세요. KB라는 관리형 서비스를 거치지 않고 모델을 직접 부르기 때문입니다.

챗봇 UI는 챗봇 과정의 모듈을 그대로 재사용

답변을 화면에 보여 주는 대화 UI는 챗봇 과정에서 만든 chat_ui 모듈을 그대로 가져다 씁니다. 새로 UI를 만들 필요가 없습니다. 우리가 할 일은 “질문을 받아 답을 돌려주는 함수” 하나를 구현해서 chat_ui.run()에 넘기는 것뿐입니다. 콘솔 UI 기반이라 별도의 웹 화면을 만들 필요가 없습니다.

RAG의 첫 단추는 긴 문서를 검색 단위로 자르는 청킹(chunking)입니다. 왜 자를까요? 문서 전체를 통째로 모델에 넣으면 토큰 비용이 커지고, 정작 답에 필요한 한 문단이 수많은 무관한 내용에 묻혀 검색 정확도가 떨어집니다. 그래서 의미가 끊기지 않는 적당한 크기로 잘라, 질문에 맞는 조각만 골라 쓰는 것입니다.

이번 데이터는 문단 경계가 곧 의미 경계이므로, 빈 줄(\n\n) 기준으로 자르는 가장 단순한 방법을 씁니다.

def load_chunks(path: str) -> list[str]:
    """빈 줄(문단 경계) 기준으로 문서를 청크로 나눈다."""
    text = open(path, encoding="utf-8").read()
    return [c.strip() for c in text.split("\n\n") if c.strip()]

• open(path, ...).read() — 문서 전체를 하나의 문자열로 읽습니다.
• text.split("\n\n") — 빈 줄(연속된 두 개의 줄바꿈)을 기준으로 문단별로 쪼갭니다.
• c.strip() — 각 조각의 앞뒤 공백을 정리합니다.
• if c.strip() — 빈 조각(내용 없는 줄)은 버립니다.

이 함수가 돌려주는 것은 10개의 문자열이 든 리스트입니다. 콘솔 KB의 “Fixed-size 청킹”이 하던 일을, 여기서는 데이터 특성에 맞춰 더 단순하게 한 셈입니다. 청킹 전략은 RAG 품질을 좌우하는 핵심 변수입니다 — 뒤의 한계 관찰에서 이 부분을 바꿔 보며 그 영향을 직접 확인합니다.

이제 각 청크를 벡터(숫자 배열)로 바꿉니다. 임베딩이란 “비슷한 뜻을 가진 텍스트는 서로 가까운 벡터가 되도록” 텍스트를 좌표 공간의 한 점으로 옮기는 것입니다. Titan Text Embeddings V2는 텍스트 하나를 1024개의 실수로 이루어진 벡터로 변환합니다. 이렇게 해 두면 “의미가 비슷한가”를 “벡터가 가까운가”라는 계산 문제로 바꿀 수 있습니다.

def embed(text: str) -> list[float]:
    """Titan 임베딩으로 텍스트를 벡터로 변환한다."""
    body = {"inputText": text}
    resp = rt.invoke_model(modelId=EMBED_MODEL, body=json.dumps(body))
    return json.loads(resp["body"].read())["embedding"]

• body = {"inputText": text} — Titan 임베딩 모델이 요구하는 요청 형식입니다. 여기서 챗봇 과정의 그 불편함이 다시 등장합니다. 챗봇 과정에서 Claude를 invoke_model로 부를 때는 messages, anthropic_version 같은 Anthropic 전용 body를 만들어야 했죠. Titan은 또 다른 형식(inputText)을 요구합니다. 저수준 invoke_model은 이렇게 모델마다 body 형식이 제각각이라는 게 핵심 불편함입니다. (Converse API가 이 불편을 통일해 주지만, 임베딩 모델은 Converse를 지원하지 않아 여기서는 invoke_model을 씁니다.)
• rt.invoke_model(modelId=EMBED_MODEL, body=json.dumps(body)) — body를 JSON 문자열로 직렬화해서 호출합니다.
• json.loads(resp["body"].read())["embedding"] — 응답 body도 스트림이라 .read()로 읽고, JSON으로 파싱한 뒤 embedding 키에서 1024차원 벡터를 꺼냅니다. 이 “응답을 직접 파싱해야 하는” 번거로움 역시 저수준 API의 특징입니다.

임베딩 모델을 부를 때마다 네트워크 호출이 한 번씩 일어납니다. 청크가 10개면 인덱스를 만들 때 Titan을 10번 부르고, 검색할 때 질문을 임베딩하느라 한 번 더 부릅니다. 소규모라 수 초면 끝나지만, “임베딩에는 비용과 시간이 든다”는 감각을 가져 두세요.

벡터 두 개가 “얼마나 비슷한가”를 재는 표준 척도가 코사인 유사도(cosine similarity)입니다. 두 벡터가 이루는 각도의 코사인 값으로, 방향이 같을수록(=의미가 비슷할수록) 1에 가깝고, 무관할수록 0에 가깝습니다. 이론 시간에 “검색 = 거리 계산”이라고 했던 그 거리가, 여기서는 코드 몇 줄로 구현됩니다.

def cosine(a: list[float], b: list[float]) -> float:
    """두 벡터의 코사인 유사도 (1에 가까울수록 비슷)."""
    dot = sum(x * y for x, y in zip(a, b))
    na = math.sqrt(sum(x * x for x in a))
    nb = math.sqrt(sum(y * y for y in b))
    return dot / (na * nb)

• dot — 두 벡터의 내적(대응하는 원소끼리 곱해서 모두 더함). 두 벡터가 같은 방향을 가리킬수록 커집니다.
• na, nb — 각 벡터의 크기(norm). 원소를 제곱해 더한 뒤 제곱근을 취합니다.
• dot / (na * nb) — 내적을 두 크기의 곱으로 나누면 벡터의 길이에 무관한, 순수한 “방향의 비슷함”만 남습니다.

이 한 함수가 벡터 검색 엔진의 심장입니다. ChromaDB나 S3 Vectors 같은 벡터 DB도 내부에서는 결국 이런 유사도 계산을 (훨씬 빠르게, 대규모로) 하고 있을 뿐입니다. 우리는 청크가 10개뿐이라 전부 일일이 계산해도 충분합니다.

인덱스 구축

이제 1~2단계를 합쳐, 모든 청크를 임베딩해 (청크, 벡터) 쌍의 리스트를 만듭니다. 이 리스트가 우리의 인덱스입니다 — 콘솔이 S3 Vectors에 저장하던 바로 그것의 손바닥만 한 버전입니다.

INDEX: list[tuple[str, list[float]]] = []

def build_index() -> int:
    """모든 청크를 임베딩해 INDEX 를 채운다. 청크 수를 돌려준다."""
    global INDEX
    INDEX = [(c, embed(c)) for c in load_chunks(DATA_PATH)]
    return len(INDEX)

load_chunks()로 자른 각 청크 c에 대해 embed(c)를 호출해 벡터를 얻고, (c, 벡터) 튜플로 묶어 INDEX에 저장합니다. 이 리스트 컴프리헨션 한 줄이 “문서를 잘라 임베딩해 벡터 스토어에 적재한다”는 콘솔 Sync 작업 전체에 대응합니다.

벡터 검색

검색은 단순합니다. 질문도 똑같이 임베딩한 뒤, 인덱스의 모든 청크 벡터와 코사인 유사도를 재서, 점수가 높은 순으로 top_k개를 돌려줍니다.

def search(query: str, top_k: int = 3) -> list[tuple[float, str]]:
    """질문과 가장 비슷한 청크 top_k 개를 (점수, 청크)로 돌려준다."""
    qv = embed(query)
    scored = [(cosine(qv, v), c) for c, v in INDEX]
    scored.sort(key=lambda x: x[0], reverse=True)
    return scored[:top_k]

• qv = embed(query) — 질문을 청크와 같은 임베딩 모델로 벡터화합니다. 같은 공간에 놓아야 거리를 잴 수 있습니다.
• scored = [(cosine(qv, v), c) for c, v in INDEX] — 인덱스의 모든 청크에 대해 (유사도 점수, 청크) 쌍을 만듭니다.
• scored.sort(..., reverse=True) — 점수 높은 순(가장 비슷한 순)으로 정렬합니다.
• return scored[:top_k] — 상위 top_k개만 잘라 돌려줍니다. 기본값은 3입니다.

검색 정확도 확인

인덱스를 만든 뒤 실제로 검색이 제대로 되는지 확인해 봅니다. 예를 들어 “AWS는 어떻게 시작됐어?”라고 물으면, “AWS의 탄생” 청크가 가장 높은 점수로 top에 잡혀야 합니다. 마찬가지로 “아마존이 말하는 Day 1이 뭐야?”라고 물으면 ‘Day 1’ 철학 청크가 1위로 올라와야 합니다. 점수와 함께 어떤 청크가 잡히는지 눈으로 확인하세요 — 이게 콘솔이 벡터 검색으로 하던 일의 실체입니다.

이제 마지막 부품입니다. 검색해서 찾은 청크들을 ‘참고 문서’로 묶어 모델에게 주고, “이 문서에 근거해서만 답하라”고 지시합니다. 이 단계가 RAG의 ‘G(Generation)‘이자, 환각을 억제하는 핵심 장치입니다.

def rag_answer(history: list[dict]) -> str:
    """검색 → 컨텍스트 조립 → 생성. chat_ui 의 응답 함수로 쓰인다."""
    query = history[-1]["text"]
    hits = search(query, top_k=3)
    context = "\n\n".join(f"[{i + 1}] {c}" for i, (s, c) in enumerate(hits))

    system = (
        "너는 주어진 '참고 문서'에 근거해서만 답하는 도우미야. "
        "문서에 없는 내용은 지어내지 말고 '문서에 없다'고 솔직히 말해. "
        "한국어로 간결하게 답해."
    )
    user = f"참고 문서:\n{context}\n\n질문: {query}"

    resp = rt.converse(
        modelId=MODEL_HAIKU,
        system=[{"text": system}],
        messages=[{"role": "user", "content": [{"text": user}]}],
        inferenceConfig={"maxTokens": 512, "temperature": 0.2},
    )
    return resp["output"]["message"]["content"][0]["text"]

• query = history[-1]["text"] — chat_ui가 넘겨준 대화 기록에서 가장 최근(마지막) 사용자 질문을 꺼냅니다.
• hits = search(query, top_k=3) — 4단계의 검색을 호출해 질문과 가장 비슷한 청크 3개를 얻습니다.
• context = "\n\n".join(...) — 찾은 청크들을 [1], [2], [3]처럼 번호를 붙여 하나의 문자열로 조립합니다. 이렇게 검색 결과를 프롬프트에 끼워 넣는 것이 RAG의 ‘Augmented(증강)‘입니다.
• system 프롬프트 — “참고 문서에 근거해서만 답하고, 없으면 ‘문서에 없다’고 말하라”는 지시가 환각 억제의 핵심입니다. 모델이 아는 척 지어내지 않도록 못 박는 가드레일 역할을 합니다.
• user 프롬프트 — 참고 문서와 질문을 함께 담습니다. 모델은 이 둘을 보고 답을 씁니다.
• rt.converse(...) — Converse API로 답을 생성합니다. temperature: 0.2로 낮춰 답이 들쭉날쭉하지 않고 문서에 충실하도록 합니다.
• 반환값 resp["output"]["message"]["content"][0]["text"] — Converse 응답 구조에서 모델이 쓴 텍스트를 꺼냅니다. (임베딩의 invoke_model과 달리 Converse는 응답 구조가 통일되어 있어 파싱이 한결 일관적입니다.)

임베딩(invoke_model + inputText)과 생성(converse)의 호출 방식이 다르다는 점을 한 번 더 확인하세요. 임베딩 모델은 저수준 API로, 생성 모델은 고수준 Converse API로 부릅니다. 같은 Bedrock이지만 모델 종류에 따라 인터페이스가 다릅니다.

모든 부품을 조립했으니 실행합니다. __main__ 블록은 인덱스를 먼저 구축하고, rag_answer 함수를 chat_ui.run()에 넘겨 대화를 시작합니다.

if __name__ == "__main__":
    from chat_ui import run
    n = build_index()
    print(f"📚 인덱스 구축 완료: {n}개 청크")
    print("   예) '아마존이 말하는 Day 1 이 뭐야?' / 'AWS 는 어떻게 시작됐어?'")
    run(rag_answer, single_turn=True, title="Lab5 · from-scratch 미니 RAG")

• build_index() — 실행 즉시 모든 청크를 임베딩해 인덱스를 채웁니다. Titan을 청크 수만큼 호출하므로 잠깐 걸립니다(소규모라 수 초, 정상입니다).
• run(rag_answer, single_turn=True, ...) — 챗봇 과정의 챗봇 UI에 우리의 rag_answer를 응답 함수로 꽂습니다. single_turn=True라 매 질문이 독립적으로 검색→생성됩니다.

터미널에서 실행하세요.

# RAG 실습 폴더로 이동 (clone 받은 레포)
cd ~/nxt-workshop-code/bedrock-rag/starter

python lab5_rag_fromscratch.py

📚 인덱스 구축 완료: 10개 청크 메시지가 뜨면 인덱스가 정상 구축된 것입니다. 이제 질문을 입력해 봅니다.

먼저 데이터에 명확히 들어 있는 질문을 던져 보세요.

• “아마존이 말하는 Day 1이 뭐야?” → ‘Day 1’ 철학 청크를 근거로 한 답이 나와야 합니다
• “AWS는 어떻게 시작됐어?” → ‘AWS의 탄생’ 청크를 근거로 답해야 합니다

RAG가 잘 동작하는 것을 확인했으니, 이제 일부러 무너뜨려 한계를 관찰합니다. 이게 이 Lab에서 가장 중요한 학습 포인트입니다. RAG는 만능이 아니며, 어디서 약해지는지 알아야 제대로 쓸 수 있습니다.

관찰 1: 문서에 없는 질문 — 환각 억제 작동 확인

데이터에 전혀 없는 정보를 물어 봅니다.

• “2024년 4분기 순이익이 얼마야?”

주주서한 데이터에는 이런 구체적 수치가 없습니다. 환각 억제가 작동한다면 모델은 아는 척 숫자를 지어내지 않고, “문서에 없다”고 솔직히 답해야 합니다. 만약 그럴듯한 가짜 숫자를 만들어 낸다면, 시스템 프롬프트의 지시가 약하거나 검색이 엉뚱한 청크를 근거로 줬다는 뜻입니다. 답이 어떻게 나오는지 확인하세요.

관찰 2: 검색이 엉뚱한 청크를 집는 경우

질문을 살짝 다르게, 모호하게 바꿔 물어 봅니다. 예를 들어 직접적인 키워드 없이 빙 둘러 물으면, 코사인 유사도가 의도와 다른 청크를 top으로 집을 수 있습니다. 검색이 틀리면 그 위에서 생성된 답도 틀립니다(garbage in, garbage out). “RAG는 검색이 절반”이라는 말을 직접 체감하는 순간입니다.

관찰 3: top_k를 바꿔 보기

search의 top_k를 1로 줄여 보세요. 참고 문서가 단 한 조각으로 줄면, 답에 필요한 정보가 그 한 청크에 다 없을 때 답이 빈약해집니다. 반대로 top_k를 너무 키우면 무관한 청크가 섞여 들어와 모델이 헷갈릴 수 있습니다. 적절한 top_k는 데이터와 질문에 따라 다릅니다.

관찰 4: 청킹을 바꿔 보기

load_chunks의 분할 방식을 바꿔 청크를 더 잘게(예: 문장 단위) 혹은 더 크게(예: 여러 문단을 한 덩어리로) 만들어 보세요. 너무 잘면 의미가 끊겨 검색이 부정확해지고, 너무 크면 한 청크에 여러 주제가 섞여 정확도가 떨어지고 토큰도 낭비됩니다. 청킹은 RAG 품질의 숨은 핵심 변수입니다.

오전에 콘솔에서 클릭으로 했던 일과, 이번 Lab에서 코드로 만든 것을 나란히 놓고 보면 “관리형 RAG가 내부에서 무엇을 했는지”가 또렷해집니다.

콘솔의 한 번의 클릭 뒤에는 임베딩 호출, 유사도 계산, 프롬프트 조립이 숨어 있었습니다. 직접 만들어 보고 나니 분명해집니다 — RAG는 마법이 아니라 임베딩 + 벡터 검색 + 프롬프트 조립의 합입니다.

관리형 KB가 추가로 해 주는 일(대규모 벡터 인덱싱, 빠른 검색, 자동 동기화, 메타데이터 필터링, 인용 관리 등)도 분명히 있습니다. 하지만 그 핵심 원리는 오늘 손으로 만든 이 50줄 안에 다 들어 있습니다. 언제 직접 만들고 언제 관리형을 쓸지는, 양쪽을 모두 이해한 지금부터 스스로 판단할 수 있습니다.

• 청킹을 문단(\n\n) 대신 고정 200자로 바꾸면 검색 품질이 어떻게 변할까요?
• top_k를 키우면 항상 좋아질까요? 무관한 청크가 섞여 답이 나빠지는 지점을 찾아보세요.
• 문서에 없는 답을 모델이 지어내게 유도하려면 질문을 어떻게 꼬아야 할까요? 환각 억제 프롬프트는 얼마나 견고한가요?