# PlanitAI KPI 개발기 v13: MVP 구현 계획 > 시리즈: PlanitAI KPI 개발 여정 (13/16) > 작성일: 2024년 12월 ## 개요 완벽한 제품보다 빠른 시장 검증이 중요합니다. 이번 글에서는 PlanitAI KPI의 MVP(Minimum Viable Product) 스코프를 정의하고 구현 우선순위를 설정합니다. --- ## 1. MVP 철학 ### 1.1 MVP의 목표 ``` ┌─────────────────────────────────────────────────────────────┐ │ MVP Goals │ ├─────────────────────────────────────────────────────────────┤ │ │ │ ✅ 검증할 것: │ │ - KPI 트리 시각화의 가치 │ │ - AI 분석 인사이트의 유용성 │ │ - 목표 가격($50-100/월)의 적정성 │ │ │ │ ❌ 검증하지 않을 것: │ │ - 대규모 확장성 │ │ - 모든 연동 기능 │ │ - 완벽한 UI/UX │ │ │ └─────────────────────────────────────────────────────────────┘ ``` ### 1.2 MVP vs 최종 제품 | 기능 | MVP | 최종 제품 | |------|-----|----------| | KPI 트리 | 단일 트리, 10개 노드 제한 | 무제한 트리/노드 | | 데이터 입력 | 수동 + CSV | API + 다양한 연동 | | AI 분석 | 트렌드 + 병목 | 예측 + 시뮬레이션 + 추천 | | 사용자 | 단일 사용자 | 팀 협업 | | 보고서 | HTML 출력 | PDF + PPT + 스케줄링 | --- ## 2. MVP 스코프 정의 ### 2.1 핵심 기능 (Must Have) ```markdown ## Phase 1: Core Features (4주) ### Week 1-2: KPI 트리 기본 - [ ] Google Sheets 연동 (읽기) - [ ] KPI 트리 구조 정의 (YAML/JSON) - [ ] 기본 계산 엔진 (사칙연산) - [ ] 단순 트리 시각화 (HTML) ### Week 3: AI 분석 - [ ] Gemini API 연동 - [ ] 트렌드 분석 프롬프트 - [ ] 병목 분석 프롬프트 - [ ] 분석 결과 HTML 출력 ### Week 4: 통합 & 마무리 - [ ] CLI 인터페이스 - [ ] 설정 파일 관리 - [ ] 기본 에러 처리 - [ ] 사용 문서 ``` ### 2.2 제외 기능 (Not in MVP) ```markdown ## Deferred Features ### 사용자 관리 - 인증/인가 시스템 - 팀 협업 기능 - 역할 기반 접근 제어 ### 고급 기능 - What-If 시뮬레이션 - 예측 모델 - 알림/스케줄링 ### 연동 - freee 연동 - Slack 연동 - API 제공 ### 인프라 - 웹 애플리케이션 - 멀티 테넌트 - 자동 스케일링 ``` --- ## 3. MVP 아키텍처 ### 3.1 단순화된 구조 ``` ┌─────────────────────────────────────────────────────────────┐ │ MVP Architecture │ ├─────────────────────────────────────────────────────────────┤ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌────────────┐ │ │ │ User │────▶│ CLI Tool │────▶│ Output │ │ │ │ (Terminal) │ │ (Python) │ │ (HTML) │ │ │ └──────────────┘ └──────┬───────┘ └────────────┘ │ │ │ │ │ ┌──────────┼──────────┐ │ │ ▼ ▼ ▼ │ │ ┌──────────┐ ┌────────┐ ┌──────────┐ │ │ │ Google │ │ Config │ │ Gemini │ │ │ │ Sheets │ │ (YAML) │ │ API │ │ │ └──────────┘ └────────┘ └──────────┘ │ │ │ └─────────────────────────────────────────────────────────────┘ ``` ### 3.2 디렉토리 구조 ``` planitai-kpi/ ├── src/ │ ├── __init__.py │ ├── main.py # CLI 엔트리포인트 │ ├── config.py # 설정 관리 │ ├── sheets/ │ │ ├── __init__.py │ │ └── client.py # Google Sheets 클라이언트 │ ├── kpi/ │ │ ├── __init__.py │ │ ├── tree.py # KPI 트리 구조 │ │ ├── parser.py # 수식 파서 │ │ └── calculator.py # 계산 엔진 │ ├── ai/ │ │ ├── __init__.py │ │ ├── client.py # Gemini 클라이언트 │ │ └── prompts.py # 프롬프트 템플릿 │ └── output/ │ ├── __init__.py │ └── html.py # HTML 생성기 ├── templates/ │ ├── report.html # 보고서 템플릿 │ └── components/ # HTML 컴포넌트 ├── config/ │ └── example.yaml # 설정 예시 ├── tests/ │ └── ... ├── pyproject.toml └── README.md ``` --- ## 4. 상세 구현 계획 ### 4.1 Week 1: 기반 구축 ```python # Day 1-2: 설정 및 Google Sheets 연동 # src/config.py from dataclasses import dataclass from pathlib import Path import yaml @dataclass class KPINodeConfig: """KPI 노드 설정""" id: str name: str type: str # kgi, kpi, input formula: str | None = None unit: str = "" children: list[str] | None = None @dataclass class Config: """전체 설정""" sheet_id: str sheet_range: str tree: list[KPINodeConfig] gemini_api_key: str output_path: str = "./output" @classmethod def from_yaml(cls, path: str) -> "Config": with open(path) as f: data = yaml.safe_load(f) tree = [KPINodeConfig(**node) for node in data.get("tree", [])] return cls( sheet_id=data["sheet_id"], sheet_range=data["sheet_range"], tree=tree, gemini_api_key=data["gemini_api_key"], output_path=data.get("output_path", "./output") ) ``` ```yaml # config/example.yaml # Google Sheets 설정 sheet_id: "1ABC..." sheet_range: "Sheet1!A1:Z100" # Gemini API gemini_api_key: "${GEMINI_API_KEY}" # 출력 경로 output_path: "./output" # KPI 트리 구조 tree: - id: revenue name: 月間売上 type: kgi formula: "{contracts} * {avg_price}" unit: 円 children: [contracts, avg_price] - id: contracts name: 契約数 type: kpi formula: "{meetings} * {close_rate}" unit: 件 children: [meetings, close_rate] - id: avg_price name: 平均単価 type: input unit: 円 - id: meetings name: 商談数 type: kpi formula: "{leads} * {meeting_rate}" unit: 件 children: [leads, meeting_rate] - id: close_rate name: 成約率 type: input unit: "%" - id: leads name: リード数 type: input unit: 件 - id: meeting_rate name: 商談化率 type: input unit: "%" ``` ### 4.2 Week 1: Google Sheets 클라이언트 ```python # src/sheets/client.py from google.oauth2.service_account import Credentials from googleapiclient.discovery import build from dataclasses import dataclass @dataclass class SheetData: """시트 데이터""" headers: list[str] rows: list[dict] raw: list[list] class GoogleSheetsClient: """Google Sheets 클라이언트""" SCOPES = ['https://www.googleapis.com/auth/spreadsheets.readonly'] def __init__(self, credentials_path: str): creds = Credentials.from_service_account_file( credentials_path, scopes=self.SCOPES ) self.service = build('sheets', 'v4', credentials=creds) def read_sheet(self, sheet_id: str, range_name: str) -> SheetData: """시트 데이터 읽기""" result = self.service.spreadsheets().values().get( spreadsheetId=sheet_id, range=range_name ).execute() values = result.get('values', []) if not values: return SheetData(headers=[], rows=[], raw=[]) headers = values[0] rows = [] for row in values[1:]: # 행을 딕셔너리로 변환 row_dict = {} for i, header in enumerate(headers): row_dict[header] = row[i] if i < len(row) else None rows.append(row_dict) return SheetData( headers=headers, rows=rows, raw=values ) def get_kpi_data( self, sheet_id: str, range_name: str, period_column: str = "period" ) -> dict[str, dict]: """KPI 데이터를 기간별 딕셔너리로 변환""" data = self.read_sheet(sheet_id, range_name) result = {} for row in data.rows: period = row.get(period_column) if period: result[period] = { k: self._parse_number(v) for k, v in row.items() if k != period_column } return result def _parse_number(self, value) -> float | None: """값을 숫자로 파싱""" if value is None: return None try: # 쉼표 제거, 퍼센트 처리 clean = str(value).replace(",", "").replace("%", "") return float(clean) except ValueError: return None ``` ### 4.3 Week 2: KPI 계산 엔진 ```python # src/kpi/tree.py from dataclasses import dataclass, field from enum import Enum from typing import Optional class NodeType(Enum): KGI = "kgi" KPI = "kpi" INPUT = "input" @dataclass class KPINode: """KPI 노드""" id: str name: str node_type: NodeType formula: Optional[str] = None unit: str = "" value: Optional[float] = None target: Optional[float] = None children: list["KPINode"] = field(default_factory=list) parent: Optional["KPINode"] = None @property def achievement_rate(self) -> Optional[float]: """달성률""" if self.value is None or self.target is None or self.target == 0: return None return self.value / self.target * 100 @property def is_calculated(self) -> bool: """계산식이 있는지""" return self.formula is not None @dataclass class KPITree: """KPI 트리""" root: KPINode nodes: dict[str, KPINode] = field(default_factory=dict) @classmethod def from_config(cls, config: list[dict]) -> "KPITree": """설정에서 트리 생성""" nodes = {} # 1단계: 노드 생성 for node_config in config: node = KPINode( id=node_config["id"], name=node_config["name"], node_type=NodeType(node_config["type"]), formula=node_config.get("formula"), unit=node_config.get("unit", "") ) nodes[node.id] = node # 2단계: 부모-자식 관계 설정 for node_config in config: node = nodes[node_config["id"]] for child_id in node_config.get("children", []): if child_id in nodes: child = nodes[child_id] node.children.append(child) child.parent = node # 3단계: 루트 노드 찾기 (KGI) root = None for node in nodes.values(): if node.node_type == NodeType.KGI: root = node break if root is None: raise ValueError("KGI node not found") return cls(root=root, nodes=nodes) def set_values(self, values: dict[str, float]): """노드 값 설정""" for node_id, value in values.items(): if node_id in self.nodes: self.nodes[node_id].value = value def set_targets(self, targets: dict[str, float]): """목표 값 설정""" for node_id, target in targets.items(): if node_id in self.nodes: self.nodes[node_id].target = target ``` ```python # src/kpi/calculator.py import ast import re from .tree import KPITree, KPINode, NodeType class KPICalculator: """KPI 계산 엔진""" def __init__(self, tree: KPITree): self.tree = tree def calculate(self) -> dict[str, float]: """모든 KPI 계산""" # 위상정렬된 순서로 계산 order = self._get_calculation_order() results = {} for node_id in order: node = self.tree.nodes[node_id] if node.formula: # 수식 계산 value = self._evaluate_formula(node.formula, results) node.value = value results[node_id] = value elif node.value is not None: results[node_id] = node.value return results def _get_calculation_order(self) -> list[str]: """위상정렬 - 리프 노드부터 루트까지""" order = [] visited = set() def dfs(node: KPINode): if node.id in visited: return visited.add(node.id) # 자식 먼저 방문 for child in node.children: dfs(child) order.append(node.id) dfs(self.tree.root) return order def _evaluate_formula(self, formula: str, values: dict[str, float]) -> float: """수식 평가 (안전한 방식)""" # {variable} 형식의 변수 치환 expression = formula for var_name, value in values.items(): expression = expression.replace(f"{{{var_name}}}", str(value)) # 미치환된 변수 확인 if re.search(r"\{[^}]+\}", expression): raise ValueError(f"Unresolved variables in formula: {formula}") # AST로 안전하게 계산 try: tree = ast.parse(expression, mode='eval') return self._safe_eval(tree.body) except Exception as e: raise ValueError(f"Failed to evaluate formula: {formula}") from e def _safe_eval(self, node: ast.AST) -> float: """AST 기반 안전한 수식 계산""" if isinstance(node, ast.Constant): return float(node.value) elif isinstance(node, ast.BinOp): left = self._safe_eval(node.left) right = self._safe_eval(node.right) if isinstance(node.op, ast.Add): return left + right elif isinstance(node.op, ast.Sub): return left - right elif isinstance(node.op, ast.Mult): return left * right elif isinstance(node.op, ast.Div): if right == 0: return 0 return left / right else: raise ValueError(f"Unsupported operator: {type(node.op)}") elif isinstance(node, ast.UnaryOp): if isinstance(node.op, ast.USub): return -self._safe_eval(node.operand) else: raise ValueError(f"Unsupported node type: {type(node)}") ``` ### 4.4 Week 3: AI 분석 ```python # src/ai/prompts.py SYSTEM_PROMPT = """あなたは企業のKPI分析の専門家です。 提供されたKPIデータを分析し、経営層向けのインサイトを提供してください。 回答のルール: 1. 数値は具体的に記載(例:前月比+15.3%) 2. 日本語で回答 3. 経営層にも分かりやすい表現を使用 4. 重要度の高い順に記載 """ TREND_ANALYSIS_PROMPT = """## KPIトレンド分析 ### 分析対象期間 {period} ### KPIデータ ```json {kpi_data} ``` ### KPIツリー構造 {tree_structure} ### 分析してほしいこと 1. 各KPIの推移傾向(上昇/安定/下降) 2. 目標達成率の評価 3. 特に注目すべき変化 4. KPI間の相関関係 ### 出力形式 以下のJSON形式で回答してください: ```json {{ "summary": "全体サマリー(2-3文)", "trends": [ {{ "kpi_id": "KPI ID", "kpi_name": "KPI名", "trend": "improving/stable/declining", "change_rate": "変化率(例:+15%)", "insight": "具体的なインサイト" }} ], "highlights": ["注目ポイント1", "注目ポイント2", "注目ポイント3"] }} ``` """ BOTTLENECK_ANALYSIS_PROMPT = """## ボトルネック分析 ### 現状 KGI「{kgi_name}」の達成率: {achievement_rate}% ### 各KPIの達成状況 ```json {performance_data} ``` ### KPIツリー構造 {tree_structure} ### 分析してほしいこと 1. KGI未達の主要原因となっているKPI 2. 各ボトルネックの影響度 3. 改善の優先順位 4. 具体的な改善アクション ### 出力形式 ```json {{ "summary": "ボトルネック分析サマリー", "bottlenecks": [ {{ "kpi_id": "KPI ID", "kpi_name": "KPI名", "current": 現在値, "target": 目標値, "achievement_rate": 達成率, "impact": "high/medium/low", "root_cause": "推測される原因", "actions": ["改善アクション1", "改善アクション2"] }} ], "priority_actions": ["最優先で取り組むべきアクション"] }} ``` """ ``` ```python # src/ai/client.py import json import re import google.generativeai as genai from .prompts import SYSTEM_PROMPT, TREND_ANALYSIS_PROMPT, BOTTLENECK_ANALYSIS_PROMPT class GeminiAnalyzer: """Gemini AI 分析クライアント""" def __init__(self, api_key: str): genai.configure(api_key=api_key) self.model = genai.GenerativeModel('gemini-2.0-flash-exp') def analyze_trends( self, kpi_data: dict, tree_structure: str, period: str ) -> dict: """トレンド分析""" prompt = TREND_ANALYSIS_PROMPT.format( period=period, kpi_data=json.dumps(kpi_data, ensure_ascii=False, indent=2), tree_structure=tree_structure ) return self._analyze(prompt) def analyze_bottlenecks( self, performance_data: dict, tree_structure: str, kgi_name: str, achievement_rate: float ) -> dict: """ボトルネック分析""" prompt = BOTTLENECK_ANALYSIS_PROMPT.format( kgi_name=kgi_name, achievement_rate=f"{achievement_rate:.1f}", performance_data=json.dumps(performance_data, ensure_ascii=False, indent=2), tree_structure=tree_structure ) return self._analyze(prompt) def _analyze(self, user_prompt: str) -> dict: """AI分析実行""" full_prompt = f"{SYSTEM_PROMPT}\n\n{user_prompt}" response = self.model.generate_content( full_prompt, generation_config=genai.GenerationConfig( temperature=0.3, max_output_tokens=4096, ) ) return self._parse_response(response.text) def _parse_response(self, text: str) -> dict: """レスポンスをJSONとしてパース""" # JSONブロック抽出 json_match = re.search(r'```json\s*([\s\S]*?)\s*```', text) if json_match: try: return json.loads(json_match.group(1)) except json.JSONDecodeError: pass # JSONブロックがない場合 try: return json.loads(text) except json.JSONDecodeError: return {"raw_response": text, "parsed": False} ``` ### 4.5 Week 4: HTML 출력 & CLI ```python # src/output/html.py from pathlib import Path from jinja2 import Environment, FileSystemLoader from datetime import datetime class HTMLReportGenerator: """HTML 보고서 생성기""" def __init__(self, template_dir: str = "templates"): self.env = Environment( loader=FileSystemLoader(template_dir), autoescape=True ) def generate( self, tree_data: dict, trend_analysis: dict, bottleneck_analysis: dict, period: str, output_path: str ) -> str: """보고서 생성""" template = self.env.get_template("report.html") html = template.render( tree=tree_data, trends=trend_analysis, bottlenecks=bottleneck_analysis, period=period, generated_at=datetime.now().strftime("%Y-%m-%d %H:%M"), title="PlanitAI KPI Report" ) # 파일 저장 output_file = Path(output_path) / f"report_{period}.html" output_file.parent.mkdir(parents=True, exist_ok=True) with open(output_file, "w", encoding="utf-8") as f: f.write(html) return str(output_file) ``` ```python # src/main.py import click from pathlib import Path from .config import Config from .sheets.client import GoogleSheetsClient from .kpi.tree import KPITree from .kpi.calculator import KPICalculator from .ai.client import GeminiAnalyzer from .output.html import HTMLReportGenerator @click.command() @click.option('--config', '-c', default='config.yaml', help='Config file path') @click.option('--period', '-p', required=True, help='Analysis period (e.g., 2024-11)') @click.option('--output', '-o', default='./output', help='Output directory') def main(config: str, period: str, output: str): """PlanitAI KPI - AI-Powered KPI Analysis Tool""" click.echo("🚀 PlanitAI KPI Starting...") # 1. 설정 로드 click.echo("📁 Loading configuration...") cfg = Config.from_yaml(config) # 2. Google Sheets 데이터 가져오기 click.echo("📊 Fetching data from Google Sheets...") sheets = GoogleSheetsClient("credentials.json") kpi_data = sheets.get_kpi_data(cfg.sheet_id, cfg.sheet_range) if period not in kpi_data: click.echo(f"❌ No data found for period: {period}") return # 3. KPI 트리 구성 click.echo("🌳 Building KPI tree...") tree = KPITree.from_config([n.__dict__ for n in cfg.tree]) tree.set_values(kpi_data[period]) # 4. KPI 계산 click.echo("🔢 Calculating KPIs...") calculator = KPICalculator(tree) results = calculator.calculate() # 5. AI 분석 click.echo("🤖 Running AI analysis...") analyzer = GeminiAnalyzer(cfg.gemini_api_key) trend_analysis = analyzer.analyze_trends( kpi_data=kpi_data, tree_structure=_tree_to_text(tree), period=period ) bottleneck_analysis = analyzer.analyze_bottlenecks( performance_data=_get_performance_data(tree), tree_structure=_tree_to_text(tree), kgi_name=tree.root.name, achievement_rate=tree.root.achievement_rate or 0 ) # 6. 보고서 생성 click.echo("📝 Generating report...") generator = HTMLReportGenerator() output_file = generator.generate( tree_data=_tree_to_dict(tree), trend_analysis=trend_analysis, bottleneck_analysis=bottleneck_analysis, period=period, output_path=output ) click.echo(f"✅ Report generated: {output_file}") def _tree_to_text(tree: KPITree) -> str: """트리를 텍스트로 변환""" lines = [] def _node_to_text(node, indent=0): prefix = " " * indent lines.append(f"{prefix}- {node.name} ({node.id})") if node.formula: lines.append(f"{prefix} Formula: {node.formula}") for child in node.children: _node_to_text(child, indent + 1) _node_to_text(tree.root) return "\n".join(lines) def _tree_to_dict(tree: KPITree) -> dict: """트리를 딕셔너리로 변환""" def _node_to_dict(node): return { "id": node.id, "name": node.name, "type": node.node_type.value, "value": node.value, "target": node.target, "achievement_rate": node.achievement_rate, "unit": node.unit, "children": [_node_to_dict(c) for c in node.children] } return _node_to_dict(tree.root) def _get_performance_data(tree: KPITree) -> list[dict]: """성과 데이터 추출""" data = [] for node in tree.nodes.values(): data.append({ "id": node.id, "name": node.name, "type": node.node_type.value, "value": node.value, "target": node.target, "achievement_rate": node.achievement_rate }) return data if __name__ == "__main__": main() ``` --- ## 5. 기술 부채 관리 ### 5.1 인지된 기술 부채 | 항목 | 현재 상태 | 향후 개선 | |------|----------|----------| | 인증 | 없음 (로컬 실행) | JWT 인증 | | DB | 없음 (매번 Sheets 조회) | PostgreSQL | | 캐시 | 없음 | Redis | | 에러 처리 | 기본 | 구조화된 에러 | | 로깅 | print/click.echo | structlog | | 테스트 | 수동 | pytest 자동화 | ### 5.2 TODO 트래킹 ```python # src/kpi/calculator.py def _safe_eval(self, node: ast.AST) -> float: # TODO: 더 많은 연산자 지원 (%, **) # TODO: 함수 지원 (min, max, avg) # TODO: 조건문 지원 (if) ... # src/ai/client.py def _analyze(self, user_prompt: str) -> dict: # TODO: 재시도 로직 추가 # TODO: 토큰 사용량 추적 # TODO: 응답 캐싱 ... ``` --- ## 6. 성공 기준 ### 6.1 MVP 완료 기준 ```markdown ## Definition of Done (MVP) ### 기능 - [ ] Google Sheets에서 KPI 데이터 읽기 가능 - [ ] YAML 설정으로 KPI 트리 정의 가능 - [ ] 모든 KPI 자동 계산 완료 - [ ] AI 트렌드 분석 결과 생성 - [ ] AI 병목 분석 결과 생성 - [ ] HTML 보고서 출력 ### 품질 - [ ] 핵심 기능 수동 테스트 통과 - [ ] 에러 시 친절한 메시지 출력 - [ ] README 문서 작성 ### 검증 - [ ] 3개 이상의 실제 KPI 트리로 테스트 - [ ] 1명 이상의 외부 사용자 피드백 ``` ### 6.2 피드백 수집 ```markdown ## MVP 피드백 수집 항목 1. **가치 검증** - KPI 트리 시각화가 유용했나요? - AI 분석 인사이트가 도움이 되었나요? - 이 도구를 유료로 사용할 의향이 있나요? 2. **사용성** - 설정 파일 작성이 어려웠나요? - 보고서 출력이 만족스러웠나요? - 어떤 기능이 추가되면 좋겠나요? 3. **기술** - 실행 속도는 만족스러웠나요? - 에러 메시지가 이해하기 쉬웠나요? ``` --- ## 7. まとめ ### MVP 개발 계획 요약 | 주차 | 목표 | 산출물 | |------|------|-------| | Week 1 | 기반 구축 | 설정, Sheets 연동 | | Week 2 | 계산 엔진 | KPI 트리, 계산기 | | Week 3 | AI 분석 | Gemini 연동, 분석 | | Week 4 | 통합 | CLI, HTML 출력 | ### 次回予告 v14에서는 **로드맵 및 마일스톤**을 다룹니다: - Phase 1~3 상세 일정 - 마일스톤 정의 - 리스크 관리 --- *PlanitAI KPI - AI가 당신의 KPI를 계획하고 분석합니다*