AI 에이전트 스킬 라이브러리와 도구 학습 완벽 가이드

AI 에이전트가 다양한 스킬을 동적으로 로드하고 조합하여 복잡한 작업을 수행하는 방법을 알아봅니다. 스킬 아키텍처 설계부터 도메인 특화 개발까지, 실무에서 바로 적용할 수 있는 패턴을 다룹니다.

---

목차

1. 스킬_라이브러리_아키텍처
2. 동적_스킬_로딩
3. 도구_사용_패턴_학습
4. 스킬_조합_및_체이닝
5. 도메인_특화_스킬_개발
6. 스킬_버전_관리

---

1. 스킬 라이브러리 아키텍처

어느 날 김개발 씨는 회사에서 AI 에이전트 시스템을 구축하라는 미션을 받았습니다. 문제는 에이전트가 수행해야 할 작업이 수십 가지나 된다는 것입니다.

"이걸 어떻게 체계적으로 관리하지?" 고민하던 김개발 씨에게 선배가 한마디 던집니다. "스킬 라이브러리 아키텍처부터 설계해야지."

스킬 라이브러리 아키텍처는 AI 에이전트가 사용할 수 있는 도구들을 체계적으로 구조화하는 설계 방식입니다. 마치 대형 도서관에서 책을 주제별로 분류하고 목록화하는 것처럼, 에이전트의 능력을 모듈화하여 관리합니다.

이를 통해 확장성 있고 유지보수가 쉬운 에이전트 시스템을 구축할 수 있습니다.

다음 코드를 살펴봅시다.

from abc import ABC, abstractmethod
from typing import Dict, Any, List

# 모든 스킬의 기본이 되는 추상 클래스
class BaseSkill(ABC):
    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description
        self.metadata = {}

    @abstractmethod
    def execute(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """스킬의 핵심 실행 로직"""
        pass

    def get_schema(self) -> Dict[str, Any]:
        """LLM이 이해할 수 있는 함수 스키마 반환"""
        return {
            "name": self.name,
            "description": self.description,
            "parameters": self._get_parameters()
        }

# 스킬들을 관리하는 라이브러리 클래스
class SkillLibrary:
    def __init__(self):
        self._skills: Dict[str, BaseSkill] = {}
        self._categories: Dict[str, List[str]] = {}

    def register(self, skill: BaseSkill, category: str = "general"):
        self._skills[skill.name] = skill
        if category not in self._categories:
            self._categories[category] = []
        self._categories[category].append(skill.name)

김개발 씨는 입사 2년 차 백엔드 개발자입니다. 최근 회사에서 고객 문의를 자동으로 처리하는 AI 에이전트를 만들라는 과제를 받았습니다.

이메일 발송, 데이터베이스 조회, 티켓 생성, 알림 전송 등 에이전트가 수행해야 할 작업이 무려 스물다섯 가지나 됩니다. 처음에 김개발 씨는 모든 기능을 하나의 거대한 클래스에 넣으려 했습니다.

하지만 코드가 천 줄을 넘어가자 도저히 관리할 수가 없었습니다. 기능 하나를 수정하려면 전체 코드를 뒤져야 했고, 새 기능을 추가할 때마다 기존 코드가 깨지는 일이 반복됐습니다.

그때 팀의 시니어 개발자 박시니어 씨가 다가왔습니다. "김개발 씨, 스킬 라이브러리 패턴을 써보는 게 어때요?

마이크로서비스처럼 각 기능을 독립된 스킬로 분리하는 거예요." 스킬 라이브러리 아키텍처란 무엇일까요? 쉽게 비유하자면, 대형 도서관의 장서 관리 시스템과 같습니다.

도서관에는 수십만 권의 책이 있지만, 듀이 십진분류법 같은 체계 덕분에 원하는 책을 쉽게 찾을 수 있습니다. 각 책은 고유한 분류 번호를 가지고, 주제별 서가에 배치됩니다.

스킬 라이브러리도 마찬가지입니다. 이 아키텍처에서는 모든 스킬이 BaseSkill이라는 공통 인터페이스를 상속받습니다.

이렇게 하면 어떤 스킬이든 동일한 방식으로 호출하고 관리할 수 있습니다. 마치 모든 도서관 책이 같은 형식의 카탈로그 카드를 가지는 것과 같습니다.

위 코드를 살펴보면, BaseSkill 클래스는 세 가지 핵심 요소를 정의합니다. 첫째, 스킬의 이름과 설명을 저장하는 생성자입니다.

둘째, 실제 작업을 수행하는 execute 메서드입니다. 이 메서드는 추상 메서드로 선언되어, 모든 하위 스킬이 반드시 구현해야 합니다.

셋째, get_schema 메서드는 LLM이 이 스킬을 이해할 수 있도록 JSON 스키마 형태로 정보를 제공합니다. SkillLibrary 클래스는 중앙 관리자 역할을 합니다.

마치 도서관 사서가 새로 들어온 책을 분류하고 등록하듯이, 새로운 스킬을 시스템에 등록하고 카테고리별로 정리합니다. register 메서드를 통해 스킬을 추가하면, 해당 스킬은 이름으로 조회할 수 있게 되고 카테고리에도 자동으로 분류됩니다.

실제 현업에서는 이 구조가 어떻게 활용될까요? 예를 들어 고객 서비스 에이전트를 만든다면, "이메일 스킬", "데이터베이스 스킬", "결제 스킬" 등을 각각 독립적인 모듈로 개발합니다.

각 팀이 담당하는 스킬만 개발하고 테스트하면 되므로, 협업이 훨씬 수월해집니다. 주의할 점도 있습니다.

스킬을 너무 잘게 쪼개면 오히려 관리가 복잡해질 수 있습니다. 하나의 스킬은 하나의 명확한 책임을 가져야 하지만, 그렇다고 모든 함수를 스킬로 만들 필요는 없습니다.

적절한 추상화 수준을 찾는 것이 중요합니다. 김개발 씨는 박시니어 씨의 조언을 따라 스킬 라이브러리 구조로 코드를 리팩터링했습니다.

이제 새 기능을 추가할 때 기존 코드를 건드릴 필요가 없어졌습니다. 단순히 새로운 스킬 클래스를 만들고 라이브러리에 등록하면 끝입니다.

실전 팁

💡 - 스킬 이름은 동사로 시작하여 수행하는 작업을 명확히 표현하세요 (예: send_email, query_database)

• 메타데이터에 버전, 작성자, 의존성 정보를 포함시켜 관리 편의성을 높이세요
• 스킬 간 순환 의존성이 생기지 않도록 설계 단계에서 주의하세요

---

2. 동적 스킬 로딩

김개발 씨의 AI 에이전트가 성장하면서 스킬이 백 개가 넘어갔습니다. 그런데 문제가 생겼습니다.

에이전트가 시작할 때마다 모든 스킬을 메모리에 로드하느라 시간이 너무 오래 걸리는 것입니다. "필요한 스킬만 그때그때 불러올 수는 없을까?" 고민하던 김개발 씨에게 동적 로딩이라는 해답이 떠올랐습니다.

동적 스킬 로딩은 에이전트가 모든 스킬을 미리 메모리에 올려두지 않고, 필요한 시점에 해당 스킬만 불러오는 기법입니다. 마치 스마트폰 앱이 필요할 때만 데이터를 다운로드하는 것처럼, 리소스를 효율적으로 사용하면서도 필요한 기능을 즉시 활용할 수 있습니다.

대규모 에이전트 시스템에서 필수적인 최적화 기법입니다.

다음 코드를 살펴봅시다.

import importlib
from pathlib import Path
from typing import Optional, Dict, Any

class DynamicSkillLoader:
    def __init__(self, skills_directory: str = "./skills"):
        self.skills_dir = Path(skills_directory)
        self._loaded_skills: Dict[str, BaseSkill] = {}
        self._skill_registry = self._scan_available_skills()

    def _scan_available_skills(self) -> Dict[str, str]:
        """스킬 디렉토리를 스캔하여 사용 가능한 스킬 목록 생성"""
        registry = {}
        for skill_file in self.skills_dir.glob("*_skill.py"):
            skill_name = skill_file.stem.replace("_skill", "")
            registry[skill_name] = str(skill_file)
        return registry

    def load_skill(self, skill_name: str) -> Optional[BaseSkill]:
        """요청 시점에 스킬을 동적으로 로드"""
        if skill_name in self._loaded_skills:
            return self._loaded_skills[skill_name]

        if skill_name not in self._skill_registry:
            raise SkillNotFoundError(f"스킬 '{skill_name}'을 찾을 수 없습니다")

        # 모듈을 동적으로 임포트
        module_path = self._skill_registry[skill_name]
        spec = importlib.util.spec_from_file_location(skill_name, module_path)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)

        # 스킬 인스턴스 생성 및 캐싱
        skill_class = getattr(module, f"{skill_name.title()}Skill")
        skill_instance = skill_class()
        self._loaded_skills[skill_name] = skill_instance

        return skill_instance

