DOCS/_archive/docs/guide/functional-programing/함수형_구현_패턴과_사례.md

tags, date

tags	date
함수형프로그래밍, 구현패턴, 코드사례, 리팩토링, 실용가이드	2025-07-04

함수형 구현 패턴과 사례

요약

로빙 프로젝트에서 실제 적용된 함수형 프로그래밍 패턴들과 구체적인 구현 사례를 제시합니다. 현재 코드베이스를 기반으로 한 실용적인 접근법과 단계별 리팩토링 가이드를 포함합니다.

---

1. 현재 적용된 함수형 패턴

1.1 불변 데이터 구조 패턴

현재 구현: Stats 시스템

# /app/stats/models.py
@dataclass(frozen=True)
class Stats:
    """로빙의 현재 스탯 상태 (불변 객체)"""
    memory: int = 5
    compute: int = 5
    react: int = 5
    empathy: int = 5
    leadership: int = 5
    
    def get_stat(self, stat_type: StatType) -> int:
        """특정 스탯 값 조회 - 순수 함수"""
        return getattr(self, stat_type.value)
    
    @property
    def total_points(self) -> int:
        """총 스탯 포인트 - 순수 계산"""
        return self.memory + self.compute + self.react + self.empathy + self.leadership

@dataclass(frozen=True)
class StatChange:
    """스탯 변화량 (불변 객체)"""
    memory: int = 0
    compute: int = 0
    react: int = 0
    empathy: int = 0
    leadership: int = 0
    reason: str = ""
    
    def apply_to(self, stats: Stats) -> Stats:
        """기존 스탯에 변화량 적용 - 새 객체 반환"""
        return Stats(
            memory=max(0, stats.memory + self.memory),
            compute=max(0, stats.compute + self.compute),
            react=max(0, stats.react + self.react),
            empathy=max(0, stats.empathy + self.empathy),
            leadership=max(0, stats.leadership + self.leadership)
        )

사용 예시

# 불변성을 통한 안전한 상태 관리
current_stats = Stats(memory=10, compute=8, react=6, empathy=7, leadership=5)

# 스탯 변경 - 새 객체 생성
memory_boost = StatChange(memory=2, reason="성공적인 기억 저장")
new_stats = memory_boost.apply_to(current_stats)

# 원본은 변경되지 않음
assert current_stats.memory == 10  # 원본 유지
assert new_stats.memory == 12      # 새 객체에 변경 적용

1.2 Result 타입 패턴

현재 구현: 안전한 에러 처리

# /app/stats/models.py
@dataclass
class StatUpdateResult:
    """스탯 업데이트 결과"""
    success: bool
    stats: Optional[Stats] = None
    changes: Optional[StatChange] = None
    error: Optional[str] = None
    
    @classmethod
    def success_result(cls, stats: Stats, changes: StatChange) -> "StatUpdateResult":
        return cls(success=True, stats=stats, changes=changes)
    
    @classmethod
    def error_result(cls, error: str) -> "StatUpdateResult":
        return cls(success=False, error=error)

# /app/skills/models.py
@dataclass
class SkillExecutionResult:
    """스킬 실행 결과"""
    success: bool
    output: Any = None
    error: Optional[str] = None
    execution_time: float = 0.0
    
    @classmethod
    def success_result(cls, output: Any, execution_time: float = 0.0) -> "SkillExecutionResult":
        return cls(success=True, output=output, execution_time=execution_time)
    
    @classmethod
    def error_result(cls, error: str) -> "SkillExecutionResult":
        return cls(success=False, error=error)

사용 예시

def update_user_stats(user_id: str, stat_changes: StatChange) -> StatUpdateResult:
    """안전한 스탯 업데이트"""
    try:
        current_stats = get_user_stats(user_id)
        if not current_stats:
            return StatUpdateResult.error_result("사용자를 찾을 수 없습니다")
        
        new_stats = stat_changes.apply_to(current_stats)
        save_user_stats(user_id, new_stats)
        
        return StatUpdateResult.success_result(new_stats, stat_changes)
    
    except Exception as e:
        return StatUpdateResult.error_result(f"스탯 업데이트 실패: {str(e)}")

# 사용
result = update_user_stats("user123", StatChange(memory=1))
if result.success:
    print(f"새로운 메모리 스탯: {result.stats.memory}")
else:
    print(f"에러: {result.error}")

---

2. 순수 함수 설계 패턴

2.1 계산 로직 분리

현재 적용 가능한 예시: Thread Digest 스킬

