RAG 환각이란 무엇인가?

저는 AI API 통합 프로젝트를 진행하면서 가장 많이 마주치는 문제가 바로 RAG 환각입니다. RAG(Retrieval-Augmented Generation)는 검색 결과를 기반으로 답변을 생성하는 기술인데, 검색된 정보를 잘못 해석하거나, 존재하지 않는 내용을凭空捏造하는 경우가 발생합니다.

예를 들어, 특정 문서에 "2023년 매출 100억"이라는 정보가 있는데, AI가 "2023년 매출 200억"이라고 생성하는 것이죠. 이 현상을 환각(Hallucination)이라고 부릅니다.

RAG 시스템 신뢰도 검증 아키텍처

저의 경험상, 신뢰도 높은 RAG 시스템을 구축하려면 3단계 검증 체계가 필요합니다:

핵심 구현 코드

1단계: 신뢰도 점수 시스템 구현

"""
RAG 답변 신뢰도 검증 시스템
HolySheep AI API를 사용한 인용 추적 및 점수 산출
"""

import requests
import json
from typing import List, Dict, Tuple

HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

class RAGConfidenceValidator:
    """RAG 시스템의 답변 신뢰도를 검증하는 클래스"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    def calculate_relevance_score(self, query: str, retrieved_docs: List[Dict]) -> List[Dict]:
        """검색 결과와 질문 간의 관련성 점수 계산"""
        
        prompt = f"""다음 질문과 검색된 문서들의 관련성을 0.0~1.0 점수로 평가하세요.

질문: {query}

문서들:
{chr(10).join([f"[{i+1}] {doc.get('content', '')[:200]}" for i, doc in enumerate(retrieved_docs)])}

JSON 형식으로 출력:
{{"scores": [{{"doc_index": 0, "score": 0.85, "reason": "이유"}}]}}
"""
        
        response = requests.post(
            f"{HOLYSHEEP_BASE_URL}/chat/completions",
            headers=self.headers,
            json={
                "model": "gpt-4.1",
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0.1,
                "max_tokens": 500
            }
        )
        
        result = response.json()
        content = result["choices"][0]["message"]["content"]
        
        # JSON 파싱하여 점수 추출
        try:
            parsed = json.loads(content)
            return parsed.get("scores", [])
        except:
            return [{"doc_index": i, "score": 0.5} for i in range(len(retrieved_docs))]
    
    def validate_answer(self, question: str, answer: str, citations: List[Dict]) -> Dict:
        """답변과 인용 문서의 일관성 검증"""
        
        prompt = f"""다음 질문에 대한 답변과 인용된 문서들 간의 일관성을 검증하세요.

질문: {question}
답변: {answer}

인용 문서들:
{chr(10).join([f"[{c['doc_index']}] {c['content'][:300]}" for c in citations])}

다음 항목을 0.0~1.0 점수로 평가:
1. factual_consistency: 사실적 일관성 (답변 내용이 문서와 맞는지)
2. completeness: 완전성 (질문에 답변이 충분히 되었는지)
3. citation_accuracy: 인용 정확성 (인용된 문서가 답변을 뒷받침하는지)

