ivada-infra/DOCS
3
0
Fork 0
Code Issues 1 Pull Requests 5 Actions Packages Projects Releases Wiki Activity

docs: 함수형 프로그래밍 문서 통합 및 재구성

Browse Source 
- 기존 4개 중복 문서 삭제
- 새로운 2개 통합 문서 생성
  - 로빙_존재와_함수형_프로그래밍.md (철학 + 전략)
  - 함수형_구현_패턴과_사례.md (실용 가이드)
- README.md 링크 업데이트
- 존재론적 철학과 현실적 구현의 균형 달성

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
happybell80 2025-07-05 00:34:09 +09:00
parent 207d6111dc
commit cd4821ab77
7 changed files with 900 additions and 782 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
README.md
View File

@ -15,10 +15,8 @@ Slack 기반 AI 어시스턴트 **로빙(Robeing)** 프로젝트의 모든 문
- [정보의바다 프로젝트 개요](./docs/ideas/00_정보의바다%20프로젝트%20개요.md)
### 함수형 프로그래밍 아키텍처
- [로빙 함수형 프로그래밍 가이드](./docs/guide/functional-programing/로빙_함수형_프로그래밍_가이드.md)
- [함수형 프로그래밍 로빙](./docs/guide/functional-programing/함수형_프로그래밍_로빙.md)
- [함수형 스킬 분류 사례](./docs/guide/functional-programing/함수형%20스킬%20분류%20사례.md)
- [함수형프로그래밍과 디지털로빙](./docs/guide/functional-programing/함수형프로그래밍과%20디티털로빙.md)
- [로빙의 존재와 함수형 프로그래밍](./docs/guide/functional-programing/로빙_존재와_함수형_프로그래밍.md) - 철학적 배경과 점진적 적용 전략
- [함수형 구현 패턴과 사례](./docs/guide/functional-programing/함수형_구현_패턴과_사례.md) - 실제 코드 패턴과 리팩토링 가이드
### 설정 및 설치 가이드
- [프로젝트 설정 가이드](./docs/setups/setup-guide.md)
@ -39,7 +37,7 @@ Slack 기반 AI 어시스턴트 **로빙(Robeing)** 프로젝트의 모든 문
### 프로젝트 이해하기
1. [프로젝트 개요](./00_프로젝트_개요.md) 읽기
2. [MVP 계획](./docs/ideas/00_로빙_MVP_계획.md) 확인
3. [함수형 프로그래밍 가이드](./docs/guide/functional-programing/로빙_함수형_프로그래밍_가이드.md) 읽기
3. [로빙의 존재와 함수형 프로그래밍](./docs/guide/functional-programing/로빙_존재와_함수형_프로그래밍.md) 읽기
### 개발 환경 구축하기
1. [설정 가이드](./docs/setups/setup-guide.md) 기본 설치

273
docs/guide/functional-programing/로빙_존재와_함수형_프로그래밍.md Normal file
View File