김개발 씨의 에이전트 시스템은 나날이 성장했습니다. 처음에는 스무 개 정도였던 스킬이 어느새 백오십 개를 넘어섰습니다.

각 부서에서 자신들만의 스킬을 추가했기 때문입니다. 마케팅팀은 캠페인 분석 스킬을, 영업팀은 CRM 연동 스킬을, 개발팀은 배포 자동화 스킬을 만들었습니다.

문제는 성능이었습니다. 에이전트가 시작할 때마다 백오십 개의 스킬 모듈을 전부 메모리에 로드하느라 무려 삼십 초나 걸렸습니다.

게다가 대부분의 사용자는 그중 열 개도 채 사용하지 않았습니다. 마치 백과사전 전체를 외우고 다니면서 정작 필요한 건 한두 페이지뿐인 상황과 같았습니다.

박시니어 씨가 해결책을 제시했습니다. "동적 로딩을 써보세요.

스마트폰 앱처럼 필요할 때만 다운로드받는 거예요." 동적 스킬 로딩이란 무엇일까요? 비유하자면 도서관에서 책을 빌리는 것과 같습니다.

우리는 도서관의 모든 책을 집에 가져다 놓지 않습니다. 필요한 책만 그때그때 빌려서 읽고, 다 읽으면 반납합니다.

동적 로딩도 마찬가지로, 스킬이 실제로 필요해지는 순간에만 메모리로 불러옵니다. 위 코드의 DynamicSkillLoader 클래스를 살펴보겠습니다.

생성자에서는 스킬 파일들이 저장된 디렉토리 경로를 받습니다. 그리고 _scan_available_skills 메서드를 호출하여 어떤 스킬들이 사용 가능한지 목록만 만들어둡니다.

