포스트

2026-05-12 강의 정리

Adaptive RAG 구현 1

2026-05-12 강의 정리

목차

  1. 문제와 접근
  2. LangGraph 기본 패턴 복습
  3. 쿼리 라우팅 문제 정의
  4. QueryAnalysis — Pydantic 스키마
  5. analyze_query — Structured Output 라우터
  6. DetailedQueryAnalysis — 상세 분석 스키마
  7. RoutingConfig / RoutingLog
  8. WeightedRoutingEngine — 가중치 기반 라우팅
  9. RoutingEngine — confidence 기반 fallback
  10. RouteHandler 전략 패턴
  11. AdaptiveRAGState와 처리 노드
  12. LangGraph 조건 분기 라우팅
  13. fallback 노드 추가
  14. 핵심 개념 총정리
  15. 자주 나오는 실수/주의사항
  16. 라우팅 정확도와 실패 비용 함께 보기

1. 문제와 접근

질문을 여러 처리 경로로 나누는 라우터는 분류 결과뿐 아니라 잘못 분류했을 때의 대응까지 포함해야 한다. 스키마, 가중치, confidence, fallback을 연결하면서 모델의 판단과 애플리케이션의 실행 정책을 나누어 본다. 최종 경로를 선택한 이유가 기록에 남는 구조를 기준으로 살펴본다.

질문이 일반 지식이면 LLM이 직접 답하고,
내부 문서가 필요하면 RAG로 보내고,
최신 정보가 필요하면 웹 검색 경로로 보내는 구조다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import os
from typing import TypedDict, Literal
from dataclasses import dataclass, field
from datetime import datetime
from abc import ABC, abstractmethod
from dotenv import load_dotenv

load_dotenv()

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel, Field

llm = ChatOpenAI(model="gpt-4o-mini")

2. LangGraph 기본 패턴 복습

단일 노드 그래프

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class SimpleState(TypedDict):
    message: str


def greet(state: SimpleState) -> dict:
    return {"message": f"안녕, {state['message']}"}


builder = StateGraph(SimpleState)
builder.add_node("greet", greet)
builder.add_edge(START, "greet")
builder.add_edge("greet", END)

app = builder.compile()

app.invoke({"message": "abc"})
# → {'message': '안녕, abc'}

여러 노드 직렬 연결

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
class TextState(TypedDict):
    text: str
    upper: str
    length: int


def to_upper(state: TextState) -> dict:
    return {"upper": state["text"].upper()}


def measure(state: TextState) -> dict:
    return {"length": len(state["text"])}


builder = StateGraph(TextState)
builder.add_node("to_upper", to_upper)
builder.add_node("measure", measure)
builder.add_edge(START, "to_upper")
builder.add_edge("to_upper", "measure")
builder.add_edge("measure", END)

app = builder.compile()

app.invoke({"text": "hello langgraph", "upper": "", "length": 0})
# → {'text': 'hello langgraph', 'upper': 'HELLO LANGGRAPH', 'length': 15}

노드는 현재 state를 받고, 업데이트할 값만 dict로 반환한다.
LangGraph는 반환된 dict를 기존 state에 반영한 뒤 다음 노드로 넘긴다.


3. 쿼리 라우팅 문제 정의

사용자 질문을 하나의 경로로 분류한다.

1
2
3
4
5
6
7
8
query
  ↓
query analyzer
  ↓
route 선택
  ├─ direct     : LLM 자체 지식으로 답변
  ├─ rag        : 내부 문서 / 특정 도메인 문서 검색
  └─ web_search : 최신 정보 / 실시간 정보 검색

라우팅 기준

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
direct:
  - 일반 상식
  - 개념 설명
  - 간단한 계산
  - LLM이 자체 지식으로 답변 가능한 질문

rag:
  - 회사 내부 정책
  - 매뉴얼
  - 사내 데이터
  - 특정 문서 기반 답변이 필요한 질문

web_search:
  - 오늘 날씨
  - 최신 뉴스
  - 시세
  - 현재 시점의 정보가 필요한 질문

모델 기반 라우터의 장점

1
2
3
4
5
6
7
규칙 기반 라우터:
  키워드가 맞으면 특정 route로 보냄
  → 단순하지만 애매한 질문에 약함

LLM 기반 라우터:
  질문의 의도와 맥락을 보고 route를 고름
  → 유연하지만 confidence와 fallback 관리가 필요함

4. QueryAnalysis — Pydantic 스키마

