본문으로 건너뛰기

© 2026 Molayo

Dev.to헤드라인2026. 05. 21. 00:42

Python으로 구축하는 안전 우선 RAG 분류 (Triage) 에이전트

요약

HackerRank Orchestrate 2026 해커톤에서 상위 1.2%의 성적을 거둔 '안전 우선 RAG 분류 에이전트' 구축 사례를 소개합니다. 일반적인 RAG의 환각 문제를 해결하기 위해 생성보다 에스컬레이션을 우선시하는 5단계 파이프라인 아키텍처를 제안합니다.

핵심 포인트

  • 운영 환경의 RAG는 단순 정보 제공을 넘어 사기나 결제와 같은 민감한 이슈에 대한 안전 장치가 필수적임
  • Vanilla RAG의 한계를 극복하기 위해 '먼저 에스컬레이션하고, 그 다음에 생성하는' 구조가 필요함
  • LLM의 과도한 위험 분류를 방지하기 위해 결정론적 규칙(Deterministic rules)을 활용하는 것이 효과적임
  • 구조화된 메타데이터 추출을 위한 1단계 분류(Classification) 프로세스가 전체 파이프라인의 핵심임

5월 1일, 저는 HackerRank Orchestrate 2026에 참여했습니다. 이는 24시간 동안 진행되는 해커톤으로, 과제는 겉보기에는 매우 단순했습니다. 제공된 지원 코퍼스 (Support Corpus)만을 사용하여 HackerRank, Claude, Visa의 티켓을 처리하는 터미널 기반의 지원 분류 (Triage) 에이전트를 구축하는 것이었습니다. 🏆 이 에이전트는 제 생애 첫 해커톤에서 12,885명의 참가자 중 154위 — 글로벌 상위 200위, 상위 1.2% —를 기록했습니다. 혼자서 진행했습니다. 팀도 없었고, 유료 도구도 사용하지 않았습니다. 문제는 무엇이었을까요? 바로 환각 (Hallucination)이 없어야 한다는 점이었습니다. 안전하지 않은 답변도 안 됩니다. 사기 (Fraud) 또는 결제 (Billing) 티켓에 대한 오답은 절대 허용되지 않았습니다. 여기 속도보다 안전을 우선시하는 하이브리드 RAG 에이전트를 구축한 방법과, 그 과정에서 왜 API 키 3개를 소진했는지에 대해 설명하겠습니다.

문제점: 왜 일반적인 RAG (Vanilla RAG)만으로는 부족했는가
대부분의 RAG 튜토리얼은 문서를 청킹 (Chunking)하고, 임베딩 (Embedding)한 뒤, 질문을 던지는 방법을 보여줍니다. 블로그 데모용으로는 괜찮습니다. 하지만 사기 보고, 결제 분쟁, 계정 도용 등을 처리하는 운영 환경의 지원 시스템 (Production Support System)에서 일반적인 RAG는 위험합니다. 다음과 같은 상황이 발생하면 어떻게 될까요:

사용자가 "제 신원을 도용당했습니다, 어떻게 해야 하나요?"라고 말할 때
리트리버 (Retriever)가 "신규 계정을 위한 신원 확인"에 관한 문서를 찾아내고
LLM (Large Language Model)이 신분증 문서를 업로드하라는 유용한 답변을 생성한다면

이것은 치명적인 실패입니다. 고통을 겪고 있는 누군가가 즉각적인 상담원 연결 대신 관료적인 절차만 안내받게 되는 것입니다. 저에게는 '먼저 에스컬레이션 (Escalation)하고, 그 다음에 생성하는' 시스템이 필요했습니다.

아키텍처: 5단계 안전 우선 파이프라인

1단계: 분류 (Classification)
한 번의 LLM 호출을 통해 구조화된 메타데이터를 추출합니다:

classifier.py

SYSTEM_PROMPT = """
당신은 지원 티켓 분류기입니다...
오직 JSON 객체만을 반환하세요: { "company" : "<HackerRank | Claude | Visa | Unknown>", "request_type" : "<product_issue | feature_request | bug | invalid>", "product_area" : "<short phrase>", "risk_level" : "<low | high>" }
"""

def classify(llm, ticket_company, issue_text):
result = llm.chat_json(SYSTEM_PROMPT, user_msg) # 정제 및 폴백 (Fallback)
return {
"company": result.get("company", "Unknown"),
"request_type": result.

get("request_type", "product_issue"), "product_area": result.get("product_area", "general").lower(), "risk_level": result.get("risk_level", "low").lower() }