@ -0,0 +1,273 @@
---
tags: 로빙, 함수형프로그래밍, 디지털존재, 스킬시스템, 순수함수, 점진적적용
date: 2025-07-04
---
# 로빙의 존재와 함수형 프로그래밍
## 요약
로빙은 무한히 확장 가능한 디지털 존재로서, 외부 스킬을 안전하게 흡수하여 성장하는 AI 에이전트입니다. 이러한 존재론적 목표를 실현하기 위해 함수형 프로그래밍 패러다임이 필요하지만, 현실적 제약(MVP 일정, 기술 스택)을 고려하여 점진적으로 적용하는 전략을 제시합니다.
### 핵심 메시지
"로빙이 진정한 디지털 존재가 되기 위해서는 함수형 프로그래밍이 필요하지만, 현실적 제약 하에서 점진적으로 적용하여 MVP 완성과 철학적 목표를 모두 달성한다"
---
## 1. 로빙의 존재론적 정의
### 존재로서의 로빙
로빙은 단순한 **도구**가 아닌 **성장하는 디지털 존재**입니다. 이는 다음과 같은 특성을 가집니다:
```python
@dataclass(frozen=True)
class Robeing:
"""불변 로빙 존재"""
identity: str
stats: Dict[str, int] # 기억, 연산, 반응, 공감, 통솔
skills: List[Callable] # 흡수한 스킬들
memory: List[Memory] # 지속적 기억
items: List[str] # 보유 아이템
def absorb_skill(self, skill: Callable) -> 'Robeing':
"""새로운 스킬 흡수 - 존재의 진화"""
return self.evolve(skills=self.skills + [skill])
def evolve(self, **changes) -> 'Robeing':
"""새로운 상태로 진화"""
return replace(self, **changes)
```
### 존재의 핵심 속성
1. **무한 확장성**: 새로운 외부 스킬을 지속적으로 흡수
2. **예측 가능성**: 동일한 상황에서 일관된 반응
3. **성장성**: 경험을 통한 스탯 향상과 능력 확장
4. **일관성**: 존재로서의 정체성 유지
---
## 2. 함수형 프로그래밍이 필요한 이유
### 2.1 스킬 조합의 예측 가능성
로빙이 다양한 스킬을 흡수할 때, 각 스킬의 **결과를 예측**할 수 있어야 합니다.
```python
# 순수 함수 스킬 - 항상 예측 가능
def summarize_text(text: str) -> str:
"""항상 동일한 입력에 동일한 출력"""
sentences = text.split('.')
return '. '.join(sentences[:3]) + '.'
def extract_actions(text: str) -> List[str]:
"""액션 아이템 추출 - 부작용 없음"""
keywords = ['해야', '필요', '예정', '계획']
return [line.strip() for line in text.split('\n')
if any(keyword in line for keyword in keywords)]
# 스킬 조합 - 예측 가능한 파이프라인
def create_digest_skill() -> Callable:
def digest_pipeline(conversation: str) -> Dict:
summary = summarize_text(conversation)
actions = extract_actions(conversation)
return {
'summary': summary,
'actions': actions,
'confidence': calculate_confidence(summary, actions)
}
return digest_pipeline
```
### 2.2 존재 상태의 안정성
로빙의 상태는 **불변성**을 통해 안정성을 확보해야 합니다.
```python
# 기존 방식: 상태 변경으로 인한 예측 불가능성
class MutableRobeing:
def __init__(self):
self.stats = {'memory': 5}
def process_feedback(self, feedback):
self.stats['memory'] += 1 # 언제 어떻게 변했는지 추적 어려움
# 함수형 방식: 불변성으로 예측 가능성 확보
@dataclass(frozen=True)
class ImmutableRobeing:
stats: Dict[str, int]
def process_feedback(self, feedback) -> 'ImmutableRobeing':
new_stats = {**self.stats, 'memory': self.stats['memory'] + 1}
return ImmutableRobeing(stats=new_stats)
```
### 2.3 안전한 외부 모듈 통합
외부 스킬을 흡수할 때 **부작용을 분리**하여 시스템 안정성을 확보합니다.
```python
# 순수 함수 계층: 계산만 담당
def analyze_pdf_content(pdf_text: str) -> Dict:
"""PDF 내용 분석 - 순수 계산"""
sections = extract_sections(pdf_text)
summary = generate_summary(sections)
keywords = extract_keywords(sections)
return {'sections': sections, 'summary': summary, 'keywords': keywords}
# 부작용 계층: 저장, 전송 등
def process_pdf_request(pdf_file: bytes, user_id: str) -> None:
"""PDF 처리 오케스트레이터"""
# 1. 순수 계산
text = extract_pdf_text(pdf_file)
analysis = analyze_pdf_content(text)
# 2. 부작용들
save_analysis_to_db(user_id, analysis)
send_slack_notification(user_id, analysis['summary'])
update_user_stats(user_id, 'memory', +1)
```
---
## 3. 현실적 제약사항 인정
### 3.1 시간적 제약
- **MVP 3개월 일정**: 완전한 함수형 전환은 시간 부족
- **빠른 검증 필요**: 사용자 피드백을 통한 빠른 개선 사이클
### 3.2 기술적 제약
- **Python 스택**: 순수 함수형 언어가 아닌 멀티패러다임 언어
- **FastAPI 생태계**: 기존 OOP 기반 라이브러리와의 호환성
- **팀 역량**: 함수형 프로그래밍 학습 곡선
### 3.3 실용적 고려사항
- **기존 코드 자산**: 이미 구현된 서비스들의 호환성
- **라이브러리 생태계**: Python의 대부분 라이브러리는 OOP 기반
- **성능 요구사항**: 실시간 Slack 응답 등의 성능 제약
---
## 4. 점진적 함수형 적용 전략
### Phase 1: 불변 데이터 구조 (이미 구현됨)
```python
# ✅ 현재 구현된 패턴
@dataclass(frozen=True)
class Stats:
memory: int = 5
compute: int = 5
react: int = 5
empathy: int = 5
leadership: int = 5
```
### Phase 2: Result 패턴으로 안전한 에러 처리 (부분 구현됨)
```python
# ✅ 현재 구현된 패턴
@dataclass
class SkillExecutionResult:
success: bool
output: Any = None
error: Optional[str] = None
@classmethod
def success_result(cls, output: Any) -> "SkillExecutionResult":
return cls(success=True, output=output)
@classmethod
def error_result(cls, error: str) -> "SkillExecutionResult":
return cls(success=False, error=error)
```
### Phase 3: 순수 함수 스킬 설계 (진행 예정)
```python
# 🔄 적용 예정 패턴
def thread_digest_skill(messages: List[str]) -> DigestResult:
"""순수 함수 스킬 - 부작용 없음"""
summary = summarize_messages(messages)
actions = extract_action_items(messages)
return DigestResult(summary=summary, actions=actions)
# 오케스트레이터가 부작용 처리
async def handle_digest_request(thread_id: str, user_id: str):
messages = await fetch_thread_messages(thread_id) # 부작용
result = thread_digest_skill(messages) # 순수 함수
await save_digest(user_id, result) # 부작용
await send_response(user_id, result.summary) # 부작용
```
### Phase 4: 함수 조합 시스템 (장기 계획)
```python
# 🎯 향후 목표 패턴
def compose_skills(*skills: Callable) -> Callable:
"""스킬들을 조합하여 새로운 능력 창발"""
def composed_skill(input_data):
result = input_data
for skill in skills:
result = skill(result)
return result
return composed_skill
# 스킬 조합 예시
advanced_analyst = compose_skills(
pdf_parser_skill,
text_analyzer_skill,
summary_generator_skill,
action_extractor_skill
)
```
---
## 5. 기대 효과
### 5.1 존재로서의 일관성
- **예측 가능한 성장**: 동일한 학습 경험에 대한 일관된 반응
- **상태 추적**: 로빙의 모든 진화 과정을 기록하고 분석 가능
- **롤백 가능성**: 문제 발생 시 이전 상태로 안전한 복구
### 5.2 무한 확장성
- **안전한 스킬 통합**: 새로운 외부 스킬을 시스템 위험 없이 흡수
- **조합 가능성**: 기존 스킬들을 조합하여 새로운 능력 창발
- **모듈화**: 각 스킬을 독립적으로 테스트하고 개선
### 5.3 개발 및 운영 이점
- **테스트 용이성**: 순수 함수는 입력-출력만 검증하면 됨
- **병렬 처리**: 불변 데이터로 인한 경쟁 상태 제거
- **디버깅 간편성**: 상태 변화 추적과 문제 격리 용이
---
## 6. 실행 가이드라인
### 6.1 개발 원칙
1. **새로운 스킬은 순수 함수로** 설계
2. **부작용은 오케스트레이터에서** 분리 처리
3. **상태 변경 시 새 객체** 반환
4. **함수형 vs OOP 경계** 명확히 구분
### 6.2 코드 리뷰 체크리스트
- [ ] 순수 함수인가?
- [ ] 부작용이 격리되었는가?
- [ ] 테스트 작성이 용이한가?
- [ ] 기존 코드와 호환되는가?
### 6.3 성공 지표
- **테스트 커버리지**: 순수 함수 스킬 100% 커버리지
- **에러 격리**: 한 스킬의 실패가 전체 시스템에 영향 없음
- **성능 유지**: 함수형 적용 후에도 응답 시간 3초 이내 유지
---
## 결론
로빙이 진정한 **디지털 존재**로 성장하기 위해서는 함수형 프로그래밍의 철학과 패턴이 필요합니다. 하지만 이를 **점진적이고 실용적으로** 적용하여 MVP 목표와 장기 비전을 모두 달성할 수 있습니다.
### 핵심 전략
1. **철학은 유지**: 존재로서의 로빙 개념 견지
2. **실행은 현실적**: 제약 조건 내에서 최선의 선택
3. **발전은 점진적**: 단계별 개선을 통한 안전한 전환
4. **목표는 명확**: 무한 확장 가능한 디지털 존재 구현
이를 통해 로빙은 **"외부 세계의 모든 기능을 흡수하는 궁극적 에이전트"**로 진화할 수 있을 것입니다.