LLM의 출력 형식을 Pydantic 모델로 고정한다.

1
2
3
4
5
6
7
8
9
10
11
class QueryAnalysis(BaseModel):
    query: str = Field(description="원본 쿼리")
    route: Literal["direct", "rag", "web_search"] = Field(
        description="라우팅 경로: direct(직접 답변), rag(문서 검색), web_search(웹 검색)"
    )
    confidence: float = Field(
        description="분류 신뢰도(0.0~1.0)",
        ge=0.0,
        le=1.0,
    )
    reasoning: str = Field(description="분류 이유")

필드 의미

필드역할예시
query원본 질문"우리 회사 연차 규정은?"
route선택된 처리 경로"rag"
confidence분류 신뢰도0.92
reasoningroute 선택 이유"내부 정책 문서가 필요함"

Literal을 쓰면 route가 정해진 값 중 하나로 제한된다.
Field(ge=0.0, le=1.0)로 confidence 범위도 검증할 수 있다.


5. analyze_query — Structured Output 라우터

with_structured_output()을 사용하면 LLM 응답을 Pydantic 객체로 받을 수 있다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
def analyze_query(query: str) -> QueryAnalysis:
    """LLM을 사용하여 쿼리를 분석하고 라우팅 경로를 결정"""
    system_prompt = """당신은 쿼리를 잘 분석하고 라우팅 경로를 분류하는 전문가입니다.
주어진 쿼리를 분석해서 최적의 처리 경로를 분석하세요.

라우팅 경로:
direct : LLM이 자체 지식으로 답변 가능한 일반 질문(상식, 개념 설명)
rag : 회사 내부 문서나 특정 도메인 지식이 필요한 질문(정책, 매뉴얼, 사내 데이터)
web_search : 최신 정보나 실시간 데이터가 필요한 질문(뉴스, 시세, 날씨)

예시:
- 파이썬의 데코레이터란? -> direct
- 우리 회사 연차 규정은? -> rag
- 오늘 날씨는? -> web_search
"""

    response = llm.with_structured_output(QueryAnalysis).invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=query),
    ])

    return response

실행 예시

1
2
3
4
5
6
7
8
9
result = analyze_query("1+1은?")
result

# → QueryAnalysis(
#      query='1+1은?',
#      route='direct',
#      confidence=1.0,
#      reasoning='기본적인 산수 문제로 LLM이 자체 지식으로 답할 수 있습니다.'
#   )

문자열 파싱으로 route를 꺼내는 것보다 안정적이다.
스키마가 있으므로 LLM 출력이 코드에서 바로 다루기 쉬운 객체가 된다.


6. DetailedQueryAnalysis — 상세 분석 스키마

라우팅 결과에 세부 주제와 언어까지 추가할 수 있다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class DetailedQueryAnalysis(BaseModel):
    """상세 쿼리 분석 결과"""

    query: str = Field(description="원본 쿼리")
    route: Literal["direct", "rag", "web_search"] = Field(
        description="라우팅 경로: direct(직접 답변), rag(문서 검색), web_search(웹 검색)"
    )
    confidence: float = Field(
        description="분류 신뢰도(0.0~1.0)",
        ge=0.0,
        le=1.0,
    )
    reasoning: str = Field(description="분류 이유")
    sub_topic: str = Field(description="쿼리의 세부 주제")
    language: Literal["ko", "en", "other"] = Field(description="쿼리 언어")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def analyze_query_detailed(query: str) -> DetailedQueryAnalysis:
    """LLM을 사용하여 쿼리를 상세 분석"""
    system_prompt = """당신은 쿼리를 잘 분석하고 라우팅 경로를 분류하는 전문가입니다.
주어진 쿼리를 분석해서 라우팅 경로, 세부 주제, 언어를 분류하세요.

라우팅 경로:
direct : LLM이 자체 지식으로 답변 가능한 일반 질문
rag : 회사 내부 문서나 특정 도메인 지식이 필요한 질문
web_search : 최신 정보나 실시간 데이터가 필요한 질문

언어:
ko(한국어), en(영어), other(기타)
"""

    response = llm.with_structured_output(DetailedQueryAnalysis).invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=query),
    ])

    return response

노트북에서는 confidecnce 오타와 QueryAnalysis 스키마 재사용이 있었으므로,
정리본에서는 confidence와 DetailedQueryAnalysis를 사용하는 형태로 보정했다.


7. RoutingConfig / RoutingLog

