AI Agent 플랫폼 인터페이스 설계

BaseAgent와 Template Method Pattern

AI Agent 플랫폼의 핵심 인터페이스 설계를 다룬다. BaseAgent 추상 클래스 설계, Template Method Pattern 적용, AgentRegistry를 통한 Dependency Inversion 구현, Orchestrator로 Agent 실행 관리 방법을 구체적으로 설명한다. 실제 Agent 구현 예시와 함께 표준 인터페이스가 왜 중요한지, 어떻게 설계해야 유지보수성과 확장성을 모두 확보할 수 있는지 제시한다.

Engineering
System
Architecture Design
Agent
Platform
저자

Kwangmin Kim

공개

2026년 01월 29일

1 들어가며

1.1 왜 인터페이스 설계가 중요한가?

Phase 1-4를 거쳐 여러 Agent를 만들었다. 이제 문제는:

질문들: - “모든 Agent가 따라야 할 표준 인터페이스는 무엇인가?” - “새 Agent 추가 시 기존 코드를 수정하지 않으려면?” - “Agent 실행, 평가, 모니터링을 어떻게 통일할 것인가?”

이 글의 목표: - BaseAgent 추상 클래스 설계 - Template Method Pattern 적용 이유 - AgentRegistry로 Dependency Inversion 구현 - Orchestrator로 Agent 실행 관리 - 실제 Agent 구현 예시

1.2 앞선 글 요약

3번 글 (저장소 전략): - Monorepo 선택 근거 - core/shared/agents 모듈 분리 - 의존성 규칙: agents → shared → core - 빌드 도구: Poetry → Nx → Bazel 로드맵

핵심 질문: “core/에 어떤 인터페이스를 정의할 것인가?”

2 BaseAgent 인터페이스 설계

2.1 설계 원칙

2.1.1 원칙 1: 표준 입출력 형식

문제: 각 Agent가 다른 입출력 형식 사용

# ❌ 나쁜 예: Agent마다 다른 시그니처
class DataAgent:
    def run(self, schema: str) -> str:  # 문자열 입출력
        ...

class CodeAgent:
    def analyze(self, code_obj: CodeObject) -> List[Issue]:  # 객체 입출력
        ...

class KnowledgeAgent:
    def query(self, question: str, context: Dict) -> Dict:  # 혼합
        ...

# 문제: Agent 체이닝 불가능
result1 = data_agent.run(schema)
result2 = code_agent.analyze(result1)  # 타입 에러!

해결: Dict[str, Any] 표준 형식

# ✅ 좋은 예: 표준 입출력
class BaseAgent(ABC):
    @abstractmethod
    def process(self, input: Dict[str, Any]) -> Dict[str, Any]:
        pass

# 모든 Agent가 동일한 시그니처
data_result = data_agent.process({'data': schema})
code_result = code_agent.process(data_result)  # 체이닝 가능

표준 입력 구조:

input = {
    'task': 'standardization',  # 작업 유형
    'data': {...},              # 실제 데이터
    'config': {                 # 실행 옵션
        'temperature': 0.0,
        'max_tokens': 1000
    },
    'metadata': {               # 컨텍스트
        'source': 'previous_agent',
        'timestamp': 1234567890
    }
}

표준 출력 구조:

output = {
    'result': {...},            # 실행 결과
    'confidence': 0.95,         # 신뢰도 (0-1)
    'metadata': {               # 실행 정보
        'agent': 'data_standardization',
        'version': '1.0.0',
        'execution_time': 2.5
    },
    'errors': []                # 에러 목록 (있다면)
}

2.1.2 원칙 2: Template Method Pattern

문제: 공통 로직(검증, 로깅, 메트릭)을 매번 구현

# ❌ 나쁜 예: 각 Agent가 중복 코드
class DataAgent:
    def process(self, input):
        # 입력 검증 (중복)
        if not isinstance(input, dict):
            raise ValueError("...")
        
        # 시간 측정 (중복)
        start = time.time()
        
        # 실제 로직
        result = self._do_work(input)
        
        # 로깅 (중복)
        elapsed = time.time() - start
        logger.info(f"Elapsed: {elapsed}")
        
        return result

class CodeAgent:
    def process(self, input):
        # 동일한 검증, 측정, 로깅 코드 반복
        ...

해결: Template Method로 공통 로직 추출