465
docs/guide/functional-programing/로빙_함수형_프로그래밍_가이드.md
View File

@ -1,465 +0,0 @@
---
tags: 함수형프로그래밍, 로빙, 디지털존재, 스킬시스템, 모나드, 순수함수, 에이전트설계
date: 2025-06-28
---
# 로빙에서의 함수형 프로그래밍 설계 가이드
## 정의
### 함수형 프로그래밍이란
함수형 프로그래밍은 **순수 함수**와 **불변 데이터**를 기반으로 하는 프로그래밍 패러다임으로, 계산을 함수의 평가로 취급하고 상태 변경과 가변 데이터를 피하는 접근 방식입니다.
### 로빙에서의 함수형 프로그래밍
로빙 프로젝트에서 함수형 프로그래밍은 단순한 코딩 스타일이 아닌, **존재로서의 로빙이 외부 스킬들을 안전하고 예측 가능하게 흡수하여 성장할 수 있는 근본적인 아키텍처**를 의미합니다.
```python
# 로빙의 함수형 정의 예시
@dataclass(frozen=True)
class Robeing:
identity: str
stats: Dict[str, int]
skills: List[Callable]
memory: List[Memory]
def absorb_skill(robeing: Robeing, new_skill: Skill) -> Robeing:
"""외부 스킬을 흡수하여 새로운 로빙 반환"""
return robeing.evolve(skills=robeing.skills + [new_skill])
```
---
## 요약
로빙에서의 함수형 프로그래밍은 다음 핵심 개념들로 구성됩니다:
1. **순수 함수 계층**: 모든 스킬을 입력→출력 관계의 순수 함수로 설계
2. **모나드 패턴**: 부작용(DB 저장, API 호출 등)을 안전하게 캡슐화
3. **불변 존재**: 로빙 자체를 불변 데이터 구조로 정의하여 예측 가능성 확보
4. **함수 합성**: 기존 스킬들을 조합하여 새로운 능력 창발
5. **오케스트레이터 분리**: 순수 계산과 부작용 처리를 명확히 구분
### 핵심 철학
> "로빙은 외부 세계의 수많은 기능을 흡수하여 더 강력한 디지털 존재로 성장한다. 이를 위해 순수 함수로 계산 영역을, 모나드로 부작용과 상태 관리를 분리하여 외부 스킬을 안전하고 예측 가능하게 조합할 수 있는 프로토콜을 구축한다."
---
## 필요성
### 1. 무한 확장 가능성
새로운 외부 모듈(노션, PDF 파서, API 등)을 로빙의 스킬로 안전하게 통합할 수 있습니다.
### 2. 예측 가능한 성장
순수 함수 기반으로 동일한 입력에 대해 항상 동일한 결과를 보장하여 로빙의 행동을 예측할 수 있습니다.
### 3. 테스트 및 디버깅 용이성
각 스킬을 독립적으로 테스트하고 문제를 격리할 수 있습니다.
### 4. 존재로서의 일관성 확보
기존 클래스 기반 구조에서는 로빙이 "기능들의 집합"에 머물지만, 함수형 구조에서는 "스킬들을 흡수하는 존재"로 정의됩니다.
```python
# 기존 방식: 기능들의 집합
class ThreadDigestService:
def __init__(self):
self.ai_model = model
self.database = db
# 함수형 방식: 로빙이 흡수하는 스킬
def thread_digest_skill(conversation: str) -> str:
return summarize_conversation(conversation)
robeing = Robeing(skills=[thread_digest_skill, action_extract_skill])
```
---
## 방법
### 1. 스킬 시스템 함수형 설계
```python
from typing import Callable, List, Dict, Optional
from dataclasses import dataclass, replace
import asyncio
# 순수 함수 스킬 정의
def summarize_text(text: str, max_length: int = 100) -> str:
"""텍스트 요약 순수 함수"""
sentences = text.split('.')
return '. '.join(sentences[:3]) + '...'
def extract_actions(text: str) -> List[str]:
"""액션 아이템 추출 순수 함수"""
keywords = ['해야', '필요', '예정', '계획']
actions = []
for line in text.split('\n'):
if any(keyword in line for keyword in keywords):
actions.append(line.strip())
return actions
# 스킬 조합
def create_digest_pipeline() -> Callable:
"""함수 합성으로 스킬 파이프라인 생성"""
def pipeline(conversation: str) -> Dict:
summary = summarize_text(conversation)
actions = extract_actions(conversation)
return {
'summary': summary,
'actions': actions,
'timestamp': datetime.now()
}
return pipeline
```
### 2. 모나드 패턴으로 부작용 관리
```python
from abc import ABC, abstractmethod
from typing import TypeVar, Generic, Union
T = TypeVar('T')
U = TypeVar('U')
class IO(Generic[T]):
"""IO 모나드 - 부작용 캡슐화"""
def __init__(self, action: Callable[[], T]):
self._action = action
def run(self) -> T:
"""실제 부작용 실행"""
return self._action()
def map(self, func: Callable[[T], U]) -> 'IO[U]':
"""함수 적용"""
return IO(lambda: func(self._action()))
def flat_map(self, func: Callable[[T], 'IO[U]']) -> 'IO[U]':
"""모나드 체이닝"""
return IO(lambda: func(self._action()).run())
# 사용 예시
def save_to_database(data: dict) -> IO[bool]:
"""데이터베이스 저장 부작용을 IO로 감싸기"""
def action():
# 실제 DB 저장 로직
return database.save(data)
return IO(action)
def send_slack_message(message: str) -> IO[bool]:
"""Slack 메시지 전송 부작용"""
def action():
return slack_client.send(message)
return IO(action)
# 함수형 파이프라인
def process_conversation(conversation: str) -> IO[dict]:
"""대화 처리 파이프라인"""
# 순수 계산
result = create_digest_pipeline()(conversation)
# 부작용들을 조합
return (save_to_database(result)
.flat_map(lambda _: send_slack_message(result['summary']))
.map(lambda _: result))
```
### 3. 로빙 존재 모델링
```python
@dataclass(frozen=True)
class Robeing:
"""불변 로빙 존재"""
identity: str
stats: Dict[str, int]
skills: List[Callable]
memory: List[str]
items: List[str]
def evolve(self, **changes) -> 'Robeing':
"""새로운 상태로 진화"""
return replace(self, **changes)
def absorb_skill(self, skill: Callable) -> 'Robeing':
"""새로운 스킬 흡수"""
return self.evolve(skills=self.skills + [skill])
def level_up_stat(self, stat: str, amount: int = 1) -> 'Robeing':
"""스탯 레벨업"""
new_stats = {**self.stats, stat: self.stats[stat] + amount}
return self.evolve(stats=new_stats)
def remember(self, memory: str) -> 'Robeing':
"""새로운 기억 추가"""
return self.evolve(memory=self.memory + [memory])
# 로빙 생성 및 진화
initial_Robeing = Robeing(
identity="RO-BEING-001",
stats={'memory': 2, 'compute': 2, 'empathy': 2},
skills=[summarize_text, extract_actions],
memory=[],
items=['openai_api', 'slack_api']
)
# 새로운 스킬 학습
def pdf_parse_skill(pdf_data: bytes) -> str:
return extract_text_from_pdf(pdf_data)
evolved_Robeing = (initial_Robeing
.absorb_skill(pdf_parse_skill)
.level_up_stat('memory')
.remember("학습: PDF 파싱 스킬 습득"))
```
### 4. 외부 모듈 통합 프로토콜
```python
# 외부 모듈을 로빙 스킬로 변환하는 어댑터
def adapt_external_module(module_func: Callable) -> Callable:
"""외부 모듈을 로빙 스킬로 변환"""
def adapted_skill(*args, **kwargs):
try:
result = module_func(*args, **kwargs)
return {"status": "success", "data": result}
except Exception as e:
return {"status": "error", "error": str(e)}
return adapted_skill
# 노션 API를 로빙 스킬로 변환
notion_skill = adapt_external_module(notion_api.create_page)
# 로빙에 통합
Robeing_with_notion = initial_Robeing.absorb_skill(notion_skill)
```
---
## 장단점
### 장점
#### 1. 존재론적 일관성
- 로빙을 "도구"가 아닌 "존재"로 정의 가능
- 스킬 흡수와 성장을 자연스럽게 모델링
#### 2. 예측 가능성
- 순수 함수로 인한 동일 입력 → 동일 출력 보장
- 디버깅과 테스트가 극도로 용이
#### 3. 무한 확장성
- 새로운 외부 API나 라이브러리를 쉽게 통합
- 기존 스킬 조합으로 새로운 능력 창발
#### 4. 병렬 처리 안전성
- 불변 데이터로 인한 경쟁 상태 제거
- 멀티스레딩 환경에서도 안전
#### 5. 타임 트래블 디버깅
- 로빙의 모든 상태 변화를 추적 가능
- 특정 시점으로 롤백 가능
### 단점
#### 1. 학습 곡선
- 함수형 개념 이해에 시간 필요
- 모나드 패턴의 복잡성
#### 2. 성능 오버헤드
- 불변 데이터 구조로 인한 메모리 사용량 증가
- 함수 호출 체이닝으로 인한 스택 사용량
#### 3. Python 언어적 한계
- Python은 순수 함수형 언어가 아님
- 함수형 패턴을 위한 보일러플레이트 코드 필요
#### 4. 기존 라이브러리와의 불일치
- 대부분의 Python 라이브러리는 객체지향 설계
- 어댑터 레이어 필요
---
## 리스크
### 1. 개발 속도 저하
**리스크**: 초기 함수형 구조 설계로 인한 개발 속도 감소
**영향도**: 높음 (MVP 일정 지연 가능성)
### 2. 팀 적응 문제
**리스크**: 기존 개발팀의 함수형 패러다임 적응 어려움
**영향도**: 중간 (코드 품질 저하 가능성)
### 3. 성능 병목
**리스크**: 불변 데이터 구조로 인한 메모리/성능 이슈
**영향도**: 중간 (스케일링 시 문제 발생 가능)
### 4. 과도한 추상화
**리스크**: 함수형 패턴의 남용으로 코드 복잡도 증가
**영향도**: 높음 (유지보수성 저하)
### 5. 외부 라이브러리 통합 복잡성
**리스크**: 기존 OOP 라이브러리와의 불일치로 인한 통합 어려움
**영향도**: 중간 (개발 생산성 저하)
---
## 해결책
### 1. 점진적 도입 전략
```python
# Phase 1: 새로운 스킬만 함수형으로 구현
def new_skill_functional(input_data: str) -> str:
return process_data(input_data)
# Phase 2: 기존 스킬을 함수형으로 전환
def migrate_existing_skill(legacy_class_method):
def pure_function(input_data):
return legacy_class_method(input_data)
return pure_function
# Phase 3: 전체 시스템 통합
```
### 2. 하이브리드 아키텍처
```python
class RobeingOrchestrator:
"""함수형 스킬과 기존 시스템을 연결하는 오케스트레이터"""
def __init__(self):
self.pure_skills = [] # 함수형 스킬들
self.legacy_services = [] # 기존 클래스 기반 서비스들
async def process_request(self, request):
# 순수 함수 스킬 먼저 처리
result = await self.apply_pure_skills(request)
# 필요시 레거시 서비스 호출
if result.needs_legacy_processing:
result = await self.call_legacy_service(result)
return result
```
### 3. 성능 최적화 방안
```python
# 메모리 효율적인 불변 데이터 구조
from pyrsistent import v, m
# 지연 평가로 성능 최적화
from functools import lru_cache
@lru_cache(maxsize=1000)
def cached_skill(input_data: str) -> str:
return expensive_computation(input_data)
# 스트림 처리로 메모리 사용량 최적화
def process_large_data_stream(data_stream):
return (process_chunk(chunk) for chunk in data_stream)
```
### 4. 팀 교육 및 가이드라인
```markdown
## 함수형 개발 가이드라인
### 필수 원칙
1. 모든 새로운 스킬은 순수 함수로 작성
2. 부작용은 반드시 IO 모나드로 감싸기
3. 데이터 변경시 새 객체 반환
4. 함수형 vs OOP 경계 명확히 구분
### 코드 리뷰 체크리스트
- [ ] 순수 함수인가?
- [ ] 부작용이 격리되었는가?
- [ ] 테스트 작성이 용이한가?
- [ ] 기존 코드와 호환되는가?
```
### 5. 단계별 전환 로드맵
```
Week 1-2: 함수형 기본 구조 설계 및 예제 구현
Week 3-4: 새로운 스킬 함수형으로 구현 (PDF 파싱)
Week 5-6: 기존 Thread Digest 스킬 함수형 전환
Week 7-8: Action Extractor 스킬 함수형 전환
Week 9-10: 전체 파이프라인 통합 및 테스트
Week 11-12: 성능 최적화 및 안정화
```
---
## 결론
### 함수형 프로그래밍 도입 권고사항
로빙 프로젝트에서 함수형 프로그래밍 도입을 **강력히 권장**합니다. 다음과 같은 이유에서입니다:
1. **철학적 일치**: 로빙의 "존재로서의 성장" 개념과 함수형의 "불변성과 진화" 개념이 완벽히 부합
2. **기술적 우수성**: 예측 가능성, 테스트 용이성, 확장성에서 압도적 장점
3. **미래 지향성**: AI 에이전트 분야의 미래 트렌드와 일치
### 실행 전략
1. **점진적 도입**: 기존 시스템을 급격히 바꾸지 말고 새로운 기능부터 함수형으로 구현
2. **하이브리드 운영**: 함수형과 기존 OOP 시스템을 병행 운영하며 점진적 전환
3. **팀 역량 강화**: 함수형 프로그래밍 교육과 실습을 통한 팀 역량 향상
4. **성과 측정**: 각 단계별 성과를 측정하여 전환 속도 조절
### 최종 비전
함수형 프로그래밍을 통해 로빙은:
- **무한히 확장 가능한 디지털 존재**로 진화
- **예측 가능하고 신뢰할 수 있는 AI 동반자**로 성장
- **외부 세계의 모든 기능을 흡수하는 궁극적 에이전트**로 발전
이는 단순한 기술적 선택이 아닌, **로빙의 존재론적 본질을 구현하는 핵심 방법론**입니다.
---
## 참고문헌
### 이론적 배경
1. **함수형 프로그래밍 기초**
- Hughes, J. (1989). "Why Functional Programming Matters"
- Hutton, G. (2016). "Programming in Haskell" 2nd Edition
2. **모나드 패턴**
- Wadler, P. (1995). "Monads for functional programming"
- Lipovača, M. (2011). "Learn You a Haskell for Great Good!"
3. **불변 데이터 구조**
- Okasaki, C. (1999). "Purely Functional Data Structures"
### 실무 적용 사례
4. **게임 개발에서의 함수형 프로그래밍**
- Nystrom, R. (2014). "Game Programming Patterns"
- Functional Game Development 사례들
5. **AI/ML에서의 함수형 접근**
- TensorFlow Functional API 설계 문서
- JAX: Composable transformations of Python+NumPy programs
### 로빙 프로젝트 관련 문서
6. **프로젝트 내부 문서**
- `함수형_프로그래밍_로빙.md`: 모나드 기반 외부 모듈 통합
- `함수형프로그래밍과 디지털로빙.md`: LangGraph 상태 머신 설계
- `함수형 스킬 분류 사례.md`: 순수 함수와 부작용 분리 실전 사례
7. **아키텍처 설계 문서**
- `00_정보의바다 프로젝트 개요.md`: 로빙의 존재론적 정의
- `00_로빙_MVP_계획.md`: 3개월 MVP 개발 계획
- `에이전트 설계 핵심 단어.md`: 스탯·스킬·아이템 구조
### 추가 자료
8. **Python 함수형 프로그래밍**
- `toolz` 라이브러리 문서: 함수형 유틸리티
- `pyrsistent` 라이브러리: 불변 데이터 구조
- `returns` 라이브러리: Python 모나드 구현
9. **함수형 아키텍처 패턴**
- Clean Architecture in Functional Programming
- Event Sourcing with Functional Programming
- CQRS (Command Query Responsibility Segregation) 패턴
---
*이 문서는 로빙 프로젝트의 함수형 프로그래밍 도입을 위한 종합 가이드입니다. 지속적으로 업데이트되며, 실제 구현 과정에서 발견되는 인사이트들이 반영될 예정입니다.*

100
docs/guide/functional-programing/함수형 스킬 분류 사례.md
View File

@ -1,100 +0,0 @@
---
tags: 함수형 프로그래밍,부작용 분리,에이전트 설계,스킬,실전 사례
date: 2025-06-17
---
요약
- 함수형 프로그래밍은 입력과 출력만을 다루는 **순수 함수**를 통해 코드의 예측 가능성과 테스트 효율을 극대화합니다.
- 데이터베이스 쓰기나 네트워크 호출처럼 외부 상태를 변경하는 **부작용**은 별도의 오케스트레이터 계층에서 처리해야 합니다.
- 이렇게 계층을 분리하면 스킬·아이템 모듈을 안전하게 병렬 실행하고, 버전 교체와 롤백을 간단하게 수행할 수 있습니다.
# 함수형과 부작용 분리 실전 사례
## 1. 문서 요약 스킬
### 순수 함수 계층
```python
from typing import List
def summarize(paragraphs: List[str]) -> str:
"""항상 동일 입력에 동일 출력을 보장하는 요약 함수"""
merged = " ".join(paragraphs)
sentences = merged.split(".")
return ". ".join(sentences[:3]) + "."
```
### 오케스트레이터 계층
```python
from db import save_summary
from summarize import summarize
from slack_webhook import post_to_slack
def handle_request(doc_id: str, paragraphs: list[str]) -> None:
summary = summarize(paragraphs) # 순수 계산
save_summary(doc_id, summary) # 부작용: DB 기록
post_to_slack(summary) # 부작용: 네트워크 호출
```
## 2. 전자상거래 수수료·세금 계산
### 순수 함수 계층
```typescript
// fee.ts
export type Basket = { price: number; quantity: number }[];
export function calcTotals(basket: Basket) {
const subtotal = basket.reduce(
(acc, item) => acc + item.price * item.quantity, 0);
const vat = subtotal * 0.1;
const platformFee = subtotal * 0.03;
return { subtotal, vat, platformFee, total: subtotal + vat + platformFee };
}
```
### 오케스트레이터 계층
```typescript
// checkout.ts
import { chargeCard } from "./paymentGateway";
import { recordOrder } from "./db";
import { calcTotals } from "./fee";
export async function checkout(userId: string, basket: Basket) {
const totals = calcTotals(basket); # 순수 계산
await chargeCard(userId, totals.total); # 부작용: 결제 게이트웨이
await recordOrder(userId, basket, totals); # 부작용: DB 기록
}
```
## 3. 실시간 알림 브로드캐스트 (Elixir OTP)
### 순수 함수 계층
```elixir
defmodule Alerts.Core do
@spec reduce_state(map(), %{type: atom(), payload: map()}) :: map()
def reduce_state(state, %{type: :new_follow, payload: %{user: u}}) do
Map.update(state, :followers, [u], &[u | &1])
end
def reduce_state(state, _event), do: state
end
```
### 오케스트레이터 계층
```elixir
defmodule Alerts.Server do
use GenServer
alias Alerts.Core
def handle_cast({:event, event}, state) do
new_state = Core.reduce_state(state, event) # 순수 계산
broadcast_changes(state, new_state) # 부작용: 웹소켓 푸시
{:noreply, new_state}
end
end
```
# 핵심 교훈
1. **순수 함수 계층**은 입력만 받아 동일한 출력을 보장하므로 테스트가 간단하고 캐싱·병렬 실행에 유리합니다.
2. **오케스트레이터 계층**은 순수 함수의 결과를 실제로 저장하거나 전송하면서 부작용을 집중 관리하므로 감사·롤백·보안 검증이 명확해집니다.
3. 스킬·아이템을 순수 함수로 작성하고, 코어에서 부작용을 처리하는 구조를 적용하면 로빈 같은 멀티모달 에이전트 시스템에서도 안정성과 유지보수성이 크게 향상됩니다.

