매일 5만 건의 고객 문의를 처리하는 이커머스 AI 고객 서비스. 출시 3개월 차, 월간 LLM API 비용이 처음 예상한 금액의 400%를 초과했다. 원인은 명확했습니다. 사용자 한 명이 "내 주문 상태 알려줘"라고 입력하면, 시스템은 전체 데이터베이스의 관련 문서를 전부 컨텍스트에 밀어 넣고 있었습니다. 이것이 바로 오늘 여러분이 해결할 문제입니다.
왜 컨텍스트 길이가 비용을 좌우하는가
LLM 비용 구조는 대부분 토큰 기반입니다. 입력 토큰 수 × 모델 단가 = LLM 호출 비용. 단순한 수학이지만, 여기서 간과하기 쉬운 사실이 있습니다.
# 토큰 수에 따른 비용 비교 (Claude Sonnet 4.5 기준: $15/MTok)
❌ 비최적화: 전체 컨텍스트 전달
input_tokens = 50_000 # 50K 토큰
output_tokens = 500 # 500 토큰
cost_per_call = (input_tokens + output_tokens) * 15 / 1_000_000
print(f"비최적화 호출 비용: ${cost_per_call:.4f}")
결과: $0.7575 per call
✅ 최적화: 관련 청크만 선별
input_tokens = 3_000 # 3K 토큰 (94% 감소)
output_tokens = 500 # 500 토큰
cost_per_call = (input_tokens + output_tokens) * 15 / 1_000_000
print(f"최적화 호출 비용: ${cost_per_call:.4f}")
결과: $0.0525 per call
하루 5만 건 호출 기준 월간 비용
daily_calls = 50_000
unoptimized_monthly = cost_per_call * daily_calls * 30 # 이미 비최적화로 계산
optimized_monthly = 0.0525 * daily_calls * 30
print(f"\n비최적화 월 비용: ${unoptimized_monthly:,.2f}")
print(f"최적화 월 비용: ${optimized_monthly:,.2f}")
print(f"절감액: ${unoptimized_monthly - optimized_monthly:,.2f} (93% 절감)")
핵심은 명확합니다. 컨텍스트를 줄이면 비용이 줄어듭니다. 다만, 줄이는 과정에서 검색 품질이 저하되면 안 됩니다. 7가지 실전 전략을 하나씩 살펴보겠습니다.
전략 1: 지능형 청킹(Chunking) 설계
RAG 시스템의 첫 번째 병목은 대개 잘못된 청킹 전략입니다. 너무 크면 불필요한 컨텍스트가 포함되고, 너무 작으면 맥락이 분절됩니다.
# HolySheep AI를 사용한 최적의 청킹 구현
import os
from openai import OpenAI
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
def semantic_chunking(text: str, max_tokens: int = 512, overlap: int = 64) -> list[dict]:
"""
의미론적 청킹: 문장 경계와 의미 단위를 고려하여 분할
- max_tokens: 청크당 최대 토큰 (일반적으로 512 이하 권장)
- overlap: 인접 청크 간 중첩 토큰 수 (맥락 유지를 위해)
"""
# 문장 단위로 분리
sentences = text.replace('!', '.').replace('?', '.').split('.')
chunks = []
current_chunk = []
current_tokens = 0
for sentence in sentences:
sentence = sentence.strip()
if not sentence:
continue
# 토큰 추정 (한국어 기준 1글자 ≈ 1~2 토큰)
estimated_tokens = len(sentence) // 2
if current_tokens + estimated_tokens > max_tokens:
# 현재 청크 저장
if current_chunk:
chunk_text = '. '.join(current_chunk) + '.'
chunks.append({
"content": chunk_text,
"tokens": current_tokens,
"index": len(chunks)
})
# 중첩 적용 (마지막 문장 다음 청크에 포함)
current_chunk = current_chunk[-2:] if len(current_chunk) > 2 else current_chunk
current_tokens = sum(len(s) // 2 for s in current_chunk)
current_chunk.append(sentence)
current_tokens += estimated_tokens
# 마지막 청크 추가
if current_chunk:
chunk_text = '. '.join(current_chunk) + '.'
chunks.append({
"content": chunk_text,
"tokens": current_tokens,
"index": len(chunks)
})
return chunks
사용 예시
sample_product_review = """
이 제품의 배송 속도가 매우 빠르습니다. 주문 다음 날에 도착했어요.
포장이 깨끗했고 제품 상태도 완벽했습니다. 다만 설명서 없이 사용하기가 조금 어려웠습니다.
가격 대비 품질은 만족스럽습니다. 다음에 또 구매하고 싶습니다.
"""
chunks = semantic_chunking(sample_product_review)
print(f"생성된 청크 수: {len(chunks)}")
for i, chunk in enumerate(chunks):
print(f"\n[청크 {i+1}] 토큰 수: {chunk['tokens']}")
print(f" 내용: {chunk['content'][:80]}...")
전략 2: 하이브리드 검색 + 상위-K 필터링
단일 검색 방법(Semantic Only 또는 Keyword Only)보다 하이브리드 검색이 더 정확한 결과를 제공하면서도 불필요한 결과를 걸러냅니다.
# 하이브리드 검색 + BM25 + 유사도 상위-K 필터링
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
def hybrid_search(query: str, documents: list[str], top_k: int = 3,
semantic_weight: float = 0.6) -> list[dict]:
"""
BM25(키워드) + 코사인 유사도(시맨틱)를 결합한 하이브리드 검색
- semantic_weight: 시맨틱 검색 가중치 (0~1)
- top_k: 최종 반환할 상위 결과 수 (LLM 컨텍스트 길이 제어를 위해 필수)
"""
# 1단계: BM25 스코어 계산 (키워드 매칭)
bm25_scores = bm25_score(query, documents)
# 2단계: 코사인 유사도 계산 (의미론적 유사도)
semantic_scores = semantic_score(query, documents)
# 3단계: 결합 스코어 계산
combined_scores = (
(1 - semantic_weight) * normalize(bm25_scores) +
semantic_weight * normalize(semantic_scores)
)
# 4단계: 상위-K 선별 (컨텍스트 길이 제한 핵심)
top_indices = np.argsort(combined_scores)[::-1][:top_k]
results = []
for rank, idx in enumerate(top_indices):
results.append({
"rank": rank + 1,
"document_id": idx,
"content": documents[idx],
"combined_score": float(combined_scores[idx]),
"bm25_score": float(bm25_scores[idx]),
"semantic_score": float(semantic_scores[idx])
})
return results
def bm25_score(query: str, documents: list[str]) -> np.ndarray:
"""BM25 알고리즘으로 키워드 매칭 스코어 계산"""
vectorizer = TfidfVectorizer(stop_words='english')
tfidf_matrix = vectorizer.fit_transform(documents + [query])
query_vec = tfidf_matrix[-1]
doc_vecs = tfidf_matrix[:-1]
scores = cosine_similarity(doc_vecs, query_vec).flatten()
return scores
def semantic_score(query: str, documents: list[str]) -> np.ndarray:
"""OpenAI 임베딩을 사용한 시맨틱 유사도 계산"""
response = client.embeddings.create(
model="text-embedding-3-small",
input=[query] + documents
)
query_embedding = np.array(response.data[0].embedding)
doc_embeddings = np.array([d.embedding for d in response.data[1:]])
similarities = cosine_similarity(
[query_embedding], doc_embeddings
).flatten()
return similarities
def normalize(scores: np.ndarray) -> np.ndarray:
"""스코어 Min-Max 정규화 (0~1 범위)"""
min_s, max_s = scores.min(), scores.max()
if max_s - min_s == 0:
return np.zeros_like(scores)
return (scores - min_s) / (max_s - min_s)
실제 사용 예시
documents = [
"배송이 매우 빠르고 포장도 꼼꼼했습니다. 다음에도 구매하겠습니다.",
"상품 설명과 실제 제품의 색상이 조금 달랐습니다.",
"고객센터 응대가 친절하고 반품 처리도 빠르게 되었습니다.",
"가격 대비 품질이 우수하며 재구매 의향이 있습니다.",
"결제 시스템에 오류가 있어 주문을 여러 번 시도해야 했습니다."
]
query = "배송 관련 고객 후기"
results = hybrid_search(query, documents, top_k=2, semantic_weight=0.7)
print("=== 하이브리드 검색 결과 ===")
for result in results:
print(f"\n#{result['rank']} (결합 스코어: {result['combined_score']:.3f})")
print(f" BM25: {result['bm25_score']:.3f} | 시맨틱: {result['semantic_score']:.3f}")
print(f" 내용: {result['content']}")
전략 3: 쿼리 압축(Query Compression)
사용자 쿼리가 장황하거나 모호할수록 검색 결과가 부정확해지고 더 많은 문서를 포함해야 합니다. 쿼리를 압축하면 검색 효율성과 정확도가 동시에 향상됩니다.
# LLM 기반 쿼리 압축으로 검색 효율성 향상
def compress_query(user_query: str, compression_ratio: float = 0.4) -> str:
"""
LLM을 사용하여 사용자 쿼리를 핵심 의도로 압축
- 원본 쿼리가 길수록 압축 이점 증가
- 불필요한 수사, 반복, 잡담 제거
"""
system_prompt = """당신은 쿼리 압축 전문가입니다.
사용자의 자연어 질문을 핵심 검색 의도로 변환하세요.
규칙:
1. 핵심 키워드와 의도만 유지
2. 불필요한 인사, 수사적 표현, 반복 제거
3. 검색에 유리한 짧은 형태로 변환
4. 한국어로 출력"""
response = client.chat.completions.create(
model="gpt-4.1",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"원본 쿼리: {user_query}"}
],
temperature=0.3, # 일관된 결과
max_tokens=50 # 짧게 유지
)
compressed = response.choices[0].message.content.strip()
return compressed
비교 테스트
test_queries = [
"안녕하세요 선생님, 혹시 제가 이번에 주문한产品在 배송중인데 언제쯤 받을 수 있을까요?",
"어제 밤에 세탁기에 빨래를 돌렸는데 아침에 확인해보니까 이상하게 소리가 나요. 이거 고장인 건가요?",
"저기요~ 혹시 이번 달 달력에 빨간 날이 있는지 확인하고 싶은데 알려주세요!"
]
print("=== 쿼리 압축 결과 ===\n")
for query in test_queries:
compressed = compress_query(query)
original_tokens = len(query) // 2
compressed_tokens = len(compressed) // 2
reduction = (1 - compressed_tokens / original_tokens) * 100
print(f"원본: {query}")
print(f"압축: {compressed}")
print(f"토큰 감소: {reduction:.1f}%\n")
전략 4: Reranker로 관련성 재순위 매기기
초기 검색에서 가져온 후보 문서가 20개일 때, 전부 LLM에 보내면 비용이 폭발합니다. Reranker는 가장 관련성 높은 2~5개만 선별합니다.
# Reranker를 활용한 정밀 필터링
def rerank_documents(query: str, retrieved_docs: list[str],
top_n: int = 3) -> list[dict]:
"""
교차 인코더 Reranker로 문서 재순위 매기기
- retrieved_docs: 초기 검색 결과 (보통 10~20개)
- top_n: 최종 LLM에 전달할 문서 수 (보통 2~5개)
"""
from sentence_transformers import CrossEncoder
# Cross-Encoder 모델 로드 (고품질 재순위)
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
# 쿼리-문서 쌍 생성
pairs = [(query, doc) for doc in retrieved_docs]
# 일괄 재순위 scoring
scores = reranker.predict(pairs)
# 상위 N개 선별
ranked_indices = np.argsort(scores)[::-1][:top_n]
results = []
for rank, idx in enumerate(ranked_indices):
results.append({
"rank": rank + 1,
"document": retrieved_docs[idx],
"rerank_score": float(scores[idx]),
"original_position": idx
})
return results
파이프라인 통합
def optimized_rag_pipeline(query: str, vector_store: list[str]) -> dict:
"""
완전한 최적화 RAG 파이프라인:
1. 쿼리 압축 → 2. 하이브리드 검색 → 3. Reranker 필터링 → 4. LLM 응답
"""
# 단계 1: 쿼리 압축
compressed_query = compress_query(query)
# 단계 2: 초기 검색 (넉넉하게 20개)
initial_results = hybrid_search(
compressed_query, vector_store, top_k=20
)
retrieved_docs = [r["content"] for r in initial_results]
# 단계 3: Reranker로 3개로 축소
final_docs = rerank_documents(compressed_query, retrieved_docs, top_n=3)
# 단계 4: LLM 응답 생성
context = "\n".join([f"[{d['rank']}] {d['document']}"
for d in final_docs])
response = client.chat.completions.create(
model="gpt-4.1",
messages=[
{"role": "system", "content": "제공된 참고 문서를 바탕으로 정확하게 답변하세요."},
{"role": "user", "content": f"질문: {query}\n\n참고:\n{context}"}
],
max_tokens=500,
temperature=0.5
)
return {
"compressed_query": compressed_query,
"retrieved_count": len(retrieved_docs),
"final_count": len(final_docs),
"response": response.choices[0].message.content
}
print("=== 최적화 RAG 파이프라인 ===")
result = optimized_rag_pipeline(
"배송이 늦어지고 있어요. 주문번호 12345번 상태 알려주세요.",
documents
)
print(f"압축된 쿼리: {result['compressed_query']}")
print(f"검색 후 후보: {result['retrieved_count']}개")
print(f"LLM 전달 문서: {result['final_count']}개")
print(f"응답: {result['response']}")
전략 5: 계층적(Hierarchical) 검색 아키텍처
대규모 문서 데이터베이스에서 전체를 단일 레벨로 검색하면 효율이 떨어집니다. 카탈로그 → 서브카테고리 → 문서 구조의 계층적 접근이 검색 품질과 비용 모두를 개선합니다.
# 계층적 검색 시스템
class HierarchicalSearch:
def __init__(self, catalog: list[dict], subcatalog_index: dict,
document_index: dict):
self.catalog = catalog # 상위 카테고리 임베딩
self.subcatalog_index = subcatalog_index # 중위 인덱스
self.document_index = document_index # 하위 문서
def search(self, query: str, depth: int = 2) -> list[str]:
"""
계층적 검색:
- depth=1: 카탈로그 수준만 검색 (매우 넓은 범위)
- depth=2: 카탈로그 → 문서 (표준)