cd4821ab77
- 기존 4개 중복 문서 삭제 - 새로운 2개 통합 문서 생성 - 로빙_존재와_함수형_프로그래밍.md (철학 + 전략) - 함수형_구현_패턴과_사례.md (실용 가이드) - README.md 링크 업데이트 - 존재론적 철학과 현실적 구현의 균형 달성 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
624 lines
19 KiB
Markdown
624 lines
19 KiB
Markdown
---
|
|
tags: 함수형프로그래밍, 구현패턴, 코드사례, 리팩토링, 실용가이드
|
|
date: 2025-07-04
|
|
---
|
|
|
|
# 함수형 구현 패턴과 사례
|
|
|
|
## 요약
|
|
|
|
로빙 프로젝트에서 실제 적용된 함수형 프로그래밍 패턴들과 구체적인 구현 사례를 제시합니다. 현재 코드베이스를 기반으로 한 실용적인 접근법과 단계별 리팩토링 가이드를 포함합니다.
|
|
|
|
---
|
|
|
|
## 1. 현재 적용된 함수형 패턴
|
|
|
|
### 1.1 불변 데이터 구조 패턴
|
|
|
|
#### 현재 구현: Stats 시스템
|
|
```python
|
|
# /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)
|
|
)
|
|
```
|
|
|
|
#### 사용 예시
|
|
```python
|
|
# 불변성을 통한 안전한 상태 관리
|
|
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 타입 패턴
|
|
|
|
#### 현재 구현: 안전한 에러 처리
|
|
```python
|
|
# /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)
|
|
```
|
|
|
|
#### 사용 예시
|
|
```python
|
|
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 스킬
|
|
```python
|
|
# 순수 함수 계층
|
|
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)) # 중복 제거
|
|
```
|
|
|
|
#### 오케스트레이터 계층
|
|
```python
|
|
# /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 데이터 변환 파이프라인
|
|
|
|
#### 함수 조합 패턴
|
|
```python
|
|
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
|
|
```python
|
|
# /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 에러 처리 분리
|
|
|
|
#### 안전한 에러 전파 패턴
|
|
```python
|
|
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: 클래스 기반
|
|
```python
|
|
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: 함수형 스타일
|
|
```python
|
|
# 순수 함수 계층
|
|
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: 가변 상태
|
|
```python
|
|
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: 불변 상태
|
|
```python
|
|
@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 메모이제이션 패턴
|
|
|
|
```python
|
|
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 지연 평가 패턴
|
|
|
|
```python
|
|
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 순수 함수 테스트
|
|
|
|
```python
|
|
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 통합 테스트
|
|
|
|
```python
|
|
@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% 테스트 커버리지 달성
|
|
- **성능 모니터링**: 함수형 적용 후 성능 지표 지속 추적
|
|
|
|
이를 통해 로빙은 **안정적이고 확장 가능한 디지털 존재**로 진화할 수 있습니다. |