이 단계에서는 실제 모듈을 로드하지 않습니다. 단지 "이메일 스킬은 여기에, 데이터베이스 스킬은 저기에 있다"는 정보만 기록해둡니다.

핵심은 load_skill 메서드입니다. 에이전트가 특정 스킬을 사용하려 할 때 이 메서드가 호출됩니다.

먼저 이미 로드된 스킬인지 확인합니다. 캐시에 있다면 바로 반환합니다.

그렇지 않다면 importlib을 사용하여 해당 Python 파일을 동적으로 임포트합니다. Python의 importlib 모듈은 런타임에 모듈을 임포트할 수 있게 해주는 강력한 도구입니다.

spec_from_file_location으로 모듈 명세를 만들고, module_from_spec으로 모듈 객체를 생성한 뒤, exec_module로 실제 코드를 실행합니다. 이 과정을 거쳐야 비로소 해당 스킬이 메모리에 로드됩니다.

로드된 스킬은 _loaded_skills 딕셔너리에 캐싱됩니다. 같은 스킬을 다시 요청하면 파일을 다시 읽지 않고 캐시에서 바로 반환합니다.

이렇게 하면 한 번 사용한 스킬은 빠르게 재사용할 수 있습니다. 실무에서는 이 패턴을 더 발전시킬 수 있습니다.

예를 들어 자주 사용되는 스킬은 에이전트 시작 시 미리 로드하고, 드물게 사용되는 스킬은 동적 로딩으로 처리하는 하이브리드 방식을 적용할 수 있습니다. 또한 스킬 사용 통계를 수집하여 자동으로 프리로드 목록을 조정하는 방법도 있습니다.

주의할 점은 동적 로딩에는 오버헤드가 있다는 것입니다. 처음 스킬을 로드할 때는 파일 I/O와 모듈 초기화 시간이 필요합니다.

따라서 응답 시간이 매우 중요한 상황에서는 핵심 스킬을 미리 로드해두는 것이 좋습니다. 김개발 씨는 동적 로딩을 적용한 후 에이전트 시작 시간이 삼십 초에서 이 초로 줄어드는 것을 확인했습니다.

"이제 사용자가 기다리다 지쳐서 떠나는 일은 없겠네요!" 김개발 씨는 만족스럽게 웃었습니다.

실전 팁

💡 - 핫 스킬(자주 사용되는 스킬)은 통계 기반으로 프리로드 목록에 추가하세요

• 스킬 로드 실패 시 폴백 메커니즘을 구현하여 에이전트가 멈추지 않도록 하세요
• 개발 환경에서는 핫 리로드를 지원하여 스킬 수정 후 에이전트 재시작 없이 테스트할 수 있게 하세요

---

3. 도구 사용 패턴 학습

김개발 씨는 에이전트에게 스킬을 가르쳤지만, 에이전트가 언제 어떤 스킬을 써야 할지 잘 판단하지 못하는 문제에 부딪혔습니다. "데이터를 저장해줘"라는 요청에 데이터베이스 스킬 대신 파일 저장 스킬을 호출하는 일이 빈번했습니다.

"에이전트가 도구 사용 패턴을 스스로 학습하게 할 수는 없을까?"

도구 사용 패턴 학습은 AI 에이전트가 과거의 성공적인 도구 사용 기록을 바탕으로, 주어진 상황에 가장 적합한 도구를 선택하는 능력을 키우는 과정입니다. 마치 신입사원이 선배의 업무 처리 방식을 관찰하며 노하우를 배우는 것처럼, 에이전트도 축적된 경험 데이터를 통해 더 현명한 결정을 내릴 수 있게 됩니다.

다음 코드를 살펴봅시다.

from dataclasses import dataclass
from typing import List, Tuple
import numpy as np

@dataclass
class ToolUsageRecord:
    """도구 사용 기록을 저장하는 데이터 클래스"""
    user_intent: str
    context: dict
    selected_tool: str
    success: bool
    feedback_score: float  # 0.0 ~ 1.0

class ToolUsagePatternLearner:
    def __init__(self, embedding_model):
        self.embedding_model = embedding_model
        self.usage_history: List[ToolUsageRecord] = []
        self.tool_embeddings: dict = {}

    def record_usage(self, record: ToolUsageRecord):
        """도구 사용 기록을 저장하고 패턴 업데이트"""
        self.usage_history.append(record)
        self._update_tool_affinity(record)

    def suggest_tools(self, intent: str, context: dict, top_k: int = 3) -> List[Tuple[str, float]]:
        """사용자 의도에 가장 적합한 도구 추천"""
        intent_embedding = self.embedding_model.encode(intent)

        scores = []
        for tool_name, tool_data in self.tool_embeddings.items():
            # 의미적 유사도 계산
            similarity = self._cosine_similarity(intent_embedding, tool_data["embedding"])
            # 과거 성공률 반영
            success_rate = tool_data["success_count"] / max(tool_data["total_count"], 1)
            # 최종 점수: 유사도 70% + 성공률 30%
            final_score = 0.7 * similarity + 0.3 * success_rate
            scores.append((tool_name, final_score))

        return sorted(scores, key=lambda x: x[1], reverse=True)[:top_k]