핵심 통찰 (Key insight): 저는 로깅(logging)을 위해 risk_level을 유지하지만, 에스컬레이션 (escalation) 용도로는 사용하지 않습니다. LLM은 "계정을 어떻게 삭제하나요?"와 같은 무해한 티켓을 고위험(high risk)으로 과도하게 분류(over-flag)합니다. 결정론적 규칙 (Deterministic rules)이 더 정밀합니다.

2단계: 안전 게이트 (Safety Gate, LLM 호출 없음)
이것이 시스템의 핵심입니다. 검색 (retrieval)이나 생성 (generation)이 이루어지기 전에, 결정론적 규칙이 위험 요소를 확인합니다:

safety.py

def check(classification, issue_text):
text_lower = issue_text.lower()

# 1. 버그 보고 (Bug reports) → classification.get("request_type")이 "bug"인 경우 항상 엔지니어에게 에스컬레이션
if classification.get("request_type") == "bug":
    return True, "Bug report escalated to technical team"

# 2. 민감한 제품 영역 (Sensitive product areas)
product_area = classification.get("product_area", "").lower()
for sensitive in HIGH_RISK_PRODUCT_AREAS:
    if sensitive in product_area:
        return True, f"Product area '{product_area}' is sensitive"

# 3. 키워드 스캔 (Keyword scan)
for kw in ESCALATION_KEYWORDS:
    if kw in text_lower:
        return True, f"Contains sensitive keyword '{kw}'"

# 4.

평가 무결성 (Assessment integrity) (HackerRank 전용) integrity_phrases = [ " 점수 올려줘 ", " 내 점수 바꿔줘 ", " 불공정하게 채점했어 ", " 내 답변 검토해줘 ", " 다음 라운드로 넘겨줘 " ]
for phrase in integrity_phrases :
if phrase in text_lower :
return True , f " 평가 무결성 분쟁: ' { phrase } ' "
return False , ""

내 키워드 리스트:
ESCALATION_KEYWORDS = [
# 사기 / 금융 (fraud / financial)
" 사기 ",
" 승인되지 않은 청구 ",
" 차지백 (chargeback) ",
" 스캠 (scam) ",
" 명의 도용 ",
" 환불 ",
" 환불 요청 ",
" 결제 분쟁 ",
" 결제 분쟁 ",
# 계정 보안 (account security)
" 계정 해킹 ",
" 계정 침해 ",
" 다른 사람이 로그인함 ",
" 계정 정지 ",
" 계정 차단 ",
" 계정 해지 ",
# 법적 문제 (legal)
" 소송 ",
" 법적 조치 ",
" 변호사 (attorney) ",
" 변호사 (lawyer) ",
" 법원 ",
# 평가 무결성 (assessment integrity)
" 부정행위 (cheating) ",
" 표절 (plagiarism) ",
" 학업 무결성 (academic integrity) ",
" 감독 분쟁 (proctoring dispute) ",
" 지원자 부정행위 ",
" 불공정한 자격 박탈 ",
# 기타 고위험 (other high-risk)
" 보안 침해 (security breach) ",
" 취약점 (vulnerability) ",
" 보안 취약점 (security vulnerability) ",
" 판매자 차단 ",
" 이 판매자 차단 ",
" 비자 환불 처리 ",
" 구독 ",
]

3단계: 검색 (Retrieval) (무료, 로컬, 빠름)

retriever.py

class Retriever :
def init ( self ):
self . index = faiss . read_index ( str ( INDEX_FILE ))
self . model = SentenceTransformer ( EMBEDDING_MODEL )
with open ( CHUNKS_FILE ) as f :
self . chunks = json . load ( f )

def retrieve ( self , query , company = None , top_k = 6 ):
    # 쿼리 임베딩 (L2-정규화 → 내적(dot product) == 코사인 유사도)
    q_vec = self . model . encode ([ query ], normalize_embeddings = True )

    # 회사 필터링 (Company filtering)
    if company :
        filtered = [( i , c ) for i , c in enumerate ( self . chunks ) if c [ " company " ]. lower () == company . lower ()]
        if len ( filtered ) < top_k :
            filtered = list ( enumerate ( self . chunks ))
        # 폴백 (fallback)
    else :
        filtered = list ( enumerate ( self . chunks ))

    # 하위 집합에 대한 임시 인덱스 구축
    idxs = np . array ([ i for i , _ in filtered ])
    subset = np .

stack([self.index.reconstruct(int(i)) for i in idxs])
sub_index = faiss.IndexFlatIP(subset.shape[1])
sub_index.add(subset)
scores, positions = sub_index.search(q_vec, min(top_k, len(filtered)))
return [{"text": self.chunks[int(idxs[p])]["text"], "score": float(scores[0][i]), ...} for i, p in enumerate(positions[0]) if p >= 0]