# ✅ 좋은 예: BaseAgent가 공통 로직 처리
class BaseAgent(ABC):
    def execute(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """Template Method (final - 수정 불가)"""
        # 1. 공통 검증
        self._validate_input(input)
        
        # 2. 전처리 hook
        input = self._pre_process(input)
        
        # 3. 시간 측정
        start_time = time.time()
        
        # 4. 실제 로직 (하위 클래스 구현)
        result = self.process(input)
        
        # 5. 후처리 hook
        result = self._post_process(result)
        
        # 6. 메트릭 수집
        self._collect_metrics(time.time() - start_time)
        
        return result
    
    @abstractmethod
    def process(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """하위 클래스가 구현해야 할 핵심 로직"""
        pass

# Agent는 process()만 구현
class DataAgent(BaseAgent):
    def process(self, input):
        # 비즈니스 로직만 집중
        return {'result': standardized_data}

2.1.3 원칙 3: Hook Methods

선택적 커스터마이징:

class BaseAgent(ABC):
    def _pre_process(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """전처리 hook (선택적 override)"""
        return input
    
    def _post_process(self, output: Dict[str, Any]) -> Dict[str, Any]:
        """후처리 hook (선택적 override)"""
        return output

# Agent가 필요하면 override
class DataAgent(BaseAgent):
    def _pre_process(self, input):
        # 입력 정규화
        input['data'] = input['data'].lower()
        return input
    
    def process(self, input):
        # 핵심 로직
        return {'result': ...}

2.2 BaseAgent 전체 구현

# core/base_agent.py
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional, List
from dataclasses import dataclass
import time
import logging

logger = logging.getLogger(__name__)

@dataclass
class AgentMetadata:
    """Agent 메타데이터"""
    name: str
    version: str
    domain: str
    description: str
    author: str
    created_at: str
    tags: List[str] = None

class AgentMetrics:
    """Agent 성능 메트릭"""
    def __init__(self):
        self.execution_count = 0
        self.total_time = 0.0
        self.errors: List[str] = []
        self.success_count = 0
    
    def record_success(self, elapsed_time: float):
        self.execution_count += 1
        self.success_count += 1
        self.total_time += elapsed_time
    
    def record_error(self, error: str):
        self.execution_count += 1
        self.errors.append(error)
    
    def get_stats(self) -> Dict[str, Any]:
        return {
            'executions': self.execution_count,
            'success_rate': self.success_count / max(self.execution_count, 1),
            'avg_time': self.total_time / max(self.success_count, 1),
            'error_count': len(self.errors)
        }

class BaseAgent(ABC):
    """모든 Agent의 기본 클래스
    
    설계 원칙:
    1. 표준 입출력: Dict[str, Any] (확장성)
    2. Template Method: execute()는 final
    3. Hook Methods: 선택적 커스터마이징
    4. 자동 메트릭: 성능 추적
    5. 평가 인터페이스: 도메인별 구현
    """
    
    def __init__(self, metadata: AgentMetadata):
        self.metadata = metadata
        self.metrics = AgentMetrics()
        logger.info(f"Initialized Agent: {metadata.name} v{metadata.version}")
    
    @abstractmethod
    def process(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """핵심 비즈니스 로직 (하위 클래스 구현 필수)
        
        Args:
            input: 표준 입력 딕셔너리
                - task: 작업 유형
                - data: 실제 데이터
                - config: 실행 옵션 (선택)
                - metadata: 컨텍스트 (선택)
        
        Returns:
            표준 출력 딕셔너리
                - result: 실행 결과
                - confidence: 신뢰도 (0-1)
                - errors: 에러 목록 (선택)
        """
        pass
    
    @abstractmethod
    def evaluate(self, ground_truth: Any, prediction: Any) -> float:
        """성능 평가 (하위 클래스 구현 필수)
        
        각 Agent는 도메인에 맞는 평가 메트릭 구현
        
        Examples:
            - 데이터 표준화: 스키마 매칭 정확도
            - 코드 분석: AST 파싱 성공률
            - 지식 QnA: ROUGE 스코어
        
        Returns:
            평가 점수 (0-1)
        """
        pass
    
    def execute(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """Agent 실행 (Template Method - 수정 불가)
        
        실행 흐름:
        1. 입력 검증
        2. 전처리 hook
        3. 실제 process() 호출
        4. 후처리 hook
        5. 메트릭 수집
        6. 메타데이터 추가
        
        Returns:
            표준 출력 + 메타데이터
        """
        try:
            # 1. 입력 검증
            self._validate_input(input)
            
            # 2. 전처리
            processed_input = self._pre_process(input)
            
            # 3. 시간 측정 시작
            start_time = time.time()
            
            # 4. 핵심 로직 실행
            result = self.process(processed_input)
            
            # 5. 후처리
            processed_result = self._post_process(result)
            
            # 6. 실행 시간 계산
            elapsed = time.time() - start_time
            
            # 7. 메트릭 기록
            self.metrics.record_success(elapsed)
            
            # 8. 메타데이터 추가
            processed_result['metadata'] = {
                'agent': self.metadata.name,
                'version': self.metadata.version,
                'execution_time': elapsed,
                'timestamp': time.time()
            }
            
            logger.info(f"Agent {self.metadata.name} succeeded in {elapsed:.2f}s")
            
            return processed_result
            
        except Exception as e:
            error_msg = f"{self.metadata.name} failed: {str(e)}"
            logger.error(error_msg)
            self.metrics.record_error(error_msg)
            raise
    
    def _validate_input(self, input: Dict[str, Any]):
        """입력 검증 (공통 규칙)"""
        if not isinstance(input, dict):
            raise ValueError("Input must be a dictionary")
        
        if 'task' not in input:
            raise ValueError("Input must contain 'task' field")
        
        if 'data' not in input:
            raise ValueError("Input must contain 'data' field")
    
    def _pre_process(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """전처리 hook (선택적 override)
        
        예: 입력 정규화, 기본값 설정
        """
        return input
    
    def _post_process(self, output: Dict[str, Any]) -> Dict[str, Any]:
        """후처리 hook (선택적 override)
        
        예: 출력 형식 변환, 에러 핸들링
        """
        return output
    
    def get_metrics(self) -> Dict[str, Any]:
        """Agent 성능 메트릭 조회"""
        return {
            'name': self.metadata.name,
            **self.metrics.get_stats()
        }
    
    def reset_metrics(self):
        """메트릭 초기화"""
        self.metrics = AgentMetrics()
        logger.info(f"Reset metrics for {self.metadata.name}")

2.3 설계 의사결정 설명

2.3.1 왜 Dict[str, Any]인가?

대안 1: Pydantic 모델

class AgentInput(BaseModel):
    task: str
    data: Dict[str, Any]
    config: Optional[Dict[str, Any]]

class BaseAgent(ABC):
    def process(self, input: AgentInput) -> AgentOutput:
        pass

문제: - 타입이 고정됨 → 새 필드 추가 시 BaseAgent 수정 필요 - Agent마다 다른 입력 필요 시 상속 계층 복잡

Dict[str, Any]의 장점: - 확장성: 새 필드 자유롭게 추가 - Agent별 커스터마이징 가능 - JSON 직렬화 간단

트레이드오프: - 타입 안전성 낮음 - IDE 자동완성 없음

해결책: 문서화 + 런타임 검증

def _validate_input(self, input: Dict[str, Any]):
    """입력 검증으로 타입 안전성 보완"""
    required_fields = ['task', 'data']
    for field in required_fields:
        if field not in input:
            raise ValueError(f"Missing required field: {field}")

2.3.2 왜 execute()와 process()를 분리하는가?

Template Method Pattern의 핵심:

# execute(): BaseAgent가 제어 (final)
def execute(self, input):
    self._validate_input(input)  # 공통 로직
    result = self.process(input)  # 하위 클래스 로직
    self._collect_metrics()       # 공통 로직
    return result

# process(): 하위 클래스가 구현
@abstractmethod
def process(self, input):
    pass

이점: 1. 일관성: 모든 Agent가 검증, 메트릭 수집 자동 수행 2. 단순성: Agent 개발자는 process()만 구현 3. 안전성: 공통 로직 수정 시 모든 Agent에 자동 반영

2.3.3 왜 evaluate()를 인터페이스에 포함하는가?

문제: 각 Agent의 성능을 어떻게 측정?

# ❌ 나쁜 예: 평가 로직이 Agent 밖에 있음
def evaluate_data_agent(agent, test_data):
    # 평가 로직이 외부에 있어 유지보수 어려움
    ...

def evaluate_code_agent(agent, test_data):
    # 다른 평가 로직 (중복)
    ...

해결: Agent가 자신의 평가 방법 정의

# ✅ 좋은 예: Agent가 evaluate() 구현
class DataAgent(BaseAgent):
    def evaluate(self, ground_truth, prediction):
        # 스키마 매칭 정확도
        return schema_matching_score(ground_truth, prediction)

class CodeAgent(BaseAgent):
    def evaluate(self, ground_truth, prediction):
        # AST 파싱 성공률
        return ast_parsing_success_rate(ground_truth, prediction)

# 플랫폼이 일관되게 평가
for agent in agents:
    score = agent.evaluate(test_data.truth, test_data.pred)
    print(f"{agent.metadata.name}: {score:.2f}")

3 AgentRegistry: Dependency Inversion 구현

3.1 문제: Core가 Agent에 의존하면?

# ❌ 나쁜 예: Orchestrator가 Agent를 직접 import
# core/orchestrator.py
from agents.data_standardization import DataAgent
from agents.code_analysis import CodeAgent
from agents.knowledge_qna import KnowledgeAgent

class Orchestrator:
    def __init__(self):
        self.agents = [
            DataAgent(),
            CodeAgent(),
            KnowledgeAgent()
        ]
    
    def add_agent(self, agent):
        # 새 Agent 추가 시 이 파일 수정 필요
        # → Core 모듈이 불안정해짐
        pass

문제점: 1. Core → Agents 의존 (의존성 방향 위반) 2. 새 Agent 추가 시 Core 코드 수정 3. 순환 의존 가능성

3.2 해결: Registry Pattern

# ✅ 좋은 예: Agent가 스스로 등록
# core/registry.py
class AgentRegistry:
    """Agent 등록 및 조회 (Singleton)"""
    
    _instance = None
    _agents: Dict[str, BaseAgent] = {}
    
    @classmethod
    def get_instance(cls):
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance
    
    def register(self, agent: BaseAgent):
        """Agent 등록"""
        name = agent.metadata.name
        
        if name in self._agents:
            raise ValueError(f"Agent {name} already registered")
        
        self._agents[name] = agent
        logger.info(f"✓ Registered: {name} v{agent.metadata.version}")
    
    def get(self, name: str) -> BaseAgent:
        """Agent 조회"""
        if name not in self._agents:
            available = ', '.join(self._agents.keys())
            raise ValueError(f"Agent {name} not found. Available: {available}")
        
        return self._agents[name]
    
    def list_all(self) -> Dict[str, BaseAgent]:
        """모든 Agent 목록"""
        return self._agents.copy()
    
    def unregister(self, name: str):
        """Agent 제거 (테스트용)"""
        if name in self._agents:
            del self._agents[name]
            logger.info(f"✗ Unregistered: {name}")

# agents/data_standardization/__init__.py
from core.registry import AgentRegistry
from core.base_agent import AgentMetadata
from .agent import DataAgent

# 모듈 import 시 자동 등록
metadata = AgentMetadata(
    name="data_standardization",
    version="1.0.0",
    domain="data",
    description="데이터 스키마 표준화",
    author="Platform Team",
    created_at="2026-02-02"
)
agent = DataAgent(metadata)
AgentRegistry.get_instance().register(agent)

# core/orchestrator.py
class Orchestrator:
    def __init__(self):
        self.registry = AgentRegistry.get_instance()
    
    def run(self, agent_name: str, input: Dict) -> Dict:
        # Agent를 직접 import하지 않음
        agent = self.registry.get(agent_name)
        return agent.execute(input)

의존성 방향:

agents/data_standardization/ → core/registry.py ✅
core/orchestrator.py → agents/ ❌ (불가능, import 없음)

3.3 실전 예시: Agent 추가

새 Agent 추가 시 Core 수정 불필요:

# 1. 새 Agent 디렉토리 생성
mkdir agents/recommendation

# 2. Agent 구현
# agents/recommendation/agent.py
class RecommendationAgent(BaseAgent):
    def process(self, input):
        return {'result': recommendations}
    
    def evaluate(self, truth, pred):
        return accuracy_score(truth, pred)

# 3. 자동 등록
# agents/recommendation/__init__.py
metadata = AgentMetadata(name="recommendation", ...)
agent = RecommendationAgent(metadata)
AgentRegistry.get_instance().register(agent)

# 4. 즉시 사용 가능 (Core 수정 없음)
orchestrator = Orchestrator()
result = orchestrator.run("recommendation", input_data)

4 Orchestrator: Agent 실행 관리

4.1 단일 Agent 실행

# core/orchestrator.py
from typing import Dict, List, Any
from .registry import AgentRegistry
import logging

logger = logging.getLogger(__name__)

class AgentOrchestrator:
    """Agent 실행 및 조율
    
    기능:
    1. 단일 Agent 실행
    2. Agent 체이닝 (순차 실행)
    3. 병렬 실행
    4. 조건부 실행
    """
    
    def __init__(self):
        self.registry = AgentRegistry.get_instance()
    
    def run(
        self,
        agent_name: str,
        input_data: Dict[str, Any],
        timeout: Optional[float] = None
    ) -> Dict[str, Any]:
        """단일 Agent 실행
        
        Args:
            agent_name: Agent 이름
            input_data: 입력 데이터
            timeout: 타임아웃 (초)
        
        Returns:
            Agent 실행 결과
        """
        logger.info(f"Running agent: {agent_name}")
        
        agent = self.registry.get(agent_name)
        result = agent.execute(input_data)
        
        logger.info(f"Agent {agent_name} completed")
        return result

4.2 Agent 체이닝

def chain(
    self,
    agent_names: List[str],
    input_data: Dict[str, Any],
    pass_intermediate: bool = True
) -> Dict[str, Any]:
    """Agent 체이닝 (순차 실행)
    
    Args:
        agent_names: Agent 이름 목록
        input_data: 초기 입력
        pass_intermediate: 중간 결과를 다음 Agent에 전달 여부
    
    Returns:
        최종 결과
    
    Example:
        orchestrator.chain(
            ['data_standardization', 'code_analysis', 'recommendation'],
            {'task': 'analyze', 'data': schema}
        )
    """
    result = input_data
    
    for agent_name in agent_names:
        logger.info(f"Chain step: {agent_name}")
        
        agent = self.registry.get(agent_name)
        result = agent.execute(result)
        
        if not pass_intermediate:
            # 원본 입력 유지, 결과만 누적
            result = {
                **input_data,
                f'{agent_name}_result': result
            }
    
    return result

실행 예시:

# 데이터 표준화 → 코드 분석 → 추천
orchestrator = AgentOrchestrator()

result = orchestrator.chain(
    agent_names=['data_standardization', 'code_analysis', 'recommendation'],
    input_data={
        'task': 'analyze_and_recommend',
        'data': {'schema': raw_schema}
    }
)

# 결과
{
    'result': [...],  # 최종 추천 결과
    'metadata': {
        'agent': 'recommendation',  # 마지막 Agent
        'execution_time': 5.2
    }
}

4.3 병렬 실행

import concurrent.futures

def parallel(
    self,
    agent_names: List[str],
    input_data: Dict[str, Any],
    max_workers: int = 4
) -> List[Dict[str, Any]]:
    """병렬 실행 (동일 입력, 여러 Agent)
    
    Args:
        agent_names: Agent 이름 목록
        input_data: 입력 데이터 (모든 Agent 공통)
        max_workers: 최대 병렬 워커 수
    
    Returns:
        Agent별 실행 결과 리스트
    
    Use Case:
        - 여러 Agent의 결과 비교 (앙상블)
        - A/B 테스트
    """
    def run_agent(agent_name: str) -> Dict[str, Any]:
        agent = self.registry.get(agent_name)
        return agent.execute(input_data)
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(run_agent, name): name
            for name in agent_names
        }
        
        results = []
        for future in concurrent.futures.as_completed(futures):
            agent_name = futures[future]
            try:
                result = future.result()
                results.append(result)
                logger.info(f"Agent {agent_name} completed")
            except Exception as e:
                logger.error(f"Agent {agent_name} failed: {e}")
        
        return results

실행 예시:

# 3개 Agent를 병렬로 실행하여 결과 비교
results = orchestrator.parallel(
    agent_names=['agent_v1', 'agent_v2', 'agent_v3'],
    input_data={'task': 'analyze', 'data': test_data}
)

# 결과 비교
for result in results:
    agent_name = result['metadata']['agent']
    confidence = result['confidence']
    print(f"{agent_name}: {confidence:.2f}")

4.4 조건부 실행

def conditional_chain(
    self,
    agent_configs: List[Dict[str, Any]],
    input_data: Dict[str, Any]
) -> Dict[str, Any]:
    """조건부 Agent 체이닝
    
    Args:
        agent_configs: Agent 설정 목록
            [
                {'name': 'agent1', 'condition': lambda r: r['confidence'] > 0.8},
                {'name': 'agent2', 'condition': lambda r: 'error' not in r}
            ]
        input_data: 초기 입력
    
    Returns:
        최종 결과
    """
    result = input_data
    
    for config in agent_configs:
        agent_name = config['name']
        condition = config.get('condition', lambda r: True)
        
        # 조건 확인
        if not condition(result):
            logger.info(f"Skipping {agent_name} (condition not met)")
            continue
        
        # Agent 실행
        agent = self.registry.get(agent_name)
        result = agent.execute(result)
    
    return result

실행 예시:

# 신뢰도 높을 때만 다음 Agent 실행
result = orchestrator.conditional_chain(
    agent_configs=[
        {'name': 'data_standardization'},
        {
            'name': 'code_analysis',
            'condition': lambda r: r.get('confidence', 0) > 0.8
        },
        {
            'name': 'recommendation',
            'condition': lambda r: 'errors' not in r
        }
    ],
    input_data={'task': 'analyze', 'data': schema}
)

5 실제 Agent 구현 예시

5.1 DataStandardizationAgent

# agents/data_standardization/agent.py
from core.base_agent import BaseAgent, AgentMetadata
from shared.llm.client import LLMClient
from typing import Dict, Any

class DataStandardizationAgent(BaseAgent):
    """데이터 스키마 표준화 Agent
    
    기능:
    - 비표준 스키마를 표준 형식으로 변환
    - 필드명 정규화 (예: "환자번호" → "patient_id")
    - 데이터 타입 추론
    """
    
    def __init__(self, metadata: AgentMetadata):
        super().__init__(metadata)
        
        # Agent 전용 LLM 클라이언트
        self.llm = LLMClient(
            provider="openai",
            model="gpt-4",
            cache_enabled=True
        )
    
    def process(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """스키마 표준화"""
        schema = input['data']
        
        # LLM 프롬프트 생성
        prompt = self._build_prompt(schema)
        
        # LLM 호출
        standardized = self.llm.generate(
            prompt=prompt,
            temperature=0.0  # 결정론적 출력
        )
        
        # 결과 파싱
        result = self._parse_result(standardized)
        
        return {
            'result': result,
            'confidence': self._calculate_confidence(result)
        }
    
    def _build_prompt(self, schema: Dict) -> str:
        """프롬프트 생성"""
        return f"""다음 스키마를 표준화하세요:

입력 스키마:
{schema}

규칙:
1. 필드명을 영문 snake_case로 변환
2. 데이터 타입을 명시 (string, int, float, date 등)
3. 설명 추가

출력 형식 (JSON):
{{
    "fields": [
        {{"name": "field_name", "type": "string", "description": "..."}}
    ]
}}"""
    
    def _parse_result(self, llm_output: str) -> Dict:
        """LLM 출력 파싱"""
        import json
        return json.loads(llm_output)
    
    def _calculate_confidence(self, result: Dict) -> float:
        """신뢰도 계산"""
        # 필드 개수, 타입 명시 여부 등으로 계산
        if 'fields' not in result:
            return 0.0
        
        fields = result['fields']
        if len(fields) == 0:
            return 0.0
        
        # 모든 필드가 type과 description을 가지면 신뢰도 1.0
        complete_fields = sum(
            1 for f in fields
            if 'type' in f and 'description' in f
        )
        
        return complete_fields / len(fields)
    
    def evaluate(self, ground_truth: Dict, prediction: Dict) -> float:
        """스키마 매칭 정확도"""
        truth_fields = {f['name'] for f in ground_truth['fields']}
        pred_fields = {f['name'] for f in prediction['fields']}
        
        # Jaccard similarity
        intersection = len(truth_fields & pred_fields)
        union = len(truth_fields | pred_fields)
        
        return intersection / union if union > 0 else 0.0
    
    def _pre_process(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """전처리: 입력 정규화"""
        # 스키마를 소문자로 변환
        if 'data' in input and isinstance(input['data'], dict):
            normalized = {
                k.lower(): v
                for k, v in input['data'].items()
            }
            input['data'] = normalized
        
        return input

5.2 Agent 등록

# agents/data_standardization/__init__.py
from core.registry import AgentRegistry
from core.base_agent import AgentMetadata
from .agent import DataStandardizationAgent

# 메타데이터 정의
metadata = AgentMetadata(
    name="data_standardization",
    version="1.0.0",
    domain="data",
    description="데이터 스키마 표준화 Agent",
    author="Platform Team",
    created_at="2026-02-02",
    tags=["data", "standardization", "schema"]
)

# Agent 생성 및 등록
agent = DataStandardizationAgent(metadata)
AgentRegistry.get_instance().register(agent)

5.3 사용 예시

# 플랫폼 초기화
from core.orchestrator import AgentOrchestrator
import agents.data_standardization  # Agent 자동 등록

# Orchestrator 생성
orchestrator = AgentOrchestrator()

# Agent 실행
result = orchestrator.run(
    agent_name="data_standardization",
    input_data={
        'task': 'standardization',
        'data': {
            '환자번호': '12345',
            '진료일자': '2026-01-15',
            'Age': '45'
        }
    }
)

# 결과
{
    'result': {
        'fields': [
            {'name': 'patient_id', 'type': 'string', 'description': '환자 고유 번호'},
            {'name': 'visit_date', 'type': 'date', 'description': '진료 방문 일자'},
            {'name': 'age', 'type': 'int', 'description': '환자 나이'}
        ]
    },
    'confidence': 1.0,
    'metadata': {
        'agent': 'data_standardization',
        'version': '1.0.0',
        'execution_time': 2.3,
        'timestamp': 1738485600.0
    }
}

# 메트릭 확인
agent = orchestrator.registry.get("data_standardization")
metrics = agent.get_metrics()
print(metrics)
# {'name': 'data_standardization', 'executions': 1, 'success_rate': 1.0, 'avg_time': 2.3, 'error_count': 0}

6 핵심 설계 결정 요약

6.1 BaseAgent 인터페이스

  1. 표준 입출력: Dict[str, Any] (확장성)
  2. Template Method: execute()는 final, process()만 구현
  3. Hook Methods: _pre_process(), _post_process() 선택적 커스터마이징
  4. 자동 메트릭: 실행 시간, 성공률, 에러 추적
  5. 평가 인터페이스: evaluate() 구현 필수

6.2 AgentRegistry

  1. Singleton 패턴: 전역 Registry
  2. Dependency Inversion: Agent가 스스로 등록
  3. Core 독립성: Core → Agents 의존 없음

6.3 Orchestrator

  1. 단일 실행: run()
  2. 체이닝: chain() (순차)
  3. 병렬 실행: parallel()
  4. 조건부 실행: conditional_chain()

6.4 다음 단계

이 글에서 인터페이스 설계를 완료했다. 다음 글에서는:

  • 5번 글: 데이터 표준화 계층 (프롬프트, 벡터 데이터, 메타데이터 관리)
  • 6번 글: 플랫폼 운영 (CI/CD, 모니터링, 배포 전략)

6.5 참고문헌

Design Patterns: - Gamma, E., et al. (1994). “Design Patterns: Elements of Reusable Object-Oriented Software.” Addison-Wesley. - Freeman, E., & Freeman, E. (2004). “Head First Design Patterns.” O’Reilly.

Software Architecture: - Martin, R. C. (2017). “Clean Architecture.” Prentice Hall. - Evans, E. (2003). “Domain-Driven Design.” Addison-Wesley.

Python Best Practices: - Ramalho, L. (2021). “Fluent Python.” O’Reilly. - Martin, R. C. (2008). “Clean Code.” Prentice Hall.

역할: Agent 생명주기 관리, 표준 인터페이스 정의

core/
├── agent/
│   ├── base_agent.py          # BaseAgent 추상 클래스
│   ├── lifecycle.py           # Agent 초기화, 종료
│   └── registry.py            # Agent 등록 및 조회
├── orchestrator.py            # Agent 실행 관리
├── monitoring.py              # 메트릭 수집
└── evaluation.py              # 공통 평가 프레임워크

6.5.1 BaseAgent 설계 (2번 글 확장)

2번 글에서 제시한 BaseAgent를 더 구체화한다:

# core/agent/base_agent.py
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
import time
from dataclasses import dataclass

@dataclass
class AgentMetadata:
    """Agent 메타데이터"""
    name: str
    version: str
    domain: str
    description: str
    author: str
    created_at: str

class BaseAgent(ABC):
    """모든 Agent가 상속해야 하는 기본 클래스
    
    설계 원칙:
    1. 표준 입출력: Dict[str, Any] (확장성)
    2. Template Method Pattern: execute()는 final
    3. Hook Methods: _pre_process, _post_process
    """
    
    def __init__(self, metadata: AgentMetadata):
        self.metadata = metadata
        self.execution_count = 0
        self.total_time = 0.0
        self.errors = []
    
    @abstractmethod
    def process(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """핵심 비즈니스 로직
        
        Args:
            input: 표준화된 입력 딕셔너리
                - 'task': 작업 유형
                - 'data': 실제 데이터
                - 'config': 실행 옵션
        
        Returns:
            출력 딕셔너리
                - 'result': 실행 결과
                - 'metadata': 실행 정보
                - 'confidence': 신뢰도 (0-1)
        """
        pass
    
    @abstractmethod
    def evaluate(self, ground_truth: Any, prediction: Any) -> float:
        """성능 평가
        
        각 Agent는 도메인에 맞는 평가 메트릭 구현
        예: 데이터 표준화 - 스키마 매칭 정확도
            코드 분석 - AST 파싱 성공률
        """
        pass
    
    def execute(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """Agent 실행 (Template Method - 수정 불가)
        
        실행 흐름:
        1. 입력 검증
        2. 전처리 (hook)
        3. 실제 process() 호출
        4. 후처리 (hook)
        5. 메트릭 수집
        """
        # 1. 입력 검증
        self._validate_input(input)
        
        # 2. 전처리 (선택적 override)
        input = self._pre_process(input)
        
        # 3. 실행 시간 측정
        start_time = time.time()
        
        try:
            # 4. 핵심 로직 실행
            result = self.process(input)
            
            # 5. 후처리
            result = self._post_process(result)
            
            # 6. 성공 메트릭
            elapsed = time.time() - start_time
            self.execution_count += 1
            self.total_time += elapsed
            
            # 7. 메타데이터 추가
            result['metadata'] = {
                'agent': self.metadata.name,
                'version': self.metadata.version,
                'execution_time': elapsed,
                'timestamp': time.time()
            }
            
            return result
            
        except Exception as e:
            self.errors.append(str(e))
            raise
    
    def _validate_input(self, input: Dict[str, Any]):
        """입력 검증 (공통 규칙)"""
        if not isinstance(input, dict):
            raise ValueError("Input must be a dictionary")
        if 'task' not in input:
            raise ValueError("Input must contain 'task' field")
    
    def _pre_process(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """전처리 hook (선택적 override)"""
        return input
    
    def _post_process(self, output: Dict[str, Any]) -> Dict[str, Any]:
        """후처리 hook (선택적 override)"""
        return output
    
    def get_metrics(self) -> Dict[str, Any]:
        """Agent 성능 메트릭"""
        return {
            'name': self.metadata.name,
            'executions': self.execution_count,
            'avg_time': self.total_time / max(self.execution_count, 1),
            'error_rate': len(self.errors) / max(self.execution_count, 1)
        }

6.5.2 Agent Registry (Dependency Inversion 구현)

# core/agent/registry.py
from typing import Dict, Type
from .base_agent import BaseAgent

class AgentRegistry:
    """Agent 등록 및 조회
    
    Dependency Inversion 구현:
    - Core는 구체적인 Agent를 모름
    - Agent가 스스로 Registry에 등록
    """
    
    _instance = None
    _agents: Dict[str, BaseAgent] = {}
    
    @classmethod
    def get_instance(cls):
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance
    
    def register(self, agent: BaseAgent):
        """Agent 등록"""
        if agent.metadata.name in self._agents:
            raise ValueError(f"Agent {agent.metadata.name} already registered")
        
        self._agents[agent.metadata.name] = agent
        print(f"✓ Registered: {agent.metadata.name} v{agent.metadata.version}")
    
    def get(self, name: str) -> BaseAgent:
        """Agent 조회"""
        if name not in self._agents:
            raise ValueError(f"Agent {name} not found")
        return self._agents[name]
    
    def list_all(self) -> Dict[str, BaseAgent]:
        """모든 Agent 목록"""
        return self._agents.copy()

6.5.3 Orchestrator (Agent 실행 관리)

# core/orchestrator.py
from typing import Dict, List, Any
from .agent.registry import AgentRegistry

class AgentOrchestrator:
    """Agent 실행 및 조율
    
    기능:
    - Agent 실행
    - Agent 체이닝 (순차 실행)
    - 병렬 실행 (선택)
    """
    
    def __init__(self):
        self.registry = AgentRegistry.get_instance()
    
    def run(self, agent_name: str, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """단일 Agent 실행"""
        agent = self.registry.get(agent_name)
        return agent.execute(input_data)
    
    def chain(self, agent_names: List[str], input_data: Dict[str, Any]) -> Dict[str, Any]:
        """Agent 체이닝 (순차 실행)
        
        Example:
            orchestrator.chain(
                ['data_standardization', 'code_analysis', 'knowledge_qna'],
                {'task': 'analyze', 'data': schema}
            )
        """
        result = input_data
        
        for agent_name in agent_names:
            agent = self.registry.get(agent_name)
            result = agent.execute(result)
        
        return result
    
    def parallel(self, agent_names: List[str], input_data: Dict[str, Any]) -> List[Dict[str, Any]]:
        """병렬 실행 (동일 입력, 여러 Agent)"""
        results = []
        
        for agent_name in agent_names:
            agent = self.registry.get(agent_name)
            result = agent.execute(input_data)
            results.append(result)
        
        return results

6.6 shared/ 모듈: 공통 라이브러리

역할: Agent들이 재사용하는 공통 기능

shared/
├── llm/
│   ├── client.py              # LLM 클라이언트 (2번 글의 LLMClient)
│   ├── prompt_engine.py       # 프롬프트 템플릿 관리
│   └── context_manager.py     # 컨텍스트 윈도우 관리
├── validation/
│   ├── json_parser.py         # JSON 파싱 및 검증
│   └── schema_validator.py    # 스키마 검증
├── logging/
│   └── structured_logger.py   # 구조화된 로깅
├── monitoring/
│   ├── metrics.py             # 메트릭 수집
│   └── dashboard.py           # 대시보드 연동
└── document_processing/
    ├── pdf_parser.py
    ├── code_parser.py
    └── text_extractor.py

6.6.1 LLM 클라이언트 (2번 글 확장)

# shared/llm/client.py
from typing import Optional, Dict, Any
from openai import OpenAI
from anthropic import Anthropic
import time

class LLMClient:
    """통합 LLM 클라이언트
    
    2번 글의 LLMClient를 확장:
    - 토큰 카운팅
    - 비용 추적
    - 캐싱
    """
    
    def __init__(
        self,
        provider: str = "openai",
        model: str = "gpt-4",
        max_retries: int = 3,
        cache_enabled: bool = True
    ):
        self.provider = provider
        self.model = model
        self.max_retries = max_retries
        self.cache_enabled = cache_enabled
        
        # 캐시
        self._cache: Dict[str, str] = {}
        
        # 비용 추적
        self.total_tokens = 0
        self.total_cost = 0.0
        
        # 클라이언트 초기화
        if provider == "openai":
            self.client = OpenAI()
            self.cost_per_token = 0.00003  # GPT-4 가격
        elif provider == "anthropic":
            self.client = Anthropic()
            self.cost_per_token = 0.00008  # Claude 가격
    
    def generate(
        self,
        prompt: str,
        system_message: Optional[str] = None,
        temperature: float = 0.0
    ) -> str:
        """텍스트 생성 (캐싱 + 재시도)"""
        
        # 캐시 확인
        cache_key = f"{prompt}:{system_message}:{temperature}"
        if self.cache_enabled and cache_key in self._cache:
            return self._cache[cache_key]
        
        # 재시도 로직
        for attempt in range(self.max_retries):
            try:
                if self.provider == "openai":
                    response = self._call_openai(prompt, system_message, temperature)
                elif self.provider == "anthropic":
                    response = self._call_anthropic(prompt, system_message, temperature)
                
                # 캐시 저장
                if self.cache_enabled:
                    self._cache[cache_key] = response
                
                return response
                
            except Exception as e:
                if attempt < self.max_retries - 1:
                    wait_time = 2 ** attempt
                    time.sleep(wait_time)
                else:
                    raise
    
    def _call_openai(self, prompt: str, system_message: Optional[str], temperature: float) -> str:
        """OpenAI API 호출"""
        messages = []
        if system_message:
            messages.append({"role": "system", "content": system_message})
        messages.append({"role": "user", "content": prompt})
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature
        )
        
        # 토큰 및 비용 추적
        tokens = response.usage.total_tokens
        self.total_tokens += tokens
        self.total_cost += tokens * self.cost_per_token
        
        return response.choices[0].message.content
    
    def get_stats(self) -> Dict[str, Any]:
        """사용 통계"""
        return {
            'total_tokens': self.total_tokens,
            'total_cost': self.total_cost,
            'cache_size': len(self._cache)
        }

6.7 agents/ 모듈: 도메인 Agent

역할: 실제 비즈니스 로직 구현

agents/
├── data_standardization/
│   ├── __init__.py            # Agent 자동 등록
│   ├── agent.py               # DataStandardizationAgent
│   ├── manifest.yaml          # Agent 메타데이터
│   ├── prompts/
│   │   ├── system.txt
│   │   └── user_template.txt
│   └── tests/
│       └── test_agent.py
├── code_analysis/
│   ├── __init__.py
│   ├── agent.py
│   ├── manifest.yaml
│   ├── parsers/               # Agent 전용 모듈
│   │   ├── python_parser.py
│   │   └── java_parser.py
│   └── tests/
└── knowledge_qna/
    ├── __init__.py
    ├── agent.py
    ├── manifest.yaml
    ├── rag_pipeline.py        # Agent 전용 RAG
    └── tests/

6.7.1 Agent 구현 예시

# agents/data_standardization/agent.py
from core.agent import BaseAgent, AgentMetadata
from shared.llm import LLMClient
from typing import Dict, Any

class DataStandardizationAgent(BaseAgent):
    """데이터 스키마 표준화 Agent"""
    
    def __init__(self):
        metadata = AgentMetadata(
            name="data_standardization",
            version="1.0.0",
            domain="data",
            description="데이터 스키마 표준화",
            author="Platform Team",
            created_at="2026-01-28"
        )
        super().__init__(metadata)
        
        # Agent 전용 LLM 클라이언트
        self.llm = LLMClient(provider="openai", model="gpt-4")
    
    def process(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """표준화 로직"""
        schema = input['data']
        
        # LLM 호출
        prompt = f"다음 스키마를 표준화하세요: {schema}"
        standardized = self.llm.generate(prompt)
        
        return {
            'result': standardized,
            'confidence': 0.95
        }
    
    def evaluate(self, ground_truth: Any, prediction: Any) -> float:
        """정확도 평가"""
        # 스키마 매칭 로직
        return 0.9

# agents/data_standardization/__init__.py
from core.agent.registry import AgentRegistry
from .agent import DataStandardizationAgent

# Agent 자동 등록 (Dependency Inversion)
agent = DataStandardizationAgent()
AgentRegistry.get_instance().register(agent)

6.7.2 Agent Manifest

# agents/data_standardization/manifest.yaml
name: data_standardization_agent
version: "1.0.0"
domain: data_standardization

description: |
  데이터 스키마를 표준화하여 일관된 형식으로 변환

inputs:
  - name: raw_schema
    type: dict
    required: true

outputs:
  - name: standardized_schema
    type: dict

dependencies:
  - core.agent>=1.0.0
  - shared.llm>=1.0.0
  - shared.validation>=1.0.0

resources:
  memory: 512MB
  cpu: 0.5

monitoring:
  metrics:
    - name: accuracy
      type: gauge
    - name: processing_time
      type: histogram

Subscribe

Enjoy this blog? Get notified of new posts by email: