반응형
"우리 회사 내부 문서를 학습한 AI 챗봇을 만들고 싶어요." 요즘 개발팀에서 가장 많이 나오는 요구사항 중 하나입니다. 사내 규정, 기술 문서, 과거 장애 대응 기록, 제품 스펙 — 이 모든 것을 꿰뚫고 답변하는 AI 어시스턴트. RAG(Retrieval-Augmented Generation)가 그 해답이라는 건 이제 많은 분들이 알고 있습니다.
문제는 "어떻게 구축하느냐"입니다. ChatGPT API 하나 붙이는 것과, 프로덕션 수준의 사내 RAG 서버를 구축하는 것은 완전히 다른 이야기입니다. 어떤 벡터 DB를 선택할지, Java로 갈지 Python으로 갈지, 어떤 아키텍처로 설계할지, 보안은 어떻게 챙길지 — 결정해야 할 것이 한두 가지가 아닙니다.
이 글에서는 사내 RAG 서버를 처음부터 끝까지 직접 구축하는 방법을 단계별로 설명합니다. 벡터DB 선택 및 세팅부터 Java vs Python 언어 선택, 전체 시스템 아키텍처 설계까지, 실무에서 바로 적용할 수 있는 내용을 담았습니다.
목차
- 전체 아키텍처 overview — 사내 RAG 서버의 구성요소
- 언어 선택 — Java vs Python, 무엇을 골라야 하나?
- 벡터DB 선택 가이드 — Qdrant vs pgvector vs Weaviate vs Chroma
- 벡터DB 세팅 — Qdrant와 pgvector 실전 설치
- 임베딩 모델 선택 및 서버 구성
- 문서 인덱싱 파이프라인 구축
- 검색 및 생성 API 서버 구축
- 보안 및 접근 제어 설계
- 모니터링 및 품질 관리
- 배포 전략 및 운영 팁
1. 전체 아키텍처 overview — 사내 RAG 서버의 구성요소
본격적인 구축에 앞서, 사내 RAG 서버가 어떤 컴포넌트로 구성되는지 전체 그림을 그려보겠습니다.
┌─────────────────────────────────────────────────────────────────┐
│ 사내 RAG 서버 전체 아키텍처 │
│ │
│ ┌─────────────┐ ┌──────────────────────────────────────┐ │
│ │ 데이터 소스 │ │ RAG 서버 │ │
│ │ │ │ │ │
│ │ - Confluence │───▶│ ┌──────────────┐ ┌─────────────┐ │ │
│ │ - Notion │ │ │ 인덱싱 파이프 │ │ 검색 API │ │ │
│ │ - GitHub │ │ │ │ │ │ │ │
│ │ - SharePoint│ │ │ 1. 문서 로딩 │ │ 1. 쿼리 임베딩│ │ │
│ │ - PDF/Word │ │ │ 2. 청크 분할 │ │ 2. 벡터 검색 │ │ │
│ │ - DB 덤프 │ │ │ 3. 임베딩 │ │ 3. Reranking │ │ │
│ └─────────────┘ │ │ 4. 벡터 저장 │ │ 4. LLM 생성 │ │ │
│ │ └──────┬───────┘ └──────┬──────┘ │ │
│ │ │ │ │ │
│ │ ▼ ▼ │ │
│ │ ┌──────────────────────────┐ │ │
│ │ │ 벡터 DB │ │ │
│ │ │ (Qdrant / pgvector) │ │ │
│ │ └──────────────────────────┘ │ │
│ └──────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌────────────────────────┐ │
│ │ 클라이언트 │ │
│ │ - 슬랙봇 │ │
│ │ - 웹 채팅 UI │ │
│ │ - IDE 플러그인 │ │
│ └────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
핵심 컴포넌트
인덱싱 파이프라인 (Offline)
문서를 수집하고, 청크로 분할하고, 임베딩을 생성하여 벡터 DB에 저장합니다. 데이터가 추가되거나 변경될 때 실행됩니다. 실시간 처리가 필요 없으므로, 배치 작업 또는 스케줄 작업으로 구성합니다.
검색 API 서버 (Online)
사용자 질문이 들어오면, 관련 문서를 벡터 DB에서 검색하고, LLM API에 전달하여 답변을 생성하는 REST/gRPC API 서버입니다. 응답 속도가 중요하므로, 비동기 처리와 캐싱을 적극 활용합니다.
벡터 DB
임베딩 벡터를 저장하고 고속 유사도 검색을 제공하는 특화 데이터베이스입니다. RAG 시스템의 핵심 인프라로, 선택이 전체 성능에 큰 영향을 미칩니다.
임베딩 모델 서버
텍스트를 벡터로 변환하는 모델을 서빙합니다. OpenAI API를 쓸 수도 있고, 자체 서버에 오픈소스 모델을 올릴 수도 있습니다. 비용과 레이턴시의 트레이드오프가 있습니다.
LLM API
최종 답변을 생성하는 언어 모델입니다. OpenAI GPT-4, Anthropic Claude, 또는 자체 호스팅 LLM(Llama, Mistral 등)을 사용합니다.
2. 언어 선택 — Java vs Python, 무엇을 골라야 하나?
사내 RAG 서버 구축에서 가장 먼저 부딪히는 결정이 바로 언어 선택입니다. AI/ML 생태계는 Python이 압도적이지만, 사내 백엔드가 Java/Spring으로 구성된 경우가 많아 고민이 됩니다.
Java의 현실 — Spring AI의 등장
한때 "AI는 Python으로만"이라는 공식이 있었지만, 2024년부터 상황이 달라졌습니다. Spring AI 프로젝트가 정식 출시되면서, Java/Spring 환경에서도 수준급의 RAG 시스템을 구축할 수 있게 되었습니다.
// Spring AI로 RAG 구현 예시
@Service
public class RagService {
private final ChatClient chatClient;
private final VectorStore vectorStore;
public RagService(ChatClient.Builder builder, VectorStore vectorStore) {
this.chatClient = builder
.defaultAdvisors(new QuestionAnswerAdvisor(vectorStore))
.build();
this.vectorStore = vectorStore;
}
public String ask(String question) {
return chatClient.prompt()
.user(question)
.call()
.content();
}
}
Spring AI는 다음을 지원합니다:
- VectorStore 추상화: pgvector, Chroma, Pinecone, Weaviate, Qdrant 등 통일된 인터페이스
- ChatClient: OpenAI, Claude, Ollama, Azure OpenAI 등 LLM 통합
- DocumentReader: PDF, Word, 웹페이지, Tika 등 다양한 문서 형식 지원
- ETL Pipeline: 문서 로딩, 변환, 임베딩, 저장 파이프라인
Python의 강점 — 여전히 압도적인 생태계
Python은 AI/ML 생태계의 사실상 표준입니다. LangChain, LlamaIndex, Haystack 등 성숙한 RAG 프레임워크가 이미 존재하며, 최신 모델과 기법이 Python 라이브러리로 가장 먼저 출시됩니다.
# LangChain으로 RAG 구현 예시
from langchain.chains import RetrievalQA
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Qdrant
from langchain.chat_models import ChatOpenAI
import qdrant_client
client = qdrant_client.QdrantClient(host="localhost", port=6333)
embeddings = OpenAIEmbeddings()
vectorstore = Qdrant(
client=client,
collection_name="company_docs",
embeddings=embeddings
)
qa_chain = RetrievalQA.from_chain_type(
llm=ChatOpenAI(model="gpt-4o"),
retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
return_source_documents=True
)
result = qa_chain.invoke("우리 회사 휴가 정책은?")
print(result["result"])
상세 비교표
┌────────────────────────────────────────────────────────────────┐
│ Java vs Python — 사내 RAG 서버 구축 관점 │
│ │
│ 항목 Java (Spring AI) Python (LangChain) │
│ ───────────── ───────────────── ────────────────── │
│ AI 생태계 ★★★☆☆ ★★★★★ │
│ 성능/처리량 ★★★★★ ★★★☆☆ │
│ 타입 안정성 ★★★★★ ★★★☆☆ │
│ 기존 시스템 통합 ★★★★★ ★★★☆☆ │
│ 최신 기법 적용 ★★★☆☆ ★★★★★ │
│ 러닝 커브 ★★★☆☆ ★★★★★ │
│ 커뮤니티/레퍼런스 ★★★☆☆ ★★★★★ │
│ 운영 안정성 ★★★★★ ★★★★☆ │
│ 메모리 효율 ★★★★★ ★★★☆☆ │
│ 비동기 처리 ★★★★☆ ★★★★★ │
└────────────────────────────────────────────────────────────────┘
어느 쪽을 선택해야 하나?
Java를 선택해야 하는 경우:
- 사내 백엔드 스택이 이미 Java/Spring으로 구성된 경우
- 기존 Spring Security, Spring Data 등과 통합이 필요한 경우
- 높은 처리량(TPS)과 메모리 효율이 중요한 경우
- 타입 안정성과 엔터프라이즈 수준의 유지보수가 필요한 경우
- 팀의 Java 숙련도가 Python보다 높은 경우
320x100
Python을 선택해야 하는 경우:
- AI/ML 기능을 빠르게 프로토타이핑하고 실험해야 하는 경우
- 최신 RAG 기법(GraphRAG, Self-RAG 등)을 빠르게 적용해야 하는 경우
- 오픈소스 임베딩 모델을 자체 서빙해야 하는 경우 (HuggingFace Transformers)
- 데이터 전처리, 분석 작업이 많은 경우
- 팀의 Python 숙련도가 높고 AI 중심 개발 조직인 경우
Python + Java 혼합 (권장 아키텍처):
현실적으로 가장 많이 선택하는 방식입니다.
┌──────────────────────────────────────────────────────────┐
│ 혼합 아키텍처 (권장) │
│ │
│ Python FastAPI Java Spring Boot │
│ ┌─────────────────────┐ ┌────────────────────┐ │
│ │ RAG API 서버 │ │ 비즈니스 API 서버 │ │
│ │ - 임베딩 │◀───────▶│ - 인증/권한 │ │
│ │ - 벡터 검색 │ REST │ - 사용자 관리 │ │
│ │ - LLM 생성 │ │ - 로그/감사 │ │
│ │ - 인덱싱 파이프라인 │ │ - 기존 업무 로직 │ │
│ └─────────────────────┘ └────────────────────┘ │
│ │
│ "AI는 Python, 비즈니스 로직은 Java"로 역할 분리 │
└──────────────────────────────────────────────────────────┘
3. 벡터DB 선택 가이드 — Qdrant vs pgvector vs Weaviate vs Chroma
벡터DB 선택은 RAG 시스템의 성능, 운영 복잡도, 비용에 직결됩니다. 주요 벡터DB를 사내 RAG 서버 관점에서 비교해보겠습니다.
Qdrant
Rust로 작성된 고성능 벡터 검색 엔진입니다. 사내 RAG 서버에서 가장 많이 추천되는 선택지 중 하나입니다.
┌──────────────────────────────────────────────────────┐
│ Qdrant 주요 특징 │
│ │
│ ✅ 성능 Rust 기반, 초당 수만 건 검색 처리 │
│ ✅ 필터링 페이로드(메타데이터) 기반 복합 필터링 │
│ ✅ 자체 호스팅 Docker로 5분 안에 설치 가능 │
│ ✅ 클라우드 Qdrant Cloud 완전 관리형 서비스 제공 │
│ ✅ API REST + gRPC 모두 지원 │
│ ⚠️ 단점 전용 DB이므로 기존 PostgreSQL과 별도 운영 │
└──────────────────────────────────────────────────────┘
언제 Qdrant를 선택하나:
- 대용량 문서(100만 건 이상)를 처리해야 하는 경우
- 복잡한 메타데이터 필터링이 필요한 경우 (부서별, 날짜별, 분류별)
- 독립적인 벡터 검색 서비스로 분리하고 싶은 경우
- 최고 수준의 검색 성능이 필요한 경우
pgvector
PostgreSQL의 벡터 확장으로, 기존 PostgreSQL 데이터베이스에 벡터 검색 기능을 추가합니다.
┌──────────────────────────────────────────────────────┐
│ pgvector 주요 특징 │
│ │
│ ✅ 통합성 기존 PostgreSQL 그대로 사용 │
│ ✅ SQL 일반 SQL과 벡터 검색 JOIN 가능 │
│ ✅ 운영 DBA가 이미 알고 있는 PostgreSQL로 관리 │
│ ✅ 비용 추가 인프라 없이 기존 DB 확장 │
│ ⚠️ 성능 Qdrant 대비 대용량에서 성능 차이 발생 │
│ ⚠️ 한계 수백만 건 이상에서 HNSW 인덱스 튜닝 필요 │
└──────────────────────────────────────────────────────┘
언제 pgvector를 선택하나:
- 이미 PostgreSQL을 사용 중이며 인프라를 단순하게 유지하고 싶은 경우
- 벡터 검색과 일반 데이터를 JOIN하는 복합 쿼리가 필요한 경우
- 문서 수가 수십만 건 이하인 소/중규모 서비스
- DBA 없이 개발팀이 직접 운영하는 경우
Weaviate
GraphQL 기반의 오픈소스 벡터 DB로, 객체 중심의 데이터 모델을 제공합니다.
┌──────────────────────────────────────────────────────┐
│ Weaviate 주요 특징 │
│ │
│ ✅ 하이브리드 벡터 + BM25 하이브리드 검색 내장 │
│ ✅ 멀티모달 텍스트, 이미지, 오디오 벡터 통합 저장 │
│ ✅ GraphQL 직관적인 쿼리 인터페이스 │
│ ✅ 모듈 임베딩 모델 통합 모듈 내장 (text2vec) │
│ ⚠️ 복잡도 설정과 스키마 설계가 복잡한 편 │
│ ⚠️ 메모리 다른 벡터 DB 대비 메모리 사용량 높음 │
└──────────────────────────────────────────────────────┘
Chroma
Python 네이티브의 경량 벡터 DB입니다. 개발/테스트 환경에 최적화되어 있습니다.
┌──────────────────────────────────────────────────────┐
│ Chroma 주요 특징 │
│ │
│ ✅ 간편함 pip install chromadb 한 줄로 시작 │
│ ✅ Python LangChain, LlamaIndex와 완벽 통합 │
│ ✅ 인메모리 테스트/프로토타이핑에 최적 │
│ ⚠️ 규모 대규모 프로덕션 환경에서 성능 한계 │
│ ⚠️ 운영 엔터프라이즈 수준의 기능 부족 │
└──────────────────────────────────────────────────────┘
사내 RAG 서버를 위한 최종 선택 가이드
┌────────────────────────────────────────────────────────────────┐
│ 벡터DB 선택 결정 트리 │
│ │
│ 이미 PostgreSQL을 쓰고 있다? │
│ YES → 문서 수 50만 건 이하? │
│ YES → pgvector ✅ │
│ NO → Qdrant + pgvector 혼합 고려 │
│ NO → 최고 성능이 필요하다? │
│ YES → Qdrant ✅ │
│ NO → 하이브리드 검색이 중요하다? │
│ YES → Weaviate │
│ NO → 프로토타입이다? │
│ YES → Chroma │
│ NO → Qdrant ✅ │
└────────────────────────────────────────────────────────────────┘
사내 RAG 서버 1순위 추천: Qdrant
- 설치 간편 (Docker 한 줄)
- 뛰어난 성능과 필터링 기능
- Java, Python 모두 공식 클라이언트 제공
- 클라우드 관리형 서비스도 제공하여 운영 부담 최소화
2순위 추천: pgvector
- 기존 PostgreSQL 인프라 활용 가능
- 운영 복잡도 최소화
- SQL과 벡터 검색 통합 쿼리
4. 벡터DB 세팅 — Qdrant와 pgvector 실전 설치
Qdrant 설치 및 세팅
Docker로 설치 (가장 빠른 방법)
# 단일 노드 실행
docker run -d \
--name qdrant \
-p 6333:6333 \
-p 6334:6334 \
-v $(pwd)/qdrant_storage:/qdrant/storage \
qdrant/qdrant
# 확인
curl http://localhost:6333/health
# {"title":"qdrant - vector search engine","version":"1.x.x"}
Docker Compose로 설치 (권장)
# docker-compose.yml
version: '3.8'
services:
qdrant:
image: qdrant/qdrant:latest
restart: always
ports:
- "6333:6333" # REST API
- "6334:6334" # gRPC
volumes:
- qdrant_data:/qdrant/storage
environment:
QDRANT__SERVICE__API_KEY: "your-secret-api-key" # API 키 인증
deploy:
resources:
limits:
memory: 4G # 메모리 제한 설정
volumes:
qdrant_data:
컬렉션 생성 (Python)
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams
client = QdrantClient(
host="localhost",
port=6333,
api_key="your-secret-api-key"
)
# 컬렉션 생성 (1536차원 = OpenAI text-embedding-3-small)
client.create_collection(
collection_name="company_docs",
vectors_config=VectorParams(
size=1536,
distance=Distance.COSINE
)
)
# 인덱스 설정 (HNSW 파라미터 튜닝)
client.update_collection(
collection_name="company_docs",
hnsw_config={
"m": 16, # 연결 수 (높을수록 정확하지만 느림)
"ef_construct": 100 # 인덱스 빌드 품질 (높을수록 좋지만 느림)
}
)
컬렉션 생성 (Java - Spring AI)
// Spring AI + Qdrant 설정
// application.yml
spring:
ai:
vectorstore:
qdrant:
host: localhost
port: 6334
api-key: your-secret-api-key
collection-name: company_docs
use-tls: false
embedding:
openai:
api-key: ${OPENAI_API_KEY}
// Java에서 컬렉션 직접 생성
import io.qdrant.client.QdrantClient;
import io.qdrant.client.grpc.Collections;
QdrantClient qdrantClient = new QdrantClient(
QdrantGrpcClient.newBuilder("localhost", 6334, false).build()
);
qdrantClient.createCollectionAsync("company_docs",
Collections.VectorsConfig.newBuilder()
.setParams(Collections.VectorParams.newBuilder()
.setSize(1536)
.setDistance(Collections.Distance.Cosine)
.build())
.build()
).get();
pgvector 설치 및 세팅
Docker로 PostgreSQL + pgvector 설치
# pgvector 포함 PostgreSQL 이미지
docker run -d \
--name pgvector \
-e POSTGRES_PASSWORD=password \
-e POSTGRES_DB=rag_db \
-p 5432:5432 \
-v pgvector_data:/var/lib/postgresql/data \
pgvector/pgvector:pg16
기존 PostgreSQL에 pgvector 확장 설치
-- PostgreSQL에 접속 후 실행
CREATE EXTENSION IF NOT EXISTS vector;
-- 버전 확인
SELECT extversion FROM pg_extension WHERE extname = 'vector';
테이블 생성 및 인덱스 설정
-- 문서 저장 테이블
CREATE TABLE documents (
id BIGSERIAL PRIMARY KEY,
content TEXT NOT NULL,
embedding VECTOR(1536), -- 벡터 컬럼
metadata JSONB DEFAULT '{}', -- 메타데이터 (출처, 날짜, 부서 등)
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
-- 벡터 인덱스 생성 (HNSW - 고속 근사 검색)
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- 메타데이터 인덱스 (부서별 필터링 등)
CREATE INDEX ON documents USING GIN (metadata);
pgvector로 유사도 검색 (Python)
import psycopg2
import numpy as np
conn = psycopg2.connect("postgresql://user:password@localhost/rag_db")
def search_similar(query_embedding: list[float], top_k: int = 5, filter_dept: str = None):
cursor = conn.cursor()
base_query = """
SELECT content, metadata, 1 - (embedding <=> %s::vector) AS similarity
FROM documents
WHERE 1=1
"""
params = [query_embedding]
if filter_dept:
base_query += " AND metadata->>'department' = %s"
params.append(filter_dept)
base_query += " ORDER BY embedding <=> %s::vector LIMIT %s"
params.extend([query_embedding, top_k])
cursor.execute(base_query, params)
return cursor.fetchall()
pgvector 사용 (Java - Spring AI)
// application.yml
spring:
datasource:
url: jdbc:postgresql://localhost:5432/rag_db
username: user
password: password
ai:
vectorstore:
pgvector:
index-type: HNSW
distance-type: COSINE_DISTANCE
dimensions: 1536
initialize-schema: true # 자동 테이블 생성
5. 임베딩 모델 선택 및 서버 구성
임베딩 모델은 크게 두 가지 방식으로 사용할 수 있습니다: 외부 API 호출 방식과 자체 호스팅 방식입니다.
외부 API 방식 (OpenAI Embeddings)
┌──────────────────────────────────────────────────────┐
│ 외부 API 방식 │
│ │
│ 장점 단점 │
│ - 설치 불필요 - API 비용 발생 │
│ - 고품질 임베딩 - 외부 네트워크 의존 │
│ - 업데이트 자동 - 데이터 외부 전송 (보안 이슈)│
│ - 관리 부담 없음 - 레이턴시 (네트워크 RTT) │
└──────────────────────────────────────────────────────┘
from openai import OpenAI
client = OpenAI(api_key="sk-...")
def embed(texts: list[str]) -> list[list[float]]:
response = client.embeddings.create(
model="text-embedding-3-small", # 1536차원, 저렴
input=texts
)
return [item.embedding for item in response.data]
비용 계산 예시:
- text-embedding-3-small: $0.02 / 1M tokens
- 문서 1만 건, 평균 500 토큰 = 500만 토큰 = $0.10 (인덱싱 1회)
- 일일 검색 1,000건, 쿼리 평균 50 토큰 = 5만 토큰/일 = $0.001/일
자체 호스팅 방식 (HuggingFace)
사내 문서가 기밀이라면 외부 API 전송이 불가능합니다. 이 경우 임베딩 모델을 자체 서버에 올려야 합니다.
┌──────────────────────────────────────────────────────┐
│ 자체 호스팅 방식 │
│ │
│ 장점 단점 │
│ - 데이터 외부 전송 없음 - GPU 서버 필요 │
│ - 장기 비용 절감 - 모델 관리 부담 │
│ - 레이턴시 최소화 - 초기 설정 복잡 │
│ - 커스텀 파인튜닝 가능 - 모델 업데이트 직접 관리 │
└──────────────────────────────────────────────────────┘
한국어 지원 추천 모델:
- BAAI/bge-m3: 다국어, 한국어 성능 우수, 1024차원
- intfloat/multilingual-e5-large: 한국어 포함 100개 언어
- snunlp/KR-ELECTRA-discriminator: 한국어 특화
FastAPI로 임베딩 서버 구축 (Python)
# embedding_server.py
from fastapi import FastAPI
from pydantic import BaseModel
from sentence_transformers import SentenceTransformer
import torch
app = FastAPI()
model = SentenceTransformer("BAAI/bge-m3", device="cuda" if torch.cuda.is_available() else "cpu")
class EmbedRequest(BaseModel):
texts: list[str]
class EmbedResponse(BaseModel):
embeddings: list[list[float]]
@app.post("/embed", response_model=EmbedResponse)
async def embed(request: EmbedRequest):
embeddings = model.encode(
request.texts,
normalize_embeddings=True, # 코사인 유사도를 위한 정규화
batch_size=32
).tolist()
return EmbedResponse(embeddings=embeddings)
# 실행
pip install fastapi uvicorn sentence-transformers torch
uvicorn embedding_server:app --host 0.0.0.0 --port 8001 --workers 1
Docker로 임베딩 서버 배포
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
RUN pip install fastapi uvicorn sentence-transformers torch --index-url https://download.pytorch.org/whl/cpu
COPY embedding_server.py .
# 모델 사전 다운로드 (빌드 시)
RUN python -c "from sentence_transformers import SentenceTransformer; SentenceTransformer('BAAI/bge-m3')"
CMD ["uvicorn", "embedding_server:app", "--host", "0.0.0.0", "--port", "8001"]
6. 문서 인덱싱 파이프라인 구축
인덱싱 파이프라인은 사내 문서를 수집, 정제, 임베딩하여 벡터 DB에 저장하는 과정입니다. 이 파이프라인의 품질이 RAG 시스템 전체의 품질을 좌우합니다.
문서 소스별 로더
# 다양한 소스에서 문서 로딩
from langchain.document_loaders import (
PyPDFLoader, # PDF
Docx2txtLoader, # Word
ConfluenceLoader, # Confluence
NotionDBLoader, # Notion
GitLoader, # GitHub (코드, README)
WebBaseLoader, # 웹페이지
CSVLoader, # CSV
)
# Confluence 로더 예시
confluence_loader = ConfluenceLoader(
url="https://your-company.atlassian.net/wiki",
username="user@company.com",
api_key="your-confluence-api-key",
space_key="TECH", # 기술 문서 스페이스
include_attachments=False,
limit=50 # 한 번에 가져올 페이지 수
)
docs = confluence_loader.load()
청크 분할 전략
from langchain.text_splitter import RecursiveCharacterTextSplitter
# 일반 문서용 (기술 문서, 정책 문서 등)
general_splitter = RecursiveCharacterTextSplitter(
chunk_size=800,
chunk_overlap=80,
separators=["\n\n", "\n", ".", " ", ""]
)
# 코드 문서용 (GitHub, 기술 스펙 등)
from langchain.text_splitter import Language
code_splitter = RecursiveCharacterTextSplitter.from_language(
language=Language.PYTHON,
chunk_size=1000,
chunk_overlap=100
)
# 마크다운용 (Notion, Confluence 등)
md_splitter = RecursiveCharacterTextSplitter.from_language(
language=Language.MARKDOWN,
chunk_size=800,
chunk_overlap=80
)
메타데이터 설계
메타데이터는 검색 후 필터링과 출처 표시에 핵심적입니다. 사내 환경에 맞게 필드를 설계하세요.
# 메타데이터 구조 예시
{
"source": "confluence", # 문서 출처
"space": "TECH", # Confluence 스페이스
"page_id": "123456", # 원본 문서 ID
"title": "API 설계 가이드", # 문서 제목
"department": "backend", # 담당 부서
"last_updated": "2026-03-15", # 최종 수정일
"author": "kim@company.com", # 작성자
"category": "architecture", # 카테고리
"access_level": "internal", # 접근 권한 (public/internal/restricted)
"language": "ko" # 언어
}
완성된 인덱싱 파이프라인 (Python)
import asyncio
from datetime import datetime
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct
from openai import AsyncOpenAI
import uuid
qdrant = QdrantClient(host="localhost", port=6333)
openai = AsyncOpenAI()
async def embed_batch(texts: list[str]) -> list[list[float]]:
response = await openai.embeddings.create(
model="text-embedding-3-small",
input=texts
)
return [item.embedding for item in response.data]
async def index_documents(documents: list[dict]):
BATCH_SIZE = 50
for i in range(0, len(documents), BATCH_SIZE):
batch = documents[i:i+BATCH_SIZE]
texts = [doc["content"] for doc in batch]
# 배치 임베딩
embeddings = await embed_batch(texts)
# 벡터 DB에 업서트
points = [
PointStruct(
id=str(uuid.uuid4()),
vector=embedding,
payload={
"content": doc["content"],
**doc["metadata"]
}
)
for doc, embedding in zip(batch, embeddings)
]
qdrant.upsert(
collection_name="company_docs",
points=points
)
print(f"인덱싱 완료: {i+len(batch)}/{len(documents)}")
await asyncio.sleep(0.5) # API rate limit 방지
증분 업데이트 전략
문서가 추가/수정/삭제될 때 전체 재인덱싱 없이 효율적으로 업데이트하는 방법입니다.
# 문서 해시 기반 변경 감지
import hashlib
def get_doc_hash(content: str) -> str:
return hashlib.md5(content.encode()).hexdigest()
# DB에 hash 저장 → 변경된 문서만 재인덱싱
# 삭제된 문서 → source_id로 Qdrant에서 삭제
qdrant.delete(
collection_name="company_docs",
points_selector={"filter": {"must": [{"key": "source_id", "match": {"value": "deleted_page_id"}}]}}
)
7. 검색 및 생성 API 서버 구축
Python FastAPI로 RAG API 서버 구축
# rag_server.py
from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from openai import AsyncOpenAI
from qdrant_client import QdrantClient
import asyncio
app = FastAPI(title="사내 RAG API", version="1.0.0")
security = HTTPBearer()
qdrant = QdrantClient(host="localhost", port=6333)
openai = AsyncOpenAI()
class QueryRequest(BaseModel):
question: str
department: str | None = None # 부서별 필터링
top_k: int = 5
class QueryResponse(BaseModel):
answer: str
sources: list[dict]
confidence: float
async def embed_query(text: str) -> list[float]:
response = await openai.embeddings.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
def build_filter(department: str | None) -> dict | None:
if not department:
return None
return {
"must": [
{"key": "department", "match": {"value": department}},
{"key": "access_level", "match": {"any": ["public", "internal"]}}
]
}
@app.post("/query", response_model=QueryResponse)
async def query(
request: QueryRequest,
credentials: HTTPAuthorizationCredentials = Depends(security)
):
# 1. 질문 임베딩
query_embedding = await embed_query(request.question)
# 2. 벡터 검색
search_results = qdrant.search(
collection_name="company_docs",
query_vector=query_embedding,
query_filter=build_filter(request.department),
limit=request.top_k,
with_payload=True,
score_threshold=0.6 # 유사도 0.6 이하는 제외
)
if not search_results:
raise HTTPException(status_code=404, detail="관련 문서를 찾을 수 없습니다.")
# 3. 컨텍스트 구성
context = "\n\n".join([
f"[출처: {r.payload.get('title', 'Unknown')}]\n{r.payload['content']}"
for r in search_results
])
# 4. LLM으로 답변 생성
response = await openai.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": """당신은 사내 문서 기반의 AI 어시스턴트입니다.
반드시 제공된 참고 문서를 기반으로만 답변하세요.
확실하지 않은 내용은 추측하지 말고 "해당 내용은 문서에서 확인되지 않습니다"라고 하세요."""
},
{
"role": "user",
"content": f"[참고 문서]\n{context}\n\n[질문]\n{request.question}"
}
],
temperature=0.1,
max_tokens=1000
)
answer = response.choices[0].message.content
avg_score = sum(r.score for r in search_results) / len(search_results)
return QueryResponse(
answer=answer,
sources=[
{
"title": r.payload.get("title"),
"source": r.payload.get("source"),
"score": round(r.score, 3),
"url": r.payload.get("url")
}
for r in search_results
],
confidence=round(avg_score, 3)
)
Java Spring Boot로 RAG API 서버 구축
// RagController.java
@RestController
@RequestMapping("/api/v1")
@RequiredArgsConstructor
public class RagController {
private final RagService ragService;
@PostMapping("/query")
public ResponseEntity<QueryResponse> query(
@RequestBody @Valid QueryRequest request,
@AuthenticationPrincipal UserDetails user
) {
QueryResponse response = ragService.query(request, user);
return ResponseEntity.ok(response);
}
}
// RagService.java
@Service
@RequiredArgsConstructor
public class RagService {
private final ChatClient chatClient;
private final VectorStore vectorStore;
public QueryResponse query(QueryRequest request, UserDetails user) {
// 1. 유사 문서 검색
SearchRequest searchRequest = SearchRequest.query(request.getQuestion())
.withTopK(request.getTopK())
.withSimilarityThreshold(0.6)
.withFilterExpression(buildFilter(request.getDepartment(), user));
List<Document> docs = vectorStore.similaritySearch(searchRequest);
if (docs.isEmpty()) {
throw new NoResultException("관련 문서를 찾을 수 없습니다.");
}
// 2. RAG 답변 생성 (QuestionAnswerAdvisor 활용)
String answer = chatClient.prompt()
.advisors(new QuestionAnswerAdvisor(vectorStore, searchRequest))
.user(request.getQuestion())
.call()
.content();
return QueryResponse.builder()
.answer(answer)
.sources(docs.stream().map(this::toSource).collect(toList()))
.build();
}
private String buildFilter(String department, UserDetails user) {
// Spring AI Filter Expression
if (department != null) {
return String.format("department == '%s' && access_level in ['public', 'internal']",
department);
}
return "access_level in ['public', 'internal']";
}
}
8. 보안 및 접근 제어 설계
사내 RAG 서버에서 보안은 특히 중요합니다. 기밀 문서가 권한 없는 사람에게 노출되어서는 안 됩니다.
문서 레벨 접근 제어
# 메타데이터 기반 접근 제어
ACCESS_LEVELS = {
"public": 0, # 전체 공개
"internal": 1, # 임직원 전체
"department": 2, # 특정 부서
"restricted": 3, # 특정 팀/개인
}
def build_access_filter(user: User) -> dict:
"""사용자 권한에 맞는 Qdrant 필터 생성"""
allowed_conditions = [
{"key": "access_level", "match": {"value": "public"}},
{"key": "access_level", "match": {"value": "internal"}},
]
# 부서 권한
allowed_conditions.append({
"key": "department",
"match": {"value": user.department}
})
# 특정 팀 권한
for team in user.teams:
allowed_conditions.append({
"key": "allowed_teams",
"match": {"any": [team]}
})
return {"should": allowed_conditions}
API 인증/인가
# JWT 토큰 기반 인증
from fastapi import Depends
from jose import jwt
SECRET_KEY = "your-secret-key"
def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
token = credentials.credentials
payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
user_id = payload.get("sub")
if not user_id:
raise HTTPException(status_code=401, detail="Invalid token")
return payload
감사 로그 (Audit Log)
"누가, 언제, 어떤 질문을 했는지" 기록은 컴플라이언스와 보안 감사에 필수입니다.
# 모든 쿼리 로깅
import logging
import json
from datetime import datetime
audit_logger = logging.getLogger("audit")
@app.post("/query")
async def query(request: QueryRequest, user = Depends(verify_token)):
start_time = datetime.now()
response = await process_query(request, user)
# 감사 로그 기록
audit_logger.info(json.dumps({
"timestamp": start_time.isoformat(),
"user_id": user["sub"],
"department": user.get("department"),
"question_hash": hashlib.sha256(request.question.encode()).hexdigest(),
"sources_accessed": [s["title"] for s in response.sources],
"response_time_ms": (datetime.now() - start_time).microseconds // 1000
}))
return response
9. 모니터링 및 품질 관리
핵심 지표 모니터링
┌──────────────────────────────────────────────────────┐
│ RAG 서버 모니터링 지표 │
│ │
│ 성능 지표 │
│ - 평균 응답 시간 (목표: < 3초) │
│ - 벡터 검색 레이턴시 (목표: < 100ms) │
│ - LLM API 레이턴시 │
│ - TPS (초당 처리 요청 수) │
│ │
│ 품질 지표 │
│ - 검색 결과 없음 비율 (높으면 문서 부족) │
│ - 사용자 피드백 👍/👎 비율 │
│ - 낮은 신뢰도 응답 비율 (score < 0.6) │
│ │
│ 운영 지표 │
│ - 벡터 DB 디스크 사용량 │
│ - 임베딩 API 비용 │
│ - LLM API 비용 │
│ - 일일 활성 사용자 수 │
└──────────────────────────────────────────────────────┘
Prometheus + Grafana로 메트릭 수집
from prometheus_client import Counter, Histogram, generate_latest
from fastapi import Response
# 메트릭 정의
query_count = Counter("rag_queries_total", "Total RAG queries", ["department", "status"])
query_latency = Histogram("rag_query_duration_seconds", "RAG query latency",
buckets=[0.5, 1.0, 2.0, 3.0, 5.0, 10.0])
search_score = Histogram("rag_search_score", "Vector search score distribution",
buckets=[0.5, 0.6, 0.7, 0.8, 0.9, 1.0])
@app.post("/query")
async def query(request: QueryRequest):
with query_latency.time():
try:
result = await process_query(request)
query_count.labels(department=request.department, status="success").inc()
search_score.observe(result.confidence)
return result
except Exception as e:
query_count.labels(department=request.department, status="error").inc()
raise
@app.get("/metrics")
async def metrics():
return Response(generate_latest(), media_type="text/plain")
10. 배포 전략 및 운영 팁
Docker Compose 풀스택 배포
# docker-compose.prod.yml
version: '3.8'
services:
# 벡터 DB
qdrant:
image: qdrant/qdrant:latest
restart: always
volumes:
- qdrant_data:/qdrant/storage
environment:
QDRANT__SERVICE__API_KEY: ${QDRANT_API_KEY}
ports:
- "6333:6333"
# 임베딩 서버
embedding:
build: ./embedding-server
restart: always
deploy:
resources:
reservations:
devices:
- driver: nvidia
count: 1
capabilities: [gpu]
# RAG API 서버
rag-api:
build: ./rag-server
restart: always
environment:
QDRANT_HOST: qdrant
QDRANT_API_KEY: ${QDRANT_API_KEY}
OPENAI_API_KEY: ${OPENAI_API_KEY}
EMBEDDING_SERVER_URL: http://embedding:8001
ports:
- "8000:8000"
depends_on:
- qdrant
- embedding
# Nginx 리버스 프록시
nginx:
image: nginx:alpine
volumes:
- ./nginx.conf:/etc/nginx/conf.d/default.conf
ports:
- "443:443"
depends_on:
- rag-api
volumes:
qdrant_data:
단계별 구축 로드맵
┌──────────────────────────────────────────────────────────┐
│ 사내 RAG 서버 구축 로드맵 │
│ │
│ Week 1-2: MVP 구축 │
│ - Qdrant 또는 pgvector 설치 │
│ - OpenAI Embeddings + GPT-4o 연동 │
│ - 핵심 문서 소스 1개만 인덱싱 (예: Confluence) │
│ - 기본 검색 API 구현 │
│ - 슬랙봇 연동 │
│ │
│ Week 3-4: 품질 개선 │
│ - 하이브리드 검색 (BM25 + 벡터) 적용 │
│ - Reranking 추가 │
│ - 메타데이터 필터링 (부서별) │
│ - RAGAS로 품질 측정 시작 │
│ │
│ Month 2: 운영 안정화 │
│ - 접근 제어 (JWT, 문서 레벨 권한) │
│ - 감사 로그 구축 │
│ - Prometheus + Grafana 모니터링 │
│ - 문서 소스 확대 (GitHub, Notion 등) │
│ │
│ Month 3+: 고도화 │
│ - 자체 임베딩 모델 호스팅 (보안 강화) │
│ - 자체 호스팅 LLM 검토 (Llama, Mistral) │
│ - 멀티턴 대화 지원 │
│ - 사용자 피드백 기반 품질 개선 │
└──────────────────────────────────────────────────────────┘
운영 핵심 팁
1. 작게 시작하라
처음부터 모든 사내 문서를 인덱싱하려 하지 마세요. 가장 자주 묻는 질문에 답할 수 있는 핵심 문서 1-2개 소스부터 시작해서 품질을 검증한 후 확장합니다.
2. 청크 크기는 반드시 실험하라
500토큰, 800토큰, 1000토큰으로 각각 실험하고 RAGAS로 품질을 비교하세요. 문서 유형마다 최적 크기가 다릅니다.
3. 문서 업데이트 자동화를 빠르게 구축하라
인덱싱 파이프라인을 수동으로 실행하면 금방 방치됩니다. Confluence webhook, GitHub Actions 등으로 문서 변경 시 자동 재인덱싱을 구축하세요.
4. 검색 결과 없음을 친절하게 처리하라
유사도 임계값 이하인 경우 "관련 문서를 찾지 못했습니다. 직접 문의해주세요"로 답하게 하세요. 환각보다 낫습니다.
5. 사용자 피드백을 수집하라
👍/👎 버튼 하나로 품질 데이터를 수집합니다. 이 데이터가 시스템 개선의 나침반이 됩니다.
마무리 — 사내 RAG 서버, 생각보다 가깝다
사내 RAG 서버 구축이 복잡하게 느껴질 수 있지만, 핵심은 단순합니다:
┌──────────────────────────────────────────────────────┐
│ 사내 RAG 서버 구축 핵심 정리 │
│ │
│ 언어 선택 │
│ 기존 스택이 Java → Spring AI로 구축 │
│ AI 중심 팀 → Python + FastAPI │
│ 현실적 선택 → Python RAG + Java 비즈니스 로직 분리 │
│ │
│ 벡터 DB 선택 │
│ 독립 서비스, 대용량 → Qdrant │
│ PostgreSQL 이미 사용 중 → pgvector │
│ │
│ 구축 순서 │
│ MVP (1-2주) → 품질 개선 → 운영 안정화 → 고도화 │
│ │
│ 성공 비결 │
│ "작게 시작, 빠르게 검증, 점진적 확장" │
└──────────────────────────────────────────────────────┘
Qdrant를 Docker로 띄우고, OpenAI 임베딩으로 문서 10개를 인덱싱하고, FastAPI로 검색 API 하나 만드는 데 이틀이면 충분합니다. 거기서부터 시작하세요. 완벽한 시스템을 처음부터 만들려다 시작도 못하는 것보다, 불완전하지만 작동하는 RAG 서버가 훨씬 가치 있습니다.
반응형
'프로그래밍 PROGRAMMING > 인공지능 AI' 카테고리의 다른 글
| Claude Design 시작하기: 말로 만들고, 대화로 다듬는 새로운 디자인 방식 (1) | 2026.04.22 |
|---|---|
| 출처가 있어도 답변이 틀리는 이유 — Citation이 신뢰를 보장하지 않는 순간들 (0) | 2026.04.21 |
| AI에서 RAG란 무엇인가? — 환각을 없애고 최신 지식을 주입하는 검색 증강 생성 (3) | 2026.04.17 |
| 클로드 코드 team-onboarding 스킬 - 신규 팀원 온보딩을 자동화하는 최신 기능 (5) | 2026.04.16 |
| claude code 생명주기 - 프롬프트에서 결과까지, 내부에서 무슨 일이 일어나는가 (0) | 2026.04.14 |