개요
- 사내 HR FAQ 챗봇의 검색 품질을 개선하기 위해 기존 single-shot RAG 파이프라인을 Agentic RAG 로 전환한 내용 정리
- 기존: 사용자 질문 → embedding → OpenSearch 검색 1회 → LLM 답변 생성
- 변경: 사용자 질문 → Claude 가 검색 전략을 판단 → tool 호출로 검색 → 결과 평가 → 필요 시 재검색 → 답변 생성
- 핵심은 검색을 누가 제어하느냐의 변경.
기존 구조 — Single-shot RAG
사용자 질문 → query embedding 생성 → OpenSearch 검색 (semantic + vector) → 결과를 prompt 에 삽입 → gpt-4o-mini 답변 생성
- 한계
- query rewriting 부재: 사용자 입력("연차 며칠?")이 그대로 검색 쿼리로 사용. HR 문서 제목/본문과의 lexical gap 발생
- 검색 결과 fusion 미흡: semantic search 와 vector search 결과를 단순 concat. score normalization 이나 deduplication 없음
- 재검색 불가: 검색 결과가 부적절해도 그대로 LLM 에 전달. 결과 품질을 평가하는 단계가 없음
- 필터 미사용: category, company 등 structured metadata 를 활용한 필터링 로직이 하드코딩되어 있지 않음
개선 구조 — Agentic RAG
사용자 질문 → Claude API 호출 (system prompt + tools + messages)
→ stop_reason == "tool_use" → OpenSearch 검색 실행 → 결과를 messages 에 추가 → Claude API 재호출
→ stop_reason == "end_turn" → 최종 답변 반환
- Claude 가 검색 쿼리를 직접 구성하고, 결과를 평가하여 재검색 여부를 판단하는 agentic loop 구조
- 주요 변경 사항
- OpenSearch 검색을 Claude tool 로 정의. Claude 가 query, category, company 파라미터를 직접 구성
- RRF (Reciprocal Rank Fusion) 로 semantic + vector 검색 결과를 rank 기반으로 합산
- category/company post-filter 적용. 필터 결과가 없으면 전체 반환 (fail-safe)
- 대화 히스토리를 Claude messages 포맷으로 전달
- LLM: gpt-4o-mini → Claude Sonnet 으로 교체
search_hr_docs — Claude 가 호출하는 검색 tool
TOOLS = [
{
"name": "search_hr_docs",
"description": (
"HR 문서를 검색합니다. "
"query에 검색 키워드를 넣고, 선택적으로 category나 company로 필터링할 수 있습니다."
),
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "검색 키워드. 사용자 질문을 검색에 적합한 형태로 변환하여 입력.",
},
"category": {
"type": "string",
"description": "카테고리 필터 (선택). 예: 급여, 복리후생, 휴가, 평가 등.",
},
"company": {
"type": "string",
"description": "회사 필터 (선택).",
},
"top_k": {
"type": "integer",
"description": "반환할 검색 결과 수. 기본값 5.",
"default": 5,
},
},
"required": ["query"],
},
},
]
- tool 을 이렇게 정의하면 Claude 가 사용자 의도를 분석하여 파라미터를 자동으로 구성
- ex) "메포 육아휴직 기간" →
query="육아휴직 기간", company="SGP" 로 분리하여 호출
Agentic Loop 구현
- Claude API 호출 →
stop_reason 에 따라 분기하는 loop 패턴
async def _answer_query_traced(self, query: str, user_id: str) -> HRAgentResponse:
messages = self.get_recent_history(user_id) + [
{"role": "user", "content": query}
]
for iteration in range(self.max_tool_iterations):
response = await self._call_claude_api(iteration, messages)
if response.stop_reason == "end_turn":
reply = next(
(b.text for b in response.content if hasattr(b, "text")), ""
)
return HRAgentResponse(response=reply, referenced_docs=unique_docs)
if response.stop_reason == "tool_use":
messages.append({"role": "assistant", "content": response.content})
tool_blocks = [b for b in response.content if b.type == "tool_use"]
tool_results = []
for block in tool_blocks:
result_text = await asyncio.get_event_loop().run_in_executor(
None, self._execute_tool, block.name, block.input
)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result_text,
})
messages.append({"role": "user", "content": tool_results})
max_tool_iterations 로 최대 반복 횟수를 제한. 기본값 10- OpenSearch 검색은 sync 함수이므로
run_in_executor 로 async 호환
검색 품질 개선 — RRF
- 기존에는 semantic search 결과와 vector search 결과를 단순 concat. 동일 문서가 중복되거나 score 체계가 다른 두 결과의 순위가 뒤섞이는 문제가 있었음
- RRF (Reciprocal Rank Fusion) 를 적용하여 두 검색 결과를 rank 기반으로 합산
def _rrf_merge(self, results_a, results_b, top_k=5, k=60):
scores = {}
doc_map = {}
for rank, doc in enumerate(results_a):
doc_id = doc["_id"]
scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + rank + 1)
doc_map[doc_id] = doc
for rank, doc in enumerate(results_b):
doc_id = doc["_id"]
scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + rank + 1)
doc_map[doc_id] = doc
sorted_ids = sorted(scores, key=lambda x: scores[x], reverse=True)[:top_k]
return [doc_map[doc_id] for doc_id in sorted_ids]
- RRF 의 핵심: 각 검색 결과에서의 순위(rank) 를 기반으로 점수를 합산. 원래 score 의 scale 차이에 영향을 받지 않음
k=60 은 RRF 논문의 기본 상수. rank 가 낮은 문서에 과도한 가중치가 부여되는 것을 방지
reference