김개발 씨가 만든 AI 에이전트는 이제 백오십 개가 넘는 스킬을 보유하게 되었습니다. 하지만 새로운 문제가 나타났습니다.

에이전트가 어떤 스킬을 언제 사용해야 할지 제대로 판단하지 못하는 것입니다. 예를 들어, 사용자가 "지난달 매출 데이터 좀 보여줘"라고 요청했습니다.

에이전트는 데이터베이스 조회 스킬을 사용해야 하는데, 엉뚱하게도 엑셀 파일 읽기 스킬을 호출했습니다. 결과는 당연히 실패였습니다.

이런 일이 하루에도 수십 번씩 일어났습니다. "에이전트가 경험을 통해 배울 수 있으면 좋겠는데..." 김개발 씨가 한숨을 쉬었습니다.

그때 박시니어 씨가 아이디어를 던졌습니다. "사람도 실수를 통해 배우잖아요.

에이전트에게도 학습 메커니즘을 붙여보는 건 어떨까요?" 도구 사용 패턴 학습의 핵심 아이디어는 간단합니다. 에이전트가 도구를 사용할 때마다 그 기록을 남기고, 성공했는지 실패했는지를 피드백으로 저장합니다.

그리고 비슷한 상황이 다시 발생하면, 과거에 성공했던 도구를 우선적으로 추천합니다. 위 코드에서 ToolUsageRecord는 도구 사용 기록 하나를 표현합니다.

사용자의 의도, 당시 상황(컨텍스트), 선택된 도구, 성공 여부, 그리고 피드백 점수가 포함됩니다. 피드백 점수는 사용자 만족도나 작업 완료 품질을 숫자로 나타낸 것입니다.

ToolUsagePatternLearner 클래스가 학습의 핵심입니다. 이 클래스는 임베딩 모델을 사용하여 사용자 의도를 벡터로 변환합니다.

임베딩이란 텍스트를 고차원 숫자 벡터로 표현하는 기술입니다. "데이터 저장해줘"와 "정보를 기록해줘"는 단어는 다르지만 의미가 비슷하므로, 임베딩 벡터도 가까운 위치에 놓이게 됩니다.

suggest_tools 메서드는 추천의 마법이 일어나는 곳입니다. 사용자 의도를 임베딩으로 변환한 뒤, 각 도구와의 코사인 유사도를 계산합니다.

코사인 유사도는 두 벡터가 얼마나 같은 방향을 가리키는지 측정합니다. 값이 1에 가까울수록 의미가 유사합니다.

여기서 중요한 것은 단순히 유사도만 보지 않는다는 점입니다. 과거 성공률도 함께 고려합니다.

의미적으로 비슷해 보여도 실제로는 실패했던 도구라면 점수가 낮아집니다. 코드에서는 유사도에 70퍼센트, 성공률에 30퍼센트 가중치를 두었습니다.

이 비율은 도메인과 상황에 따라 조정할 수 있습니다. 실무에서는 이 패턴을 더욱 정교하게 발전시킬 수 있습니다.

예를 들어, 사용자별로 선호하는 도구 패턴을 학습할 수 있습니다. 어떤 사용자는 데이터를 항상 엑셀로 받기를 원하고, 다른 사용자는 대시보드로 보기를 원할 수 있습니다.

시간대나 요일에 따른 패턴도 학습 대상이 될 수 있습니다. 주의할 점은 콜드 스타트 문제입니다.

처음에는 학습 데이터가 없으므로 추천 품질이 낮을 수밖에 없습니다. 이를 해결하기 위해 초기에는 규칙 기반 추천을 사용하고, 데이터가 쌓이면 점진적으로 학습 기반으로 전환하는 전략이 필요합니다.

김개발 씨는 패턴 학습 시스템을 도입한 후, 에이전트의 도구 선택 정확도가 65퍼센트에서 92퍼센트로 상승하는 것을 확인했습니다. "이제 에이전트가 정말 똑똒해진 것 같아요!"

실전 팁

💡 - 실패 기록도 반드시 저장하세요. 무엇을 하면 안 되는지 아는 것도 학습입니다

• 임베딩 모델은 도메인에 맞게 파인튜닝하면 정확도가 크게 향상됩니다
• A/B 테스트로 가중치 비율을 최적화하여 추천 품질을 지속적으로 개선하세요

---

4. 스킬 조합 및 체이닝

어느 날 김개발 씨에게 까다로운 요구사항이 들어왔습니다. "매주 월요일 아침에 지난주 매출 데이터를 분석하고, 요약 보고서를 작성해서, 팀장님께 이메일로 보내줘." 단일 스킬로는 해결할 수 없는 복합 작업입니다.

여러 스킬을 순서대로 연결해야 하는데, 어떻게 해야 할까요?

스킬 체이닝은 여러 개의 스킬을 파이프라인처럼 연결하여 복잡한 작업을 수행하는 패턴입니다. 마치 공장의 조립 라인에서 각 공정이 이전 공정의 결과물을 받아 처리하듯이, 앞선 스킬의 출력이 다음 스킬의 입력이 됩니다.

이를 통해 단순한 스킬들을 조합하여 정교한 워크플로우를 구성할 수 있습니다.

다음 코드를 살펴봅시다.

from typing import List, Callable, Any, Optional
from dataclasses import dataclass, field

@dataclass
class SkillChainStep:
    """체인의 각 단계를 정의"""
    skill_name: str
    input_mapper: Callable[[dict], dict] = field(default=lambda x: x)
    output_key: str = "result"
    condition: Optional[Callable[[dict], bool]] = None