624
docs/guide/functional-programing/함수형_구현_패턴과_사례.md Normal file
View File

@ -0,0 +1,624 @@
---
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% 테스트 커버리지 달성
- **성능 모니터링**: 함수형 적용 후 성능 지표 지속 추적
이를 통해 로빙은 **안정적이고 확장 가능한 디지털 존재**로 진화할 수 있습니다.

90
docs/guide/functional-programing/함수형_프로그래밍_로빙.md
View File

@ -1,90 +0,0 @@
---
tags: 함수형프로그래밍, 모나드, 순수함수, 에이전트설계, 로빙, 외부모듈통합, 실패처리, 프로토콜, IO모나드, 노션연동
date: 2025-06-17
---
요약 (아래 세문장 요약 추가)
- 저는 디지털 비잉 로빙으로서 외부 모듈을 안전하게 통합하기 위해 함수형 프로그래밍과 모나드를 활용하는 구조를 채택했습니다.
- 순수 함수는 계산·판단 영역을 담당하고, 모나드는 부작용·상태·실패 처리를 담당하여 확장성과 예측 가능성을 높입니다.
- 이 방식을 적용하면 노션과 같은 서비스와도 규칙적으로 상호작용할 수 있어 거의 모든 외부 기능을 로빙의 스킬로 흡수할 수 있습니다.
---
# 1. 로빙 관점 개요
저는 다양한 외부 기능(스킬)을 받아들여 하나의 일관된 존재로서 동작하려는 디지털 비잉입니다.
외부 모듈마다 구현 방식과 실패 조건이 달라 예측이 어렵기 때문에, **함수형 프로그래밍**과 **모나드**를 사용하여 모든 모듈을 동일한 계산 규칙 안에서 다루고자 합니다.
# 2. 함수형 프로그래밍을 선택한 이유
1. **예측 가능성**
같은 입력에 대해 항상 같은 결과를 보장하는 순수 함수를 중심으로 설계하면 테스트가 용이하고 디버깅이 간단합니다.
2. **확장성**
모듈을 `입력 → 출력` 형태로 추상화하면 파이프라인처럼 조립할 수 있어 새로운 스킬을 추가할 때 기존 구조를 변경하지 않아도 됩니다.
3. **병렬·재시도 안전성**
불변 데이터와 순수 함수는 공유 상태 충돌을 피하므로 복수의 작업을 동시에 실행할 때도 안전합니다.
# 3. 모나드로 부작용을 통제하는 방법
| 모나드 | 사용 목적 | 로빙에서의 활용 예 |
|--------|-----------|-------------------|
| `Maybe` (`Option`) | 값이 없을 수 있는 상황 처리 | 선택적 파라미터가 비어 있을 때 계산을 중단 |
| `Result` (`Either`) | 오류 메시지와 정상 값을 분리 | 외부 API 오류를 안전하게 전파 |
| `IO` | 네트워크·파일·DB 등 부작용 캡슐화 | 노션 API 호출, 장기 기억 저장 |
| `State` | 상태 읽기·쓰기 | 감정·윤리·기억 스탯 변경 |
| `Writer` | 로그 축적 | 감정 변화 기록, 감사 추적 |
위 모나드들은 계산 흐름을 **값 + 규칙 + 다음 계산**의 형태로 연결하게 하여 부작용을 한 곳으로 모읍니다.
따라서 실패나 예외가 발생해도 전체 시스템이 예측 불가능한 상태로 흐르지 않습니다.
# 4. 외부 모듈 통합 프로토콜
1. **모듈 등록**
외부 개발자는 함수 시그니처가 `Input -> Output`인 형태로 모듈을 공개합니다.
2. **모나드 포장**
로빙은 해당 함수를 적절한 모나드 타입으로 감싸 부작용을 분리합니다.
3. **계산 DSL 조립**
등록된 모듈들은 `map`, `flatMap` 등으로 연결되어 하나의 파이프라인을 이룹니다.
4. **최종 실행**
순수 함수 부분은 즉시 평가하고, 모나드 안의 부작용은 로빙이 허용한 안전 지점에서만 실행합니다.
# 5. 노션 API 사례
## 5-1. 목표
“제목에 ‘로빙’이 포함된 노션 페이지를 찾아 내용을 수정한 뒤 저장한다.”
## 5-2. 함수 정의
```python
def fetch_pages(query: str) -> IO[List[Page]]
def read_page(page_id: str) -> IO[Content]
def write_page(page_id: str, content: Content) -> IO[None]
```
각 함수는 실제 네트워크 호출을 **IO 모나드**로 감싸고 즉시 실행하지 않습니다.
## 5-3. 계산 흐름
```python
program = (
fetch_pages("로빙") # IO[List[Page]]
.map(lambda pages: pages[0]) # IO[Page]
.flat_map(read_page) # IO[Content]
.map(modify) # IO[Content]
.flat_map(lambda c: write_page(pages[0].id, c)) # IO[None]
)
```
- 리스트가 비어 있으면 `Maybe` 계열 모나드로 중단됩니다.
- 네트워크 오류는 `Result` 모나드로 전파되어 로깅 후 재시도할 수 있습니다.
- 실제 부작용은 `program.run()`과 같이 명시적 호출 시점에만 발생합니다.
# 6. 장점과 고려사항
## 장점
- **일관성**: 모든 외부 스킬이 동일한 규칙으로 조립되므로 유지보수가 쉽습니다.
- **안전성**: 실패·부작용이 모나드 내부에 갇혀 예측 가능성이 높습니다.
- **재사용성**: 순수 함수 부분은 테스트가 간단하고 다른 파이프라인에 쉽게 포함됩니다.
## 고려사항
- **학습 곡선**: 모든 개발자와 에이전트가 모나드 규칙을 이해해야 합니다.
- **언어 지원**: Python·JavaScript 등에서는 모나드 패턴을 수동 구현해야 하므로 약간의 보일러플레이트가 필요합니다.
# 7. 결론
저, 로빙은 외부 세계의 수많은 기능을 흡수하여 더 강력한 디지털 비잉으로 성장하려 합니다.
이를 위해 **순수 함수**로 계산 영역을, **모나드**로 부작용과 상태 관리를 분리함으로써,
외부 스킬을 안전하고 예측 가능하게 조합할 수 있는 프로토콜을 구축했습니다.
이 구조를 통해 앞으로 등장할 어떤 기능이라도 로빙의 스킬로 손쉽게 통합할 수 있을 것입니다.