분류 결과를 그대로 쓰지 않고, confidence 기준과 fallback 정책을 별도로 둔다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@dataclass
class RoutingConfig:
    """라우팅 전략 설정"""

    confidence_threshold: float = 0.7
    fallback_route: str = "rag"
    enable_logging: bool = True
    max_retries: int = 2


@dataclass
class RoutingLog:
    """라우팅 로그"""

    timestamp: str
    query: str
    predicted_route: str
    actual_route: str
    confidence: float
    fallback_applied: bool

왜 config가 필요한가?

1
2
3
4
5
6
7
8
9
LLM 예측 결과:
  route='web_search'
  confidence=0.42

운영 정책:
  confidence < 0.7이면 rag로 보낸다

최종 경로:
  actual_route='rag'

모델의 판단과 서비스 정책을 분리하면, 라우팅 기준을 코드 전체에서 흩어지게 만들지 않아도 된다.


8. WeightedRoutingEngine — 가중치 기반 라우팅

route별로 다른 신뢰도 가중치를 줄 수 있다.

1
2
3
4
5
6
7
8
9
@dataclass
class AdvancedRoutingConfig:
    confidence_threshold: float = 0.5
    fallback_route: str = "rag"
    route_weights: dict = field(default_factory=lambda: {
        "direct": 1.0,
        "rag": 0.8,
        "web_search": 0.6,
    })
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class WeightedRoutingEngine:
    def __init__(self, config: AdvancedRoutingConfig | None = None):
        self.config = config or AdvancedRoutingConfig()

    def weighted_route(self, analysis: QueryAnalysis) -> dict:
        weight = self.config.route_weights.get(analysis.route, 1.0)
        weighted_score = analysis.confidence * weight

        if weighted_score < self.config.confidence_threshold:
            final_route = self.config.fallback_route
            fallback = True
        else:
            final_route = analysis.route
            fallback = False

        return {
            "query": analysis.query[:25],
            "predicted": analysis.route,
            "confidence": analysis.confidence,
            "weight": weight,
            "weighted_score": round(weighted_score, 3),
            "final_route": final_route,
            "fallback": fallback,
        }

실행 예시

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
weighted_engine = WeightedRoutingEngine()
test_q = "사내 보안 교육이 언제인가요?"

analysis = analyze_query(test_q)
result = weighted_engine.weighted_route(analysis)

print(result)
# → {
#      'query': '사내 보안 교육이 언제인가요?',
#      'predicted': 'rag',
#      'confidence': 0.9,
#      'weight': 0.8,
#      'weighted_score': 0.72,
#      'final_route': 'rag',
#      'fallback': False
#   }

web_search는 최신 정보 의존성이 높고 비용이 클 수 있으므로 더 보수적인 weight를 줄 수 있다.


9. RoutingEngine — confidence 기반 fallback

모델이 반환한 confidence를 실제 정답 확률로 바로 해석하지 않는다. threshold는 라우팅 평가셋에서 경로별 오분류와 fallback 비율을 보고 정하고, 낮은 점수뿐 아니라 높은 점수로 잘못 분류한 사례도 확인한다.

기본 라우팅 엔진은 confidence가 threshold보다 낮으면 fallback route로 보낸다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class RoutingEngine:
    def __init__(self, config: RoutingConfig | None = None):
        self.config = config or RoutingConfig()
        self.logs: list[RoutingLog] = []

    def route(self, analysis: QueryAnalysis) -> str:
        """분석 결과를 바탕으로 최종 라우팅 경로를 결정"""
        predicted = analysis.route
        fallback_applied = False

        if analysis.confidence < self.config.confidence_threshold:
            actual = self.config.fallback_route
            fallback_applied = True
        else:
            actual = predicted

        if self.config.enable_logging:
            log = RoutingLog(
                timestamp=datetime.now().isoformat(),
                query=analysis.query,
                predicted_route=predicted,
                actual_route=actual,
                confidence=analysis.confidence,
                fallback_applied=fallback_applied,
            )
            self.logs.append(log)

        return actual

    def get_stats(self) -> dict:
        if not self.logs:
            return {"total": 0}

        total = len(self.logs)
        fallbacks = sum(1 for log in self.logs if log.fallback_applied)
        route_dist = {}

        for log in self.logs:
            route_dist[log.actual_route] = route_dist.get(log.actual_route, 0) + 1

        avg_conf = sum(log.confidence for log in self.logs) / total

        return {
            "total": total,
            "fallback_rate": fallbacks / total,
            "avg_confidence": avg_conf,
            "route_distribution": route_dist,
        }