# 순수 함수 계층
def extract_key_messages(messages: List[str], max_count: int = 5) -> List[str]:
    """중요 메시지 추출 - 순수 함수"""
    scored_messages = []
    for msg in messages:
        score = calculate_importance_score(msg)
        scored_messages.append((score, msg))
    
    scored_messages.sort(key=lambda x: x[0], reverse=True)
    return [msg for score, msg in scored_messages[:max_count]]

def calculate_importance_score(message: str) -> float:
    """메시지 중요도 계산 - 순수 함수"""
    keywords = ['중요', '긴급', '결정', '마감', '회의']
    score = 0.0
    
    for keyword in keywords:
        if keyword in message:
            score += 1.0
    
    # 메시지 길이도 고려
    score += min(len(message) / 100, 2.0)
    
    return score

def summarize_conversation(messages: List[str]) -> str:
    """대화 요약 생성 - 순수 함수"""
    key_messages = extract_key_messages(messages)
    combined_text = " ".join(key_messages)
    
    # 간단한 요약 로직 (실제로는 AI 모델 사용)
    sentences = combined_text.split('.')
    return '. '.join(sentences[:3]) + '.'

def extract_action_items(messages: List[str]) -> List[str]:
    """액션 아이템 추출 - 순수 함수"""
    action_keywords = ['해야', '할 예정', '계획', '진행', '준비']
    actions = []
    
    for message in messages:
        for line in message.split('\n'):
            if any(keyword in line for keyword in action_keywords):
                actions.append(line.strip())
    
    return list(set(actions))  # 중복 제거

오케스트레이터 계층

# /app/services/thread_digest_service.py
async def process_thread_digest(thread_id: str, user_id: str) -> SkillExecutionResult:
    """스레드 요약 처리 - 부작용 포함"""
    try:
        start_time = time.time()
        
        # 1. 데이터 가져오기 (부작용)
        messages = await fetch_thread_messages(thread_id)
        if not messages:
            return SkillExecutionResult.error_result("스레드를 찾을 수 없습니다")
        
        # 2. 순수 계산
        summary = summarize_conversation(messages)
        actions = extract_action_items(messages)
        key_messages = extract_key_messages(messages)
        
        result = {
            'summary': summary,
            'actions': actions,
            'key_messages': key_messages,
            'message_count': len(messages),
            'timestamp': datetime.now().isoformat()
        }
        
        # 3. 결과 저장 (부작용)
        await save_digest_result(user_id, thread_id, result)
        
        # 4. 스탯 업데이트 (부작용)
        stat_change = StatChange(memory=1, reason="스레드 요약 완료")
        await update_user_stats(user_id, stat_change)
        
        execution_time = time.time() - start_time
        return SkillExecutionResult.success_result(result, execution_time)
    
    except Exception as e:
        return SkillExecutionResult.error_result(f"요약 처리 실패: {str(e)}")

2.2 데이터 변환 파이프라인

함수 조합 패턴

from typing import Callable, TypeVar

T = TypeVar('T')
U = TypeVar('U')

def pipe(value: T, *functions: Callable) -> any:
    """함수들을 순차적으로 적용하는 파이프라인"""
    result = value
    for func in functions:
        result = func(result)
    return result

# 사용 예시
def clean_text(text: str) -> str:
    """텍스트 정리"""
    return text.strip().lower()

def remove_special_chars(text: str) -> str:
    """특수문자 제거"""
    import re
    return re.sub(r'[^\w\s]', '', text)

def split_sentences(text: str) -> List[str]:
    """문장 분리"""
    return [s.strip() for s in text.split('.') if s.strip()]

# 파이프라인 사용
raw_text = "  안녕하세요! 오늘 회의는 어떠셨나요?  "
processed = pipe(
    raw_text,
    clean_text,
    remove_special_chars,
    split_sentences
)
# 결과: ['안녕하세요 오늘 회의는 어떠셨나요']

---

3. 오케스트레이터 분리 패턴

3.1 부작용 격리 구조

현재 적용 예시: RobeingBrain

