본 콘텐츠의 이미지 및 내용은 AI로 생성되었습니다.
본 콘텐츠의 이미지 및 내용을 무단으로 복제, 배포, 수정하여 사용할 경우 저작권법에 의해 법적 제재를 받을 수 있습니다.
이미지 로딩 중...
AI Generated
2025. 12. 27. · 3 Views
Router 패턴으로 배우는 LLM 워크플로 설계
LLM 애플리케이션에서 여러 전문가 모델을 효율적으로 연결하고 태스크에 따라 적절한 처리 경로를 선택하는 Router 패턴을 배웁니다. 실무에서 바로 적용할 수 있는 도메인별 라우터와 전문가 선택 시스템을 직접 구현해봅니다.
목차
1. 다중 전문가 라우팅
김개발 씨는 회사에서 AI 챗봇 프로젝트를 맡게 되었습니다. 고객 문의가 법률, 의료, 기술 등 다양한 분야에 걸쳐 들어오는데, 하나의 모델로 모든 질문에 답하려니 정확도가 떨어졌습니다.
"각 분야 전문가에게 질문을 연결해주는 안내 데스크가 필요해!" 바로 이 순간, Router 패턴의 필요성을 깨달았습니다.
다중 전문가 라우팅은 사용자의 질문을 분석하여 가장 적합한 전문가 모델에게 연결해주는 패턴입니다. 마치 대형 병원의 접수처에서 환자의 증상을 듣고 적절한 진료과로 안내해주는 것과 같습니다.
이 패턴을 활용하면 각 분야에 특화된 모델의 강점을 최대한 살릴 수 있습니다.
다음 코드를 살펴봅시다.
from typing import Dict, Callable
class ExpertRouter:
def __init__(self):
# 전문가 모델 등록 저장소
self.experts: Dict[str, Callable] = {}
self.classifier = self._create_classifier()
def register_expert(self, domain: str, handler: Callable):
# 도메인별 전문가 핸들러 등록
self.experts[domain] = handler
def route(self, query: str) -> str:
# 질문 분석 후 적절한 전문가에게 라우팅
domain = self.classifier(query)
expert = self.experts.get(domain, self.experts["general"])
return expert(query)
김개발 씨는 입사 1년 차 주니어 개발자입니다. 최근 회사에서 고객 상담 AI 시스템을 개선하라는 미션을 받았습니다.
기존 시스템은 하나의 범용 모델이 모든 질문에 답변하고 있었는데, 법률 상담에서는 부정확한 답변이, 기술 지원에서는 피상적인 답변이 계속 나왔습니다. "이거 어떻게 해야 할까요?" 김개발 씨가 팀장에게 물었습니다.
박시니어 팀장은 화이트보드에 그림을 그리며 설명했습니다. "큰 병원에 가면 어떻게 하지?
일단 접수처에서 증상을 말하면, 내과인지 외과인지 안내해주잖아. 우리 AI도 그렇게 만들면 돼." 바로 이것이 다중 전문가 라우팅의 핵심 아이디어입니다.
쉽게 비유하자면, 다중 전문가 라우팅은 마치 대기업 콜센터의 ARS 시스템과 같습니다. 전화를 걸면 "상품 문의는 1번, 배송 문의는 2번, 기술 지원은 3번을 눌러주세요"라고 안내합니다.
각 번호를 누르면 해당 분야 전문 상담원에게 연결됩니다. 다중 전문가 라우팅도 마찬가지로 사용자의 질문을 분석한 뒤, 가장 적합한 전문가 모델에게 연결해줍니다.
이 패턴이 없던 시절에는 어땠을까요? 개발자들은 하나의 거대한 모델에 모든 지식을 담으려고 했습니다.
법률, 의료, 기술, 금융... 모든 분야를 다루다 보니 모델은 점점 커졌고, 학습 비용은 천문학적으로 늘어났습니다.
더 큰 문제는 전문성의 희석이었습니다. 이것저것 다 알려고 하다 보니 정작 깊이 있는 답변은 어려워졌습니다.
바로 이런 문제를 해결하기 위해 다중 전문가 라우팅이 등장했습니다. 이 패턴을 사용하면 각 분야별로 특화된 모델을 따로 둘 수 있습니다.
법률 전문 모델은 법률만 깊이 있게 학습하고, 의료 전문 모델은 의료만 집중적으로 다룹니다. 그리고 앞단에 있는 라우터가 질문을 분석해서 적절한 전문가에게 연결해줍니다.
위의 코드를 살펴보겠습니다. 먼저 ExpertRouter 클래스의 experts 딕셔너리가 핵심입니다.
이곳에 도메인 이름을 키로, 해당 전문가 핸들러 함수를 값으로 저장합니다. register_expert 메서드를 통해 새로운 전문가를 언제든 추가할 수 있습니다.
route 메서드가 실제 라우팅을 수행합니다. classifier가 먼저 질문의 도메인을 분류하고, 해당 도메인의 전문가를 찾아 질문을 전달합니다.
만약 적합한 전문가가 없다면 general 전문가가 기본으로 처리합니다. 실제 현업에서는 이 패턴을 어떻게 활용할까요?
글로벌 IT 기업의 고객 지원 시스템을 예로 들어보겠습니다. 고객이 "환불 규정이 어떻게 되나요?"라고 물으면 라우터가 이를 정책 도메인으로 분류하여 정책 전문 모델에게 연결합니다.
반면 "앱이 자꾸 충돌해요"라는 질문은 기술 지원 도메인으로 분류되어 기술 전문 모델이 답변합니다. 주의할 점도 있습니다.
초보 개발자들이 흔히 하는 실수는 분류기를 너무 세분화하는 것입니다. 도메인을 20개, 30개로 나누면 분류 정확도가 오히려 떨어집니다.
처음에는 3-5개의 큰 카테고리로 시작해서 점진적으로 세분화하는 것이 좋습니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.
박시니어 팀장의 조언을 듣고 다중 전문가 라우팅을 구현한 김개발 씨는 만족스러운 결과를 얻었습니다. 법률 질문의 정확도는 40% 향상되었고, 기술 지원 만족도도 크게 올랐습니다.
실전 팁
💡 - 분류기의 정확도가 전체 시스템 성능을 좌우합니다. 분류기에 충분한 투자를 하세요
- 전문가 모델이 없는 도메인을 위한 fallback 전략을 반드시 마련하세요
- 라우팅 결정 로그를 남겨두면 나중에 시스템 개선에 큰 도움이 됩니다
2. 태스크별 분기
김개발 씨의 다중 전문가 라우팅은 큰 성공을 거두었습니다. 하지만 새로운 요구사항이 생겼습니다.
같은 법률 도메인이라도 "계약서 검토해줘"와 "판례 찾아줘"는 완전히 다른 처리가 필요했습니다. 단순히 도메인만으로는 부족했습니다.
이제는 태스크 유형에 따른 분기가 필요한 시점이었습니다.
태스크별 분기는 사용자 요청의 작업 유형을 파악하여 적절한 처리 파이프라인으로 연결하는 패턴입니다. 마치 요리사가 주문을 받았을 때 굽기, 볶기, 찌기 중 어떤 조리법을 사용할지 결정하는 것과 같습니다.
같은 재료라도 조리법에 따라 완전히 다른 요리가 되듯이, 같은 도메인의 질문도 태스크에 따라 다른 처리가 필요합니다.
다음 코드를 살펴봅시다.
from enum import Enum
from typing import Callable, Dict
class TaskType(Enum):
SUMMARIZE = "summarize" # 요약 태스크
ANALYZE = "analyze" # 분석 태스크
GENERATE = "generate" # 생성 태스크
TRANSLATE = "translate" # 번역 태스크
class TaskRouter:
def __init__(self):
self.pipelines: Dict[TaskType, Callable] = {}
def register_pipeline(self, task: TaskType, pipeline: Callable):
self.pipelines[task] = pipeline
def execute(self, query: str, context: dict = None) -> str:
# 질문에서 태스크 유형 추출
task_type = self._detect_task(query)
pipeline = self.pipelines.get(task_type)
return pipeline(query, context)
김개발 씨는 다중 전문가 라우팅 시스템을 성공적으로 구축한 뒤, 뿌듯한 마음으로 운영 데이터를 분석하고 있었습니다. 그런데 이상한 점을 발견했습니다.
같은 법률 도메인으로 분류된 질문들인데, 일부는 높은 만족도를, 일부는 낮은 만족도를 기록하고 있었습니다. 자세히 들여다보니 이유가 명확해졌습니다.
"이 계약서 요약해줘"라는 요청과 "이 계약서에서 문제될 조항 분석해줘"라는 요청은 완전히 다른 처리가 필요했던 것입니다. 전자는 핵심 내용을 추출하는 요약 태스크이고, 후자는 깊이 있는 검토가 필요한 분석 태스크입니다.
박시니어 팀장은 이렇게 설명했습니다. "도메인 라우팅이 '어느 부서로 갈까'를 정한다면, 태스크 라우팅은 '어떤 방식으로 처리할까'를 정하는 거야." 태스크별 분기는 마치 레스토랑 주방의 조리 과정과 같습니다.
손님이 소고기를 주문했다고 가정해봅시다. 소고기라는 재료는 같지만, 스테이크로 구워달라고 하면 그릴에서 조리하고, 불고기로 해달라고 하면 팬에서 볶습니다.
샤브샤브로 해달라고 하면 뜨거운 육수에 살짝 데칩니다. 재료가 같아도 조리 방식이 완전히 다른 것처럼, 같은 도메인의 질문도 태스크에 따라 다른 처리 파이프라인을 거쳐야 합니다.
태스크별 분기가 없으면 어떤 문제가 생길까요? 모든 요청을 동일한 방식으로 처리하면, 요약이 필요한 곳에 과도하게 상세한 분석이 나오거나, 심층 분석이 필요한 곳에 피상적인 요약만 나올 수 있습니다.
사용자는 원하는 형태의 결과를 얻지 못하고, 시스템에 대한 신뢰도가 떨어집니다. 위의 코드에서 TaskType Enum을 살펴봅시다.
여기서는 네 가지 기본 태스크 유형을 정의했습니다. SUMMARIZE는 긴 내용을 짧게 요약하는 태스크입니다.
ANALYZE는 깊이 있는 분석과 인사이트 도출이 필요한 태스크입니다. GENERATE는 새로운 콘텐츠를 창작하는 태스크이고, TRANSLATE는 언어 변환 태스크입니다.
TaskRouter 클래스의 execute 메서드가 핵심 로직입니다. 먼저 _detect_task 메서드가 사용자의 질문을 분석하여 태스크 유형을 판별합니다.
"요약해줘", "정리해줘" 같은 키워드가 있으면 SUMMARIZE로, "분석해줘", "검토해줘"가 있으면 ANALYZE로 분류합니다. 그 다음 해당 태스크에 등록된 파이프라인을 실행합니다.
실제 현업에서는 이 두 가지 라우팅을 조합해서 사용합니다. 먼저 도메인 라우터가 "이건 법률 관련 질문이군"이라고 판단합니다.
그 다음 태스크 라우터가 "요약 태스크네"라고 판단합니다. 최종적으로 "법률 도메인의 요약 파이프라인"이 실행됩니다.
이런 계층적 라우팅을 통해 더욱 정교한 처리가 가능해집니다. 주의해야 할 점이 있습니다.
태스크 유형을 너무 많이 정의하면 분류가 어려워집니다. 또한 하나의 질문에 여러 태스크가 복합적으로 포함될 수도 있습니다.
"이 문서 요약하고 문제점도 분석해줘"처럼요. 이런 경우를 위해 복합 태스크 처리 전략도 미리 설계해두어야 합니다.
김개발 씨는 태스크별 분기를 추가한 후, 시스템 만족도가 크게 향상된 것을 확인했습니다. 이제 같은 법률 질문이라도 사용자가 원하는 형태로 결과를 제공할 수 있게 되었습니다.
실전 팁
💡 - 태스크 유형은 5개 이하로 시작하고, 필요에 따라 점진적으로 확장하세요
- 복합 태스크의 경우 순차 처리 또는 병렬 처리 전략을 미리 정의하세요
- 태스크 감지 실패 시 기본 태스크를 지정해두면 안정성이 높아집니다
3. 실습 도메인별 라우터
이론만으로는 부족합니다. 이제 직접 손으로 코드를 작성해볼 시간입니다.
김개발 씨와 함께 실제로 작동하는 도메인별 라우터를 구현해보겠습니다. 기술, 금융, 일반 세 가지 도메인을 처리하는 라우터를 만들어봅시다.
도메인별 라우터는 질문의 주제 영역을 판별하여 해당 분야 전문가에게 연결하는 실용적인 패턴입니다. 이번 실습에서는 키워드 기반 분류기와 함께 확장 가능한 구조의 라우터를 직접 구현합니다.
실무에서 바로 사용할 수 있는 완성된 코드를 작성해보겠습니다.
다음 코드를 살펴봅시다.
from typing import Dict, Callable, List
import re
class DomainRouter:
def __init__(self):
self.experts: Dict[str, Callable] = {}
self.keywords: Dict[str, List[str]] = {
"tech": ["코드", "버그", "API", "서버", "개발", "프로그램"],
"finance": ["주식", "투자", "금리", "환율", "펀드", "수익"],
}
def register(self, domain: str, handler: Callable):
self.experts[domain] = handler
def classify(self, query: str) -> str:
for domain, words in self.keywords.items():
if any(word in query for word in words):
return domain
return "general"
def route(self, query: str) -> str:
domain = self.classify(query)
return self.experts[domain](query)
김개발 씨는 노트북을 펴고 실습을 시작했습니다. 박시니어 팀장이 옆에서 지켜보며 조언을 해줍니다.
"일단 간단한 것부터 시작하자. 복잡한 AI 분류기 대신 키워드 기반으로 먼저 만들어보는 거야." 좋은 접근법입니다.
MVP(Minimum Viable Product) 원칙에 따라, 먼저 작동하는 가장 단순한 버전을 만든 뒤 점진적으로 개선하는 것이 현명합니다. DomainRouter 클래스의 구조를 살펴봅시다.
생성자에서 두 가지를 초기화합니다. experts 딕셔너리는 도메인별 전문가 핸들러를 저장하고, keywords 딕셔너리는 각 도메인을 대표하는 키워드 목록을 담습니다.
이렇게 설정과 로직을 분리하면 나중에 키워드만 수정해도 분류 기준을 바꿀 수 있습니다. register 메서드는 새로운 전문가를 등록하는 간단한 메서드입니다.
실무에서는 이 메서드를 통해 런타임에 동적으로 전문가를 추가할 수 있습니다. 예를 들어 새로운 도메인 서비스가 런칭되면, 시스템을 재시작하지 않고도 해당 전문가를 등록할 수 있습니다.
classify 메서드가 분류의 핵심입니다. 각 도메인의 키워드를 순회하면서 질문에 해당 키워드가 포함되어 있는지 확인합니다.
any 함수를 사용해 키워드 중 하나라도 매칭되면 해당 도메인으로 분류합니다. 어떤 도메인에도 매칭되지 않으면 "general"을 반환하여 범용 처리기로 넘깁니다.
route 메서드는 분류 결과를 바탕으로 실제 라우팅을 수행합니다. classify로 도메인을 판별하고, 해당 도메인의 전문가 핸들러를 호출합니다.
이 과정이 마치 전화 교환원이 전화를 받아 적절한 부서로 연결해주는 것과 같습니다. 이 코드를 어떻게 사용할까요?
먼저 각 도메인별 전문가 함수를 정의합니다. tech_expert, finance_expert, general_expert 같은 함수를 만들어 register 메서드로 등록합니다.
그 다음 route 메서드에 사용자 질문을 전달하면, 자동으로 적절한 전문가가 답변을 생성합니다. 키워드 기반 분류의 장단점을 알아야 합니다.
장점은 구현이 간단하고 속도가 빠르다는 것입니다. 별도의 AI 모델 호출 없이 문자열 매칭만으로 분류가 가능합니다.
단점은 동음이의어나 문맥을 이해하지 못한다는 것입니다. "주식"이라는 단어가 "주식회사"의 줄임말로 쓰인 건지, 투자 상품을 의미하는 건지 구분하지 못합니다.
그래서 실무에서는 하이브리드 접근을 많이 사용합니다. 1차로 키워드 기반 분류를 수행하고, 확신도가 낮은 경우에만 AI 분류기를 호출합니다.
이렇게 하면 대부분의 명확한 질문은 빠르게 처리하면서, 모호한 질문만 정밀하게 분류할 수 있습니다. 비용과 속도, 정확도 사이의 균형을 맞추는 것이 중요합니다.
김개발 씨는 이 기본 라우터를 완성한 뒤, 테스트 질문들을 넣어보았습니다. "API 연동 방법 알려줘"는 tech로, "펀드 추천해줘"는 finance로 정확히 분류되었습니다.
첫 번째 버전이 성공적으로 작동했습니다.
실전 팁
💡 - 키워드 목록은 실제 사용자 로그를 분석해서 지속적으로 업데이트하세요
- 분류 결과와 함께 신뢰도 점수를 함께 반환하면 후처리에 유용합니다
- 새 도메인 추가 시 기존 키워드와 충돌하지 않는지 확인하세요
4. 실습 전문가 선택 시스템
도메인 라우터가 잘 작동하자, 박시니어 팀장이 새로운 과제를 주었습니다. "이번엔 좀 더 똑똑한 라우터를 만들어볼까?
단순히 키워드 매칭이 아니라, 각 전문가의 역량을 고려해서 최적의 전문가를 선택하는 시스템 말이야." 김개발 씨의 눈이 반짝였습니다.
전문가 선택 시스템은 각 전문가의 역량과 현재 상태를 종합적으로 고려하여 최적의 전문가를 선택하는 고급 라우팅 패턴입니다. 단순 도메인 매칭을 넘어, 전문가의 강점, 약점, 현재 부하 상태까지 고려한 지능적인 라우팅을 구현합니다.
마치 프로젝트 매니저가 팀원들의 역량을 파악하고 적절한 업무를 배분하는 것과 같습니다.
다음 코드를 살펴봅시다.
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class Expert:
name: str
domains: List[str] # 전문 도메인 목록
confidence_scores: dict # 도메인별 자신감 점수
current_load: int = 0 # 현재 처리 중인 요청 수
class SmartExpertSelector:
def __init__(self):
self.experts: List[Expert] = []
def add_expert(self, expert: Expert):
self.experts.append(expert)
def select_best(self, domain: str, complexity: int) -> Optional[Expert]:
candidates = [e for e in self.experts if domain in e.domains]
if not candidates:
return None
# 점수와 부하를 고려한 최적 전문가 선택
return max(candidates, key=lambda e:
e.confidence_scores.get(domain, 0) - (e.current_load * 0.1))
김개발 씨는 새로운 도전에 들떠 있었습니다. 지금까지 만든 라우터는 도메인만 보고 전문가를 선택했습니다.
하지만 현실은 더 복잡합니다. 같은 기술 도메인이라도 프론트엔드에 강한 전문가가 있고, 백엔드에 강한 전문가가 있습니다.
또한 전문가마다 현재 처리 중인 요청 수가 다릅니다. 박시니어 팀장이 비유를 들어 설명했습니다.
"회사에서 새 프로젝트가 들어왔을 때 어떻게 해? 그냥 아무 개발자한테 던지지 않잖아.
누가 이 기술에 강한지, 지금 누가 여유가 있는지 다 고려하지." 바로 이것이 전문가 선택 시스템의 핵심입니다. Expert 클래스를 먼저 살펴봅시다.
dataclass 데코레이터를 사용해 간결하게 정의했습니다. domains는 해당 전문가가 다룰 수 있는 도메인 목록입니다.
한 전문가가 여러 도메인을 담당할 수 있습니다. confidence_scores는 각 도메인에 대한 자신감 점수입니다.
같은 도메인이라도 전문가마다 실력이 다르기 때문입니다. current_load는 현재 처리 중인 요청 수로, 로드 밸런싱에 사용됩니다.
SmartExpertSelector의 select_best 메서드가 핵심 알고리즘입니다. 먼저 해당 도메인을 처리할 수 있는 전문가들을 후보로 추립니다.
그 다음 각 후보의 점수를 계산합니다. 점수는 도메인별 자신감 점수에서 현재 부하에 비례하는 페널티를 뺀 값입니다.
아무리 실력이 좋아도 지금 바쁘면 점수가 낮아집니다. 이 공식의 의미를 생각해봅시다.
confidence_scores.get(domain, 0)은 해당 전문가의 도메인 전문성을 나타냅니다. current_load * 0.1은 부하 페널티입니다.
0.1이라는 가중치는 조절 가능한 값입니다. 이 값이 크면 부하를 더 중요하게 보고, 작으면 전문성을 더 중요하게 봅니다.
실제 사용 예시를 들어보겠습니다. 법률 도메인 전문가가 세 명 있다고 가정합시다.
A 전문가는 점수 0.9에 부하 5, B 전문가는 점수 0.8에 부하 1, C 전문가는 점수 0.7에 부하 0입니다. 단순히 점수만 보면 A가 최고이지만, 부하를 고려하면 A는 0.9 - 0.5 = 0.4, B는 0.8 - 0.1 = 0.7, C는 0.7 - 0 = 0.7이 됩니다.
B나 C가 선택될 가능성이 높아집니다. 이런 방식의 장점은 무엇일까요?
첫째, 자동 로드 밸런싱이 이루어집니다. 바쁜 전문가에게 계속 요청이 몰리는 것을 방지합니다.
둘째, 전문성 활용이 극대화됩니다. 각 전문가의 강점에 맞는 요청이 배분됩니다.
셋째, 확장성이 좋습니다. 새 전문가를 추가하거나 기존 전문가의 점수를 조정하기 쉽습니다.
실무에서 더 고려할 점들이 있습니다. 전문가의 응답 시간 이력, 사용자 만족도 피드백, 최근 처리한 유사 질문 등도 점수에 반영할 수 있습니다.
또한 긴급한 요청의 경우 부하 페널티를 무시하고 최고 전문가에게 바로 연결하는 우선순위 큐 개념도 도입할 수 있습니다. 김개발 씨는 이 시스템을 완성하고 테스트해보았습니다.
요청이 골고루 분산되면서도 전문성이 잘 매칭되는 것을 확인했습니다. 박시니어 팀장이 흐뭇하게 웃었습니다.
"이제 진짜 쓸만한 라우터가 됐네."
실전 팁
💡 - confidence_scores는 초기에 수동 설정 후, 실제 성과 데이터로 자동 업데이트하세요
- 부하 가중치(0.1)는 시스템 특성에 맞게 튜닝이 필요합니다
- 전문가가 없는 도메인 요청을 위한 fallback 전략을 반드시 마련하세요
이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!
댓글 (0)
함께 보면 좋은 카드 뉴스
Context Fundamentals - AI 컨텍스트의 기본 원리
AI 에이전트 개발의 핵심인 컨텍스트 관리를 다룹니다. 시스템 프롬프트 구조부터 Attention Budget, Progressive Disclosure까지 실무에서 바로 적용할 수 있는 컨텍스트 최적화 전략을 배웁니다.
Phase 1 보안 사고방식 구축 완벽 가이드
초급 개발자가 보안 전문가로 성장하기 위한 첫걸음입니다. 해커의 관점에서 시스템을 바라보는 방법부터 OWASP Top 10, 포트 스캐너 구현, 실제 침해사고 분석까지 보안의 기초 체력을 다집니다.
프로덕션 워크플로 배포 완벽 가이드
LLM 기반 애플리케이션을 실제 운영 환경에 배포하기 위한 워크플로 최적화, 캐싱 전략, 비용 관리 방법을 다룹니다. Airflow와 서버리스 아키텍처를 활용한 실습까지 포함하여 초급 개발자도 프로덕션 수준의 배포를 할 수 있도록 안내합니다.
워크플로 모니터링과 디버깅 완벽 가이드
LLM 기반 워크플로의 실행 상태를 추적하고, 문제를 진단하며, 성능을 최적화하는 방법을 다룹니다. LangSmith 통합부터 커스텀 모니터링 시스템 구축까지 실무에서 바로 적용할 수 있는 내용을 담았습니다.
LlamaIndex Workflow 완벽 가이드
LlamaIndex의 워크플로 시스템을 활용하여 복잡한 RAG 파이프라인을 구축하는 방법을 알아봅니다. 이벤트 기반 워크플로부터 멀티 인덱스 쿼리까지 단계별로 학습합니다.