class SkillChain:
    def __init__(self, skill_library: SkillLibrary):
        self.library = skill_library
        self.steps: List[SkillChainStep] = []
        self.context: dict = {}

    def add_step(self, step: SkillChainStep) -> "SkillChain":
        """체인에 새로운 단계 추가 (빌더 패턴)"""
        self.steps.append(step)
        return self

    def execute(self, initial_input: dict) -> dict:
        """체인 전체 실행"""
        self.context = {"input": initial_input}

        for i, step in enumerate(self.steps):
            # 조건부 실행: 조건이 있고 만족하지 않으면 스킵
            if step.condition and not step.condition(self.context):
                continue

            # 입력 매핑: 이전 결과를 현재 스킬 입력 형식으로 변환
            skill_input = step.input_mapper(self.context)

            # 스킬 실행
            skill = self.library.get_skill(step.skill_name)
            result = skill.execute(skill_input)

            # 결과를 컨텍스트에 저장
            self.context[step.output_key] = result
            self.context["last_result"] = result

        return self.context

# 사용 예시: 주간 보고서 자동화 체인
report_chain = (
    SkillChain(skill_library)
    .add_step(SkillChainStep(
        skill_name="database_query",
        input_mapper=lambda ctx: {"query": "SELECT * FROM sales WHERE week = last_week"},
        output_key="sales_data"
    ))
    .add_step(SkillChainStep(
        skill_name="data_analyzer",
        input_mapper=lambda ctx: {"data": ctx["sales_data"]},
        output_key="analysis"
    ))
    .add_step(SkillChainStep(
        skill_name="report_generator",
        input_mapper=lambda ctx: {"analysis": ctx["analysis"], "template": "weekly"},
        output_key="report"
    ))
    .add_step(SkillChainStep(
        skill_name="email_sender",
        input_mapper=lambda ctx: {"to": "manager@company.com", "body": ctx["report"]},
        output_key="email_result"
    ))
)

김개발 씨는 난감했습니다. 팀장님이 요청한 주간 보고서 자동화는 단순히 스킬 하나로 해결될 문제가 아니었습니다.

데이터 조회, 분석, 보고서 작성, 이메일 발송이라는 네 가지 작업을 순서대로 처리해야 했습니다. 그것도 앞 단계의 결과물을 다음 단계에서 사용해야 했습니다.

"이런 건 어떻게 해야 하죠?" 김개발 씨가 박시니어 씨에게 물었습니다. "스킬 체이닝을 써야지.

마치 레고 블록을 연결하듯이 스킬들을 이어붙이는 거야." 스킬 체이닝은 공장의 조립 라인을 떠올리면 이해하기 쉽습니다. 자동차 공장에서 차체가 라인을 따라 이동하면서 각 공정을 거칩니다.

첫 번째 공정에서 프레임을 만들고, 두 번째에서 엔진을 얹고, 세 번째에서 도장을 하고, 마지막에 검수를 합니다. 각 공정은 이전 공정의 결과물을 받아 자기 일을 처리합니다.

위 코드에서 SkillChainStep은 체인의 각 단계를 정의합니다. 중요한 속성들이 있습니다.

skill_name은 어떤 스킬을 사용할지, input_mapper는 이전 결과를 어떻게 현재 스킬의 입력으로 변환할지, output_key는 결과를 어떤 이름으로 저장할지를 지정합니다. input_mapper가 핵심입니다.

각 스킬은 자신만의 입력 형식을 가지고 있습니다. 데이터 분석 스킬은 데이터 배열을 기대하고, 이메일 스킬은 수신자와 본문을 기대합니다.

입력 매퍼는 현재 컨텍스트에서 필요한 값을 꺼내 해당 스킬이 이해할 수 있는 형태로 변환합니다. condition 속성은 조건부 실행을 가능하게 합니다.

예를 들어, 매출 데이터가 특정 기준 이하일 때만 경고 이메일을 보내고 싶다면, 해당 단계에 조건을 추가하면 됩니다. 조건을 만족하지 않으면 그 단계는 건너뜁니다.

execute 메서드를 보면, 체인은 context라는 공유 저장소를 가집니다. 각 스킬의 실행 결과는 지정된 키로 컨텍스트에 저장됩니다.

다음 스킬은 이 컨텍스트에서 필요한 데이터를 꺼내 사용합니다. 마치 릴레이 경주에서 바통을 넘기듯이 데이터가 전달됩니다.

사용 예시를 보면, 빌더 패턴을 활용하여 체인을 구성합니다. add_step 메서드가 자기 자신을 반환하므로, 메서드 체이닝으로 여러 단계를 우아하게 추가할 수 있습니다.

코드만 봐도 워크플로우가 어떻게 흘러가는지 한눈에 파악할 수 있습니다. 실무에서는 에러 처리가 매우 중요합니다.

체인 중간에 스킬이 실패하면 어떻게 할까요? 전체를 중단할 수도 있고, 실패한 단계만 건너뛸 수도 있고, 재시도할 수도 있습니다.

실제 시스템에서는 각 단계에 에러 핸들러를 추가하고, 실패 시 롤백 로직도 구현해야 합니다. 병렬 실행도 고려할 수 있습니다.

