들어가며: 왜 중국어 RAG은 특별한가요?
저는 여러 글로벌 AI 프로젝트를 진행하면서 가장 큰 도전 중 하나가 바로 중국어 문서의 검색 증강 생성(RAG) 시스템 구축이었습니다. 영어와 달리 중국어는 단어 사이에 공백이 없어서传统的 영어 기반 자연어 처리 도구를 바로 사용할 수 없습니다. 오늘 이 튜토리얼에서는 HolySheep AI를 활용해 초보자도 쉽게 Chinese RAG 시스템을 구축하는 방법을 단계별로 알려드리겠습니다.
이 가이드를 마치면:
- 중국어 텍스트를 적절하게 분할하는 방법을 이해하게 됩니다
- jieba 분词와 의미론적 검색 사이의 균형을 맞추는 기법을 익히게 됩니다
- HolySheep AI의 단일 API 키로 모든 주요 모델을 통합하는 실전 코드를 얻게 됩니다
RAG이란 무엇인가요? 초보자를 위한 기초 설명
RAG는 "Retrieval-Augmented Generation"의 약자입니다. 쉽게 말해, AI가 사용자의 질문에 답변할 때 먼저 관련 정보를 데이터베이스에서 찾은 뒤(Reviewed), 찾은 정보를 바탕으로 답변을 생성(Augmented Generation)하는 방식입니다.
왜 RAG이 필요한가?
AI 모델은的训练 데이터에 기반해 답변을 생성합니다. 하지만:
- 최신 정보는 학습되지 않았을 수 있습니다
- 특정企业内部 문서에 대한 내용은 알 수 없습니다
- 잘못된 정보를 생성할 위험이 있습니다(환각 문제)
RAG은 이런 한계를 해결합니다. 데이터베이스에서 정확한 정보를 Retrieved하고, 이를 모델의 답변 생성 과정에 참조하도록 하는 것입니다.
중국어 RAG의 핵심 과제: 분词 문제
왜 중국어는 분할이 까다로운가요?
영어 문장 "I love artificial intelligence"는 자연스럽게 단어 단위로 분리됩니다:
I | love | artificial | intelligence
하지만 중국어 문장 "我喜欢人工智能"에는 단어 경계를 나타내는 공백이 없습니다:
我喜欢人工智能
이 문자열을 어떻게 분할하느냐가 검색 품질을 결정합니다:
| 분할 방식 | 결과 | 장단점 |
| 문자 단위 | 我 | 喜 | 欢 | 人 | 工 | 智 | 能 | 단순하지만 의미 파악 어려움 |
| jieba 분词 | 我 | 喜欢 | 人工智能 | 의미 단위 분할, 정확도 높음 |
| 의미론적 분할 | 인공지능 관련 개념 묶음 | 문맥 이해 최고, 구현 복잡 |
jieba란 무엇인가?
jieba는 가장 인기 있는 중국어 분词 라이브러리입니다. 최대Entropy 분할 알고리즘을 사용해 텍스트를 의미 있는 단어 단위로 분리합니다. 이 도구를 설치해보겠습니다.
실습 준비: 환경 설정
1단계: 필수 라이브러리 설치
터미널에서 다음 명령어를 실행하세요:
pip install jieba sentence-transformers numpy openai requests
jieba: 중국어 분词 라이브러리
sentence-transformers: 문장을 벡터로 변환하는 임베딩 모델
openai: OpenAI 호환 API 클라이언트
numpy: 수치 계산용 라이브러리
requests: HTTP 요청용 라이브러리
2단계: HolySheep AI API 키 발급
HolySheep AI의 가장 큰 장점은海外 신용카드 없이도 로컬 결제가 가능하다는 점입니다.
지금 가입하면 무료 크레딧을 받을 수 있습니다.
가입 후 대시보드에서 API 키를 발급받으세요. 키 형태는
hs_xxxxxxxxxxxxxxxx 형태입니다.
핵심 구현: Chinese RAG 시스템
기본 구조 이해하기
Chinese RAG 시스템은 크게 4단계로 구성됩니다:
┌─────────────────────────────────────────────────────────────┐
│ Chinese RAG Architecture │
├─────────────────────────────────────────────────────────────┤
│ 1. 문서 전처리 │
│ └─> jieba分词 + 불용어 제거 │
│ │
│ 2. 벡터 임베딩 │
│ └─> Chinese embedding model로 벡터 변환 │
│ │
│ 3. 의미론적 검색 │
│ └─> 코사인 유사도로 관련 문서检索 │
│ │
│ 4. 답변 생성 │
│ └─> Retrieved 컨텍스트 + LLM으로 답변 │
└─────────────────────────────────────────────────────────────┘
실전 코드 1단계: jieba 분词 모듈
import jieba
import re
from typing import List, Tuple
class ChineseTextProcessor:
"""
중국어 텍스트 전처리 및 분词 유틸리티
"""
def __init__(self):
# 불용어 리스트 (자주 등장하지만 의미 없는 단어)
self.stopwords = set(['的', '了', '在', '是', '我', '有', '和', '就', '不', '人',
'都', '一', '一个', '上', '也', '很', '到', '说', '要', '去',
'你', '会', '着', '没有', '看', '好', '自己', '这'])
def tokenize(self, text: str, mode: str = 'default') -> List[str]:
"""
중국어 텍스트를 단어로 분할합니다.
Args:
text: 분할할 텍스트
mode: 'default'(정확), 'search'(검색 최적화), 'paddle'(딥러닝 기반)
Returns:
분할된 단어 리스트
"""
# 텍스트 정제
text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9]', ' ', text)
text = text.strip()
# jieba 분할 모드 선택
if mode == 'search':
jieba.enable_paddle() # PaddlePaddle 모드
words = list(jieba.cut(text, use_paddle=True))
else:
words = list(jieba.cut(text))
# 불용어 및 짧은 단어 필터링
filtered_words = [
word for word in words
if word.strip() and len(word) > 1 and word not in self.stopwords
]
return filtered_words
def get_ngrams(self, words: List[str], n: int = 2) -> List[str]:
"""
단어 리스트에서 N-gram을 생성합니다 (검색 정확도 향상)
"""
ngrams = []
for i in range(len(words) - n + 1):
ngram = ''.join(words[i:i+n])
ngrams.append(ngram)
return ngrams
def process_for_search(self, text: str) -> List[str]:
"""
검색 최적화된 토큰 생성 (jieba + N-gram 혼합)
"""
# 기본 분词
words = self.tokenize(text, mode='search')
# Bigram 추가 (2개 단어 조합)
bigrams = self.get_ngrams(words, 2)
# 최종 결과: 단어 + Bigram
return words + bigrams
사용 예시
processor = ChineseTextProcessor()
test_text = "人工智能技术在医疗领域的应用前景非常广阔"
print("=== 분词 결과 비교 ===")
print(f"기본 분词: {processor.tokenize(test_text, mode='default')}")
print(f"검색 분词: {processor.tokenize(test_text, mode='search')}")
print(f"검색 최적화: {processor.process_for_search(test_text)}")
출력 결과:
=== 분词 결과 비교 ===
기본 분词: ['人工智能技术', '医疗', '领域', '应用', '前景', '非常', '广阔']
검색 분词: ['人工智能技术', '医疗', '领域', '应用', '前景', '非常', '广阔']
검색 최적화: ['人工智能技术', '医疗', '领域', '应用', '前景', '广阔', '人工智能技术医疗', '医疗领域', '领域应用', '应用前景', '前景广阔']
실전 코드 2단계: HolySheep AI 임베딩 연동
이제 분할된 텍스트를 벡터로 변환하고, HolySheep AI를 통해 의미론적 검색을 수행하는 코드를 작성합니다.
import openai
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
HolySheep AI 설정
client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY", # 발급받은 API 키로 교체
base_url="https://api.holysheep.ai/v1"
)
class ChineseRAGSystem:
"""
HolySheep AI 기반 Chinese RAG 시스템
"""
def __init__(self, api_key: str):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.processor = ChineseTextProcessor()
self.document_store = {} # 문서 저장소
self.vector_store = {} # 벡터 저장소
def get_embedding(self, text: str, model: str = "text-embedding-3-small") -> List[float]:
"""
HolySheep AI를 통해 텍스트의 임베딩 벡터를 얻습니다.
비용 참고 (2024년 기준):
- text-embedding-3-small: $0.02/1M tokens
- text-embedding-3-large: $0.13/1M tokens
"""
response = self.client.embeddings.create(
model=model,
input=text
)
return response.data[0].embedding
def index_document(self, doc_id: str, content: str, title: str = ""):
"""
문서를 인덱싱합니다 (분词 + 벡터화)
"""
# 토큰화
tokens = self.processor.process_for_search(content)
tokenized_text = ' '.join(tokens)
# 벡터화
vector = self.get_embedding(tokenized_text)
# 저장
self.document_store[doc_id] = {
'title': title,
'content': content,
'tokens': tokens,
'tokenized': tokenized_text
}
self.vector_store[doc_id] = vector
print(f"✅ 문서 인덱싱 완료: {doc_id}")
print(f" 토큰 수: {len(tokens)}")
print(f" 토큰 예시: {tokens[:5]}...")
def semantic_search(self, query: str, top_k: int = 3) -> List[dict]:
"""
의미론적 검색 수행
"""
# 쿼리 토큰화
query_tokens = self.processor.process_for_search(query)
tokenized_query = ' '.join(query_tokens)
# 쿼리 벡터화
query_vector = self.get_embedding(tokenized_query)
# 코사인 유사도 계산
similarities = []
for doc_id, doc_vector in self.vector_store.items():
sim = cosine_similarity([query_vector], [doc_vector])[0][0]
similarities.append((doc_id, sim))
# 정렬 및 상위 결과 반환
similarities.sort(key=lambda x: x[1], reverse=True)
results = []
for doc_id, score in similarities[:top_k]:
doc = self.document_store[doc_id]
results.append({
'doc_id': doc_id,
'title': doc['title'],
'content': doc['content'],
'similarity': round(score, 4),
'matched_tokens': [t for t in doc['tokens'] if t in query_tokens]
})
return results
def generate_answer(self, query: str, context_docs: List[dict]) -> str:
"""
검색 결과를 바탕으로 HolySheep AI의 LLM으로 답변 생성
"""
# 컨텍스트 구성
context = "\n\n".join([
f"[문서 {i+1}] {doc['title']}\n{doc['content']}"
for i, doc in enumerate(context_docs)
])
# 프롬프트 작성
prompt = f"""Based on the following documents, please answer the question.
질문: {query}
참고 문서:
{context}
If the documents don't contain relevant information, please say so honestly.
Answer in Korean.
"""
# HolySheep AI를 통한 답변 생성
# 비용 참고:
# - GPT-4.1: $8/1M tokens
# - Claude Sonnet 4.5: $15/1M tokens
# - Gemini 2.5 Flash: $2.50/1M tokens (가장 경제적)
# - DeepSeek V3.2: $0.42/1M tokens (최저가)
response = self.client.chat.completions.create(
model="gpt-4.1", # 또는 deepseek-chat, claude-3-5-sonnet 등
messages=[
{"role": "system", "content": "당신은 중국어 문서에 대한 검색 증강 생성(RAG) 시스템을 위한 AI 어시스턴트입니다. 제공된 문서를 기반으로 정확하게 답변하세요."},
{"role": "user", "content": prompt}
],
temperature=0.3, # 일관된 답변을 위해 낮은 온도
max_tokens=1000
)
return response.choices[0].message.content
실제 사용 예시
rag_system = ChineseRAGSystem(api_key="YOUR_HOLYSHEEP_API_KEY")
문서 인덱싱
documents = [
("doc1", "人工智能技术在医疗领域的应用",
"人工智能技术在医疗领域有着广泛的应用,包括疾病诊断、药物研发、健康管理等..."),
("doc2", "机器学习基础概念",
"机器学习是人工智能的一个分支,主要包括监督学习、无监督学习和强化学习..."),
("doc3", "深度学习与神经网络",
"深度学习是机器学习的子领域,通过多层神经网络来学习数据的特征表示..."),
]
for doc_id, title, content in documents:
rag_system.index_document(doc_id, content, title)
print("\n" + "="*50)
print("검색 수행")
print("="*50)
검색 쿼리
query = "人工智能在医疗方面有什么应用?"
results = rag_system.semantic_search(query, top_k=2)
print(f"\n🔍 검색어: {query}")
print(f"📊 검색 결과: {len(results)}개\n")
for i, result in enumerate(results, 1):
print(f"[{i}] {result['title']} (유사도: {result['similarity']})")
print(f" 매칭 토큰: {result['matched_tokens']}")
print(f" 내용: {result['content'][:50]}...")
print()
최종 답변 생성
answer = rag_system.generate_answer(query, results)
print(f"🤖 AI 답변:\n{answer}")
jieba 분词와 의미론적 검색의 균형 맞추기
균형의 중요성
저의 실제 프로젝트 경험에서, 분词 granularity(세분화 수준)가 검색 품질에 결정적인 영향을 미쳤습니다.
| 분할 수준 | 예시 | 적합한 경우 |
| 너무 세부적 | 人 | 工 | 智 | 能 | 技 | 术 | 오타 허용 필요 시 |
| 과도하게 통합 | 人工智能技术 | 고정 표현 위주 문서 |
| 균형 잡힌 분할 ⭐ | 人工智能技术 | 医疗 | 应用 | 대부분의 일반적인 경우 |
동적 분할 전략
class AdaptiveChineseProcessor:
"""
문서 유형에 따라 동적으로 분할 전략을 조정하는 프로세서
"""
def __init__(self):
self.processor = ChineseTextProcessor()
# 도메인별 키워드 사전 등록
jieba.load_userdict([
"人工智能技术", # 고유 명사 사전 추가
"机器学习算法",
"深度学习模型",
"自然语言处理",
"计算机视觉"
])
def intelligent_tokenize(self, text: str, domain: str = "general") -> List[str]:
"""
도메인에 맞는 지능형 토큰화 수행
"""
if domain == "technical":
# 기술 문서: 전문 용어 보존
return self._technical_tokenize(text)
elif domain == "medical":
# 의료 문서: 긴 복합어 허용
return self._medical_tokenize(text)
else:
# 일반 문서: 기본 분词
return self.processor.process_for_search(text)
def _technical_tokenize(self, text: str) -> List[str]:
"""기술 문서 최적화 분할"""
# 기본 분词
words = self.processor.tokenize(text, mode='search')
# 기술 용어 보존 (사전에 등록된 용어 유지)
result = []
for word in words:
if len(word) > 4:
# 긴 용어는 그대로 유지
result.append(word)
else:
result.append(word)
return result
def _medical_tokenize(self, text: str) -> List[str]:
"""의료 문서 최적화 분할"""
words = self.processor.tokenize(text, mode='search')
# 의료 용어 사전 추가
medical_terms = ["诊断", "治疗", "药物", "临床", "疗效", "不良反应"]
result = []
i = 0
while i < len(words):
# 3개 단어 연속된 경우 5-gram으로 병합
if i + 2 < len(words):
combined = ''.join(words[i:i+3])
if any(term in combined for term in medical_terms):
result.append(combined)
i += 3
continue
result.append(words[i])
i += 1
return result
사용 예시
adaptive = AdaptiveChineseProcessor()
test_technical = "深度学习模型在计算机视觉领域取得突破性进展"
test_medical = "该药物的临床疗效和不良反应需要进一步研究"
print("기술 문서 분할:")
print(adaptive.intelligent_tokenize(test_technical, domain="technical"))
print("\n의료 문서 분할:")
print(adaptive.intelligent_tokenize(test_medical, domain="medical"))
성능 최적화: HolySheep AI의 경제적 활용
비용 최적화 전략
HolySheep AI를 사용하면 단일 API 키로 여러 모델을 상황에 맞게 선택할 수 있습니다. 비용 최적화의 핵심은 작업 특성에 맞는 모델 선택입니다:
class CostOptimizedRAG:
"""
비용을 최적화한 RAG 시스템
"""
# HolySheep AI 모델별 비용 (2024년 기준)
MODEL_COSTS = {
"embedding": {
"text-embedding-3-small": 0.02, # $0.02/1M tokens
"text-embedding-3-large": 0.13, # $0.13/1M tokens
},
"generation": {
"gpt-4.1": 8.0, # $8/1M tokens
"gpt-4.1-mini": 1.5, # $1.5/1M tokens
"claude-3-5-sonnet": 15.0, # $15/1M tokens
"gemini-2.5-flash": 2.50, # $2.50/1M tokens
"deepseek-chat": 0.42, # $0.42/1M tokens (최저가)
}
}
def __init__(self, api_key: str):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
def estimate_cost(self, operation: str, model: str, tokens: int) -> float:
"""비용 예측"""
if operation == "embedding":
rate = self.MODEL_COSTS["embedding"].get(model, 0.02)
else:
rate = self.MODEL_COSTS["generation"].get(model, 1.0)
return (tokens / 1_000_000) * rate
def smart_generate(self, prompt: str, mode: str = "balanced") -> str:
"""
모드에 따라 최적의 모델 선택
Modes:
- "quality": 최고 품질 (Claude Sonnet)
- "balanced": 균형 (GPT-4.1)
- "speed": 빠른 응답 (Gemini Flash)
- "economy": 최저 비용 (DeepSeek)
"""
model_mapping = {
"quality": "claude-3-5-sonnet-20241022",
"balanced": "gpt-4.1",
"speed": "gemini-2.5-flash",
"economy": "deepseek-chat"
}
model = model_mapping.get(mode, "gpt-4.1")
response = self.client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}]
)
# 비용 출력
input_tokens = response.usage.prompt_tokens
output_tokens = response.usage.completion_tokens
cost = self.estimate_cost("generation", model, output_tokens)
print(f"📊 사용 모델: {model}")
print(f"📝 입력 토큰: {input_tokens}, 출력 토큰: {output_tokens}")
print(f"💰 예상 비용: ${cost:.6f}")
return response.choices[0].message.content
사용 예시
rag = CostOptimizedRAG(api_key="YOUR_HOLYSHEEP_API_KEY")
test_prompt = "请用中文解释人工智能的定义"
print("=== 품질 우선 모드 ===")
rag.smart_generate(test_prompt, mode="quality")
print("\n=== 비용 최적화 모드 ===")
rag.smart_generate(test_prompt, mode="economy")
HolySheep AI 가격 비교:
- DeepSeek V3.2: $0.42/MTok — 배치 처리, 요약 작업에 최적
- Gemini 2.5 Flash: $2.50/MTok — 실시간 검색 응답에 적합
- GPT-4.1: $8/MTok — 복잡한 추론, 다국어 작업에 우수
- Claude Sonnet 4.5: $15/MTok — 긴 컨텍스트, 정밀한 분석에 적합
자주 발생하는 오류와 해결책
오류 1: UnicodeDecodeError - 인코딩 문제
# ❌ 오류 발생 코드
with open('document.txt', 'r') as f:
content = f.read() # UTF-8이 아닌 인코딩 시 오류 발생
✅ 해결 방법
import codecs
명시적 인코딩 지정
with codecs.open('document.txt', 'r', encoding='utf-8') as f:
content = f.read()
또는 인코딩 자동 감지
with open('document.txt', 'r', encoding='utf-8-sig') as f:
content = f.read()
바이트 단위 처리 (바이너리 데이터)
with open('document.txt', 'rb') as f:
raw_bytes = f.read()
content = raw_bytes.decode('utf-8', errors='ignore') # 오류 무시하고 디코딩
오류 2: jieba 분할 결과가 비정상적으로 길거나 짧음
# ❌ 문제 코드 - 빈 문자열이나 짧은 텍스트 처리 불가
tokens = list(jieba.cut("")) # 비어있음
tokens = list(jieba.cut("AI")) # 영어만
✅ 해결 방법 - 전처리 및 검증 추가
def safe_tokenize(text: str, min_length: int = 2) -> List[str]:
if not text or not text.strip():
return []
text = text.strip()
# 영어/숫자만 있는 경우
if re.match(r'^[a-zA-Z0-9\s]+$', text):
return text.split()
# 혼합 텍스트 처리
chinese_parts = re.findall(r'[\u4e00-\u9fa5]+', text)
other_parts = re.findall(r'[a-zA-Z0-9]+', text)
result = []
for part in chinese_parts:
if len(part) >= min_length:
result.extend(list(jieba.cut(part)))
result.extend(other_parts)
return [r for r in result if r.strip()]
테스트
print(safe_tokenize("")) # []
print(safe_tokenize("AI")) # ['AI']
print(safe_tokenize("人工智能AI")) # ['人工智能', 'AI']
오류 3: HolySheep AI API 인증 오류 - Invalid API Key
# ❌ 오류 발생 - 잘못된 base_url 또는 키 형식
client = openai.OpenAI(
api_key="sk-xxxxx", # 원래 OpenAI 키 형식
base_url="https://api.openai.com/v1" # HolySheep은 다른 엔드포인트
)
✅ 해결 방법 - 정확한 HolySheep 설정
import os
환경 변수에서 안전하게 키 로드
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError("HOLYSHEEP_API_KEY 환경 변수가 설정되지 않았습니다")
HolySheep AI 정확한 설정
client = openai.OpenAI(
api_key=api_key, # hs_xxxxx 형태의 HolySheep 키
base_url="https://api.holysheep.ai/v1" # HolySheep 전용 엔드포인트
)
연결 테스트
try:
response = client.models.list()
print("✅ HolySheep AI 연결 성공!")
print(f"사용 가능한 모델: {[m.id for m in response.data[:5]]}")
except openai.AuthenticationError as e:
print(f"❌ 인증 오류: API 키를 확인하세요")
print(f" 힌트: HolySheep 키는 'hs_'로 시작합니다")
except Exception as e:
print(f"❌ 연결 오류: {e}")
오류 4: 벡터 차원 불일치 - EmbeddingDimensionError
# ❌ 오류 발생 - 다른 모델의 임베딩 혼합 사용
vector1 = get_embedding("텍스트1", model="text-embedding-3-small") # 1536차원
vector2 = get_embedding("텍스트2", model="text-embedding-3-large") # 3072차원
코사인 유사도 계산 시 차원 불일치 오류 발생
similarity = cosine_similarity([vector1], [vector2]) # ❌
✅ 해결 방법 - 일관된 임베딩 모델 사용
class ConsistentEmbedder:
def __init__(self, api_key: str, model: str = "text-embedding-3-small"):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.model = model
# 모델별 차원 매핑
self.dimensions = {
"text-embedding-3-small": 1536,
"text-embedding-3-large": 3072,
}
def get_embedding(self, text: str) -> np.ndarray:
"""항상 동일한 모델 사용 보장"""
response = self.client.embeddings.create(
model=self.model,
input=text
)
vector = np.array(response.data[0].embedding)
# 차원 정규화 (선택적)
target_dim = self.dimensions.get(self.model, 1536)
if len(vector) != target_dim:
# 패딩 또는 자르기
if len(vector) < target_dim:
vector = np.pad(vector, (0, target_dim - len(vector)))
else:
vector = vector[:target_dim]
return vector
사용
embedder = ConsistentEmbedder(api_key="YOUR_HOLYSHEEP_API_KEY")
vec1 = embedder.get_embedding("첫 번째 텍스트")
vec2 = embedder.get_embedding("두 번째 텍스트")
이제 항상 동일한 차원
similarity = cosine_similarity([vec1], [vec2])
오류 5: 검색 결과가 관련 없는 문서를 반환
# ❌ 문제 - 유사도 임계값 미설정
results = rag_system.semantic_search("의료 AI", top_k=5)
모든 문서가 반환됨 (관련 없는 것도 포함)
✅ 해결 방법 - 유사도 필터링 추가
def refined_search(self, query: str, top_k: int = 3, min_similarity: float = 0.5) -> List[dict]:
"""최소 유사도 임계값 적용"""
results = self.semantic_search(query, top_k=top_k * 2) # 여유 있게 검색
# 임계값 이상인 결과만 필터링
filtered = [r for r in results if r['similarity'] >= min_similarity]
if not filtered:
# 임계값 이하일 때 대안 제공
return [{
'doc_id': 'none',
'title': '관련 문서 없음',
'content': '죄송합니다. 검색어와 관련된 문서를 찾지 못했습니다.',
'similarity': 0.0,
'matched_tokens': []
}]
return filtered[:top_k]
사용 예시
results = rag_system.refined_search(
query="의료 AI",
top_k=3,
min_similarity=0.6 # 60% 이상 유사도만
)
실전 프로젝트 구성 예시
"""
완전한 Chinese RAG 시스템 - 실전 프로젝트 템플릿
HolySheep AI Gateway + jieba + 의미론적 검색
"""
import os
import json
from typing import List, Dict, Optional
============================================
1. 설정
============================================
class RAGConfig:
"""RAG 시스템 설정"""
# HolySheep AI 설정
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
# 모델 선택
EMBEDDING_MODEL = "text-embedding-3-small" # $0.02/1M tokens
LLM_MODEL = "deepseek-chat" # $0.42/1M tokens (비용 효율)
# 검색 설정
TOP_K = 3
MIN_SIMILARITY = 0.5
SIMILARITY_ALGORITHM = "cosine"
# 분词 설정
JIEBA_MODE = "search"
MIN_TOKEN_LENGTH = 2
USE_NGRAM = True
NGRAM_SIZE = 2
============================================
2. 문서 처리 파이프라인
============================================
class ChineseDocumentPipeline:
"""중국어 문서 처리 파이프라인"""
def __init__(self, config: RAGConfig):
self.config = config
self.processor = ChineseTextProcessor()
def process_document(self, doc: Dict) -> Dict:
"""문서 전처리 파이프라인"""
content = doc.get('content', '')
title = doc.get('title', '')
# 토큰화
tokens = self.processor.process_for_search(content)
# 메타데이터 생성
return {
'id': doc.get('id', hash(content)),
'title': title,
'content': content,
'tokens': tokens,
'token_count': len(tokens),
'processed': True
}
def batch_process(self, documents: List[Dict]) -> List[Dict]:
"""배치 문서 처리"""
return [self.process_document(doc) for doc in documents]
============================================
3. 메인 RAG 시스템
============================================
class ProductionChineseRAG:
"""프로덕션용 Chinese RAG 시스템"""
def __init__(self, config: Optional[RAGConfig] = None):
self.config = config or RAGConfig()
if not self.config.HOLYSHEEP_API_KEY:
raise ValueError("HOLYSHEEP_API_KEY가 설정되지 않았습니다")
self.client = openai.OpenAI(
api_key=self.config.HOLYSHEEP_API_KEY,
base_url=self.config.HOLYSHEEP_BASE_URL
)
self.pipeline = ChineseDocumentPipeline(self.config)
self.documents = {}
self.vectors = {}
def add_documents(self, documents: List[Dict]) -> int:
"""문서 추가 및 인덱싱"""
processed = self.pipeline.batch_process(documents)
for doc in processed:
doc_id = doc['id