# 로빙(Robing) 슬랙 요약 스킬 아키텍처 설계 ## 1. 개요 ### 1.1 배경 - 로빙은 스탯 기반 성장형 AI 에이전트로, 회사별로 독립적인 컨테이너로 배포됨 - 로빙 브레인은 스킬 라우팅과 결과 판단에 집중하며, 실제 처리는 공용 스킬 서버에서 수행 - 프롬프트용 기억은 컨테이너 외부의 공용 저장소에 보관 ### 1.2 설계 원칙 - **경량화**: 회사별 로빙 컨테이너는 최소한의 리소스(512MB)로 운영 - **확장성**: 스킬 서버는 독립적으로 스케일링 가능 - **비용 효율성**: LLM API는 공용 서버에서만 호출하여 비용 최적화 - **독립성**: 각 스킬은 마이크로서비스로 분리되어 독립적 배포/업데이트 가능 ## 2. 전체 아키텍처 ### 2.1 멀티테넌트 구조 ``` ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ 회사 A 로빙 │ │ 회사 B 로빙 │ │ 회사 C 로빙 │ │ (가벼운 라우터) │ │ (가벼운 라우터) │ │ (가벼운 라우터) │ └────────┬────────┘ └────────┬────────┘ └────────┬────────┘ │ │ │ └───────────────────────┴───────────────────────┘ │ ┌────────────▼────────────┐ │ 공용 스킬 서버 클러스터 │ ├─────────────────────────┤ │ • 슬랙 요약 서비스 │ │ • 이메일 파싱 서비스 │ │ • 일정 관리 서비스 │ │ • LLM 처리 (Gemini/GPT) │ └─────────────────────────┘ │ ┌────────────▼────────────┐ │ 공용 데이터 저장소 │ ├─────────────────────────┤ │ • ChromaDB (벡터 DB) │ │ • PostgreSQL (메타데이터) │ │ • Redis (캐싱) │ └─────────────────────────┘ ``` ### 2.2 데이터 흐름 ``` 1. Slack Event → 회사별 로빙 컨테이너 2. 로빙 브레인이 의도 분석 (키워드 기반, LLM 없이) 3. 적절한 스킬 서버로 라우팅 4. 스킬 서버에서 LLM 처리 및 결과 생성 5. 결과를 로빙 컨테이너로 반환 6. 로빙이 Slack으로 응답 전송 ``` ## 3. 로빙 컨테이너 설계 ### 3.1 핵심 역할 - **Slack 이벤트 게이트웨이**: 회사별 Slack 이벤트 수신 및 ACK 응답 - **간단한 의도 분석**: 키워드 기반 빠른 분류 (LLM 없이) - **스킬 라우팅**: 적절한 스킬 서버로 요청 전달 - **회사별 컨텍스트 관리**: 회사 설정, 사용자 스탯, 활성 스킬 관리 - **학습 욕구 기록**: 처리하지 못한 요청 로컬 기록 ### 3.2 구조 ```python app/ ├── core/ │ ├── config.py # 회사별 설정 │ ├── auth.py # 인증/인가 │ └── router.py # 스킬 라우팅 ├── brain/ │ └── intent.py # 의도 분석 (경량) ├── api/ │ └── slack.py # Slack 이벤트 수신 └── main.py # FastAPI 앱 ``` ### 3.3 로빙 브레인 구현 ```python class RobeingBrain: """스킬 라우터 + 의사결정자""" def __init__(self, company_id: str): self.company_id = company_id self.skill_registry = { "요약": "http://skill-summary:8001", "이메일": "http://skill-email:8002", "일정": "http://skill-calendar:8003" } async def route_message(self, message: str, context: dict): # 1. 간단한 의도 분석 (LLM 없이) intent = self._quick_intent_check(message) # 2. 스킬 결정 skill_url = self.skill_registry.get(intent) # 3. 스킬 서버로 전달 if skill_url: return await self._call_skill(skill_url, message, context) else: self._record_learning_desire(message) return "아직 그 기능은 없어요" ``` ## 4. 슬랙 요약 스킬 마이크로서비스 ### 4.1 프로젝트 구조 ``` skill-slack-summary/ ├── Dockerfile ├── requirements.txt ├── src/ │ ├── api/ │ │ └── endpoints.py # REST API │ ├── core/ │ │ ├── config.py │ │ └── security.py # API 키 검증 │ ├── services/ │ │ ├── slack_client.py # Slack API 통신 │ │ ├── summarizer.py # LLM 요약 로직 │ │ └── memory.py # ChromaDB 인터페이스 │ ├── models/ │ │ └── summary.py # 데이터 모델 │ └── utils/ │ ├── cache.py # Redis 캐싱 │ └── rate_limiter.py # API 제한 관리 └── main.py ``` ### 4.2 API 엔드포인트 ```python from fastapi import FastAPI, Depends from pydantic import BaseModel app = FastAPI() class SummaryRequest(BaseModel): company_id: str channel_id: str time_range: str # "1h", "2h", "today", etc. user_id: str user_stats: Dict[str, int] @app.post("/summarize") async def summarize_messages( request: SummaryRequest, auth: str = Depends(verify_auth) ): """슬랙 메시지 요약""" # 1. 메시지 수집 messages = await slack_client.fetch_messages( request.channel_id, request.time_range ) # 2. LLM으로 요약 summary = await summarizer.process(messages) # 3. 메모리 저장 await memory_service.save(request.company_id, summary) return { "summary": summary.content, "key_points": summary.key_points, "action_items": summary.action_items } ``` ### 4.3 스킬 메타데이터 ```python class SkillMetadata(BaseModel): name: str = "slack_summary" version: str = "1.0.0" description: str = "슬랙 회의 내용 요약" required_stats: Dict[str, int] = {"memory": 5} triggers: List[str] = ["회의 요약", "미팅 정리", "summary"] ``` ## 5. 통신 프로토콜 ### 5.1 로빙 → 스킬 서비스 요청 ```json { "company_id": "company_a", "user_id": "U123456", "skill": "slack_summary", "auth_token": "jwt_token", "payload": { "channel_id": "C789012", "time_range": "2h", "thread_ts": "1234567890.123456" }, "context": { "user_stats": {"memory": 15, "compute": 10}, "workspace_id": "T123456" } } ``` ### 5.2 스킬 서비스 → 로빙 응답 ```json { "status": "success", "data": { "summary": "오늘 회의에서는 다음 분기 제품 로드맵에 대해 논의했습니다...", "key_points": [ "Q2 신제품 출시 일정 확정", "마케팅 예산 20% 증액 승인" ], "action_items": [ { "assignee": "@김철수", "task": "프로토타입 데모 준비", "due_date": "2025-07-15" } ] }, "metadata": { "processing_time": 2.5, "tokens_used": 1500, "cache_hit": false } } ``` ## 6. LLM 처리 및 비용 최적화 ### 6.1 LLM 관리 전략 ```python class LLMManager: """중앙 집중식 LLM 관리""" def __init__(self): self.gemini_client = GeminiClient() self.openai_client = OpenAIClient() self.cache = Redis() self.rate_limiter = RateLimiter() async def summarize(self, messages: List[str], company_id: str): # 1. 캐시 확인 cache_key = f"summary:{company_id}:{hash(messages)}" if cached := await self.cache.get(cache_key): return cached # 2. Rate limiting await self.rate_limiter.check(company_id) # 3. LLM 호출 summary = await self.gemini_client.summarize(messages) # 4. 캐시 저장 await self.cache.set(cache_key, summary, ttl=3600) return summary ``` ### 6.2 Rate Limit 대응 1. **다중 API 키 풀**: 여러 API 키를 로테이션하여 사용 2. **큐 기반 처리**: 우선순위 큐로 유료 고객 우선 처리 3. **배치 처리**: 여러 요청을 하나의 프롬프트로 묶어 처리 4. **계층적 처리**: 부하에 따라 다른 모델 사용 (Pro vs Flash) 5. **스마트 캐싱**: 유사한 요청에 대한 캐시 재사용 ## 7. 메모리 저장 구조 ### 7.1 ChromaDB Collection ```json { "collection": "meeting_summaries", "documents": [ { "id": "summary_company_a_20250701_123456", "content": "오늘 회의에서는 다음 분기 계획을 논의했습니다...", "metadata": { "company_id": "company_a", "user_id": "U123456", "channel_id": "C789012", "timestamp": "2025-07-01T14:30:00", "participants": ["user1", "user2", "user3"], "key_points": ["포인트1", "포인트2"], "action_items": ["할일1", "할일2"], "skill_version": "1.0.0" } } ] } ``` ## 8. 배포 구성 ### 8.1 Docker Compose - 공용 스킬 서버 ```yaml # docker-compose.skills.yml version: '3.8' services: skill-slack-summary: build: ./skills/slack-summary image: robing-skills/slack-summary:latest ports: - "8001:8001" environment: - GEMINI_API_KEY=${GEMINI_API_KEY} - OPENAI_API_KEY=${OPENAI_API_KEY} - REDIS_URL=redis://redis:6379 - CHROMADB_HOST=chromadb - MAX_REQUESTS_PER_MINUTE=60 deploy: replicas: 3 resources: limits: memory: 2G redis: image: redis:7-alpine volumes: - redis_data:/data chromadb: image: chromadb/chroma volumes: - chroma_data:/chroma/chroma ``` ### 8.2 Docker Compose - 회사별 로빙 ```yaml # docker-compose.company-a.yml version: '3.8' services: robing-company-a: image: robing/core:latest environment: - COMPANY_ID=company_a - SLACK_BOT_TOKEN=${COMPANY_A_SLACK_TOKEN} - SLACK_SIGNING_SECRET=${COMPANY_A_SIGNING_SECRET} - SKILL_SERVER_URL=https://skills.robing.ai - JWT_SECRET=${JWT_SECRET} ports: - "10001:8000" deploy: resources: limits: memory: 512M cpus: '0.5' healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8000/health"] interval: 30s timeout: 10s retries: 3 ``` ## 9. 확장성 고려사항 ### 9.1 수평 확장 - 로빙 컨테이너: 회사별로 독립 배포, 필요시 여러 인스턴스 가능 - 스킬 서버: 부하에 따라 레플리카 수 조정 - 데이터베이스: 읽기 전용 복제본 추가 ### 9.2 지역별 배포 - 회사 위치에 따라 가까운 리전에 로빙 컨테이너 배포 - 스킬 서버는 중앙 또는 주요 리전에 배포 ### 9.3 모니터링 - Prometheus + Grafana로 실시간 모니터링 - 회사별 사용량 추적 - API 사용량 및 비용 대시보드 ## 10. 보안 고려사항 ### 10.1 인증/인가 - 로빙-스킬 간 통신: JWT 토큰 기반 인증 - 회사별 데이터 격리: company_id 기반 접근 제어 - API 키 관리: 환경 변수로 관리, 로빙 컨테이너에는 노출하지 않음 ### 10.2 데이터 보호 - 전송 중 암호화: HTTPS 사용 - 저장 시 암호화: 민감한 데이터는 암호화하여 저장 - 로그 마스킹: 개인정보가 로그에 노출되지 않도록 처리 ## 11. 성능 최적화 ### 11.1 캐싱 전략 - Redis를 통한 LLM 응답 캐싱 - 자주 요청되는 요약에 대한 사전 처리 - 유사 요청 감지 및 재사용 ### 11.2 비동기 처리 - 모든 I/O 작업은 비동기로 처리 - 백그라운드 태스크로 무거운 작업 분리 - 웹훅 방식으로 결과 전달 옵션 ## 12. 향후 로드맵 ### Phase 1 (현재) - 기본 슬랙 요약 기능 구현 - 단일 스킬 서버 운영 ### Phase 2 - 다국어 지원 (한국어/영어) - 실시간 요약 스트리밍 - 더 많은 스킬 추가 ### Phase 3 - AI 기반 요약 품질 개선 - 사용자별 요약 스타일 학습 - 스킬 마켓플레이스 구축 ## 13. 결론 이 아키텍처는 다음과 같은 이점을 제공합니다: 1. **확장성**: 회사가 증가해도 효율적으로 대응 가능 2. **비용 효율성**: LLM API 사용을 최적화하여 비용 절감 3. **독립성**: 각 컴포넌트가 독립적으로 배포/업데이트 가능 4. **안정성**: 장애 격리로 전체 시스템 안정성 향상 5. **성능**: 캐싱과 배치 처리로 응답 속도 향상 이 설계를 통해 로빙은 수백 개의 회사를 효율적으로 서비스할 수 있는 확장 가능한 플랫폼으로 성장할 수 있습니다.