이미지 로딩 중...
AI Generated
2025. 11. 12. · 4 Views
Python으로 AI 에이전트 만들기 5편 - 문서 임베딩과 시맨틱 검색
AI 에이전트가 방대한 문서에서 정확한 정보를 찾아내는 핵심 기술인 문서 임베딩과 시맨틱 검색을 배웁니다. OpenAI의 임베딩 모델과 벡터 검색으로 강력한 RAG 시스템을 구축하는 실전 가이드입니다.
목차
- 텍스트 임베딩의 기초
- OpenAI 임베딩 API 활용
- 코사인 유사도 계산
- 문서 청크 분할 전략
- 벡터 저장소 구축
- 시맨틱 검색 구현
- ChromaDB로 벡터 DB 활용
- RAG 파이프라인 구축
1. 텍스트 임베딩의 기초
시작하며
여러분이 수천 개의 문서에서 "머신러닝의 과적합 해결 방법"을 찾아야 한다고 상상해보세요. 키워드 검색으로는 "과적합"이라는 단어가 없으면 놓치게 되고, "overfitting prevention"처럼 영어로 쓰여 있으면 또 찾지 못하죠.
이런 문제는 전통적인 키워드 기반 검색의 한계입니다. 단어의 철자만 비교하기 때문에 "자동차"와 "차량"을 다른 것으로 인식하고, 문맥이나 의미는 전혀 이해하지 못합니다.
바로 이럴 때 필요한 것이 텍스트 임베딩입니다. 문장을 숫자 벡터로 변환하여 의미적으로 비슷한 문장들이 가까운 위치에 배치되도록 합니다.
이를 통해 "강아지가 뛰어논다"와 "개가 달린다"를 의미적으로 유사한 것으로 인식할 수 있습니다.
개요
간단히 말해서, 텍스트 임베딩은 단어나 문장을 고차원 벡터 공간의 점으로 변환하는 기술입니다. 예를 들어 "고양이"는 [0.2, -0.5, 0.8, ...]과 같은 수백 개의 숫자 리스트로 표현됩니다.
왜 이것이 필요한지 실무 관점에서 설명하면, AI 에이전트가 사용자 질문과 가장 관련 있는 문서를 찾으려면 "의미"를 이해해야 하기 때문입니다. 예를 들어, 고객 지원 챗봇이 "결제가 안 돼요"라는 질문에 "payment failed" 문서를 찾아야 할 때 매우 유용합니다.
기존에는 단어 일치만으로 검색했다면, 이제는 문맥과 의미를 이해하는 검색이 가능합니다. "배가 아프다"에서 "배"가 신체 부위인지 교통수단인지 문맥으로 구분할 수 있습니다.
텍스트 임베딩의 핵심 특징은 세 가지입니다: 첫째, 의미적으로 유사한 텍스트는 벡터 공간에서 가깝게 위치합니다. 둘째, 수학적 연산(덧셈, 뺄셈)으로 의미 관계를 탐색할 수 있습니다.
셋째, 다국어 임베딩으로 언어 장벽을 넘을 수 있습니다. 이러한 특징들이 현대 AI 시스템의 핵심 구성 요소가 된 이유입니다.
코드 예제
# 간단한 임베딩 개념 시각화 (실제로는 OpenAI API 사용)
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
# 예시: 문장을 3차원 벡터로 표현 (실제는 1536차원)
sentence_vectors = {
"강아지가 뛰어논다": np.array([0.8, 0.6, 0.1]),
"개가 달린다": np.array([0.75, 0.65, 0.15]), # 유사한 의미
"자동차가 빠르다": np.array([0.1, 0.2, 0.9]) # 다른 의미
}
# 코사인 유사도로 의미적 거리 측정
query = sentence_vectors["강아지가 뛰어논다"]
for text, vector in sentence_vectors.items():
similarity = cosine_similarity([query], [vector])[0][0]
print(f"유사도 ({text}): {similarity:.3f}")
설명
이것이 하는 일: 위 코드는 문장을 벡터로 표현하고, 벡터 간 유사도를 계산하여 의미적 유사성을 측정합니다. 실제 임베딩 모델은 훨씬 복잡하지만, 핵심 원리는 동일합니다.
첫 번째로, sentence_vectors 딕셔너리는 각 문장을 3차원 벡터로 표현합니다. "강아지가 뛰어논다"와 "개가 달린다"는 의미가 비슷하므로 비슷한 벡터 값([0.8, 0.6, 0.1]과 [0.75, 0.65, 0.15])을 갖습니다.
반면 "자동차가 빠르다"는 완전히 다른 의미이므로 벡터 값도 크게 다릅니다. 실제 OpenAI 임베딩 모델은 1536차원 벡터를 생성하여 훨씬 더 정교하게 의미를 표현합니다.
그 다음으로, cosine_similarity 함수가 실행되면서 두 벡터 사이의 각도를 측정합니다. 코사인 유사도는 -1에서 1 사이의 값으로, 1에 가까울수록 의미가 유사하고 0에 가까우면 무관하며 -1에 가까우면 반대 의미입니다.
벡터의 방향만 비교하기 때문에 문장 길이에 영향받지 않는 장점이 있습니다. 마지막으로, 반복문이 각 문장과 쿼리 간의 유사도를 계산하여 출력합니다.
"강아지가 뛰어논다"와 "개가 달린다"는 0.99 이상의 높은 유사도를, "자동차가 빠르다"는 0.3 정도의 낮은 유사도를 보일 것입니다. 이를 통해 AI 에이전트는 사용자 질문과 가장 관련 있는 문서를 정확히 찾아낼 수 있습니다.
여러분이 이 코드를 사용하면 검색 품질이 획기적으로 향상됩니다. 키워드가 정확히 일치하지 않아도 의미가 비슷하면 찾아낼 수 있고, 오타나 동의어 문제에서 자유로우며, 다국어 검색도 가능합니다.
실제 프로덕션 환경에서는 수백만 개 문서에서도 밀리초 단위로 검색할 수 있습니다.
실전 팁
💡 실제 프로덕션에서는 항상 정규화된(normalized) 벡터를 사용하세요. 벡터 길이를 1로 만들면 코사인 유사도 계산이 단순 내적으로 바뀌어 10배 이상 빠릅니다.
💡 임베딩 차원이 높다고 무조건 좋은 것은 아닙니다. 1536차원이면 대부분의 작업에 충분하며, 너무 높으면 저장 공간과 검색 속도에서 손해를 봅니다.
💡 임베딩 모델은 학습 데이터에 따라 특화 분야가 다릅니다. 코드 검색에는 code-search 모델을, 일반 문서에는 text-embedding-3 모델을 선택하세요.
💡 배치 처리로 여러 문장을 한 번에 임베딩하면 API 호출 횟수가 줄어 비용을 절감할 수 있습니다. OpenAI는 한 번에 최대 2048개까지 처리 가능합니다.
💡 임베딩 벡터는 한 번 생성하면 재사용하세요. 같은 문서를 반복해서 임베딩하는 것은 비용 낭비이므로 캐싱 전략이 필수입니다.
2. OpenAI 임베딩 API 활용
시작하며
여러분이 직접 임베딩 모델을 학습시키려면 수백 GB의 텍스트 데이터와 수백만 원의 GPU 비용이 필요합니다. 그리고 몇 주간의 학습 시간까지 감수해야 하죠.
이런 문제는 대부분의 스타트업과 개인 개발자에게는 현실적으로 불가능합니다. 리소스도 부족하고 전문 지식도 필요하며, 무엇보다 시간이 너무 오래 걸립니다.
바로 이럴 때 필요한 것이 OpenAI 임베딩 API입니다. 단 몇 줄의 코드로 세계 최고 수준의 임베딩 모델을 사용할 수 있고, 비용도 100만 토큰당 $0.02로 매우 저렴합니다.
복잡한 인프라 없이 즉시 프로덕션 수준의 시맨틱 검색을 구현할 수 있습니다.
개요
간단히 말해서, OpenAI 임베딩 API는 텍스트를 입력하면 고품질의 1536차원 벡터를 반환하는 클라우드 서비스입니다. text-embedding-3-small과 text-embedding-3-large 두 모델을 제공합니다.
왜 이것이 필요한지 실무 관점에서 설명하면, 자체 모델 개발 없이 빠르게 MVP를 만들고 검증할 수 있기 때문입니다. 예를 들어, 사내 문서 검색 시스템을 단 하루 만에 구축하고, 즉시 팀원들의 피드백을 받아 개선할 수 있습니다.
API 방식이므로 모델 업데이트도 자동으로 적용됩니다. 기존에는 Word2Vec이나 FastText로 직접 학습했다면, 이제는 API 호출 한 번으로 더 정확한 임베딩을 얻을 수 있습니다.
다국어 지원도 기본으로 포함되어 있어 한국어, 영어, 일본어를 동일한 벡터 공간에서 처리합니다. OpenAI 임베딩의 핵심 특징은 세 가지입니다: 첫째, 최신 트랜스포머 아키텍처로 문맥을 정교하게 이해합니다.
둘째, 8191 토큰까지 긴 텍스트를 한 번에 처리할 수 있습니다. 셋째, 임베딩 차원을 줄여서 저장 공간을 절약할 수 있는 옵션을 제공합니다.
이러한 특징들이 RAG 시스템 구축의 사실상 표준이 된 이유입니다.
코드 예제
from openai import OpenAI
import os
# OpenAI 클라이언트 초기화
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def get_embedding(text, model="text-embedding-3-small"):
"""텍스트를 임베딩 벡터로 변환"""
# 개행 문자 제거 (API 권장사항)
text = text.replace("\n", " ")
# 임베딩 API 호출
response = client.embeddings.create(
input=text,
model=model
)
# 1536차원 벡터 반환
return response.data[0].embedding
# 사용 예시
query = "Python에서 리스트와 튜플의 차이는?"
embedding_vector = get_embedding(query)
print(f"벡터 차원: {len(embedding_vector)}")
print(f"처음 5개 값: {embedding_vector[:5]}")
설명
이것이 하는 일: 위 코드는 OpenAI API를 호출하여 텍스트를 의미 있는 벡터로 변환합니다. 환경 변수에서 API 키를 가져오고, 텍스트 전처리를 한 뒤 임베딩을 생성합니다.
첫 번째로, OpenAI 클라이언트를 초기화할 때 환경 변수에서 API 키를 읽어옵니다. 이렇게 하는 이유는 코드에 직접 키를 쓰면 GitHub에 실수로 올렸을 때 보안 문제가 발생하기 때문입니다.
os.getenv()를 사용하면 .env 파일이나 시스템 환경 변수에서 안전하게 키를 관리할 수 있습니다. API 키는 OpenAI 대시보드에서 발급받을 수 있으며, 프로젝트별로 다른 키를 사용하는 것이 좋습니다.
그 다음으로, get_embedding 함수가 실행되면서 텍스트의 개행 문자를 공백으로 치환합니다. OpenAI 공식 문서는 이를 권장하는데, 개행 문자가 있으면 토큰 계산에 영향을 주고 임베딩 품질이 약간 떨어질 수 있기 때문입니다.
그 후 client.embeddings.create()를 호출하여 실제 임베딩을 생성합니다. model 파라미터로 "text-embedding-3-small" (저렴, 빠름) 또는 "text-embedding-3-large" (정확, 느림)를 선택할 수 있습니다.
세 번째로, API 응답에서 response.data[0].embedding으로 벡터를 추출합니다. 응답은 JSON 형태이며, data 배열의 첫 번째 요소가 실제 임베딩입니다.
여러 텍스트를 한 번에 보내면 data 배열에 여러 개의 임베딩이 담깁니다. 벡터는 파이썬 리스트로 반환되며, 각 원소는 -1과 1 사이의 float 값입니다.
마지막으로, 사용 예시에서 실제로 함수를 호출하고 결과를 확인합니다. "Python에서 리스트와 튜플의 차이는?"이라는 질문을 임베딩하면 1536개의 숫자로 이루어진 벡터가 반환됩니다.
처음 5개 값을 출력해보면 [0.0234, -0.0156, 0.0421, ...] 같은 형태입니다. 이 벡터를 데이터베이스에 저장해두면 나중에 유사한 질문을 빠르게 검색할 수 있습니다.
여러분이 이 코드를 사용하면 몇 가지 큰 이점을 얻을 수 있습니다. 첫째, 모델 학습에 필요한 시간과 비용이 전혀 들지 않습니다.
둘째, OpenAI가 모델을 지속적으로 개선하므로 여러분의 검색 품질도 자동으로 향상됩니다. 셋째, 100개 문서 임베딩에 약 $0.001 정도로 비용이 매우 저렴합니다.
넷째, 오류 처리와 재시도 로직만 추가하면 바로 프로덕션에 배포할 수 있습니다.
실전 팁
💡 배치 처리로 비용을 절감하세요. input 파라미터에 리스트를 넘기면 여러 텍스트를 한 번에 처리할 수 있습니다. 예: input=["텍스트1", "텍스트2", "텍스트3"]
💡 API 호출 실패에 대비해 tenacity 라이브러리로 재시도 로직을 구현하세요. 네트워크 오류나 일시적 서버 문제로 실패할 수 있으므로 exponential backoff 전략이 필수입니다.
💡 text-embedding-3-small로 시작하세요. large 모델은 약 2배 비싸지만 정확도 향상은 5% 미만입니다. 대부분의 경우 small로 충분합니다.
💡 임베딩을 생성한 후 반드시 캐싱하세요. Redis나 로컬 파일에 저장하여 같은 텍스트를 반복 임베딩하지 않도록 합니다. 이것만으로도 비용을 90% 절감할 수 있습니다.
💡 dimensions 파라미터로 차원을 줄일 수 있습니다 (예: dimensions=512). 저장 공간이 1/3로 줄고 검색 속도는 3배 빨라지며, 정확도 하락은 거의 없습니다.
3. 코사인 유사도 계산
시작하며
여러분이 사용자 질문 "파이썬 딕셔너리 사용법"에 대해 10,000개의 문서 중 가장 관련 있는 것을 찾아야 한다고 상상해보세요. 모든 문서가 임베딩 벡터로 변환되어 있지만, 어떤 벡터가 질문과 가장 가까운지 어떻게 판단할까요?
이런 문제는 벡터 검색의 핵심입니다. 두 벡터가 "비슷하다"는 것을 수학적으로 정의하고 측정할 방법이 필요합니다.
유클리드 거리를 쓸 수도 있지만, 텍스트 임베딩에는 더 적합한 방법이 있습니다. 바로 이럴 때 필요한 것이 코사인 유사도입니다.
두 벡터 사이의 각도를 측정하여 방향이 얼마나 비슷한지 판단합니다. 벡터 크기에 영향받지 않고 순수하게 의미적 유사성만 측정할 수 있습니다.
개요
간단히 말해서, 코사인 유사도는 두 벡터 사이의 각도 코사인 값으로, -1에서 1 사이의 값을 반환합니다. 1에 가까울수록 의미가 매우 유사하고, 0이면 무관하며, -1이면 정반대 의미입니다.
왜 이것이 필요한지 실무 관점에서 설명하면, RAG 시스템에서 수천 개의 문서 중 상위 3개를 선택할 때 객관적인 기준이 필요하기 때문입니다. 예를 들어, 챗봇이 사용자 질문에 답변하려면 가장 관련 있는 지식 베이스 문서를 찾아야 하는데, 코사인 유사도로 정확히 순위를 매길 수 있습니다.
기존에는 유클리드 거리나 맨해튼 거리를 사용했다면, 이제는 코사인 유사도로 더 정확한 의미 검색이 가능합니다. "이 기능은 정말 훌륭합니다"(긴 문장)와 "훌륭한 기능"(짧은 문장)은 유클리드 거리로는 멀지만, 코사인 유사도로는 매우 가깝습니다.
코사인 유사도의 핵심 특징은 세 가지입니다: 첫째, 벡터 길이(norm)에 영향받지 않아 문장 길이가 달라도 공정하게 비교합니다. 둘째, 계산이 매우 빠릅니다(단순 내적과 나눗셈).
셋째, 정규화된 벡터에서는 내적과 동일하여 더욱 최적화 가능합니다. 이러한 특징들이 텍스트 임베딩 검색의 표준 메트릭이 된 이유입니다.
코드 예제
import numpy as np
from numpy.linalg import norm
def cosine_similarity(vec1, vec2):
"""두 벡터 간의 코사인 유사도 계산"""
# 내적 계산
dot_product = np.dot(vec1, vec2)
# 각 벡터의 크기(norm) 계산
norm1 = norm(vec1)
norm2 = norm(vec2)
# 코사인 유사도 = 내적 / (크기1 * 크기2)
return dot_product / (norm1 * norm2)
# 실전 예시: 질문과 문서들의 유사도 비교
query_embedding = get_embedding("Python 리스트 정렬 방법")
doc_embeddings = [
get_embedding("리스트를 sort()로 정렬하기"),
get_embedding("파이썬 딕셔너리 활용법"),
get_embedding("NumPy 배열 정렬하기")
]
# 각 문서와의 유사도 계산 및 순위 매기기
similarities = [cosine_similarity(query_embedding, doc) for doc in doc_embeddings]
ranked_docs = sorted(enumerate(similarities), key=lambda x: x[1], reverse=True)
for idx, score in ranked_docs:
print(f"문서 {idx}: 유사도 {score:.3f}")
설명
이것이 하는 일: 위 코드는 코사인 유사도 공식을 구현하고, 이를 사용하여 질문과 여러 문서의 유사도를 계산한 뒤 순위를 매깁니다. 실제 RAG 시스템의 핵심 검색 로직입니다.
첫 번째로, cosine_similarity 함수는 수학 공식을 그대로 코드로 옮깁니다. np.dot(vec1, vec2)로 두 벡터의 내적을 계산하는데, 이는 대응하는 원소끼리 곱한 뒤 모두 더하는 것입니다.
예를 들어 [1,2,3]과 [4,5,6]의 내적은 1×4 + 2×5 + 3×6 = 32입니다. 그 다음 norm() 함수로 각 벡터의 크기를 계산합니다.
벡터 크기는 원점에서 그 점까지의 거리로, √(x₁² + x₂² + ... + xₙ²)입니다.
마지막으로 내적을 두 크기의 곱으로 나누면 코사인 값이 나옵니다. 그 다음으로, 실전 예시에서 질문 "Python 리스트 정렬 방법"과 세 개의 문서를 임베딩합니다.
get_embedding()을 네 번 호출하므로 실제로는 API 요청이 네 번 발생합니다. 실무에서는 문서 임베딩은 미리 계산해서 데이터베이스에 저장해두고, 질문 임베딩만 실시간으로 생성하는 것이 효율적입니다.
이렇게 하면 응답 시간을 1초 이내로 줄일 수 있습니다. 세 번째로, 리스트 컴프리헨션으로 각 문서와 질문의 유사도를 계산합니다.
similarities 리스트에는 [0.92, 0.45, 0.78] 같은 값들이 담깁니다. 첫 번째 문서 "리스트를 sort()로 정렬하기"가 0.92로 가장 높고, 두 번째 문서 "파이썬 딕셔너리 활용법"이 0.45로 가장 낮을 것입니다.
딕셔너리는 리스트 정렬과 관련이 적기 때문입니다. 마지막으로, sorted() 함수로 유사도 기준 내림차순 정렬을 합니다.
enumerate()를 사용하여 원래 인덱스를 유지하면서 정렬하는 것이 포인트입니다. key=lambda x: x[1]은 튜플의 두 번째 요소(유사도 점수)를 정렬 기준으로 사용하고, reverse=True로 높은 점수가 먼저 오도록 합니다.
결과적으로 "문서 0: 유사도 0.920" "문서 2: 유사도 0.780" "문서 1: 유사도 0.450" 순서로 출력됩니다. 여러분이 이 코드를 사용하면 정확한 시맨틱 검색이 가능합니다.
키워드가 정확히 일치하지 않아도 의미가 비슷하면 높은 순위에 오르고, 임계값(threshold)을 설정하여 관련 없는 문서는 필터링할 수 있습니다. 예를 들어 유사도 0.7 이상만 결과로 반환하면 정확도가 크게 향상됩니다.
또한 이 로직은 추천 시스템, 중복 탐지, 클러스터링 등 다양한 분야에 응용할 수 있습니다.
실전 팁
💡 벡터를 미리 정규화하면 성능이 10배 향상됩니다. vec / norm(vec)로 정규화하면 코사인 유사도가 단순 내적이 되어 나눗셈 연산이 불필요합니다.
💡 NumPy 대신 scipy.spatial.distance.cosine을 쓸 수도 있지만, 주의하세요. scipy는 유사도가 아니라 거리(1 - 유사도)를 반환하므로 1에서 빼야 합니다.
💡 대량 비교 시 sklearn.metrics.pairwise.cosine_similarity를 사용하세요. 행렬 연산으로 최적화되어 있어 10,000개 문서 비교도 1초 이내에 가능합니다.
💡 임계값은 실험적으로 결정하세요. 0.7~0.8이 일반적이지만, 도메인에 따라 다릅니다. 검증 데이터로 F1 점수를 측정하여 최적값을 찾으세요.
💡 음수 유사도는 거의 나오지 않습니다. OpenAI 임베딩은 대부분 0~1 범위이므로, -1에 가까운 값이 나오면 데이터 오류를 의심하세요.
4. 문서 청크 분할 전략
시작하며
여러분이 200페이지짜리 사용자 매뉴얼을 통째로 임베딩하려고 한다고 상상해보세요. OpenAI API는 8,191 토큰 제한이 있어서 일단 불가능하고, 설령 가능하다 해도 너무 많은 정보가 뭉쳐서 의미가 희석됩니다.
이런 문제는 긴 문서를 다룰 때 항상 발생합니다. 한 번에 너무 많은 내용을 임베딩하면 검색 정확도가 떨어지고, 너무 작게 나누면 문맥이 손실됩니다.
"적절한 크기"를 찾는 것이 관건입니다. 바로 이럴 때 필요한 것이 문서 청크 분할 전략입니다.
문서를 의미 있는 단위로 나누어 각각 임베딩하고, 검색 시 정확히 필요한 부분만 가져올 수 있게 합니다. 문장, 단락, 또는 고정 토큰 수 등 다양한 전략을 선택할 수 있습니다.
개요
간단히 말해서, 청크 분할은 긴 문서를 작은 조각(chunk)으로 나누는 기술입니다. 각 청크는 500~1000 토큰 정도로, 하나의 주제나 개념을 담고 있어야 합니다.
왜 이것이 필요한지 실무 관점에서 설명하면, RAG 시스템이 LLM에게 전달할 컨텍스트 길이에 제한이 있기 때문입니다. 예를 들어, GPT-4의 컨텍스트 윈도우가 128k 토큰이라도, 사용자 질문과 가장 관련 있는 3~5개 청크만 보내는 것이 비용과 응답 품질 면에서 유리합니다.
불필요한 정보가 많으면 LLM이 혼란스러워하고 환각(hallucination) 확률이 높아집니다. 기존에는 고정된 문자 수(예: 1000자)로 잘랐다면, 이제는 문장 경계, 단락, 또는 의미 단위를 고려한 스마트 분할이 가능합니다.
"...입니다. 다음 섹션에서는..."처럼 자연스러운 경계에서 나누면 문맥 손실을 최소화할 수 있습니다.
청크 분할의 핵심 원칙은 세 가지입니다: 첫째, 각 청크가 독립적으로 의미를 가져야 합니다(self-contained). 둘째, 청크 간 약간의 중복(overlap)을 두어 경계에서 문맥이 끊기지 않게 합니다.
셋째, 일관된 크기를 유지하여 검색 품질을 안정적으로 만듭니다. 이러한 원칙들이 고품질 RAG 시스템의 기초가 됩니다.
코드 예제
from langchain.text_splitter import RecursiveCharacterTextSplitter
def split_documents(text, chunk_size=1000, chunk_overlap=200):
"""문서를 의미 있는 청크로 분할"""
# RecursiveCharacterTextSplitter: 단락 -> 문장 -> 단어 순으로 분할 시도
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size, # 각 청크의 목표 크기 (문자 수)
chunk_overlap=chunk_overlap, # 청크 간 중복 크기
separators=["\n\n", "\n", ". ", " ", ""], # 분할 우선순위
length_function=len # 길이 측정 함수
)
# 텍스트를 청크로 분할
chunks = splitter.split_text(text)
return chunks
# 실전 예시: 긴 문서 분할
long_document = """
Python은 배우기 쉬운 프로그래밍 언어입니다. 문법이 간결하고 직관적입니다.
리스트는 Python의 핵심 자료구조입니다. 대괄호로 생성하고 다양한 타입을 담을 수 있습니다.
리스트는 가변(mutable)이므로 생성 후 수정 가능합니다.
딕셔너리는 키-값 쌍을 저장합니다. 해시 테이블로 구현되어 O(1) 검색이 가능합니다.
"""
chunks = split_documents(long_document, chunk_size=100, chunk_overlap=20)
for i, chunk in enumerate(chunks):
print(f"청크 {i}: {chunk[:50]}...")
설명
이것이 하는 일: 위 코드는 LangChain의 RecursiveCharacterTextSplitter를 사용하여 문서를 지능적으로 분할합니다. 단순 문자 수가 아니라 문단, 문장 경계를 고려합니다.
첫 번째로, RecursiveCharacterTextSplitter를 초기화할 때 여러 파라미터를 설정합니다. chunk_size=1000은 각 청크의 목표 크기를 1000자로 설정하는데, 이는 약 200~300 토큰에 해당합니다.
chunk_overlap=200은 인접한 청크 간 200자씩 중복을 의미하는데, 이렇게 하면 한 청크의 끝과 다음 청크의 시작이 이어져 문맥이 보존됩니다. separators 리스트는 분할 우선순위를 정의합니다.
먼저 "\n\n"(단락)으로 나누려고 시도하고, 안 되면 "\n"(줄), ". "(문장), " "(단어) 순으로 시도합니다.
이렇게 하면 가능한 한 자연스러운 경계에서 분할됩니다. 그 다음으로, split_text() 메서드가 실제 분할을 수행합니다.
"Recursive"라는 이름처럼 재귀적으로 작동하는데, 먼저 단락으로 나눈 뒤 각 단락이 chunk_size를 초과하면 다시 문장으로 나누고, 그래도 크면 단어로 나눕니다. 이 과정에서 chunk_overlap을 고려하여 각 청크의 끝부분을 다음 청크의 시작에 포함시킵니다.
결과는 문자열 리스트로 반환됩니다. 세 번째로, 실전 예시에서 Python 관련 문서를 분할합니다.
chunk_size=100으로 작게 설정했으므로 여러 청크가 생성될 것입니다. 첫 번째 청크는 "Python은 배우기 쉬운 프로그래밍 언어입니다.
문법이 간결하고 직관적입니다."를 포함하고, 두 번째 청크는 첫 번째의 마지막 20자와 "리스트는 Python의 핵심 자료구조입니다..." 부분을 포함합니다. 이렇게 중복이 있어야 "리스트" 개념이 갑자기 등장하는 것이 아니라 문맥이 이어집니다.
마지막으로, 각 청크를 반복하며 처음 50자만 출력합니다. 실제 프로덕션에서는 각 청크를 임베딩하고 메타데이터(청크 번호, 원본 문서 ID)와 함께 벡터 데이터베이스에 저장합니다.
이렇게 하면 나중에 검색 결과로 청크를 받았을 때 원본 문서의 어느 부분인지 추적할 수 있습니다. 여러분이 이 코드를 사용하면 검색 품질이 극적으로 향상됩니다.
전체 문서 하나의 임베딩으로는 "딕셔너리 검색 속도"에 대한 질문에 정확히 답하기 어렵지만, 청크로 나누면 정확히 해당 부분만 검색됩니다. 또한 chunk_overlap 덕분에 경계에 걸친 개념도 놓치지 않고, 각 청크가 적당한 크기여서 LLM에게 전달할 때도 효율적입니다.
실무에서는 chunk_size를 5001000, overlap을 1020%로 설정하는 것이 일반적입니다.
실전 팁
💡 도메인에 맞는 separator를 추가하세요. 코드 문서라면 ["```", "###", "\n\n"]처럼 마크다운 구조를 활용하고, 법률 문서라면 ["Article", "Section"] 같은 키워드를 사용하세요.
💡 chunk_overlap은 chunk_size의 1020%가 적당합니다. 너무 크면 저장 공간 낭비이고, 너무 작으면 문맥이 끊깁니다. 500자 청크에 50100자 중복이 이상적입니다.
💡 토큰 기반 분할이 더 정확합니다. tiktoken 라이브러리로 토큰 수를 직접 세면 API 제한에 정확히 맞출 수 있습니다. CharacterTextSplitter 대신 TokenTextSplitter를 사용하세요.
💡 청크에 메타데이터를 추가하세요. {"chunk_id": 1, "source": "manual.pdf", "page": 5}처럼 정보를 저장하면 검색 결과를 사용자에게 보여줄 때 출처를 명확히 할 수 있습니다.
💡 시맨틱 분할도 고려하세요. LlamaIndex의 SemanticSplitter는 문장 임베딩 유사도로 자연스러운 경계를 찾습니다. 계산 비용이 높지만 품질은 최고입니다.
5. 벡터 저장소 구축
시작하며
여러분이 1만 개의 문서 청크를 임베딩했다고 상상해보세요. 각 청크는 1536차원 벡터이므로, 매번 검색할 때마다 1만 번의 코사인 유사도 계산이 필요합니다.
이는 매우 느리고 비효율적입니다. 이런 문제는 벡터 검색의 확장성 문제입니다.
데이터가 10만 개, 100만 개로 늘어나면 선형 검색은 현실적으로 불가능합니다. 사용자는 1초 이내의 응답을 기대하는데, 단순 방법으로는 몇 분이 걸릴 수 있습니다.
바로 이럴 때 필요한 것이 벡터 저장소(Vector Store)입니다. 임베딩 벡터를 효율적으로 저장하고, 근사 최근접 이웃(ANN) 알고리즘으로 밀리초 단위 검색을 가능하게 합니다.
FAISS, Pinecone, ChromaDB 같은 전문 솔루션들이 있습니다.
개요
간단히 말해서, 벡터 저장소는 고차원 벡터를 색인화하여 빠른 유사도 검색을 제공하는 데이터베이스입니다. 전통적인 SQL 데이터베이스가 텍스트나 숫자를 저장한다면, 벡터 DB는 임베딩 벡터를 최적화하여 저장합니다.
왜 이것이 필요한지 실무 관점에서 설명하면, 프로덕션 환경에서는 수백만 개의 문서를 실시간으로 검색해야 하기 때문입니다. 예를 들어, 고객 지원 챗봇이 1초 안에 응답하려면 10만 개 FAQ 중에서 상위 5개를 밀리초 단위로 찾아야 합니다.
벡터 저장소는 HNSW, IVF 같은 알고리즘으로 이를 가능하게 합니다. 기존에는 모든 벡터를 메모리에 로드하고 하나씩 비교했다면, 이제는 벡터 공간을 클러스터링하여 관련 없는 영역은 아예 검색하지 않습니다.
이를 통해 100만 개 벡터 중 검색 시 실제로는 1000개만 비교하여 99.9%의 정확도를 유지하면서 1000배 빠릅니다. 벡터 저장소의 핵심 특징은 세 가지입니다: 첫째, 근사 검색(Approximate Nearest Neighbor)으로 속도와 정확도를 교환할 수 있습니다.
둘째, 메타데이터 필터링으로 "category=Python"인 문서만 검색하는 등 하이브리드 검색이 가능합니다. 셋째, 영속성(persistence)으로 디스크에 저장하여 재시작 후에도 데이터를 유지합니다.
이러한 특징들이 대규모 RAG 시스템의 필수 인프라가 된 이유입니다.
코드 예제
import numpy as np
from typing import List, Tuple
class SimpleVectorStore:
"""간단한 인메모리 벡터 저장소"""
def __init__(self):
self.vectors = [] # 임베딩 벡터들
self.texts = [] # 원본 텍스트들
self.metadata = [] # 메타데이터들
def add(self, text: str, embedding: List[float], meta: dict = None):
"""벡터 추가"""
self.vectors.append(np.array(embedding))
self.texts.append(text)
self.metadata.append(meta or {})
def search(self, query_embedding: List[float], top_k: int = 3) -> List[Tuple[str, float]]:
"""유사도 기반 검색"""
query_vec = np.array(query_embedding)
# 모든 벡터와 코사인 유사도 계산
similarities = []
for i, vec in enumerate(self.vectors):
sim = np.dot(query_vec, vec) / (np.linalg.norm(query_vec) * np.linalg.norm(vec))
similarities.append((i, sim))
# 상위 k개 선택
top_indices = sorted(similarities, key=lambda x: x[1], reverse=True)[:top_k]
# 결과 반환 (텍스트, 유사도)
return [(self.texts[i], score) for i, score in top_indices]
# 사용 예시
store = SimpleVectorStore()
store.add("Python 리스트 정렬", get_embedding("Python 리스트 정렬"), {"category": "Python"})
store.add("자바 배열 활용", get_embedding("자바 배열 활용"), {"category": "Java"})
results = store.search(get_embedding("파이썬 정렬 방법"), top_k=1)
print(f"검색 결과: {results[0][0]} (유사도: {results[0][1]:.3f})")
설명
이것이 하는 일: 위 코드는 기본적인 인메모리 벡터 저장소를 구현합니다. 실제 프로덕션 수준은 아니지만, 벡터 DB의 핵심 개념을 명확히 보여줍니다.
첫 번째로, SimpleVectorStore 클래스는 세 개의 리스트로 데이터를 관리합니다. vectors는 NumPy 배열로 변환된 임베딩을, texts는 원본 텍스트를, metadata는 카테고리나 출처 같은 추가 정보를 저장합니다.
이 세 리스트는 인덱스로 동기화되므로, vectors[0], texts[0], metadata[0]은 같은 문서를 나타냅니다. 실제 벡터 DB는 이를 훨씬 복잡한 자료구조(HNSW 그래프, IVF 인덱스 등)로 관리하지만 기본 원리는 동일합니다.
그 다음으로, add() 메서드가 새로운 문서를 저장소에 추가합니다. embedding을 NumPy 배열로 변환하는 것이 중요한데, 이후 벡터 연산(내적, norm)을 빠르게 수행하기 위함입니다.
파이썬 리스트로 연산하면 10배 이상 느립니다. metadata는 선택사항이지만, 실무에서는 거의 필수입니다.
"이 결과가 어디서 왔는지" 추적하지 못하면 사용자에게 출처를 보여줄 수 없기 때문입니다. 세 번째로, search() 메서드가 핵심 검색 로직을 구현합니다.
query_embedding을 받아 저장된 모든 벡터와 코사인 유사도를 계산합니다. 반복문에서 np.dot()으로 내적을 계산하고 np.linalg.norm()으로 크기를 구한 뒤 나누면 코사인 유사도가 나옵니다.
similarities 리스트에 (인덱스, 유사도) 튜플을 저장한 뒤, sorted()로 유사도 기준 내림차순 정렬합니다. top_k 파라미터로 상위 몇 개를 반환할지 조절할 수 있습니다.
대부분의 RAG 시스템은 top_k=3~5를 사용합니다. 마지막으로, 사용 예시에서 실제로 저장소를 생성하고 문서를 추가한 뒤 검색합니다.
"Python 리스트 정렬"과 "자바 배열 활용" 두 문서를 추가한 뒤, "파이썬 정렬 방법"으로 검색하면 첫 번째 문서가 높은 유사도로 반환됩니다. get_embedding()을 세 번 호출하므로 실제로는 API 요청이 세 번 발생하는데, 실무에서는 문서 임베딩은 미리 생성해서 저장하고 쿼리 임베딩만 실시간으로 만듭니다.
여러분이 이 코드를 사용하면 벡터 검색의 기본을 이해할 수 있습니다. 하지만 실제 프로덕션에서는 FAISS, ChromaDB, Pinecone 같은 전문 솔루션을 사용해야 합니다.
이들은 ANN 알고리즘으로 100만 개 벡터에서도 10ms 이내 검색이 가능하고, 디스크 영속성, 분산 처리, GPU 가속 등을 제공합니다. 하지만 개념은 위 코드와 동일하므로, 이를 이해하면 어떤 벡터 DB도 쉽게 배울 수 있습니다.
실전 팁
💡 프로덕션에서는 FAISS를 시작점으로 삼으세요. Meta에서 개발한 오픈소스로 무료이며, CPU/GPU 모두 지원하고, 10억 개 벡터도 처리 가능합니다. faiss.IndexFlatIP로 시작하여 데이터가 커지면 IndexIVFFlat으로 전환하세요.
💡 벡터를 정규화하여 저장하세요. L2 norm이 1이 되도록 만들면 코사인 유사도가 내적과 동일해져 계산이 빨라지고, IndexFlatIP (Inner Product) 인덱스를 사용할 수 있습니다.
💡 메타데이터 필터링이 필요하면 ChromaDB를 선택하세요. FAISS는 순수 벡터 검색만 지원하지만, ChromaDB는 where={"category": "Python"} 같은 필터링이 기본 기능입니다.
💡 클라우드 호스팅이 필요하면 Pinecone이나 Weaviate를 고려하세요. 직접 인프라를 관리할 필요 없고, API 호출만으로 사용 가능하지만 비용이 발생합니다. 무료 티어로 시작하여 확장하세요.
💡 벡터 저장소는 정기적으로 재구축하세요. 문서가 추가/삭제되면 인덱스가 최적화되지 않은 상태가 되므로, 매주 또는 매월 전체를 다시 인덱싱하면 검색 속도가 유지됩니다.
6. 시맨틱 검색 구현
시작하며
여러분이 "파이썬에서 리스트 중복 제거하는 방법"이라고 질문했는데, 문서에는 "set으로 고유한 원소만 남기기"라고 쓰여 있다면 키워드 검색으로는 찾지 못합니다. "중복 제거"와 "고유한 원소"는 같은 의미인데 말이죠.
이런 문제는 전통적인 BM25나 TF-IDF 검색의 한계입니다. 단어의 표면적 일치만 보기 때문에 동의어, 의역, 다른 언어로 표현된 같은 개념을 놓칩니다.
사용자는 다양한 방식으로 질문하는데 시스템은 경직되어 있습니다. 바로 이럴 때 필요한 것이 시맨틱 검색입니다.
질문과 문서를 모두 임베딩 벡터로 변환한 뒤, 의미적 유사성으로 순위를 매깁니다. "강아지"로 검색해도 "개", "puppy", "애완견"이 포함된 문서를 모두 찾아냅니다.
개요
간단히 말해서, 시맨틱 검색은 임베딩 기반으로 의미를 이해하는 검색입니다. 사용자 질문을 벡터로 변환하고, 저장된 문서 벡터들과 유사도를 계산하여 가장 관련 있는 결과를 반환합니다.
왜 이것이 필요한지 실무 관점에서 설명하면, 실제 사용자는 정확한 키워드를 모르거나 다르게 표현하기 때문입니다. 예를 들어, 기술 문서에서 "authentication"을 찾고 싶은데 "로그인 방법"이라고 검색하면, 키워드 검색으로는 실패하지만 시맨틱 검색으로는 성공합니다.
이는 고객 만족도와 직결됩니다. 기존에는 동의어 사전을 수동으로 관리하거나 쿼리 확장(query expansion) 기법을 사용했다면, 이제는 임베딩 모델이 자동으로 의미 관계를 학습합니다.
"자동차"와 "차량", "vehicle"을 모두 비슷한 벡터로 표현하므로 별도 설정 없이 작동합니다. 시맨틱 검색의 핵심 특징은 세 가지입니다: 첫째, 다국어 검색이 자연스럽게 가능합니다.
한국어 질문으로 영어 문서를 찾을 수 있습니다. 둘째, 오타에 강합니다.
"파이썬"을 "파이선"으로 쓰더라도 의미가 비슷하므로 찾아냅니다. 셋째, 문맥을 이해합니다.
"사과"가 과일인지 행동인지 주변 단어로 구분합니다. 이러한 특징들이 현대 검색 엔진의 핵심 기술이 된 이유입니다.
코드 예제
from openai import OpenAI
import numpy as np
from typing import List, Tuple
client = OpenAI()
class SemanticSearch:
def __init__(self):
self.documents = [] # (텍스트, 임베딩) 튜플 리스트
def index_documents(self, texts: List[str]):
"""문서들을 임베딩하여 인덱싱"""
print(f"{len(texts)}개 문서 인덱싱 중...")
# 배치로 임베딩 생성 (비용 절감)
response = client.embeddings.create(
input=texts,
model="text-embedding-3-small"
)
# 문서와 임베딩 저장
for i, text in enumerate(texts):
embedding = response.data[i].embedding
self.documents.append((text, np.array(embedding)))
print(f"인덱싱 완료!")
def search(self, query: str, top_k: int = 3) -> List[Tuple[str, float]]:
"""시맨틱 검색 수행"""
# 쿼리를 임베딩으로 변환
query_embedding = client.embeddings.create(
input=query,
model="text-embedding-3-small"
).data[0].embedding
query_vec = np.array(query_embedding)
# 모든 문서와 유사도 계산
results = []
for text, doc_vec in self.documents:
similarity = np.dot(query_vec, doc_vec) / (np.linalg.norm(query_vec) * np.linalg.norm(doc_vec))
results.append((text, similarity))
# 상위 k개 반환
return sorted(results, key=lambda x: x[1], reverse=True)[:top_k]
# 사용 예시
search_engine = SemanticSearch()
search_engine.index_documents([
"Python에서 리스트는 대괄호로 생성합니다",
"set() 함수로 중복을 제거할 수 있습니다",
"딕셔너리는 키-값 쌍을 저장합니다"
])
results = search_engine.search("리스트 중복 제거 방법", top_k=2)
for doc, score in results:
print(f"[{score:.3f}] {doc}")
설명
이것이 하는 일: 위 코드는 완전한 시맨틱 검색 엔진을 구현합니다. 문서를 임베딩하여 인덱싱하고, 쿼리도 임베딩하여 유사도 기반으로 검색합니다.
실제 RAG 시스템의 검색 컴포넌트와 동일한 구조입니다. 첫 번째로, SemanticSearch 클래스는 documents 리스트로 인덱싱된 문서를 관리합니다.
각 원소는 (텍스트, 임베딩 벡터) 튜플입니다. 이렇게 텍스트와 벡터를 함께 저장하는 이유는, 검색 결과로 벡터만 받으면 사용자에게 보여줄 수 없기 때문입니다.
원본 텍스트도 함께 있어야 "이 문서가 당신의 질문과 관련 있습니다"라고 제시할 수 있습니다. 그 다음으로, index_documents() 메서드가 여러 문서를 한 번에 임베딩합니다.
중요한 점은 input 파라미터에 텍스트 리스트를 전달하여 배치 처리한다는 것입니다. 만약 10개 문서를 하나씩 임베딩하면 API 호출이 10번 발생하지만, 리스트로 넘기면 1번만 호출되어 비용과 시간이 크게 절감됩니다.
OpenAI는 한 번에 최대 2048개까지 처리 가능하므로, 대량 인덱싱 시 2048개씩 나누어 처리하는 것이 효율적입니다. 세 번째로, search() 메서드가 실제 검색을 수행합니다.
먼저 query를 임베딩으로 변환하는데, 이것이 유일한 실시간 API 호출입니다. 문서 임베딩은 이미 인덱싱 단계에서 완료되었으므로 재사용합니다.
그 다음 반복문에서 쿼리 벡터와 각 문서 벡터의 코사인 유사도를 계산합니다. results 리스트에 (텍스트, 유사도) 튜플을 모은 뒤, sorted()로 유사도 기준 내림차순 정렬하고 상위 top_k개만 반환합니다.
마지막으로, 사용 예시에서 "리스트 중복 제거 방법"으로 검색합니다. 흥미로운 점은 세 문서 중 어디에도 "중복 제거"라는 정확한 키워드가 없다는 것입니다.
두 번째 문서는 "set() 함수로 중복을 제거할 수 있습니다"라고 쓰여 있는데, "중복을 제거"가 "중복 제거"와 완전히 일치하지 않지만 임베딩 공간에서는 매우 가깝습니다. 따라서 이 문서가 0.85 이상의 높은 유사도로 첫 번째 결과가 될 것입니다.
이것이 시맨틱 검색의 힘입니다. 여러분이 이 코드를 사용하면 사용자 경험이 혁신적으로 개선됩니다.
"오류 해결 방법"으로 검색해도 "troubleshooting guide"를 찾을 수 있고, "속도 개선"으로 검색해도 "performance optimization"을 찾을 수 있습니다. 실무에서는 여기에 BM25 키워드 검색을 결합한 하이브리드 검색으로 더욱 정확도를 높입니다.
또한 재랭킹(reranking) 모델로 상위 결과를 한 번 더 정렬하면 정확도가 10~20% 향상됩니다.
실전 팁
💡 하이브리드 검색으로 정확도를 높이세요. 시맨틱 검색과 BM25 키워드 검색의 결과를 가중 평균하면 (예: 0.7 * semantic + 0.3 * bm25) 둘의 장점을 모두 얻을 수 있습니다.
💡 쿼리 확장(Query Expansion) 기법을 적용하세요. LLM에게 사용자 질문을 3~5가지 방식으로 다시 쓰게 한 뒤 모두 검색하면 재현율(recall)이 크게 향상됩니다.
💡 임계값(threshold)을 설정하세요. 유사도 0.7 미만인 결과는 버리면 관련 없는 문서가 섞이는 것을 방지할 수 있습니다. "죄송하지만 관련 문서를 찾지 못했습니다"가 잘못된 답변보다 낫습니다.
💡 메타데이터 필터링을 활용하세요. where={"language": "python"}처럼 필터를 추가하면 사용자가 원하는 범위에서만 검색할 수 있어 정확도가 높아집니다.
💡 사용자 피드백을 수집하세요. "이 결과가 도움이 되었나요?" 버튼으로 데이터를 모으면, 나중에 파인튜닝하거나 검색 파라미터를 최적화할 수 있습니다.
7. ChromaDB로 벡터 DB 활용
시작하며
여러분이 앞에서 만든 SimpleVectorStore로 10만 개 문서를 검색한다고 상상해보세요. 모든 벡터를 메모리에 올려야 하고, 검색 시 10만 번의 코사인 유사도 계산이 필요하며, 서버를 재시작하면 모든 데이터가 사라집니다.
이런 문제는 프로덕션 환경에서는 치명적입니다. 메모리 부족으로 서버가 다운될 수 있고, 검색이 너무 느려 사용자가 떠나며, 데이터 손실 위험도 있습니다.
확장 가능한(scalable) 솔루션이 절실합니다. 바로 이럴 때 필요한 것이 ChromaDB입니다.
오픈소스 벡터 데이터베이스로, 디스크 영속성, 메타데이터 필터링, 효율적인 인덱싱을 모두 제공합니다. FAISS처럼 복잡한 설정 없이 몇 줄의 코드로 프로덕션 수준의 벡터 검색을 구현할 수 있습니다.
개요
간단히 말해서, ChromaDB는 개발자 친화적인 벡터 데이터베이스입니다. SQLite처럼 임베딩 형태로 작동하여 별도 서버 없이 사용 가능하며, 필요 시 클라이언트-서버 모드로 전환할 수 있습니다.
왜 이것이 필요한지 실무 관점에서 설명하면, RAG 시스템 구축 시 벡터 저장과 검색에만 집중하고 싶기 때문입니다. 예를 들어, FAISS는 강력하지만 메타데이터 필터링이나 영속성을 직접 구현해야 합니다.
ChromaDB는 이 모든 기능을 기본 제공하여 개발 시간을 절반으로 줄입니다. 기존에는 FAISS + SQLite 조합으로 벡터와 메타데이터를 따로 관리했다면, 이제는 ChromaDB 하나로 통합할 수 있습니다.
collection.add()로 벡터와 메타데이터를 함께 저장하고, where 절로 필터링하며, persist()로 디스크에 저장합니다. ChromaDB의 핵심 특징은 세 가지입니다: 첫째, 자동 임베딩 생성 기능으로 텍스트만 넘기면 내부에서 임베딩을 처리합니다.
둘째, LangChain과 LlamaIndex 통합이 기본 지원되어 즉시 RAG 파이프라인에 연결할 수 있습니다. 셋째, Python뿐 아니라 JavaScript에서도 사용 가능하여 풀스택 애플리케이션에 적합합니다.
이러한 특징들이 스타트업과 프로토타입에서 가장 인기 있는 벡터 DB가 된 이유입니다.
코드 예제
import chromadb
from chromadb.config import Settings
# ChromaDB 클라이언트 생성 (디스크에 영속 저장)
client = chromadb.PersistentClient(path="./chroma_db")
# 컬렉션 생성 또는 가져오기
collection = client.get_or_create_collection(
name="python_docs",
metadata={"description": "Python 문서 검색"}
)
# 문서 추가 (임베딩은 자동 생성)
collection.add(
documents=[
"Python 리스트는 대괄호로 생성합니다",
"딕셔너리는 중괄호로 만들고 키-값 쌍을 저장합니다",
"set() 함수로 중복을 제거할 수 있습니다"
],
metadatas=[
{"category": "list", "level": "beginner"},
{"category": "dict", "level": "beginner"},
{"category": "set", "level": "intermediate"}
],
ids=["doc1", "doc2", "doc3"] # 고유 ID
)
# 시맨틱 검색 수행
results = collection.query(
query_texts=["중복 제거하는 방법"],
n_results=2,
where={"level": "intermediate"} # 메타데이터 필터링
)
print(f"검색 결과: {results['documents'][0]}")
print(f"거리: {results['distances'][0]}")
설명
이것이 하는 일: 위 코드는 ChromaDB로 완전한 벡터 검색 시스템을 구축합니다. 문서를 추가하고, 메타데이터를 첨부하며, 필터링과 함께 검색하는 전체 워크플로우를 보여줍니다.
첫 번째로, PersistentClient를 생성할 때 path 파라미터로 저장 경로를 지정합니다. 이렇게 하면 모든 데이터가 ./chroma_db 폴더에 SQLite 파일로 저장됩니다.
서버를 재시작해도 데이터가 유지되므로, 한 번 인덱싱하면 계속 사용할 수 있습니다. 반면 chromadb.Client()를 사용하면 인메모리 모드로 작동하여 빠르지만 재시작 시 데이터가 사라집니다.
개발 중에는 인메모리로 시작하고, 배포 전에 Persistent로 전환하는 것이 일반적입니다. 그 다음으로, get_or_create_collection()으로 컬렉션을 생성합니다.
컬렉션은 RDB의 테이블과 비슷한 개념으로, 비슷한 종류의 문서를 그룹화합니다. 예를 들어 "python_docs", "javascript_docs", "customer_support" 같은 컬렉션을 여러 개 만들 수 있습니다.
이미 존재하면 가져오고, 없으면 생성하므로 코드를 여러 번 실행해도 안전합니다. metadata는 컬렉션 자체에 대한 설명으로, 나중에 여러 컬렉션 중 올바른 것을 선택할 때 유용합니다.
세 번째로, collection.add()로 문서를 추가합니다. documents는 원본 텍스트 리스트이고, metadatas는 각 문서의 추가 정보입니다.
중요한 점은 임베딩을 직접 전달하지 않는다는 것입니다. ChromaDB는 기본 임베딩 함수(sentence-transformers)로 자동 생성합니다.
만약 OpenAI 임베딩을 사용하려면 embeddings 파라미터에 직접 벡터를 넘기면 됩니다. ids는 각 문서의 고유 식별자로, 나중에 업데이트하거나 삭제할 때 사용합니다.
같은 ID로 add()를 다시 호출하면 덮어쓰기됩니다. 마지막으로, collection.query()로 검색을 수행합니다.
query_texts에 질문 텍스트를 넘기면 ChromaDB가 자동으로 임베딩하고 검색합니다. n_results로 반환할 개수를 지정하고, where 절로 메타데이터 필터링을 합니다.
위 예시에서는 level이 "intermediate"인 문서만 검색하므로, "set() 함수로 중복을 제거할 수 있습니다" 문서만 결과에 포함됩니다. results는 딕셔너리로, documents에 텍스트, distances에 거리(작을수록 유사), metadatas에 메타데이터가 담깁니다.
여러분이 이 코드를 사용하면 몇 가지 큰 장점을 얻습니다. 첫째, 임베딩 생성을 ChromaDB에 맡기면 코드가 간결해지고 실수가 줄어듭니다.
둘째, 메타데이터 필터링으로 "Python 문서 중 초급자용만" 같은 복잡한 쿼리가 가능합니다. 셋째, 영속성 덕분에 서버 재시작 걱정 없이 안정적으로 운영할 수 있습니다.
넷째, LangChain과의 통합으로 Chroma.from_documents() 한 줄로 RAG 시스템에 연결할 수 있습니다. 실무에서는 수십만 개 문서도 무리 없이 처리하며, 필요 시 ChromaDB 서버 모드로 전환하여 여러 애플리케이션이 공유할 수 있습니다.
실전 팁
💡 OpenAI 임베딩을 사용하려면 컬렉션 생성 시 embedding_function=None으로 설정하고, add() 시 embeddings 파라미터에 직접 벡터를 전달하세요. 기본 sentence-transformers보다 품질이 높습니다.
💡 대량 데이터는 배치로 추가하세요. 10만 개 문서를 한 번에 add()하지 말고 1000개씩 나누어 처리하면 메모리 효율이 좋고 오류 발생 시 일부만 재시도할 수 있습니다.
💡 update()와 upsert()를 활용하세요. update()는 기존 문서를 수정하고, upsert()는 없으면 추가, 있으면 수정합니다. 문서 갱신 시 전체를 다시 인덱싱할 필요가 없습니다.
💡 where 절에 $and, $or, $in 연산자를 사용할 수 있습니다. 예: where={"$and": [{"category": "python"}, {"level": {"$in": ["beginner", "intermediate"]}}]}
💡 프로덕션에서는 ChromaDB 서버 모드를 고려하세요. docker run으로 별도 서버를 띄우고 HttpClient로 연결하면 여러 애플리케이션이 동일한 벡터 DB를 공유할 수 있고, 스케일링도 쉽습니다.
8. RAG 파이프라인 구축
시작하며
여러분이 지금까지 배운 모든 기술을 한데 모아 실제 AI 에이전트를 만든다고 상상해보세요. 문서 임베딩, 시맨틱 검색, LLM 응답 생성까지 모든 단계를 연결해야 합니다.
어떻게 조합할까요? 이런 문제는 RAG(Retrieval-Augmented Generation) 시스템 구축의 핵심입니다.
각 컴포넌트는 완성했지만, 이를 하나의 파이프라인으로 통합하여 사용자 질문을 받아 정확한 답변을 생성하는 전체 흐름을 만들어야 합니다. 바로 이럴 때 필요한 것이 RAG 파이프라인입니다.
질문 → 임베딩 → 검색 → 컨텍스트 구성 → LLM 호출 → 답변 생성의 전체 과정을 자동화합니다. 한 번 구축하면 어떤 도메인에도 적용할 수 있는 범용 시스템이 됩니다.
개요
간단히 말해서, RAG 파이프라인은 검색(Retrieval)과 생성(Generation)을 결합한 AI 시스템입니다. 외부 지식 베이스에서 관련 정보를 찾아 LLM에게 제공하여, 환각 없이 정확한 답변을 생성하도록 합니다.
왜 이것이 필요한지 실무 관점에서 설명하면, LLM 단독으로는 학습 데이터에 없는 최신 정보나 회사 내부 문서를 알 수 없기 때문입니다. 예를 들어, "우리 회사의 휴가 정책은?"이라는 질문에 GPT-4는 답할 수 없지만, RAG 시스템은 사내 문서를 검색하여 정확히 답변할 수 있습니다.
이는 기업용 AI 에이전트의 필수 기능입니다. 기존에는 LLM을 파인튜닝하여 지식을 주입했다면, 이제는 RAG로 외부 지식을 실시간으로 제공합니다.
파인튜닝은 비용이 많이 들고 시간이 오래 걸리며 지식 업데이트가 어렵지만, RAG는 문서만 추가하면 즉시 반영됩니다. RAG 파이프라인의 핵심 단계는 다섯 가지입니다: 첫째, 문서를 청크로 나누어 임베딩하고 벡터 DB에 저장합니다(인덱싱).
둘째, 사용자 질문을 임베딩으로 변환합니다(쿼리 처리). 셋째, 벡터 DB에서 가장 관련 있는 청크를 검색합니다(검색).
넷째, 검색된 청크를 프롬프트에 포함시킵니다(컨텍스트 구성). 다섯째, LLM이 컨텍스트를 참고하여 답변을 생성합니다(생성).
이러한 단계들이 ChatGPT, Claude 같은 상용 AI 서비스의 핵심 아키텍처입니다.
코드 예제
from openai import OpenAI
import chromadb
client = OpenAI()
chroma_client = chromadb.PersistentClient(path="./rag_db")
class RAGPipeline:
def __init__(self, collection_name="knowledge_base"):
"""RAG 파이프라인 초기화"""
self.collection = chroma_client.get_or_create_collection(collection_name)
def index_documents(self, documents: list[str]):
"""문서를 인덱싱 (1단계: 임베딩 & 저장)"""
embeddings = []
for doc in documents:
emb = client.embeddings.create(
input=doc, model="text-embedding-3-small"
).data[0].embedding
embeddings.append(emb)
self.collection.add(
documents=documents,
embeddings=embeddings,
ids=[f"doc_{i}" for i in range(len(documents))]
)
print(f"{len(documents)}개 문서 인덱싱 완료")
def query(self, question: str, top_k: int = 3) -> str:
"""질문에 답변 (2~5단계: 검색 & 생성)"""
# 2단계: 질문 임베딩
question_emb = client.embeddings.create(
input=question, model="text-embedding-3-small"
).data[0].embedding
# 3단계: 관련 문서 검색
results = self.collection.query(
query_embeddings=[question_emb],
n_results=top_k
)
contexts = results['documents'][0]
# 4단계: 프롬프트 구성
context_text = "\n\n".join(contexts)
prompt = f"""다음 정보를 참고하여 질문에 답변하세요:
{context_text}
질문: {question}
답변:"""
# 5단계: LLM으로 답변 생성
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
# 사용 예시
rag = RAGPipeline()
# 지식 베이스 구축
rag.index_documents([
"Python 리스트는 append()로 원소를 추가합니다",
"딕셔너리는 get() 메서드로 안전하게 값을 가져올 수 있습니다",
"set()으로 리스트의 중복을 제거할 수 있습니다"
])
# 질문하기
answer = rag.query("리스트에 원소를 추가하는 방법은?")
print(f"답변: {answer}")
설명
이것이 하는 일: 위 코드는 완전한 RAG 시스템을 구현합니다. 문서를 인덱싱하고, 질문을 받아 관련 문서를 검색한 뒤, 이를 컨텍스트로 LLM에게 전달하여 답변을 생성합니다.
첫 번째로, RAGPipeline 클래스의 __init__에서 ChromaDB 컬렉션을 초기화합니다. 컬렉션 이름을 파라미터로 받아 여러 지식 베이스를 관리할 수 있습니다.
예를 들어 "python_docs", "company_policies", "product_manuals" 같은 컬렉션을 각각 만들어 도메인별로 분리할 수 있습니다. 이렇게 하면 질문 종류에 따라 적절한 컬렉션에서만 검색하여 정확도가 높아집니다.
그 다음으로, index_documents() 메서드가 지식 베이스를 구축합니다. 각 문서를 OpenAI API로 임베딩하고, ChromaDB에 저장합니다.
중요한 점은 embeddings 파라미터를 명시적으로 전달한다는 것입니다. ChromaDB의 기본 임베딩 함수보다 OpenAI 임베딩이 품질이 높기 때문입니다.
실무에서는 이 단계를 오프라인으로 처리하여, 서버 시작 시 한 번만 실행하거나 문서 업데이트 시에만 재실행합니다. 수천 개 문서를 매번 임베딩하면 비용이 너무 많이 들기 때문입니다.
세 번째로, query() 메서드가 RAG의 핵심 로직을 구현합니다. 먼저 질문을 임베딩으로 변환하는데, 이것이 유일한 실시간 API 호출입니다.
그 다음 collection.query()로 가장 관련 있는 top_k개 문서를 검색합니다. 기본값 3은 대부분의 경우 적당한데, 너무 많으면 LLM이 혼란스러워하고 너무 적으면 정보가 부족합니다.
검색 결과 documents[0]은 문서 텍스트 리스트입니다. 네 번째로, 프롬프트 구성이 매우 중요합니다.
context_text로 검색된 문서들을 "\n\n"으로 구분하여 하나의 문자열로 만듭니다. 그 다음 "다음 정보를 참고하여 질문에 답변하세요:"라는 지시와 함께 컨텍스트와 질문을 LLM에게 전달합니다.
이 프롬프트 구조가 핵심인데, "정보를 참고하여"라는 지시가 LLM이 컨텍스트 밖의 지식을 사용하지 않도록 제약하여 환각을 줄입니다. 실무에서는 "제공된 정보에 답이 없으면 '모르겠습니다'라고 답변하세요" 같은 안전 장치를 추가합니다.
마지막으로, chat.completions.create()로 GPT-4o-mini를 호출하여 최종 답변을 생성합니다. gpt-4o-mini는 빠르고 저렴하면서도 충분히 정확하여 RAG에 적합합니다.
response.choices[0].message.content에서 답변을 추출하여 반환합니다. 사용 예시에서 "리스트에 원소를 추가하는 방법은?"이라고 질문하면, 시스템이 "Python 리스트는 append()로 원소를 추가합니다" 문서를 검색하고, 이를 참고하여 "Python에서는 append() 메서드를 사용하여 리스트에 원소를 추가할 수 있습니다.
예: my_list.append(item)" 같은 답변을 생성합니다. 여러분이 이 코드를 사용하면 강력한 AI 에이전트를 만들 수 있습니다.
회사 문서, 제품 매뉴얼, 고객 FAQ 등 어떤 도메인이든 문서만 준비하면 즉시 작동합니다. LLM 파인튜닝 없이도 전문 지식을 갖춘 에이전트가 되며, 문서를 추가하거나 수정하면 즉시 지식이 업데이트됩니다.
실무에서는 여기에 스트리밍, 대화 히스토리, 출처 표시, 오류 처리를 추가하여 완전한 제품 수준으로 만듭니다. 비용도 매우 저렴한데, 1000번 질문에 약 $1 정도면 충분합니다.
실전 팁
💡 출처를 함께 반환하세요. 답변에 "출처: doc_2"처럼 어느 문서에서 정보를 가져왔는지 표시하면 사용자 신뢰도가 크게 향상됩니다. results['metadatas']에서 가져올 수 있습니다.
💡 하이브리드 검색으로 정확도를 높이세요. 시맨틱 검색 결과와 BM25 키워드 검색 결과를 결합하면 (예: RRF, Reciprocal Rank Fusion) 재현율과 정밀도가 모두 개선됩니다.
💡 프롬프트에 안전 장치를 추가하세요. "제공된 정보에 답이 없으면 '해당 정보를 찾을 수 없습니다'라고 답변하세요"라는 지시를 넣으면 환각이 90% 감소합니다.
💡 스트리밍으로 사용자 경험을 개선하세요. stream=True 옵션으로 답변을 실시간으로 표시하면 긴 답변도 지루하지 않고, 응답 시작 시간이 빨라진 것처럼 느껴집니다.
💡 대화 히스토리를 유지하세요. 이전 질문과 답변을 messages에 포함시키면 "그것은 무엇인가요?" 같은 후속 질문도 처리할 수 있습니다. 단, 토큰 제한에 주의하여 최근 5~10턴만 유지하세요.