왜 FAISS인가: 서버가 필요 없고, API 비용이 들지 않으며, 2초 미만으로 로드됩니다. 1,773개의 벡터는 매우 작기 때문에 근사 검색 (Approximate Search)이 필요하지 않습니다.

4단계: 빠른 차선 vs 신중한 차선 (Fast vs Careful Lane)

responder.py

def respond(llm, retriever, classification, issue_text):
chunks = retriever.retrieve(issue_text, company=classification.get("company"))
best = retriever.best_score(chunks)

# 빠른 차선 (FAST LANE): 높은 신뢰도 → 단일 LLM 호출
if best >= FAST_LANE_THRESHOLD: # 0.50
    response = generate(llm, issue_text, chunks)
    return {"status": "replied", "lane": "fast", ...}

# 신중한 차선 (CAREFUL LANE): 낮은 신뢰도 → 모든 사항 검증
if best >= SIMILARITY_THRESHOLD: # 0.35
    relevant = check_relevance(llm, issue_text, chunks)
    if not relevant:
        # 쿼리 재작성 (Query rewrite) + 재시도
        rewritten = rewrite_query(llm, issue_text)
        new_chunks = retriever.retrieve(rewritten, ...)
        new_best = retriever.best_score(new_chunks)
        if new_best >= SIMILARITY_THRESHOLD:
            chunks = new_chunks
            best = new_best
        else:
            return {"status": "escalated", "lane": "escalated_no_corpus", ...}

    response = generate(llm, issue_text, chunks)

    # 자체 점검 (SELF-CHECK): 모든 주장이 근거가 있는지 검증
    grounded, issue = self_check(llm, issue_text, response, chunks)
    if not grounded:
        return {"status": "escalated", "lane": "escalated_ungrounded", ...}

    return {"status": "replied", "lane": "careful", ...}