JSON으로 출력:
{{
    "factual_consistency": 0.95,
    "completeness": 0.88,
    "citation_accuracy": 0.92,
    "overall_confidence": 0.92,
    "warnings": ["주의사항이 있으면列出"]
}}
"""
        
        response = requests.post(
            f"{HOLYSHEEP_BASE_URL}/chat/completions",
            headers=self.headers,
            json={
                "model": "gpt-4.1",
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0.1,
                "max_tokens": 600
            }
        )
        
        return response.json()["choices"][0]["message"]["content"]


사용 예시

validator = RAGConfidenceValidator(HOLYSHEEP_API_KEY)

question = "회사 연간 매출은 얼마인가요?"
retrieved_docs = [
    {"content": "2023년 연간 매출 150억 원, 전년 대비 20% 증가", "source": "annual_report.txt"},
    {"content": "2024년 1분기 매출 45억 원 기록", "source": "quarterly_report.txt"}
]


관련성 점수 계산

scores = validator.calculate_relevance_score(question, retrieved_docs)
print(f"검색 관련성 점수: {scores}")

2단계: 인용 추적 및 하이라이트 시스템

"""
인용 가능한 RAG 응답 생성기
생성된 답변의 각 부분이 어느 문서에서 왔는지 추적
"""

import re
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class Citation:
    """인용 정보 클래스"""
    doc_id: str
    content: str
    confidence: float
    position: Tuple[int, int]  # (start_idx, end_idx)

class CitationAwareRAG:
    """인용 추적이 가능한 RAG 시스템"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    def generate_with_citations(self, query: str, documents: List[Dict]) -> Dict:
        """
        인용이 포함된 답변 생성
        각 답변 세그먼트에 출처 문서 연결
        """
        
        # 문서 컨텍스트 구성 (문서에 ID 부여)
        doc_context = "\n\n".join([
            f"[문서-{i}](출처: {doc.get('source', 'unknown')}):\n{doc['content']}"
            for i, doc in enumerate(documents)
        ])
        
        prompt = f"""아래 문서들을 기반으로 질문에 답변하세요.

[문서 컨텍스트]
{doc_context}

[질문]
{query}

답변 규칙:
1. 반드시 위 문서에서 직접적인 증거를 기반으로 답변하세요
2. 답변의 각 주장 뒤에 [|문서-ID|] 형식으로 인용을 표시하세요
3. 문서에 없는 정보는 절대凭空捏造하지 마세요
4. 불확실한 정보는 [|문서-ID| 확률:XX%] 형태로 표시하세요

예시 출력:
"2023년 매출은 150억 원입니다[|문서-0|]。이는 전년 대비 20% 증가한 수치입니다[|문서-0|]."
"""
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json={
                "model": "gpt-4.1",
                "messages": [
                    {"role": "system", "content": "당신은 정확한 정보만 제공하는 AI 어시스턴트입니다."},
                    {"role": "user", "content": prompt}
                ],
                "temperature": 0.2,
                "max_tokens": 1000
            }
        )
        
        raw_answer = response.json()["choices"][0]["message"]["content"]
        
        # 인용 추출 및 파싱
        parsed_answer = self._parse_citations(raw_answer, documents)
        
        return {
            "answer": parsed_answer["text"],
            "citations": parsed_answer["citations"],
            "confidence": self._calculate_answer_confidence(parsed_answer["citations"]),
            "used_documents": list(set([c["doc_id"] for c in parsed_answer["citations"]]))
        }
    
    def _parse_citations(self, text: str, documents: List[Dict]) -> Dict:
        """답변 텍스트에서 인용 패턴 추출"""
        
        # [|문서-ID|] 패턴 찾기
        pattern = r'\[\|문서-(\d+)\|\]'
        matches = list(re.finditer(pattern, text))
        
        citations = []
        parsed_segments = []
        last_end = 0
        
        for match in matches:
            # 인용 전 텍스트 추가
            if match.start() > last_end:
                parsed_segments.append({
                    "text": text[last_end:match.start()],
                    "citation": None
                })
            
            doc_idx = int(match.group(1))
            if doc_idx < len(documents):
                citation = Citation(
                    doc_id=f"문서-{doc_idx}",
                    content=documents[doc_idx]["content"][:200],
                    confidence=0.95,
                    position=(match.start(), match.end())
                )
                citations.append(citation)
                parsed_segments.append({
                    "text": text[match.start():match.end()],
                    "citation": {
                        "doc_id": citation.doc_id,
                        "source": documents[doc_idx].get("source", "unknown"),
                        "confidence": citation.confidence
                    }
                })
            last_end = match.end()
        
        # 마지막 세그먼트 추가
        if last_end < len(text):
            parsed_segments.append({
                "text": text[last_end:],
                "citation": None
            })
        
        return {
            "text": text,
            "segments": parsed_segments,
            "citations": [vars(c) for c in citations]
        }
    
    def _calculate_answer_confidence(self, citations: List[Dict]) -> float:
        """답변 전체의 신뢰도 점수 산출"""
        if not citations:
            return 0.0
        
        total_confidence = sum([c["confidence"] for c in citations])
        avg_confidence = total_confidence / len(citations)
        
        # 인용 밀도 보너스 (인용이 많을수록 신뢰도 상승)
        citation_density = len(citations) / max(1, len(citations[0]["content"]) / 100)
        density_bonus = min(0.1, citation_density * 0.05)
        
        return min(1.0, avg_confidence + density_bonus)