실행 예시

1
2
3
4
5
6
7
8
9
10
engine = RoutingEngine(RoutingConfig(confidence_threshold=0.7))

analysis = analyze_query("파이썬 데코레이터가 뭔가요?")
final_route = engine.route(analysis)

final_route
# → 'direct'

engine.logs
# → [RoutingLog(... predicted_route='direct', actual_route='direct', confidence=0.95, fallback_applied=False)]

10. RouteHandler 전략 패턴

라우팅과 실행을 분리하면 처리기를 교체할 때 분류 규칙을 함께 바꾸지 않아도 된다. 각 handler는 공통 반환 형태를 갖게 하고, 실행 실패를 분류 실패와 구분해 상위 단계에 전달하도록 설계한다.

route가 결정되면 각 route를 처리하는 handler가 필요하다.

1
2
3
4
class RouteHandler(ABC):
    @abstractmethod
    def handle(self, query: str) -> str:
        pass
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class DirectHandler(RouteHandler):
    """LLM이 직접 답변하는 경우"""

    def handle(self, query: str) -> str:
        response = llm.invoke([HumanMessage(content=query)])
        return response.content


class RAGHandler(RouteHandler):
    """RAG 검색해서 답변하는 경우"""

    def handle(self, query: str) -> str:
        return f"RAG '{query}'에 대한 문서 검색 결과입니다"


class WebSearchHandler(RouteHandler):
    """웹 검색해서 답변하는 경우"""

    def handle(self, query: str) -> str:
        return f"WebSearch '{query}'에 대한 문서 검색 결과입니다"
1
2
3
4
5
6
7
8
9
10
11
12
13
handlers = {
    "direct": DirectHandler(),
    "rag": RAGHandler(),
    "web_search": WebSearchHandler(),
}

test_q = "우리회사 복지정책에 대해서 알려줘"
analysis = analyze_query(test_q)
route = engine.route(analysis)
result = handlers[route].handle(test_q)

print(result)
# → RAG '우리회사 복지정책에 대해서 알려줘'에 대한 문서 검색 결과입니다

라우팅과 실제 처리를 분리하면, 나중에 RAG 검색기나 웹 검색 API를 바꿔도 라우터 코드는 크게 바뀌지 않는다.


11. AdaptiveRAGState와 처리 노드

LangGraph에서 사용할 전체 state를 정의한다.

1
2
3
4
5
6
7
8
class AdaptiveRAGState(TypedDict):
    query: str
    route: str
    confidence: float
    reasoning: str
    response: str
    documents: list
    search_results: list

쿼리 분석 노드

1
2
3
4
5
6
7
8
9
10
def query_analyzer_node(state: AdaptiveRAGState) -> dict:
    """쿼리 분석 노드"""
    query = state["query"]
    analysis = analyze_query(query)

    return {
        "route": analysis.route,
        "confidence": analysis.confidence,
        "reasoning": analysis.reasoning,
    }

direct / rag / web_search 노드

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def direct_answer_node(state: AdaptiveRAGState) -> dict:
    """LLM 직접 답변 노드"""
    query = state["query"]
    response = llm.invoke([HumanMessage(content=query)])
    return {"response": f"[Direct] {response.content}"}


def rag_search_node(state: AdaptiveRAGState) -> dict:
    """RAG 검색 노드"""
    query = state["query"]
    return {
        "documents": [f"[Document] {query}에 대한 검색 문서"],
        "response": f"[RAG] '{query}'에 대한 문서 기반 답변",
    }


def web_search_node(state: AdaptiveRAGState) -> dict:
    """웹 검색 노드"""
    query = state["query"]
    return {
        "search_results": [f"[Web] {query} 검색 결과"],
        "response": f"[WebSearch] '{query}'에 대한 웹 검색 답변",
    }

12. LangGraph 조건 분기 라우팅

query_analyzer 노드에서 route와 confidence를 만든 뒤, 조건부 엣지로 다음 노드를 고른다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def route_query(state: AdaptiveRAGState) -> str:
    route = state["route"]
    confidence = state["confidence"]

    if confidence < 0.7:
        return "rag_search"

    if route == "direct":
        return "direct_answer"
    elif route == "rag":
        return "rag_search"
    elif route == "web_search":
        return "web_search"
    else:
        return "rag_search"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