만약 두 스킬이 서로 독립적이라면, 순차적으로 실행할 필요 없이 동시에 실행하여 시간을 절약할 수 있습니다. 이런 최적화는 대규모 워크플로우에서 성능을 크게 향상시킵니다.

김개발 씨는 스킬 체이닝을 활용하여 주간 보고서 자동화를 완성했습니다. 이제 매주 월요일 아침, 에이전트가 알아서 데이터를 분석하고 보고서를 작성하여 팀장님께 이메일을 보냅니다.

"드디어 월요일 아침이 한결 여유로워졌어요!"

실전 팁

💡 - 각 단계는 독립적으로 테스트할 수 있도록 설계하세요. 단위 테스트가 훨씬 쉬워집니다

• 컨텍스트에 너무 많은 데이터를 저장하면 메모리 문제가 생길 수 있으니 필요한 것만 저장하세요
• 디버깅을 위해 각 단계의 입출력을 로깅하는 기능을 추가하세요

---

5. 도메인 특화 스킬 개발

김개발 씨의 에이전트 시스템이 회사 전체로 확산되면서, 각 부서에서 자신들만의 스킬을 요청하기 시작했습니다. 법무팀은 계약서 분석 스킬을, 인사팀은 채용 프로세스 자동화 스킬을 원했습니다.

범용 스킬로는 한계가 있었습니다. "각 도메인에 특화된 스킬을 어떻게 체계적으로 개발하지?"

도메인 특화 스킬은 특정 업무 영역의 지식과 로직을 캡슐화한 전문 스킬입니다. 마치 의사, 변호사, 회계사처럼 각자의 전문 분야가 있듯이, 도메인 스킬도 해당 분야의 용어, 규칙, 워크플로우를 깊이 이해하고 처리합니다.

이를 통해 에이전트는 단순한 범용 도우미를 넘어 진정한 업무 전문가로 거듭날 수 있습니다.

다음 코드를 살펴봅시다.

from abc import abstractmethod
from typing import Dict, Any, List
from pydantic import BaseModel, Field

# 도메인 스킬의 기반이 되는 추상 클래스
class DomainSkill(BaseSkill):
    def __init__(self, domain: str, name: str, description: str):
        super().__init__(name, description)
        self.domain = domain
        self.domain_vocabulary: Dict[str, str] = {}
        self.business_rules: List[Callable] = []

    def add_vocabulary(self, term: str, definition: str):
        """도메인 전문 용어 등록"""
        self.domain_vocabulary[term] = definition

    def add_business_rule(self, rule: Callable[[dict], bool], description: str):
        """비즈니스 규칙 추가"""
        self.business_rules.append({"check": rule, "description": description})

    def validate_input(self, params: dict) -> List[str]:
        """비즈니스 규칙에 따른 입력 검증"""
        violations = []
        for rule in self.business_rules:
            if not rule["check"](params):
                violations.append(rule["description"])
        return violations

# 법무 도메인 특화 스킬 예시
class ContractAnalyzerSkill(DomainSkill):
    def __init__(self):
        super().__init__(
            domain="legal",
            name="contract_analyzer",
            description="계약서를 분석하여 핵심 조항과 위험 요소를 추출합니다"
        )
        # 법률 용어 등록
        self.add_vocabulary("갑", "계약에서 서비스나 재화를 제공받는 당사자")
        self.add_vocabulary("을", "계약에서 서비스나 재화를 제공하는 당사자")
        self.add_vocabulary("손해배상", "계약 위반 시 발생하는 금전적 보상 의무")

        # 비즈니스 규칙 등록
        self.add_business_rule(
            lambda p: p.get("contract_text", "").strip() != "",
            "계약서 내용이 비어있으면 안 됩니다"
        )
        self.add_business_rule(
            lambda p: len(p.get("contract_text", "")) >= 100,
            "계약서는 최소 100자 이상이어야 합니다"
        )

    def execute(self, params: Dict[str, Any]) -> Dict[str, Any]:
        # 입력 검증
        violations = self.validate_input(params)
        if violations:
            return {"success": False, "errors": violations}

        contract_text = params["contract_text"]

        # 핵심 조항 추출 로직
        key_clauses = self._extract_key_clauses(contract_text)
        risk_factors = self._identify_risks(contract_text)

        return {
            "success": True,
            "key_clauses": key_clauses,
            "risk_factors": risk_factors,
            "domain_terms_found": self._find_domain_terms(contract_text)
        }

법무팀의 이대리 씨가 김개발 씨를 찾아왔습니다. "우리 팀에서 계약서 검토하는 데 시간이 너무 많이 걸려요.

AI가 도와줄 수 있을까요?" 김개발 씨는 기존 텍스트 분석 스킬을 사용해봤지만, 결과가 영 신통치 않았습니다. "손해배상 조항"을 단순히 "손해"와 "배상"이라는 단어로 쪼개 분석하는 바람에, 법적 맥락을 완전히 놓쳐버린 것입니다.

"범용 스킬로는 한계가 있어요. 법무 도메인에 특화된 스킬이 필요합니다." 박시니어 씨가 조언했습니다.

도메인 특화 스킬이란 무엇일까요? 비유하자면, 종합병원의 전문의와 같습니다.

일반의도 환자를 진료할 수 있지만, 심장 질환은 심장내과 전문의가, 뇌 질환은 신경외과 전문의가 더 정확하게 진단하고 치료합니다. 각 전문의는 해당 분야의 깊은 지식, 전문 용어, 진료 프로토콜을 알고 있기 때문입니다.

