완전한 RAG 시스템 구축 튜토리얼: ArXiv Paper Curator로 프로덕션 AI 만들기
⏱️ 예상 읽기 시간: 12분
들어가며: 프로덕션 환경의 RAG 시스템 구축
RAG(Retrieval-Augmented Generation)는 현대 AI 애플리케이션의 핵심이 되었으며, LLM이 방대한 지식 베이스에 접근하여 추론할 수 있게 해줍니다. 하지만 대부분의 튜토리얼은 확장성, 모니터링, 프로덕션 배포와 같은 실제 환경의 과제를 다루지 못하는 단순한 예제에만 집중합니다.
Jam With AI의 ArXiv Paper Curator 프로젝트는 이러한 격차를 해결하기 위해 프로덕션 환경에서 사용 가능한 RAG 시스템 구축을 위한 완전한 6주 커리큘럼을 제공합니다. 이 튜토리얼은 기본 설정부터 고급 모니터링 및 캐싱 전략까지 전체 여정을 안내합니다.
🎯 구축할 시스템
이 튜토리얼을 완료하면 다음과 같은 정교한 RAG 시스템을 구축하게 됩니다:
- ArXiv에서 연구 논문을 자동으로 수집하고 처리
- BM25 키워드 검색과 벡터 유사도를 결합한 하이브리드 검색 구현
- 로컬 LLM을 사용한 지능형 Q&A 기능 제공
- 포괄적인 관찰 가능성을 갖춘 프로덕션 모니터링 포함
- 최적 성능을 위한 캐싱 전략 구현
- 대화형 탐색을 위한 웹 인터페이스 제공
🏗️ 시스템 아키텍처 개요
ArXiv Paper Curator는 모듈식의 프로덕션 환경에 적합한 아키텍처를 따릅니다:
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ 데이터 수집 │ │ 처리 │ │ 검색 & RAG │
│ (ArXiv API) │───▶│ (PDF + 텍스트) │───▶│ (하이브리드 인덱스) │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ PostgreSQL │ │ OpenSearch │ │ Gradio UI │
│ (메타데이터) │ │ (검색 인덱스) │ │ (사용자 인터페이스) │
└─────────────────┘ └─────────────────┘ └─────────────────┘
📋 사전 요구사항
시작하기 전에 다음 사항을 확인하세요:
- Docker & Docker Compose: 컨테이너화된 서비스를 위해
- Python 3.11+: 애플리케이션 실행을 위해
- 8GB+ RAM: 모든 서비스를 로컬에서 실행하기 위해
- 10GB+ 저장공간: 논문과 인덱스를 위해
🚀 주차별 구현 가이드
1주차: 인프라 기초
첫 번째 주는 Docker Compose를 사용하여 여러 서비스를 조율하는 핵심 인프라를 구축합니다.
환경 설정
# 저장소 클론
git clone https://github.com/jamwithai/arxiv-paper-curator.git
cd arxiv-paper-curator
# 모든 서비스 시작
make start
# 상태 확인
make health
핵심 서비스 아키텍처
시스템은 여러 컨테이너화된 서비스를 활용합니다:
- PostgreSQL: 논문 메타데이터와 관계 저장
- OpenSearch: 키워드와 벡터 검색 기능 제공
- FastAPI: RESTful API 엔드포인트 서비스
- Gradio: 대화형 웹 인터페이스 제공
환경 설정
# 주요 설정 변수
DATABASE_URL="postgresql://postgres:password@localhost:5432/arxiv_papers"
OPENSEARCH_URL="http://localhost:9200"
JINA_API_KEY="your-jina-api-key" # 임베딩용
LANGFUSE_SECRET_KEY="your-key" # 모니터링용
2주차: 데이터 수집 파이프라인
2주차는 연구 논문을 자동으로 가져오고, 처리하고, 저장하는 견고한 데이터 수집 파이프라인 구축에 집중합니다.
ArXiv API 통합
시스템은 ArXiv API를 사용하여 카테고리와 검색 쿼리를 기반으로 논문을 가져옵니다:
async def fetch_papers_from_arxiv(
query: str = "cat:cs.AI",
max_results: int = 100,
start_date: Optional[datetime] = None
) -> List[ArxivPaper]:
"""
포괄적인 메타데이터와 함께 ArXiv API에서 논문 가져오기
"""
base_url = "http://export.arxiv.org/api/query"
params = {
"search_query": query,
"start": 0,
"max_results": max_results,
"sortBy": "submittedDate",
"sortOrder": "descending"
}
Docling을 사용한 PDF 처리
프로젝트는 문서 구조를 보존하는 정교한 PDF 처리를 위해 IBM의 Docling 라이브러리를 사용합니다:
from docling.document_converter import DocumentConverter
def process_pdf_with_docling(pdf_path: str) -> ProcessedDocument:
"""
연구 논문에서 구조화된 콘텐츠 추출
"""
converter = DocumentConverter()
result = converter.convert(pdf_path)
return ProcessedDocument(
title=result.document.title,
abstract=result.document.abstract,
sections=result.document.sections,
figures=result.document.figures,
tables=result.document.tables
)
데이터베이스 스키마 설계
PostgreSQL 스키마는 연구 논문 메타데이터에 최적화되어 있습니다:
-- 핵심 논문 테이블
CREATE TABLE papers (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
arxiv_id VARCHAR(50) UNIQUE NOT NULL,
title TEXT NOT NULL,
abstract TEXT,
authors TEXT[],
categories TEXT[],
published_date TIMESTAMP,
pdf_url TEXT,
content_extracted BOOLEAN DEFAULT FALSE,
created_at TIMESTAMP DEFAULT NOW()
);
-- 저자 관계 테이블
CREATE TABLE paper_authors (
paper_id UUID REFERENCES papers(id),
author_name TEXT,
author_order INTEGER,
PRIMARY KEY (paper_id, author_name)
);
3주차: BM25를 활용한 키워드 검색
3주차는 OpenSearch의 BM25 알고리즘을 사용한 정교한 키워드 검색을 구현하여 하이브리드 검색 시스템의 기반을 제공합니다.
OpenSearch 인덱스 설정
# BM25 최적화 인덱스 설정
index_settings = {
"settings": {
"number_of_shards": 1,
"number_of_replicas": 0,
"analysis": {
"analyzer": {
"scientific_analyzer": {
"type": "custom",
"tokenizer": "standard",
"filter": [
"lowercase",
"scientific_stemmer",
"stop_words_filter"
]
}
},
"filter": {
"scientific_stemmer": {
"type": "stemmer",
"language": "english"
},
"stop_words_filter": {
"type": "stop",
"stopwords": "_english_"
}
}
}
},
"mappings": {
"properties": {
"title": {
"type": "text",
"analyzer": "scientific_analyzer",
"boost": 3.0
},
"abstract": {
"type": "text",
"analyzer": "scientific_analyzer",
"boost": 2.0
},
"content": {
"type": "text",
"analyzer": "scientific_analyzer"
}
}
}
}
고급 BM25 쿼리 구성
def build_bm25_query(
query: str,
filters: Optional[Dict] = None,
boost_recent: bool = True
) -> Dict:
"""
필드 부스팅을 포함한 정교한 BM25 쿼리 구성
"""
base_query = {
"query": {
"bool": {
"must": [
{
"multi_match": {
"query": query,
"fields": [
"title^3", # 제목 매치 부스팅
"abstract^2", # 초록 매치 부스팅
"content", # 표준 콘텐츠 매칭
"authors" # 저자 매칭 포함
],
"type": "best_fields",
"fuzziness": "AUTO"
}
}
]
}
}
}
# 최신성 부스팅 추가
if boost_recent:
base_query["query"]["bool"]["should"] = [
{
"function_score": {
"functions": [
{
"gauss": {
"published_date": {
"origin": "now",
"scale": "30d",
"decay": 0.5
}
}
}
]
}
}
]
return base_query
4주차: 임베딩을 활용한 하이브리드 검색
4주차는 벡터 임베딩과 하이브리드 검색을 도입하여 뛰어난 검색 성능을 위해 의미적 유사성과 키워드 매칭을 결합합니다.
텍스트 청킹 전략
시스템은 문서 구조를 보존하는 섹션 인식 청킹을 구현합니다:
class SectionAwareChunker:
def __init__(self, max_chunk_size: int = 512, overlap: int = 50):
self.max_chunk_size = max_chunk_size
self.overlap = overlap
def chunk_document(self, document: ProcessedDocument) -> List[DocumentChunk]:
"""
섹션 경계를 존중하는 청크 생성
"""
chunks = []
# 각 섹션을 별도로 처리
for section in document.sections:
section_chunks = self._chunk_section(
section.content,
section.title,
section.level
)
chunks.extend(section_chunks)
return chunks
def _chunk_section(
self,
content: str,
section_title: str,
level: int
) -> List[DocumentChunk]:
"""
컨텍스트를 보존하면서 섹션 내용 청킹
"""
sentences = self._split_sentences(content)
chunks = []
current_chunk = []
current_length = 0
for sentence in sentences:
sentence_length = len(sentence.split())
if (current_length + sentence_length > self.max_chunk_size
and current_chunk):
# 섹션 컨텍스트와 함께 청크 생성
chunk_text = f"{section_title}\n\n" + " ".join(current_chunk)
chunks.append(DocumentChunk(
text=chunk_text,
section=section_title,
level=level,
word_count=current_length
))
# 겹침과 함께 새 청크 시작
overlap_sentences = current_chunk[-self.overlap:]
current_chunk = overlap_sentences + [sentence]
current_length = sum(len(s.split()) for s in current_chunk)
else:
current_chunk.append(sentence)
current_length += sentence_length
# 마지막 청크 추가
if current_chunk:
chunk_text = f"{section_title}\n\n" + " ".join(current_chunk)
chunks.append(DocumentChunk(
text=chunk_text,
section=section_title,
level=level,
word_count=current_length
))
return chunks
벡터 임베딩 생성
시스템은 고품질 벡터 표현을 위해 Jina AI의 임베딩 서비스를 사용합니다:
class JinaEmbeddingService:
def __init__(self, api_key: str, model: str = "jina-embeddings-v2-base-en"):
self.api_key = api_key
self.model = model
self.client = httpx.AsyncClient()
async def generate_embeddings(
self,
texts: List[str],
batch_size: int = 32
) -> List[List[float]]:
"""
효율성을 위해 배치로 임베딩 생성
"""
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
batch_embeddings = await self._embed_batch(batch)
all_embeddings.extend(batch_embeddings)
return all_embeddings
async def _embed_batch(self, texts: List[str]) -> List[List[float]]:
"""
단일 배치 텍스트 처리
"""
response = await self.client.post(
"https://api.jina.ai/v1/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.model,
"input": texts,
"encoding_format": "float"
}
)
result = response.json()
return [item["embedding"] for item in result["data"]]
하이브리드 검색 구현
하이브리드 검색은 정교한 점수 융합으로 BM25와 벡터 유사성을 결합합니다:
class HybridSearchEngine:
def __init__(
self,
opensearch_client,
embedding_service,
bm25_weight: float = 0.6,
vector_weight: float = 0.4
):
self.opensearch = opensearch_client
self.embeddings = embedding_service
self.bm25_weight = bm25_weight
self.vector_weight = vector_weight
async def search(
self,
query: str,
k: int = 10,
filters: Optional[Dict] = None
) -> List[SearchResult]:
"""
점수 융합을 통한 하이브리드 검색 수행
"""
# 쿼리 임베딩 생성
query_embedding = await self.embeddings.generate_embeddings([query])
query_vector = query_embedding[0]
# 하이브리드 쿼리 구성
hybrid_query = {
"size": k * 2, # 재순위화를 위해 더 많이 검색
"query": {
"bool": {
"should": [
# BM25 구성요소
{
"multi_match": {
"query": query,
"fields": ["title^3", "abstract^2", "content"],
"type": "best_fields"
}
},
# 벡터 유사성 구성요소
{
"knn": {
"embedding": {
"vector": query_vector,
"k": k
}
}
}
]
}
}
}
# 필터가 제공된 경우 추가
if filters:
hybrid_query["query"]["bool"]["filter"] = self._build_filters(filters)
# 검색 실행
response = await self.opensearch.search(
index="papers_hybrid",
body=hybrid_query
)
# 결과 처리 및 재순위화
results = self._process_results(response["hits"]["hits"])
reranked_results = self._rerank_results(results, query, k)
return reranked_results
def _rerank_results(
self,
results: List[SearchResult],
query: str,
k: int
) -> List[SearchResult]:
"""
여러 신호를 기반으로 한 정교한 재순위화 적용
"""
# 정규화된 점수 계산
max_bm25 = max(r.bm25_score for r in results) if results else 1.0
max_vector = max(r.vector_score for r in results) if results else 1.0
for result in results:
# 개별 점수 정규화
norm_bm25 = result.bm25_score / max_bm25
norm_vector = result.vector_score / max_vector
# 하이브리드 점수 계산
result.hybrid_score = (
self.bm25_weight * norm_bm25 +
self.vector_weight * norm_vector
)
# 추가 순위 신호 적용
result.hybrid_score *= self._calculate_quality_multiplier(result)
# 하이브리드 점수로 정렬하여 상위 k개 반환
return sorted(results, key=lambda x: x.hybrid_score, reverse=True)[:k]
5주차: 완전한 RAG 시스템
5주차는 정교한 프롬프트 엔지니어링과 응답 생성을 통해 모든 것을 완전한 RAG 시스템으로 통합합니다.
Ollama를 활용한 LLM 통합
시스템은 프라이버시와 제어를 제공하는 로컬 LLM 추론을 위해 Ollama를 사용합니다:
class OllamaRAGService:
def __init__(
self,
base_url: str = "http://localhost:11434",
model: str = "llama3.1:8b"
):
self.base_url = base_url
self.model = model
self.client = httpx.AsyncClient()
async def generate_answer(
self,
query: str,
context_chunks: List[DocumentChunk],
max_context_length: int = 4000
) -> RAGResponse:
"""
검색된 문서를 사용하여 맥락적 답변 생성
"""
# 스마트 잘라내기로 컨텍스트 준비
context = self._prepare_context(context_chunks, max_context_length)
# RAG 프롬프트 구성
prompt = self._build_rag_prompt(query, context)
# 응답 생성
response = await self._call_ollama(prompt)
# 후처리 및 검증
processed_response = self._post_process_response(
response,
query,
context_chunks
)
return processed_response
def _build_rag_prompt(self, query: str, context: str) -> str:
"""
역할 정의를 포함한 정교한 RAG 프롬프트 구성
"""
return f"""당신은 AI와 컴퓨터 과학 문헌을 전문으로 하는 전문 연구 어시스턴트입니다. 제공된 연구 논문을 바탕으로 정확하고 포괄적인 답변을 제공하는 것이 당신의 역할입니다.
**연구 논문 컨텍스트:**
{context}
**사용자 질문:**
{query}
**지침:**
1. 제공된 컨텍스트를 주로 기반으로 답변하세요
2. 컨텍스트에 충분한 정보가 없다면 명확히 언급하세요
3. 주장을 할 때 논문에 대한 구체적인 참조를 포함하세요
4. 적절한 경우 기술적 세부사항을 제공하세요
5. 방법론을 논의할 때는 명확히 설명하세요
6. 논문에서 언급된 제한사항이나 주의사항을 강조하세요
**답변:**"""
def _prepare_context(
self,
chunks: List[DocumentChunk],
max_length: int
) -> str:
"""
컨텍스트 청크를 지능적으로 선택하고 포맷
"""
# 관련성 점수로 청크 정렬
sorted_chunks = sorted(chunks, key=lambda x: x.score, reverse=True)
context_parts = []
current_length = 0
for chunk in sorted_chunks:
chunk_text = f"[{chunk.paper_title}]\n{chunk.text}\n---\n"
chunk_length = len(chunk_text.split())
if current_length + chunk_length > max_length:
break
context_parts.append(chunk_text)
current_length += chunk_length
return "\n".join(context_parts)
고급 프롬프트 엔지니어링
시스템은 다양한 쿼리 유형을 위한 전문화된 프롬프트를 포함합니다:
class PromptTemplates:
COMPARATIVE_ANALYSIS = """제공된 연구 논문을 바탕으로 다음 접근법/방법/개념을 비교 분석하세요: {concepts}
다음과 같이 응답을 구성해주세요:
1. **개요**: 각 개념에 대한 간단한 소개
2. **주요 유사점**: 이러한 접근법들의 공통점은 무엇인가요?
3. **주요 차이점**: 방법론, 가정, 결과에서 어떻게 다른가요?
4. **성능 비교**: 가능한 경우, 효과성을 비교하세요
5. **사용 사례**: 언제 하나를 다른 것보다 선택하시겠습니까?
6. **제한사항**: 각 접근법의 제약사항은 무엇입니까?
참조 논문: {paper_titles}"""
METHODOLOGY_EXPLANATION = """다음 주제에 대해 연구 논문에서 설명된 방법론을 설명하세요: {topic}
다음을 제공해주세요:
1. **문제 정의**: 어떤 문제를 다루고 있나요?
2. **접근법 개요**: 고수준 방법론 설명
3. **기술적 세부사항**: 방법의 단계별 설명
4. **구현 고려사항**: 구현의 실용적 측면
5. **평가 방법**: 접근법이 어떻게 검증되었나요?
6. **결과 요약**: 주요 발견사항과 성능 지표
7. **제한사항 및 향후 작업**: 인정된 제약사항과 다음 단계
설명 기반: {paper_titles}"""
TREND_ANALYSIS = """다음 연구 분야의 동향과 발전을 분석하세요: {research_area}
제공된 논문을 바탕으로 다음을 논의하세요:
1. **역사적 맥락**: 이 분야는 어떻게 발전해왔나요?
2. **현재 상태**: 오늘날 지배적인 접근법은 무엇인가요?
3. **새로운 패턴**: 어떤 새로운 동향이 보이나요?
4. **주요 혁신**: 언급된 획기적인 기여는 무엇인가요?
5. **열린 도전과제**: 아직 해결되지 않은 문제는 무엇인가요?
6. **미래 방향**: 어떤 연구 방향이 제안되나요?
분석된 논문: {paper_titles}"""
6주차: 프로덕션 모니터링 및 캐싱
6주차는 포괄적인 모니터링과 지능적인 캐싱 전략을 통한 프로덕션 준비에 집중합니다.
관찰 가능성을 위한 Langfuse 통합
class RAGMonitoringService:
def __init__(self, langfuse_client):
self.langfuse = langfuse_client
async def trace_rag_request(
self,
query: str,
search_results: List[SearchResult],
llm_response: RAGResponse,
execution_time: float
) -> None:
"""
포괄적인 RAG 파이프라인 추적
"""
trace = self.langfuse.trace(
name="rag_query",
input={"query": query},
output={"answer": llm_response.answer},
metadata={
"execution_time_ms": execution_time * 1000,
"num_results": len(search_results),
"model_used": llm_response.model,
"context_length": llm_response.context_length
}
)
# 검색 성능 추적
search_span = trace.span(
name="search_retrieval",
input={"query": query},
output={"num_results": len(search_results)},
metadata={
"search_type": "hybrid",
"bm25_weight": 0.6,
"vector_weight": 0.4
}
)
# LLM 생성 추적
generation_span = trace.span(
name="llm_generation",
input={"prompt_length": len(llm_response.prompt)},
output={"response_length": len(llm_response.answer)},
metadata={
"model": llm_response.model,
"temperature": llm_response.temperature,
"tokens_used": llm_response.token_count
}
)
# 품질 지표 추적
self._track_quality_metrics(trace, query, llm_response)
def _track_quality_metrics(
self,
trace,
query: str,
response: RAGResponse
) -> None:
"""
응답 품질 지표 추적
"""
metrics = {
"has_citations": len(response.citations) > 0,
"answer_length": len(response.answer.split()),
"confidence_score": response.confidence_score,
"context_utilization": response.context_utilization_score
}
trace.score(
name="response_quality",
value=self._calculate_quality_score(metrics)
)
지능적 캐싱 전략
class RAGCacheService:
def __init__(self, redis_client, ttl: int = 3600):
self.redis = redis_client
self.ttl = ttl
async def get_cached_response(
self,
query: str,
filters: Optional[Dict] = None
) -> Optional[CachedRAGResponse]:
"""
의미적 유사성 매칭을 통한 캐시된 응답 검색
"""
# 캐시 키 생성
cache_key = self._generate_cache_key(query, filters)
# 먼저 정확한 매치 확인
cached = await self.redis.get(cache_key)
if cached:
return CachedRAGResponse.parse_raw(cached)
# 임베딩 유사성을 사용한 유사한 쿼리 확인
similar_response = await self._find_similar_cached_query(query)
if similar_response and similar_response.similarity_score > 0.85:
return similar_response
return None
async def cache_response(
self,
query: str,
response: RAGResponse,
filters: Optional[Dict] = None
) -> None:
"""
지능적 검색을 위한 메타데이터와 함께 응답 캐시
"""
cache_key = self._generate_cache_key(query, filters)
cached_response = CachedRAGResponse(
query=query,
response=response,
filters=filters,
cached_at=datetime.utcnow(),
access_count=1,
quality_score=response.quality_score
)
# 품질 점수 기반 TTL로 저장
ttl = self._calculate_adaptive_ttl(response.quality_score)
await self.redis.setex(
cache_key,
ttl,
cached_response.json()
)
# 유사성 검색을 위한 쿼리 임베딩 인덱스 업데이트
await self._index_query_embedding(query, cache_key)
def _calculate_adaptive_ttl(self, quality_score: float) -> int:
"""
응답 품질에 따른 캐시 TTL 조정
"""
base_ttl = self.ttl
quality_multiplier = min(2.0, max(0.5, quality_score * 2))
return int(base_ttl * quality_multiplier)
🔧 API 엔드포인트 및 사용법
시스템은 포괄적인 RESTful API를 제공합니다:
핵심 엔드포인트
# 상태 확인
GET /health
# 논문 관리
GET /api/v1/papers # 페이지네이션으로 논문 목록
GET /api/v1/papers/{id} # 특정 논문 가져오기
POST /api/v1/papers/ingest # 수집 트리거
# 검색 엔드포인트
POST /api/v1/search # BM25 키워드 검색
POST /api/v1/hybrid-search # 하이브리드 검색 (BM25 + 벡터)
POST /api/v1/ask # RAG 질의응답
# 분석 엔드포인트
GET /api/v1/analytics/search # 검색 성능 지표
GET /api/v1/analytics/rag # RAG 시스템 지표
API 사용 예제
import httpx
async def query_rag_system():
async with httpx.AsyncClient() as client:
# 하이브리드 검색 수행
search_response = await client.post(
"http://localhost:8000/api/v1/hybrid-search",
json={
"query": "트랜스포머 어텐션 메커니즘",
"k": 5,
"filters": {
"categories": ["cs.AI", "cs.LG"],
"date_range": {
"start": "2023-01-01",
"end": "2024-12-31"
}
}
}
)
# RAG 질문 하기
rag_response = await client.post(
"http://localhost:8000/api/v1/ask",
json={
"question": "트랜스포머 어텐션 메커니즘은 어떻게 작동하며 제한사항은 무엇인가요?",
"context_k": 10,
"model": "llama3.1:8b",
"temperature": 0.7
}
)
return rag_response.json()
🎛️ Gradio 웹 인터페이스
시스템은 Gradio로 구축된 정교한 웹 인터페이스를 포함합니다:
import gradio as gr
def create_rag_interface():
with gr.Blocks(title="ArXiv Paper Curator") as interface:
gr.Markdown("# 🤖 ArXiv Paper Curator RAG 시스템")
with gr.Tab("🔍 논문 검색"):
with gr.Row():
search_query = gr.Textbox(
label="검색 쿼리",
placeholder="검색어를 입력하세요..."
)
search_type = gr.Radio(
["키워드 (BM25)", "의미적 (벡터)", "하이브리드"],
value="하이브리드",
label="검색 유형"
)
search_button = gr.Button("검색", variant="primary")
search_results = gr.JSON(label="검색 결과")
with gr.Tab("💬 질문하기"):
with gr.Row():
with gr.Column(scale=2):
question = gr.Textbox(
label="질문",
placeholder="연구 논문에 대해 질문하세요...",
lines=3
)
with gr.Row():
model_choice = gr.Dropdown(
["llama3.1:8b", "mistral:7b", "codellama:13b"],
value="llama3.1:8b",
label="모델"
)
temperature = gr.Slider(
0.0, 1.0, 0.7,
label="Temperature"
)
with gr.Column(scale=1):
context_k = gr.Slider(
1, 20, 10,
label="컨텍스트 논문 수"
)
include_citations = gr.Checkbox(
True,
label="인용 포함"
)
ask_button = gr.Button("질문하기", variant="primary")
with gr.Row():
answer = gr.Textbox(
label="답변",
lines=10,
max_lines=20
)
context_papers = gr.JSON(
label="출처 논문"
)
with gr.Tab("📊 분석"):
refresh_button = gr.Button("분석 새로고침")
with gr.Row():
total_papers = gr.Number(label="전체 논문")
indexed_papers = gr.Number(label="인덱스된 논문")
cache_hit_rate = gr.Number(label="캐시 히트율 (%)")
performance_chart = gr.Plot(label="쿼리 성능")
return interface
# 인터페이스 실행
if __name__ == "__main__":
interface = create_rag_interface()
interface.launch(server_name="0.0.0.0", server_port=7860)
🚀 배포 및 확장
Docker Compose 프로덕션 설정
version: '3.8'
services:
app:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://postgres:${POSTGRES_PASSWORD}@db:5432/arxiv_papers
- OPENSEARCH_URL=http://opensearch:9200
- REDIS_URL=redis://redis:6379
depends_on:
- db
- opensearch
- redis
volumes:
- ./data:/app/data
db:
image: postgres:15
environment:
- POSTGRES_DB=arxiv_papers
- POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
volumes:
- postgres_data:/var/lib/postgresql/data
opensearch:
image: opensearchproject/opensearch:2.8.0
environment:
- discovery.type=single-node
- "OPENSEARCH_JAVA_OPTS=-Xms2g -Xmx2g"
volumes:
- opensearch_data:/usr/share/opensearch/data
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
gradio:
build: .
command: python gradio_launcher.py
ports:
- "7860:7860"
depends_on:
- app
volumes:
postgres_data:
opensearch_data:
redis_data:
성능 최적화 팁
- 데이터베이스 최적화:
- 자주 쿼리되는 필드에 인덱스 생성
- 연결 풀링 사용
- 확장을 위한 읽기 복제본 구현
- 검색 최적화:
- OpenSearch 클러스터 설정 최적화
- 일관된 매핑을 위한 인덱스 템플릿 사용
- 검색 결과 캐싱 구현
- 임베딩 최적화:
- 임베딩 생성 배치 처리
- 재사용 콘텐츠에 대한 임베딩 캐시
- 메모리 효율성을 위한 양자화된 임베딩 사용
- LLM 최적화:
- 모델 워밍업 구현
- 가능한 경우 GPU 가속 사용
- 효율성을 위한 프롬프트 템플릿 최적화
🔍 모니터링 및 관찰 가능성
추적해야 할 주요 지표
# 성능 지표
search_latency = histogram("search_latency_seconds")
rag_latency = histogram("rag_latency_seconds")
cache_hit_rate = gauge("cache_hit_rate")
# 품질 지표
answer_relevance = histogram("answer_relevance_score")
context_utilization = histogram("context_utilization_score")
user_satisfaction = histogram("user_satisfaction_score")
# 시스템 지표
active_connections = gauge("active_database_connections")
opensearch_cluster_health = gauge("opensearch_cluster_health")
embedding_queue_size = gauge("embedding_queue_size")
Langfuse 대시보드 설정
시스템은 포괄적인 관찰 가능성을 위해 Langfuse와 통합됩니다:
- 추적 분석: 쿼리에서 응답까지의 완전한 요청 추적
- 성능 모니터링: 구성요소별 지연 시간 분석
- 품질 추적: 응답 품질 점수 및 사용자 피드백
- 비용 추적: 토큰 사용량 및 API 비용
- 오류 모니터링: 실패율 및 오류 분류
🎯 고급 기능 및 확장
1. 멀티모달 지원
논문의 그림과 표를 처리하도록 시스템 확장:
class MultiModalProcessor:
async def process_figures(self, paper_id: str) -> List[Figure]:
"""연구 논문에서 그림 추출 및 처리"""
async def process_tables(self, paper_id: str) -> List[Table]:
"""구조 보존과 함께 표 추출 및 처리"""
2. 인용 네트워크 분석
향상된 논문 발견을 위한 인용 그래프 구축:
class CitationNetworkService:
async def build_citation_graph(self) -> NetworkGraph:
"""논문 추천을 위한 인용 네트워크 구성"""
async def find_influential_papers(self, topic: str) -> List[Paper]:
"""특정 도메인에서 높은 인용을 받는 논문 식별"""
3. 개인화 엔진
사용자 선호도 학습 구현:
class PersonalizationService:
async def learn_user_preferences(self, user_id: str, interactions: List[Interaction]):
"""검색 및 읽기 패턴에서 사용자 선호도 학습"""
async def personalize_search_results(self, user_id: str, results: List[SearchResult]) -> List[SearchResult]:
"""사용자 선호도에 따른 결과 재순위화"""
🏆 모범 사례 및 교훈
1. 데이터 품질 관리
- 견고한 PDF 처리를 폴백 전략과 함께 구현
- 인덱싱 전에 추출된 콘텐츠 검증
- 추출 성공률 모니터링 및 문제가 있는 논문 식별
- 다국어 콘텐츠를 적절히 처리
2. 검색 품질 최적화
- 쿼리 유형에 따른 하이브리드 검색 가중치 조정
- 더 나은 회수율을 위한 쿼리 확장 구현
- 향상된 정밀도를 위한 재순위화 모델 사용
- 다양한 검색 전략에 대한 A/B 테스트
3. RAG 시스템 설계
- 다양한 쿼리 유형을 위한 컨텍스트 인식 프롬프트 설계
- 환각을 잡기 위한 응답 검증 구현
- 일관된 응답을 위한 구조화된 출력 형식 사용
- 지속적인 답변 품질 모니터링 및 개선
4. 프로덕션 준비
- 첫날부터 포괄적인 모니터링 구현
- 무상태 서비스로 수평 확장을 위한 설계
- 구성요소 실패 시 우아한 성능 저하 사용
- 외부 의존성에 대한 회로 차단기 구현
🔮 향후 개선사항
ArXiv Paper Curator 로드맵에는 몇 가지 흥미로운 개발이 포함되어 있습니다:
1. 고급 RAG 기법
- 검색 증강 파인튜닝: 검색과 전문화된 모델 훈련 결합
- 다단계 추론: 복잡한 쿼리를 위한 생각의 연쇄 구현
- 논문 간 종합: 여러 논문을 연결하여 통찰 생성
2. 향상된 사용자 경험
- 대화형 시각화: 논문 관계 그래프 및 동향 분석
- 협업 기능: 공유 읽기 목록 및 주석
- 모바일 애플리케이션: 이동 중 접근을 위한 네이티브 iOS/Android 앱
3. 연구 어시스턴트 기능
- 자동화된 문헌 검토: 포괄적인 서베이 논문 생성
- 연구 공백 식별: 탐구되지 않은 연구 영역 강조
- 방법론 비교: 다양한 접근법의 나란히 분석
📚 추가 자료
학습 자료
- 주차별 노트북: 각 구현 단계에 대한 상세한 Jupyter 노트북
- 비디오 워크스루: 단계별 구현 가이드
- 커뮤니티 포럼: 활발한 토론 및 지원 커뮤니티
문서
- API 레퍼런스: 완전한 OpenAPI 사양
- 아키텍처 가이드: 상세한 시스템 설계 문서
- 배포 가이드: 프로덕션 배포 모범 사례
커뮤니티 및 지원
- GitHub 토론: 커뮤니티 Q&A 및 기능 요청
- Discord 서버: 실시간 채팅 및 협업
- 오피스 아워: 개발 팀과의 주간 세션
🎉 결론
ArXiv Paper Curator는 프로덕션 환경에서 사용 가능한 RAG 시스템 구축에 대한 포괄적인 접근 방식을 나타냅니다. 이 튜토리얼을 따라하면서 다음을 학습했습니다:
- 실제 복잡성을 처리하는 확장 가능한 RAG 아키텍처 설계
- 여러 검색 전략을 결합하는 하이브리드 검색 시스템 구현
- 지속적인 콘텐츠 수집을 위한 견고한 데이터 파이프라인 구축
- 프로덕션 관찰 가능성을 위한 모니터링 솔루션 배포
- 최종 사용자 상호작용을 위한 직관적인 인터페이스 생성
여기서 배운 기술과 패턴은 대용량 문서 컬렉션에 대해 추론할 수 있는 지능형 시스템을 구축해야 하는 모든 도메인에 직접 적용할 수 있습니다. 법률 문서, 의료 연구, 기술 문서와 같은 영역에서 작업하든 원칙은 동일합니다.
다음 단계
- 다양한 도메인으로 실험: 특정 사용 사례에 맞게 시스템 적용
- 프로젝트에 기여: 개선사항과 확장을 커뮤니티와 공유
- 프로덕션으로 확장: 자체 인스턴스를 배포하고 실제 사용자 피드백 수집
- 최신 정보 유지: 새로운 기능과 개선사항을 위해 프로젝트 팔로우
프로덕션 AI 시스템 구축은 반복적인 과정이라는 것을 기억하세요. 기본부터 시작하여 모든 것을 측정하고 실제 사용 패턴을 바탕으로 지속적으로 개선하세요.
자체 RAG 시스템을 구축할 준비가 되셨나요? 저장소를 클론하고 1주차부터 시작하세요: GitHub의 ArXiv Paper Curator
최신 업데이트와 커뮤니티 토론을 위해 Discord 서버에 참여하고 Twitter에서 @jamwithai를 팔로우하세요.