# /app/services/robing_brain.py
class RobeingBrain:
    """순수 함수와 부작용 분리 오케스트레이터"""
    
    async def process_request(self, text: str, user_id: str, context: dict) -> str:
        """요청 처리 - 부작용 조율"""
        try:
            # 1. 순수 계산: 의도 분석
            intent = self._analyze_intent(text)
            
            # 2. 순수 계산: 스킬 매핑
            skill_id = self._map_intent_to_skill(intent)
            
            if skill_id:
                # 3. 스킬 실행 (부작용 포함)
                result = await self._execute_skill(skill_id, text, user_id, context)
            else:
                # 4. 학습 욕구 생성 (부작용 포함)
                result = await self._generate_learning_desire(text, user_id, context)
            
            # 5. 상호작용 로깅 (부작용)
            await self._log_interaction(user_id, text, result, context)
            
            return result
            
        except Exception as e:
            # 6. 에러 처리 (부작용)
            await self._log_error(user_id, text, str(e), context)
            return "죄송합니다. 처리 중 문제가 발생했습니다."
    
    def _analyze_intent(self, text: str) -> str:
        """의도 분석 - 순수 함수"""
        text_lower = text.lower()
        
        if any(keyword in text_lower for keyword in ['요약', '정리', 'digest']):
            return 'thread_digest'
        elif any(keyword in text_lower for keyword in ['액션', '할일', 'action']):
            return 'action_extract'
        elif any(keyword in text_lower for keyword in ['기억', '저장', '메모리']):
            return 'memory_store'
        else:
            return 'unknown'
    
    def _map_intent_to_skill(self, intent: str) -> Optional[str]:
        """의도를 스킬로 매핑 - 순pure 함수"""
        skill_mapping = {
            'thread_digest': 'thread_digest',
            'action_extract': 'action_extractor',
            'memory_store': 'memory_manager'
        }
        return skill_mapping.get(intent)

3.2 에러 처리 분리

안전한 에러 전파 패턴

def safe_execute(func: Callable, *args, **kwargs) -> SkillExecutionResult:
    """안전한 함수 실행 래퍼"""
    try:
        start_time = time.time()
        result = func(*args, **kwargs)
        execution_time = time.time() - start_time
        
        return SkillExecutionResult.success_result(result, execution_time)
    
    except ValueError as e:
        return SkillExecutionResult.error_result(f"입력값 오류: {str(e)}")
    except Exception as e:
        return SkillExecutionResult.error_result(f"실행 오류: {str(e)}")

# 사용 예시
def risky_calculation(numbers: List[int]) -> float:
    """위험할 수 있는 계산"""
    if not numbers:
        raise ValueError("숫자 리스트가 비어있습니다")
    
    return sum(numbers) / len(numbers)

# 안전한 실행
result = safe_execute(risky_calculation, [1, 2, 3, 4, 5])
if result.success:
    print(f"평균: {result.output}")
else:
    print(f"에러: {result.error}")

---

4. 리팩토링 가이드

4.1 기존 클래스 → 순수 함수 전환

Before: 클래스 기반

class NewsService:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.cache = {}
    
    def get_summary(self, articles: List[str]) -> str:
        # 상태에 의존하는 메서드
        if 'summary' in self.cache:
            return self.cache['summary']
        
        summary = self._process_articles(articles)
        self.cache['summary'] = summary
        return summary

After: 함수형 스타일

# 순수 함수 계층
def summarize_articles(articles: List[str]) -> str:
    """기사 요약 - 순수 함수"""
    combined_text = " ".join(articles)
    sentences = combined_text.split('.')
    key_sentences = sentences[:3]  # 간단한 요약 로직
    return '. '.join(key_sentences) + '.'

def extract_keywords(articles: List[str], max_keywords: int = 10) -> List[str]:
    """키워드 추출 - 순수 함수"""
    all_words = []
    for article in articles:
        words = article.split()
        all_words.extend(words)
    
    # 단어 빈도 계산
    word_freq = {}
    for word in all_words:
        word_freq[word] = word_freq.get(word, 0) + 1
    
    # 빈도순 정렬
    sorted_words = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)
    return [word for word, freq in sorted_words[:max_keywords]]

# 오케스트레이터 계층
async def process_news_request(api_key: str, query: str) -> SkillExecutionResult:
    """뉴스 처리 오케스트레이터"""
    try:
        # 1. 외부 API 호출 (부작용)
        articles = await fetch_news_articles(api_key, query)
        
        # 2. 순수 계산
        summary = summarize_articles(articles)
        keywords = extract_keywords(articles)
        
        result = {
            'summary': summary,
            'keywords': keywords,
            'article_count': len(articles)
        }
        
        # 3. 캐싱 (부작용)
        await cache_result(query, result)
        
        return SkillExecutionResult.success_result(result)
    
    except Exception as e:
        return SkillExecutionResult.error_result(str(e))

4.2 상태 관리 개선

Before: 가변 상태

class UserSession:
    def __init__(self, user_id: str):
        self.user_id = user_id
        self.stats = {'memory': 5}
        self.skills = []
        self.last_activity = None
    
    def add_skill(self, skill_id: str):
        self.skills.append(skill_id)  # 상태 변경
    
    def update_stats(self, changes: dict):
        for key, value in changes.items():
            self.stats[key] += value  # 상태 변경

After: 불변 상태