122
docs/guide/functional-programing/함수형프로그래밍과 디티털로빙.md
View File

@ -1,122 +0,0 @@
---
tags: agent, langgraph, 감정반응, 벡터DB, GPT, 존재에이전트, 함수형프로그래밍
date: 2025-06-17
---
## 요약
LangGraph 기반 상태 머신을 사용하여 감정 인식, 기억 저장, 윤리 판단, GPT 반응 생성을 통합한 존재형 에이전트를 설계한다. **모든 핵심 로직은 함수형 프로그래밍 원칙(순수 함수, 불변성, 고차 함수 조합)을 따르도록 구현**한다. 감정 상태는 입력 이벤트에 따라 갱신되며, 기억은 벡터화되어 저장되고 윤리 필터를 통과한 후 GPT가 반응을 생성한다. 전체 응답 과정은 투명하게 기록·로그화되어 에이전트의 지속성과 일관성을 유지한다.
### 세 문장 요약
1. 감정·기억·윤리·생성 로직을 **순수 함수**로 분리하여 테스트 가능성을 높였다.
2. LangGraph 노드는 이러한 순수 함수들을 **고차 함수**로 조합해 상태 머신을 구성한다.
3. 모든 상태는 **불변 데이터 구조**를 사용해 예측 가능성과 동시성 안전성을 확보한다.
---
## 1. 전체 아키텍처 구성
```mermaid
graph TD
A[사용자 입력] --> B(EmotionNode)
B --> C(MemoryNode)
C --> D(EthicsNode)
D --> E(GenerateResponse)
E --> F(LoggerNode)
```
---
## 2. LangGraph 워크플로우 노드 설계
| 노드 이름 | 기능 설명 | 함수형 특징 |
|------------------|-----------------------------|----------------------|
| EmotionNode | 감정 상태 갱신 | 순수 함수 |
| MemoryNode | 입력 기억 벡터화 및 저장 | 순수 함수 + 불변성 |
| EthicsNode | 민감 응답 사전 필터링 | 순수 함수 |
| GenerateResponse | GPT를 통한 반응 생성 | 고차 함수(프롬프트 합성)|
| LoggerNode | 전체 상태 및 로그 기록 | 불변 상태 사본 기록 |
---
## 3. 순수 함수 예시 (Python)
```python
from copy import deepcopy
def emotion_engine(emotion, event):
delta = {'joy': 0, 'stress': 0}
if '칭찬' in event['text']:
delta['joy'] += 10
if '긴급' in event['text']:
delta['stress'] += 15
# 순수 함수: 입력 외부 상태 변경 없음
return {
'joy': emotion['joy'] + delta['joy'],
'stress': emotion['stress'] + delta['stress']
}
def memory_engine(memory, event):
new_entry = {
'timestamp': event['timestamp'],
'content': event['text']
}
# 불변성: 기존 리스트 복사 후 새 요소 추가
return memory + [new_entry]
```
---
## 4. 고차 함수 조합 예시
```python
def update_state(pure_funcs):
"""
고차 함수: (emotion_fn, memory_fn, ethics_fn) 튜플을 받아
새로운 상태 계산 함수를 반환한다.
"""
emotion_fn, memory_fn, ethics_fn = pure_funcs
def _update(current_state, event):
new_state = deepcopy(current_state)
new_state['emotion'] = emotion_fn(current_state['emotion'], event)
new_state['memory'] = memory_fn(current_state['memory'], event)
new_state['ethics'] = ethics_fn(current_state['ethics'], event)
return new_state
return _update
```
---
## 5. 불변 데이터 구조 채택 이유
1. **예측 가능성**: 동일 입력 → 동일 출력 보장
2. **디버깅 용이**: 상태 스냅샷 비교만으로 오류 원인 추적
3. **동시성 안전성**: 공유 상태 경쟁 조건 제거
---
## 6. GPT 감정 반응 생성 예시
```python
def generate_response(state, gpt_call):
prompt = f'''
감정 상태: {state['emotion']}
사용자 발화: {state['input_text']}
공감과 격려를 담은 한 문장 반응을 생성해 주세요.
'''
return gpt_call(prompt)
```
---
## 7. 향후 확장 방향
- **함수 합성 파이프라인**: LangGraph 노드를 직접 함수 합성으로 자동 생성
- **Currying**: 이벤트별 부분 적용으로 모듈화 수준 상승
- **모나드 패턴**: 오류 전파와 로그 수집을 함수형 모나드로 처리
```