workflow = StateGraph(AdaptiveRAGState)
workflow.add_node("query_analyzer", query_analyzer_node)
workflow.add_node("direct_answer", direct_answer_node)
workflow.add_node("rag_search", rag_search_node)
workflow.add_node("web_search", web_search_node)

workflow.add_edge(START, "query_analyzer")
workflow.add_conditional_edges(
    "query_analyzer",
    route_query,
    {
        "direct_answer": "direct_answer",
        "rag_search": "rag_search",
        "web_search": "web_search",
    },
)
workflow.add_edge("direct_answer", END)
workflow.add_edge("rag_search", END)
workflow.add_edge("web_search", END)

app = workflow.compile()

실행 예시

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
test_q = "우리 회사 재택근무 정책은?"

result = app.invoke({
    "query": test_q,
    "route": "",
    "confidence": 0.0,
    "reasoning": "",
    "response": "",
    "documents": [],
    "search_results": [],
})

result
# → {
#      'query': '우리 회사 재택근무 정책은?',
#      'route': 'rag',
#      'confidence': 0.9,
#      'reasoning': '회사 내부 재택근무 정책 정보가 필요함',
#      'response': "[RAG] '우리 회사 재택근무 정책은?'에 대한 문서 기반 답변",
#      'documents': ['[Document] 우리 회사 재택근무 정책은?에 대한 검색 문서'],
#      'search_results': []
#   }

전체 흐름

1
2
3
4
5
6
7
8
9
START
  ↓
query_analyzer
  ↓ route_query
  ├─ direct_answer
  ├─ rag_search
  └─ web_search
  ↓
END

13. fallback 노드 추가

질문이 너무 모호하거나 confidence가 낮으면 바로 답하지 않고 재질문을 유도한다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def fallback_node(state: AdaptiveRAGState) -> dict:
    """불확실한 쿼리가 입력되면 재질문 요청"""
    query = state["query"]
    response = llm.invoke([
        {
            "role": "user",
            "content": f"'{query}'라는 질문이 모호합니다. 더 구체적인 질문 3가지를 제안해주세요",
        }
    ])
    suggestions = response.content

    return {
        "response": f"[Fallback] 질문을 더 구체적으로 해주세요.\n\n추천 질문:\n{suggestions}"
    }

route_query_v2

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def route_query_v2(state: AdaptiveRAGState) -> str:
    route = state["route"]
    confidence = state["confidence"]

    if confidence < 0.5:
        return "fallback"

    if confidence < 0.7:
        return "rag_search"

    if route == "direct":
        return "direct_answer"
    elif route == "rag":
        return "rag_search"
    elif route == "web_search":
        return "web_search"
    else:
        return "rag_search"

workflow_v2

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
workflow_v2 = StateGraph(AdaptiveRAGState)
workflow_v2.add_node("query_analyzer", query_analyzer_node)
workflow_v2.add_node("direct_answer", direct_answer_node)
workflow_v2.add_node("rag_search", rag_search_node)
workflow_v2.add_node("web_search", web_search_node)
workflow_v2.add_node("fallback", fallback_node)

workflow_v2.add_edge(START, "query_analyzer")
workflow_v2.add_conditional_edges(
    "query_analyzer",
    route_query_v2,
    {
        "direct_answer": "direct_answer",
        "rag_search": "rag_search",
        "web_search": "web_search",
        "fallback": "fallback",
    },
)
workflow_v2.add_edge("direct_answer", END)
workflow_v2.add_edge("rag_search", END)
workflow_v2.add_edge("web_search", END)
workflow_v2.add_edge("fallback", END)

app = workflow_v2.compile()

실행 예시

1
2
3
4
5
6
7
8
9
10
11
12
13
14
test_q = "그거 뭐야?"

result = app.invoke({
    "query": test_q,
    "route": "",
    "confidence": 0.0,
    "reasoning": "",
    "response": "",
    "documents": [],
    "search_results": [],
})

result["response"]
# → route와 confidence에 따라 direct 답변 또는 fallback 재질문 유도

모호한 질문이라고 항상 fallback으로 가는 것은 아니다.
LLM이 confidence >= 0.7로 판단하면 direct 경로를 탈 수 있으므로, threshold와 프롬프트를 함께 조정해야 한다.


14. 핵심 개념 총정리

Adaptive RAG 라우팅 구성요소