HolySheep AI API를 사용한 실제 호출

rag = CitationAwareRAG("YOUR_HOLYSHEEP_API_KEY")

documents = [
    {
        "content": "2023년HolySheep AI 연간 매출: 150억 원. 2022년 대비 25% 성장.",
        "source": "financial_report_2023.txt"
    },
    {
        "content": "주요 경쟁사 대비 30% 저렴한 가격으로 시장을 확대 중.",
        "source": "market_analysis.txt"
    }
]

result = rag.generate_with_citations("2023년 매출과 성장률은?", documents)

print(f"답변 신뢰도: {result['confidence']:.2%}")
print(f"사용된 문서: {result['used_documents']}")
print(f"답변: {result['answer']}")

3단계: 실전 RAG 파이프라인 통합

"""
완전한 RAG 파이프라인: 검색 → 생성 → 검증 → 재검증
HolySheep AI 단일 API 키로 모든 모델 활용
"""

import requests
from typing import List, Dict, Optional

class HolySheepRAGPipeline:
    """HolySheep AI 기반의 검증된 RAG 파이프라인"""
    
    # HolySheep AI 가격 정보 (2024년 기준)
    PRICING = {
        "gpt-4.1": {"input": 8.0, "output": 8.0, "per": "MTok", "use": "정밀 검증"},
        "gpt-4.1-mini": {"input": 1.0, "output": 4.0, "per": "MTok", "use": "빠른 검색"},
        "claude-sonnet-4": {"input": 15.0, "output": 15.0, "per": "MTok", "use": "복잡한 추론"},
        "gemini-2.5-flash": {"input": 2.50, "output": 10.0, "per": "MTok", "use": "대량 처리"},
        "deepseek-v3.2": {"input": 0.42, "output": 1.68, "per": "MTok", "use": "비용 최적화"}
    }
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    def retrieve_documents(self, query: str, vector_store: List[Dict], top_k: int = 5) -> List[Dict]:
        """의미론적 검색 수행 (단비速度快、成本低의 Gemini Flash 활용)"""
        
        # Gemini 2.5 Flash로 임베딩 유사도 계산
        response = requests.post(
            f"{self.base_url}/embeddings",
            headers=self.headers,
            json={
                "model": "gemini-2.5-flash",
                "input": query
            }
        )
        
        query_embedding = response.json().get("data", [{}])[0].get("embedding", [])
        
        # 간단한 코사인 유사도 계산
        def cosine_sim(a, b):
            return sum(x*y for x,y in zip(a,b)) / (max(sum(x**2 for x in a)**0.5, 0.0001) * max(sum(x**2 for x in b)**0.5, 0.0001))
        
        scored_docs = []
        for doc in vector_store:
            doc_emb = doc.get("embedding", [0] * 768)
            sim = cosine_sim(query_embedding, doc_emb)
            scored_docs.append((sim, doc))
        
        # 상위 k개 문서 반환
        scored_docs.sort(key=lambda x: x[0], reverse=True)
        return [doc for _, doc in scored_docs[:top_k]]
    
    def generate_answer(self, query: str, context_docs: List[Dict]) -> Dict:
        """GPT-4.1로 인용 기반 답변 생성"""
        
        context = "\n\n".join([
            f"[문서 {i+1}] 출처: {doc['source']}\n{doc['content']}"
            for i, doc in enumerate(context_docs)
        ])
        
        prompt = f"""Based ONLY on the following documents, answer the question.

Documents:
{context}

Question: {query}

Instructions:
- Answer ONLY using information from the documents above
- If you're unsure, say "I cannot find this information in the provided documents"
- Reference specific document numbers for each claim
- Rate your confidence: HIGH (all info from docs) / MEDIUM (some inference) / LOW (mostly guessing)
"""
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json={
                "model": "gpt-4.1",
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0.1,
                "max_tokens": 800
            }
        )
        
        return {
            "answer": response.json()["choices"][0]["message"]["content"],
            "context_docs": context_docs
        }
    
    def verify_answer(self, query: str, answer: str, original_docs: List[Dict]) -> Dict:
        """Claude Sonnet 4로 답변 검증 (다른 모델로 교차 검증)"""
        
        verification_prompt = f"""Verify if this answer is supported by the documents.

Original Question: {query}
Answer to Verify: {answer}

Documents:
{chr(10).join([f"[Doc {i+1}]: {doc['content']}" for i, doc in enumerate(original_docs)])}

Return JSON:
{{
    "is_supported": true/false,
    "confidence": 0.0-1.0,
    "unsupported_claims": ["list any claims not in documents"],
    "verification_notes": "explanation"
}}
"""
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json={
                "model": "claude-sonnet-4",
                "messages": [{"role": "user", "content": verification_prompt}],
                "temperature": 0.1,
                "max_tokens": 500
            }
        )
        
        try:
            import json
            result = json.loads(response.json()["choices"][0]["message"]["content"])
            return result
        except:
            return {"is_supported": True, "confidence": 0.8, "verification_notes": "Parse error"}
    
    def run_pipeline(self, query: str, vector_store: List[Dict]) -> Dict:
        """전체 RAG 파이프라인 실행"""
        
        # 1. 검색
        retrieved = self.retrieve_documents(query, vector_store, top_k=3)
        
        # 2. 생성
        generated = self.generate_answer(query, retrieved)
        
        # 3. 검증
        verification = self.verify_answer(query, generated["answer"], retrieved)
        
        # 4. 신뢰도 기반 응답 결정
        if verification["confidence"] < 0.6:
            response_text = (
                f"⚠️ 신뢰도 경고: {verification['confidence']:.0%}\n"
                f"확인되지 않은 정보가 포함되어 있을 수 있습니다.\n\n"
                f"{generated['answer']}\n\n"
                f"확인 불가 내용: {verification.get('unsupported_claims', [])}"
            )
        else:
            response_text = generated["answer"]
        
        return {
            "answer": response_text,
            "confidence": verification["confidence"],
            "is_verified": verification["is_supported"],
            "sources": [doc["source"] for doc in retrieved],
            "pipeline_stats": {
                "retrieval_count": len(retrieved),
                "verification_model": "claude-sonnet-4",
                "generation_model": "gpt-4.1"
            }
        }