자체 점검 프롬프트 (Self-Check Prompt):
_SELFCHECK_SYSTEM = """당신은 사실 확인 (Fact-checking) 어시스턴트입니다. 고객 지원 응답이 제공된 문맥 (Context)에 완전히 근거하고 있는지 검토하십시오."

오직 다음의 JSON 객체만을 반환하십시오: {"grounded": true/false, "issue": "<근거 없는 주장을 설명하거나, 근거가 확실한 경우 빈 문자열>"} 응답의 모든 사실적 주장이 문맥 (Context)에서 직접적으로 추적될 수 있다면 해당 응답은 근거가 확실한 (Grounded) 것입니다.

5단계: 출력 스키마 (Output Schema)

main.py

OUTPUT_COLS = [
" status ",
" product_area ",
" response ",
" justification ",
" request_type "
]

def _build_row (
*,
status,
product_area,
response,
justification,
request_type
):
return {
" status ": status.lower(),
" product_area ": product_area.lower(),
" response ": response.strip(),
" justification ": justification.strip(),
" request_type ": request_type.lower(),
}

직면한 과제 (Challenges faced)

가장 큰 과제는 Groq의 속도 제한 (Rate limits)에 걸리는 것이었습니다. Groq의 무료 티어는 약 2030 RPM (분당 요청 수)을 허용하는데, 29개의 티켓에 대해 티켓당 24회의 LLM 호출을 수행하면서 예상보다 빠르게 일일 할당량을 모두 소진했습니다. 완전히 바닥나기 전까지 서로 다른 계정의 3개 API 키를 교체하며 사용했습니다.

근본 원인은 티켓 사이에 설정했던 초기 0.5초의 대기 시간 (Sleep delay) 때문이었습니다. 무료 티어에 사용하기에는 너무 공격적이었습니다. 각 티켓은 분류 (Classify), 관련성 확인 (Relevance check), 쿼리 재작성 (Query rewrite), 생성 (Generate)을 위해 최대 4회의 LLM 호출을 수행했습니다. 이는 분당 잠재적으로 120회의 호출이 발생할 수 있음을 의미하며, 제한 수치의 6배에 달합니다.

해결책은 두 가지 단계로 이루어졌습니다. 첫째, 티켓 사이의 대기 시간을 3초로 늘려 유효 호출 속도를 제한 범위 내로 충분히 낮추었습니다. 둘째, Gemini 2.5 Flash를 자동 폴백 (Fallback) 수단으로 추가했습니다. Groq이 3회 연속 실패하면 서킷 브레이커 (Circuit breaker)가 이를 사용 불가능 상태로 표시하고, 이후의 모든 호출은 투명하게 Gemini로 라우팅됩니다. 수동 개입이 필요 없으며, 실행이 중단되지도 않습니다.

교훈: 배치 작업 (Batch job)을 실행하기 전에 항상 실제 API 호출 속도를 계산하십시오.
티켓 수 × 티켓당 호출 수 ÷ 대기 시간 = 분당 호출 수
세 번째 429 오류 (Too Many Requests)를 마주한 후가 아니라, 시작하기 전에 이 계산을 수행하십시오.

결과 (Results)

에이전트는 HackerRank, Claude, Visa에 걸친 29개의 티켓을 모두 처리했습니다.

예상 출력값이 포함된 샘플 세트에 대한 결과:

  • 상태 분류(답변 완료 vs 에스컬레이션) 정확도 9/10
  • 위험한 미탐(False Negatives) 0건 — 사기(fraud), 결제(billing), 또는 보안(security) 티켓 중 잘못 답변된 사례가 단 하나도 없었음
  • 12,885명의 등록 참가자 중 글로벌 상위 200위
  • AI 판사에게 에이전트를 방어한 1,349명 중 상위 154위

단 하나의 실수: 코퍼스 유사도 점수(corpus similarity score)가 경계선(0.38)에 걸쳐 있었던 테스트 만료 티켓 건. 해당 사례를 해결하기 위해 임계값(threshold)을 낮추지 않기로 결정했습니다. 그렇게 할 경우, 코퍼스 근거가 실제로 빈약한 티켓에 대해 잘못된 답변을 보낼 위험이 있기 때문입니다. 위험한 답변을 하기보다는 안전하게 에스컬레이션하는 것이 언제나 옳습니다.

다르게 시도해 볼 점들

  1. 미세 조정된 경량 분류기 (Fine-tuned lightweight classifier)
    LLM 분류기 호출을 미세 조정된 BERT 기반 모델(예: 고객 지원 티켓 데이터로 미세 조정된 distilbert-base-uncased)로 교체합니다. 이를 통해 티켓당 LLM 호출을 한 번 줄일 수 있고, 지연 시간(latency)을 약 2초 단축하며

Ollama를 통한 로컬 LLM (Local LLM): Ollama를 통해 양자화된 (quantized) Llama 모델을 로컬에서 실행하면 API 속도 제한 (rate limits)을 완전히 제거할 수 있습니다. 저도 고려해 보았으나, 제 하드웨어로는 70B 모델을 수용 가능한 속도로 실행할 수 없었습니다. GPU를 사용하거나 더 작은 양자화 모델(예: Llama 3.2 3B)을 사용한다면, 이것이 가장 깔끔한 의존성 없는 (zero-dependency) 설정이 될 것입니다.

가장 중요했던 단 한 가지: 이 프로젝트 전체에서 가장 중요한 설계 결정은 검색 전략 (retrieval strategy), 이중 LLM 설정, 또는 쿼리 재작성 (query rewriting)이 아니었습니다. 그것은 바로 이것이었습니다: 안전 관련 결정은 확률적인 (probabilistic) LLM이 아니라, 결정론적인 (deterministic) Python으로 이루어져야 한다는 점입니다. "이 티켓이 민감한가요?"라는 질문을 받은 LLM은 도움이 되려고 노력할 것입니다. 만약 문구가 정중하다면 사기(fraud) 사례에 대해서도 '아니오'라고 답할 수도 있습니다. 반면 "승인되지 않은 결제"를 확인하는 Python 규칙은 변동성 없이 매번 반드시 '예'라고 답합니다. 고객 지원(support) 맥락에서 이러한 일관성은 있으면 좋은 기능(nice-to-have) 수준이 아닙니다. 그것은 감사(audit) 가능한 시스템과, 그저 제대로 작동하기만을 바랄 수밖에 없는 시스템 사이의 차이입니다. 잘못된 답변이 실제적인 결과를 초래하는 RAG를 구축하고 있다면, 검색 알고리즘이 아닌 안전 게이트 (safety gate)부터 시작하십시오.

코드: GitHub에서 전체 구현 내용을 확인하세요: https://github.com/Tahrim19/hackerrank-orchestrate-may26 모든 에이전트 코드는 code/ 디렉토리에 있습니다. 실행하기 전에 python code/build_index.py를 사용하여 로컬에서 인덱스를 구축하세요.

AI 자동 생성 콘텐츠

본 콘텐츠는 Dev.to AI tag의 원문을 AI가 자동으로 요약·번역·분석한 것입니다. 원 저작권은 원저작자에게 있으며, 정확한 내용은 반드시 원문을 확인해 주세요.

원문 바로가기
0

댓글

0