위 코드에서 DomainSkill 클래스는 도메인 특화 스킬의 기반이 됩니다. 일반 스킬과 달리 세 가지 특별한 요소를 가집니다.

첫째, domain_vocabulary는 해당 분야의 전문 용어 사전입니다. 둘째, business_rules는 해당 도메인에서 지켜야 할 규칙들입니다.

셋째, validate_input 메서드는 이 규칙들을 기반으로 입력을 검증합니다. ContractAnalyzerSkill은 법무 도메인에 특화된 구체적인 스킬입니다.

생성자에서 법률 용어들을 등록합니다. "갑", "을", "손해배상" 같은 용어와 그 정의를 저장해두면, 스킬이 계약서를 분석할 때 이 용어들의 맥락을 이해할 수 있습니다.

비즈니스 규칙도 중요합니다. 예시에서는 계약서 내용이 비어있으면 안 되고, 최소 100자 이상이어야 한다는 규칙을 등록했습니다.

실제로는 "계약 기간이 명시되어야 한다", "양 당사자 정보가 포함되어야 한다" 같은 더 복잡한 규칙들이 추가될 것입니다. execute 메서드를 보면, 실제 처리 전에 먼저 입력 검증을 수행합니다.

비즈니스 규칙을 위반하면 에러 메시지와 함께 즉시 반환합니다. 이렇게 하면 잘못된 입력으로 인한 오류를 사전에 방지할 수 있습니다.

실무에서 도메인 스킬을 개발할 때는 해당 분야 전문가와의 협업이 필수입니다. 법무 스킬을 만들 때는 변호사와, 회계 스킬을 만들 때는 회계사와 함께 일해야 합니다.

그들의 지식을 코드로 옮기는 과정이 도메인 스킬 개발의 핵심입니다. 테스트도 달라져야 합니다.

단순히 코드가 돌아가는지만 확인해서는 안 됩니다. 실제 계약서를 가지고 전문가가 검토한 결과와 스킬의 결과를 비교해야 합니다.

이를 통해 스킬의 정확도를 측정하고 개선할 수 있습니다. 김개발 씨는 법무팀 이대리 씨와 긴밀히 협력하여 계약서 분석 스킬을 개발했습니다.

이대리 씨는 법률 용어를 정리해주고, 계약서 검토 체크리스트를 공유해주었습니다. 그 결과물은 법무팀에서 실제로 사용하는 수준의 정확도를 보여주었습니다.

"이제 단순 계약서 검토는 AI에게 맡기고, 저는 복잡한 협상에 집중할 수 있겠어요!"

실전 팁

💡 - 도메인 전문가와 정기적인 리뷰 세션을 진행하여 스킬의 정확도를 검증하세요

• 도메인 용어 사전은 지속적으로 업데이트하세요. 새로운 용어나 규정 변경을 반영해야 합니다
• 도메인별로 테스트 케이스를 분리하여 관리하고, 실제 문서 샘플로 테스트하세요

---

6. 스킬 버전 관리

6개월이 지났습니다. 김개발 씨의 에이전트 시스템에는 이제 삼백 개가 넘는 스킬이 등록되어 있습니다.

그런데 어느 날 큰 문제가 터졌습니다. 누군가 이메일 발송 스킬을 업데이트했는데, 기존에 잘 돌아가던 주간 보고서 자동화가 먹통이 된 것입니다.

"이전 버전으로 롤백할 수 있으면 좋겠는데..." 버전 관리의 필요성을 절감하는 순간이었습니다.

스킬 버전 관리는 스킬의 변경 이력을 추적하고, 필요할 때 특정 버전으로 되돌리거나 여러 버전을 동시에 운영할 수 있게 해주는 체계입니다. 마치 소프트웨어의 Git처럼, 언제 누가 무엇을 바꿨는지 기록하고, 문제가 생기면 안전하게 이전 상태로 복원할 수 있습니다.

프로덕션 환경에서 안정성을 보장하는 핵심 요소입니다.

다음 코드를 살펴봅시다.

from datetime import datetime
from typing import Dict, Any, Optional, List
from dataclasses import dataclass, field
import hashlib
import json

@dataclass
class SkillVersion:
    """스킬의 특정 버전을 표현"""
    version: str  # semantic versioning: "1.0.0", "1.1.0", "2.0.0"
    skill_class: type
    created_at: datetime
    author: str
    changelog: str
    checksum: str  # 코드 변경 감지용 해시
    deprecated: bool = False