======== 실행 예시 ========

pipeline = HolySheepRAGPipeline("YOUR_HOLYSHEEP_API_KEY")


문서 벡터 저장소 (실제로는 Pinecone, Weaviate 등 사용)

sample_vector_store = [
    {"content": "HolySheep AI의 GPT-4.1 모델은 MTok당 $8입니다.", "source": "pricing.txt", "embedding": [0.1]*768},
    {"content": "Claude Sonnet 4는 $15/MTok로 GPT보다 가격이 높습니다.", "source": "pricing.txt", "embedding": [0.2]*768},
    {"content": "Gemini 2.5 Flash는 $2.50/MTok로 가장 저렴한 옵션입니다.", "source": "pricing.txt", "embedding": [0.3]*768},
    {"content": "DeepSeek V3.2는 $0.42/MTok로 업계 최저가입니다.", "source": "pricing.txt", "embedding": [0.4]*768},
]


질문 실행

query = "가장 저렴한 AI 모델은 무엇이며 어떤 가격인가요?"
result = pipeline.run_pipeline(query, sample_vector_store)

print(f"신뢰도: {result['confidence']:.0%}")
print(f"검증 완료: {result['is_verified']}")
print(f"출처: {result['sources']}")
print(f"답변:\n{result['answer']}")

비용 최적화 팁: HolySheep AI 활용

제가 여러 프로젝트를 진행하면서 정리한 HolySheep AI 비용 최적화 전략은 다음과 같습니다:

작업 유형추천 모델가격 ($/MTok)예상 비용
문서 임베딩Gemini 2.5 Flash$2.50$0.0025/1000건
답변 생성GPT-4.1$8.00$0.008/1000건
교차 검증Claude Sonnet 4$15.00$0.015/1000건
대량 처리DeepSeek V3.2$0.42$0.0004/1000건

실전

관련 리소스

관련 문서