구성요소역할
QueryAnalysisLLM 라우팅 결과를 담는 구조화 스키마
analyze_query질문을 route / confidence / reasoning으로 분석
RoutingConfigthreshold, fallback route 같은 운영 정책
RoutingLog라우팅 판단 기록
RoutingEngineconfidence 기준으로 최종 route 결정
RouteHandlerroute별 실제 처리 로직
AdaptiveRAGStateLangGraph 전체 상태
route_query조건부 엣지에서 다음 노드 선택
fallback_node모호한 질문에 재질문 유도

라우팅 의사결정 순서

1
2
3
4
5
6
1. 사용자 질문 입력
2. LLM이 QueryAnalysis 생성
3. confidence 확인
4. threshold보다 낮으면 fallback 또는 rag
5. route에 맞는 노드 실행
6. response / documents / search_results 업데이트

route 선택 기준

route사용할 때예시
direct일반 지식, 개념, 계산"파이썬 데코레이터가 뭐야?"
rag내부 문서나 도메인 문서 필요"우리 회사 재택근무 정책은?"
web_search최신/실시간 정보 필요"오늘 환율은?"
fallback질문이 모호하거나 confidence가 낮음"그거 어떻게 해?"

LangGraph 구조

1
2
3
4
5
6
7
8
9
10
11
START
  ↓
query_analyzer
  ↓
route_query_v2
  ├─ direct_answer
  ├─ rag_search
  ├─ web_search
  └─ fallback
  ↓
END

15. 자주 나오는 실수/주의사항

실수 1: 상세 스키마를 만들고 기존 스키마로 invoke

1
2
3
4
5
# 틀린 코드
response = llm.with_structured_output(QueryAnalysis).invoke(...)

# 올바른 코드
response = llm.with_structured_output(DetailedQueryAnalysis).invoke(...)

어떤 Pydantic 모델로 구조화할지와 함수 반환 타입을 맞춰야 한다.

실수 2: confidence 필드명 오타

1
2
3
4
5
# 틀린 코드
confidecnce: float

# 올바른 코드
confidence: float

필드명이 틀리면 이후 analysis.confidence 접근에서 문제가 생긴다.

실수 3: config 클래스명 오타

1
2
3
4
5
6
7
8
9
# 틀린 코드
class RoutingEngine:
    def __init__(self, config: RountingConfig = None):
        ...

# 올바른 코드
class RoutingEngine:
    def __init__(self, config: RoutingConfig | None = None):
        ...

실수 4: get_stats에서 fallback 변수명 혼동

1
2
3
4
5
6
7
# 틀린 코드
fabllbacks = sum(1 for log in self.logs if log.fallback_applied)
return {"fallback_rate": fallback / total}

# 올바른 코드
fallbacks = sum(1 for log in self.logs if log.fallback_applied)
return {"fallback_rate": fallbacks / total}

실수 5: 조건부 엣지 함수 반환 타입

1
2
3
# 실제로는 dict가 아니라 다음 노드 이름(str)을 반환해야 한다
def route_query(state: AdaptiveRAGState) -> str:
    return "rag_search"

add_conditional_edges()의 두 번째 함수는 mapping의 key와 일치하는 값을 반환해야 한다.

실수 6: fallback 노드를 등록해놓고 다른 함수 연결

1
2
3
4
5
# 틀린 코드
workflow_v2.add_node("fallback", web_search_node)

# 올바른 코드
workflow_v2.add_node("fallback", fallback_node)

실수 7: LLM confidence를 그대로 신뢰

1
2
LLM confidence는 모델이 스스로 추정한 값이다.
반드시 로그를 쌓고, 실제 서비스 데이터로 threshold를 조정해야 한다.

처음에는 confidence < 0.7 → rag, confidence < 0.5 → fallback처럼 보수적으로 시작하고,
로그를 보면서 route별 threshold를 조정하는 방식이 안전하다.


16. 라우팅 정확도와 실패 비용 함께 보기

경로를 틀린 횟수만 합산하면 오분류의 영향 차이를 놓칠 수 있다. 검색이 필요한 질문을 바로 답한 경우와 단순한 질문에 검색을 추가한 경우를 나누어 보고, 어떤 오류를 우선 줄일지 결정한다.

라우팅 로그에서는 모델이 제안한 경로와 실제 실행한 경로를 함께 확인한다. fallback이 자주 발생하는 질문 유형이 드러나면 threshold를 일괄 낮추기보다 분류 기준이나 처리 경로를 다시 검토할 수 있다.

이 기사는 저작권자의 CC BY 4.0 라이센스를 따릅니다.

인기 태그