@dataclass
class VersionedSkillRegistry:
    """버전 관리가 적용된 스킬 레지스트리"""
    _skills: Dict[str, Dict[str, SkillVersion]] = field(default_factory=dict)
    _active_versions: Dict[str, str] = field(default_factory=dict)

    def register_version(
        self,
        skill_name: str,
        version: str,
        skill_class: type,
        author: str,
        changelog: str
    ) -> SkillVersion:
        """새 버전의 스킬 등록"""
        if skill_name not in self._skills:
            self._skills[skill_name] = {}

        # 코드 해시 생성으로 실제 변경 여부 확인
        checksum = self._calculate_checksum(skill_class)

        skill_version = SkillVersion(
            version=version,
            skill_class=skill_class,
            created_at=datetime.now(),
            author=author,
            changelog=changelog,
            checksum=checksum
        )

        self._skills[skill_name][version] = skill_version

        # 첫 버전이면 자동으로 활성화
        if skill_name not in self._active_versions:
            self._active_versions[skill_name] = version

        return skill_version

    def set_active_version(self, skill_name: str, version: str):
        """특정 버전을 활성 버전으로 설정"""
        if version not in self._skills.get(skill_name, {}):
            raise ValueError(f"버전 {version}을 찾을 수 없습니다")
        self._active_versions[skill_name] = version

    def get_skill(self, skill_name: str, version: str = None) -> BaseSkill:
        """스킬 인스턴스 반환. 버전 미지정시 활성 버전 사용"""
        if version is None:
            version = self._active_versions.get(skill_name)

        skill_version = self._skills[skill_name][version]
        return skill_version.skill_class()

    def rollback(self, skill_name: str, target_version: str):
        """이전 버전으로 롤백"""
        current = self._active_versions[skill_name]
        self.set_active_version(skill_name, target_version)
        return {"rolled_back_from": current, "rolled_back_to": target_version}

금요일 오후 5시, 퇴근을 앞두고 김개발 씨의 슬랙에 긴급 메시지가 날아왔습니다. "주간 보고서 발송이 안 돼요!

팀장님이 기다리고 계셔요!" 식은땀이 흘렀습니다. 분명 어제까지는 잘 작동했는데, 오늘 갑자기 에러가 발생한 것입니다.

원인을 추적하니, 오전에 다른 개발자가 이메일 발송 스킬을 업데이트했는데 파라미터 형식을 바꿔버린 것이었습니다. 기존에는 수신자를 문자열로 받았는데, 새 버전에서는 리스트로 받도록 변경된 것입니다.

주간 보고서 체인은 여전히 문자열을 넘기고 있었습니다. "이전 버전으로 되돌릴 수 있으면 좋겠는데..." 하지만 이전 코드는 어디에도 남아있지 않았습니다.

결국 김개발 씨는 금요일 밤을 새워 호환성 문제를 해결해야 했습니다. 그 이후로 김개발 씨는 스킬 버전 관리 시스템을 도입하기로 결심했습니다.

버전 관리란 무엇일까요? 소프트웨어 개발자라면 Git을 떠올리실 겁니다.

코드의 변경 이력을 기록하고, 문제가 생기면 이전 커밋으로 되돌릴 수 있는 시스템입니다. 스킬 버전 관리도 같은 개념입니다.

스킬이 언제, 누구에 의해, 어떻게 변경되었는지 기록하고, 필요하면 이전 버전으로 롤백할 수 있습니다. 위 코드에서 SkillVersion 클래스는 스킬의 특정 버전을 표현합니다.

version 필드는 시맨틱 버저닝을 따릅니다. "1.0.0"에서 "1.1.0"으로 바뀌면 하위 호환 가능한 기능 추가, "2.0.0"으로 바뀌면 하위 호환이 깨지는 변경이라는 의미입니다.

checksum은 코드의 해시값으로, 실제로 코드가 변경되었는지 확인하는 데 사용합니다. VersionedSkillRegistry는 버전 관리의 핵심입니다.

_skills 딕셔너리는 스킬 이름을 키로, 버전별 스킬 정보를 값으로 가집니다. 하나의 스킬에 여러 버전이 동시에 존재할 수 있는 것입니다.

_active_versions는 각 스킬의 현재 활성 버전을 추적합니다. register_version 메서드로 새 버전을 등록할 때, 작성자와 변경 이력(changelog)을 함께 기록합니다.

누가 왜 이 변경을 했는지 추적할 수 있어야 문제가 생겼을 때 빠르게 대응할 수 있습니다. rollback 메서드가 생명줄입니다.

새 버전에 문제가 발견되면, 단 한 줄의 호출로 이전 버전으로 되돌릴 수 있습니다. 김개발 씨가 그 금요일 밤에 이 기능이 있었다면, 퇴근 전에 문제를 해결할 수 있었을 것입니다.

실무에서는 카나리 배포 패턴도 함께 사용합니다. 새 버전을 등록해도 바로 활성화하지 않고, 일부 트래픽만 새 버전으로 보내 테스트합니다.

문제가 없으면 점진적으로 트래픽을 늘리고, 문제가 발견되면 즉시 롤백합니다. 스킬 간 의존성도 고려해야 합니다.

A 스킬 2.0 버전이 B 스킬 1.5 이상을 요구한다면, 이런 의존성 정보도 버전 메타데이터에 포함시켜야 합니다. 그래야 호환되지 않는 버전 조합을 미리 막을 수 있습니다.

그 이후로 김개발 씨의 팀에서는 모든 스킬 변경에 버전을 붙이고, 변경 이력을 상세히 기록하는 문화가 정착되었습니다. 덕분에 금요일 오후의 긴급 상황은 더 이상 악몽이 아니게 되었습니다.

"롤백 한 방이면 해결이니까요!"

실전 팁

💡 - 시맨틱 버저닝 규칙을 팀 전체에 공유하고 일관되게 적용하세요

• 하위 호환이 깨지는 변경(major version bump)은 충분한 공지 기간을 두고 진행하세요
• 자동화된 호환성 테스트를 구축하여 새 버전 등록 전에 기존 체인과의 호환성을 검증하세요

---

이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!