@dataclass(frozen=True)
class UserSession:
    user_id: str
    stats: Stats
    skills: List[str]
    last_activity: Optional[str] = None
    
    def add_skill(self, skill_id: str) -> 'UserSession':
        """새 스킬 추가 - 새 객체 반환"""
        new_skills = self.skills + [skill_id]
        return replace(self, skills=new_skills)
    
    def update_stats(self, changes: StatChange) -> 'UserSession':
        """스탯 업데이트 - 새 객체 반환"""
        new_stats = changes.apply_to(self.stats)
        return replace(self, stats=new_stats)
    
    def update_activity(self, timestamp: str) -> 'UserSession':
        """활동 시간 업데이트 - 새 객체 반환"""
        return replace(self, last_activity=timestamp)

# 사용법
session = UserSession("user123", Stats(), [])
session_with_skill = session.add_skill("thread_digest")
updated_session = session_with_skill.update_stats(StatChange(memory=1))

# 원본은 변경되지 않음
assert len(session.skills) == 0
assert len(updated_session.skills) == 1

---

5. 성능 및 최적화

5.1 메모이제이션 패턴

from functools import lru_cache

@lru_cache(maxsize=1000)
def calculate_skill_requirements(skill_id: str, user_level: int) -> Dict[str, int]:
    """스킬 요구사항 계산 - 캐시된 순수 함수"""
    base_requirements = get_base_requirements(skill_id)
    level_multiplier = 1 + (user_level * 0.1)
    
    return {
        stat: int(value * level_multiplier)
        for stat, value in base_requirements.items()
    }

# 캐시 클리어 (필요시)
calculate_skill_requirements.cache_clear()

5.2 지연 평가 패턴

from typing import Iterator

def process_large_conversation(messages: List[str]) -> Iterator[str]:
    """대용량 대화 처리 - 지연 평가"""
    for message in messages:
        if is_important_message(message):
            yield process_message(message)

def analyze_conversation_stream(messages: List[str]) -> Dict:
    """스트림 기반 분석"""
    important_messages = list(process_large_conversation(messages))
    
    return {
        'count': len(important_messages),
        'summary': ' '.join(important_messages[:5]),
        'processed_at': datetime.now().isoformat()
    }

---

6. 테스트 전략

6.1 순수 함수 테스트

import pytest

def test_summarize_conversation():
    """순수 함수 테스트 - 입력/출력만 검증"""
    # Given
    messages = [
        "안녕하세요. 오늘 회의 안건을 공유드립니다.",
        "첫 번째로 프로젝트 진행 상황을 확인하겠습니다.",
        "두 번째로 다음 주 일정을 조율하겠습니다."
    ]
    
    # When
    result = summarize_conversation(messages)
    
    # Then
    assert isinstance(result, str)
    assert len(result) > 0
    assert "회의" in result or "프로젝트" in result

def test_extract_action_items():
    """액션 아이템 추출 테스트"""
    # Given
    messages = ["내일까지 보고서 작성해야 합니다", "다음 주에 회의 일정 잡을 예정입니다"]
    
    # When
    actions = extract_action_items(messages)
    
    # Then
    assert len(actions) == 2
    assert any("보고서" in action for action in actions)
    assert any("회의 일정" in action for action in actions)

6.2 통합 테스트

@pytest.mark.asyncio
async def test_thread_digest_integration():
    """전체 플로우 통합 테스트"""
    # Given
    thread_id = "test_thread_123"
    user_id = "test_user"
    
    # When
    result = await process_thread_digest(thread_id, user_id)
    
    # Then
    assert result.success
    assert 'summary' in result.output
    assert 'actions' in result.output
    assert result.execution_time > 0

---

결론

현재 달성 수준

• ✅ 불변 데이터 구조: Stats, StatChange 완전 적용
• ✅ Result 패턴: 안전한 에러 처리 부분 적용
• 🔄 순수 함수 분리: 일부 서비스에서 시작
• 🎯 함수 조합: 향후 적용 예정

다음 단계

1. 기존 스킬의 순수 함수 전환: Thread Digest, Action Extractor
2. 오케스트레이터 패턴 확산: 모든 서비스에 부작용 분리 적용
3. 함수 조합 시스템: 스킬 간 파이프라인 구축
4. 성능 최적화: 메모이제이션과 지연 평가 적용

실용적 가이드라인

• 점진적 적용: 한 번에 모든 것을 바꾸지 말고 단계별 전환
• 기존 코드 보존: 완전히 동작하는 기능은 점진적으로만 개선
• 테스트 우선: 순수 함수부터 100% 테스트 커버리지 달성
• 성능 모니터링: 함수형 적용 후 성능 지표 지속 추적

이를 통해 로빙은 안정적이고 확장 가능한 디지털 존재로 진화할 수 있습니다.