MCP Localization Project

#!/usr/bin/env python3 """ Shrimp Task Manager - Metacognition System 로컬 AI의 메타인지 시스템: 자기 성찰과 품질 관리 이 모듈은 AI의 메타인지 능력을 구현합니다: - process_thought: 구조화된 사고 과정 관리 - reflect_task: 비판적 성찰 및 개선점 도출 - verify_task: 엄격한 품질 검증 및 점수 산정 메타인지는 AI가 자신의 사고 과정을 의식적으로 관찰하고 개선하는 능력으로, 로컬 AI가 GPT-4o를 뛰어넘을 수 있는 핵심 차별화 요소입니다. """ import os import re import json import time import asyncio import logging from typing import Dict, List, Optional, Any, Tuple, Set from datetime import datetime, timedelta from dataclasses import dataclass, field from enum import Enum # 로컬 모듈 임포트 from .shrimp_base import ( Task, TaskStatus, TaskPriority, TaskCategory, ThoughtProcess, ThoughtStage, Evidence, QualityScore, generate_thought_id, generate_task_id ) # 로깅 설정 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # ============================================================================= # 메타인지 관련 데이터 클래스 (Metacognition Data Classes) # ============================================================================= @dataclass class ReflectionResult: """성찰 결과""" task_id: str # 대상 작업 ID analysis_summary: str # 분석 요약 strengths_identified: List[str] # 식별된 강점들 weaknesses_identified: List[str] # 식별된 약점들 improvement_opportunities: List[str] # 개선 기회들 optimization_suggestions: List[str] # 최적화 제안들 lessons_learned: List[str] # 학습된 교훈들 pattern_recognition: Dict[str, Any] # 패턴 인식 결과 confidence_score: float # 신뢰도 점수 (0.0-1.0) reflection_depth: str # 성찰 깊이 (surface, moderate, deep) created_at: datetime = field(default_factory=datetime.now) def to_dict(self) -> Dict[str, Any]: """딕셔너리로 변환""" return { 'task_id': self.task_id, 'analysis_summary': self.analysis_summary, 'strengths_identified': self.strengths_identified, 'weaknesses_identified': self.weaknesses_identified, 'improvement_opportunities': self.improvement_opportunities, 'optimization_suggestions': self.optimization_suggestions, 'lessons_learned': self.lessons_learned, 'pattern_recognition': self.pattern_recognition, 'confidence_score': self.confidence_score, 'reflection_depth': self.reflection_depth, 'created_at': self.created_at.isoformat() } @dataclass class VerificationResult: """검증 결과""" task_id: str # 대상 작업 ID overall_score: float # 전체 점수 (0-100) # 검증 기준별 점수 (각각 0-100) requirements_compliance: float # 요구사항 충족도 (30%) technical_quality: float # 기술적 품질 (30%) integration_compatibility: float # 통합 호환성 (20%) performance_scalability: float # 성능 확장성 (20%) # 세부 분석 passed_criteria: List[str] # 통과한 기준들 failed_criteria: List[str] # 실패한 기준들 warnings: List[str] # 경고 사항들 recommendations: List[str] # 권장 사항들 # 증거 및 메타데이터 verification_evidence: List[str] # 검증 증거들 verification_method: str # 검증 방법 verifier_notes: str # 검증자 노트 verification_time: float # 검증 소요 시간 (초) created_at: datetime = field(default_factory=datetime.now) def get_quality_grade(self) -> QualityScore: """품질 등급 반환""" if self.overall_score >= 90: return QualityScore.EXCELLENT elif self.overall_score >= 70: return QualityScore.GOOD elif self.overall_score >= 50: return QualityScore.ACCEPTABLE elif self.overall_score >= 30: return QualityScore.POOR else: return QualityScore.CRITICAL def to_dict(self) -> Dict[str, Any]: """딕셔너리로 변환""" return { 'task_id': self.task_id, 'overall_score': self.overall_score, 'requirements_compliance': self.requirements_compliance, 'technical_quality': self.technical_quality, 'integration_compatibility': self.integration_compatibility, 'performance_scalability': self.performance_scalability, 'passed_criteria': self.passed_criteria, 'failed_criteria': self.failed_criteria, 'warnings': self.warnings, 'recommendations': self.recommendations, 'verification_evidence': self.verification_evidence, 'verification_method': self.verification_method, 'verifier_notes': self.verifier_notes, 'verification_time': self.verification_time, 'quality_grade': self.get_quality_grade().value, 'created_at': self.created_at.isoformat() } class ThinkingMode(Enum): """사고 모드""" ANALYTICAL = "analytical" # 분석적 사고 CREATIVE = "creative" # 창조적 사고 CRITICAL = "critical" # 비판적 사고 SYSTEMATIC = "systematic" # 체계적 사고 INTUITIVE = "intuitive" # 직관적 사고 STRATEGIC = "strategic" # 전략적 사고 # ============================================================================= # Metacognition Manager - 핵심 클래스 # ============================================================================= class MetacognitionManager: """메타인지 관리자 - AI의 자기 성찰과 품질 관리""" def __init__(self, working_directory: str = "/home/skyki/qwen2.5"): """초기화""" self.working_directory = working_directory self.thought_processes: Dict[str, ThoughtProcess] = {} self.reflection_results: Dict[str, ReflectionResult] = {} self.verification_results: Dict[str, VerificationResult] = {} # 메타인지 상태 추적 self.current_thinking_mode = ThinkingMode.ANALYTICAL self.cognitive_load = 0.0 # 인지 부하 (0.0-1.0) self.confidence_level = 0.7 # 신뢰도 수준 (0.0-1.0) self.learning_patterns: Dict[str, Any] = {} # 학습된 패턴들 # 품질 기준 self.quality_thresholds = { 'excellent': 90.0, 'good': 70.0, 'acceptable': 50.0, 'poor': 30.0 } logger.info("MetacognitionManager 초기화 완료") # ========================================================================= # process_thought: 구조화된 사고 과정 관리 # ========================================================================= async def process_thought(self, thought: str, thought_number: int, total_thoughts: int, next_thought_needed: bool, stage: str, task_id: Optional[str] = None, assumptions_challenged: List[str] = None, axioms_used: List[str] = None, tags: List[str] = None) -> ThoughtProcess: """ 구조화된 사고 과정 처리 Args: thought: 사고 내용 thought_number: 현재 사고 번호 total_thoughts: 총 사고 수 next_thought_needed: 다음 사고 필요 여부 stage: 사고 단계 task_id: 관련 작업 ID assumptions_challenged: 도전받은 가정들 axioms_used: 사용된 공리들 tags: 태그들 Returns: ThoughtProcess: 처리된 사고 과정 """ logger.info(f"🧠 사고 과정 처리 시작: {thought[:50]}... (단계: {stage})") try: # 1. 사고 단계 검증 및 변환 thinking_stage = self._validate_and_convert_stage(stage) # 2. 사고 내용 분석 thought_analysis = await self._analyze_thought_content(thought, thinking_stage) # 3. 메타인지 처리 metacognitive_insights = await self._apply_metacognitive_processing( thought, thinking_stage, thought_analysis ) # 4. 사고 과정 객체 생성 thought_process = ThoughtProcess( id=generate_thought_id(), task_id=task_id, thought=thought, stage=thinking_stage, thought_number=thought_number, total_thoughts=total_thoughts, next_thought_needed=next_thought_needed, assumptions_challenged=assumptions_challenged or [], axioms_used=axioms_used or [], tags=tags or [] ) # 5. 메타인지 인사이트 추가 self._enrich_thought_with_insights(thought_process, metacognitive_insights) # 6. 사고 연결성 분석 await self._analyze_thought_connectivity(thought_process) # 7. 학습 패턴 업데이트 self._update_learning_patterns(thought_process, thought_analysis) # 8. 저장 self.thought_processes[thought_process.id] = thought_process # 9. 인지 부하 업데이트 self._update_cognitive_load(thought_process, thought_analysis) logger.info(f"✅ 사고 과정 처리 완료: {thought_process.id}") return thought_process except Exception as e: logger.error(f"❌ 사고 과정 처리 실패: {e}") raise async def _analyze_thought_content(self, thought: str, stage: ThoughtStage) -> Dict[str, Any]: """사고 내용 분석""" logger.info("🔍 사고 내용 분석 중...") analysis = { 'complexity': 0.0, 'clarity': 0.0, 'logic_coherence': 0.0, 'evidence_usage': 0.0, 'creativity_index': 0.0, 'critical_thinking': 0.0, 'keyword_analysis': {}, 'cognitive_biases': [], 'reasoning_patterns': [] } # 복잡도 분석 analysis['complexity'] = self._analyze_complexity(thought) # 명확성 분석 analysis['clarity'] = self._analyze_clarity(thought) # 논리적 일관성 분석 analysis['logic_coherence'] = self._analyze_logic_coherence(thought) # 증거 사용 분석 analysis['evidence_usage'] = self._analyze_evidence_usage(thought) # 창의성 지수 계산 analysis['creativity_index'] = self._calculate_creativity_index(thought) # 비판적 사고 분석 analysis['critical_thinking'] = self._analyze_critical_thinking(thought) # 키워드 분석 analysis['keyword_analysis'] = self._extract_thought_keywords(thought) # 인지 편향 탐지 analysis['cognitive_biases'] = self._detect_cognitive_biases(thought) # 추론 패턴 식별 analysis['reasoning_patterns'] = self._identify_reasoning_patterns(thought) return analysis async def _apply_metacognitive_processing(self, thought: str, stage: ThoughtStage, analysis: Dict[str, Any]) -> Dict[str, Any]: """메타인지 처리 적용""" logger.info("🎯 메타인지 처리 적용 중...") insights = { 'self_assessment': {}, 'thinking_quality': {}, 'improvement_suggestions': [], 'pattern_recognition': {}, 'confidence_calibration': {} } # 자기 평가 insights['self_assessment'] = self._perform_self_assessment(thought, stage, analysis) # 사고 품질 평가 insights['thinking_quality'] = self._assess_thinking_quality(analysis) # 개선 제안 insights['improvement_suggestions'] = self._generate_improvement_suggestions(analysis) # 패턴 인식 insights['pattern_recognition'] = self._recognize_thinking_patterns(thought, analysis) # 신뢰도 보정 insights['confidence_calibration'] = self._calibrate_confidence(analysis) return insights # ========================================================================= # reflect_task: 비판적 성찰 및 개선점 도출 # ========================================================================= async def reflect_task(self, task_id: str, analysis: str, previous_reflection: Optional[str] = None) -> ReflectionResult: """ 비판적 성찰 및 개선점 도출 Args: task_id: 대상 작업 ID analysis: 분석 결과 previous_reflection: 이전 성찰 결과 Returns: ReflectionResult: 성찰 결과 """ logger.info(f"🔍 작업 성찰 시작: {task_id}") try: # 1. 작업 및 분석 데이터 수집 task_data = await self._gather_task_data(task_id) analysis_data = await self._parse_analysis_data(analysis) # 2. 강점 분석 strengths = await self._identify_strengths(task_data, analysis_data) # 3. 약점 분석 weaknesses = await self._identify_weaknesses(task_data, analysis_data) # 4. 개선 기회 식별 improvements = await self._identify_improvement_opportunities( task_data, analysis_data, strengths, weaknesses ) # 5. 최적화 제안 생성 optimizations = await self._generate_optimization_suggestions( task_data, analysis_data, improvements ) # 6. 학습된 교훈 추출 lessons = await self._extract_lessons_learned( task_data, analysis_data, strengths, weaknesses ) # 7. 패턴 인식 수행 patterns = await self._perform_pattern_recognition(task_data, analysis_data) # 8. 신뢰도 계산 confidence = self._calculate_reflection_confidence( task_data, analysis_data, strengths, weaknesses, improvements ) # 9. 성찰 깊이 결정 depth = self._determine_reflection_depth(analysis_data, confidence) # 10. 성찰 결과 생성 reflection = ReflectionResult( task_id=task_id, analysis_summary=self._create_analysis_summary(analysis_data), strengths_identified=strengths, weaknesses_identified=weaknesses, improvement_opportunities=improvements, optimization_suggestions=optimizations, lessons_learned=lessons, pattern_recognition=patterns, confidence_score=confidence, reflection_depth=depth ) # 11. 이전 성찰과 비교 (있는 경우) if previous_reflection: reflection = await self._compare_with_previous_reflection( reflection, previous_reflection ) # 12. 저장 self.reflection_results[task_id] = reflection # 13. 메타 학습 수행 await self._perform_meta_learning(reflection) logger.info(f"✅ 작업 성찰 완료: 신뢰도={confidence:.2f}, 깊이={depth}") return reflection except Exception as e: logger.error(f"❌ 작업 성찰 실패: {e}") raise async def _identify_strengths(self, task_data: Dict[str, Any], analysis_data: Dict[str, Any]) -> List[str]: """강점 식별""" logger.info("💪 강점 식별 중...") strengths = [] # 완성도 기반 강점 completion = task_data.get('completion_percentage', 0) if completion >= 90: strengths.append("높은 완성도 달성") # 품질 기반 강점 quality_score = task_data.get('quality_score', 0) if quality_score >= 80: strengths.append("우수한 품질 점수") # 문서화 강점 if task_data.get('implementation_guide') and len(task_data['implementation_guide']) > 100: strengths.append("상세한 구현 가이드 제공") if task_data.get('verification_criteria') and len(task_data['verification_criteria']) > 50: strengths.append("명확한 검증 기준 정의") # 의존성 관리 강점 if task_data.get('dependencies') and len(task_data['dependencies']) > 0: strengths.append("체계적인 의존성 관리") # 증거 기반 의사결정 강점 evidence_count = len(task_data.get('evidence', [])) if evidence_count >= 3: strengths.append("충분한 증거 기반 의사결정") # 실행 로그 품질 log_count = len(task_data.get('execution_log', [])) if log_count >= 5: strengths.append("상세한 실행 과정 기록") # 분석 기반 강점 if analysis_data.get('technical_feasibility', 0) >= 0.8: strengths.append("높은 기술적 실현 가능성") if analysis_data.get('risk_mitigation_coverage', 0) >= 0.7: strengths.append("효과적인 위험 완화 전략") return strengths async def _identify_weaknesses(self, task_data: Dict[str, Any], analysis_data: Dict[str, Any]) -> List[str]: """약점 식별""" logger.info("⚠️ 약점 식별 중...") weaknesses = [] # 완성도 기반 약점 completion = task_data.get('completion_percentage', 0) if completion < 50: weaknesses.append("낮은 완성도") # 품질 기반 약점 quality_score = task_data.get('quality_score', 0) if quality_score < 60: weaknesses.append("품질 점수 개선 필요") # 문서화 약점 if not task_data.get('implementation_guide') or len(task_data['implementation_guide']) < 50: weaknesses.append("구현 가이드 부족") if not task_data.get('verification_criteria') or len(task_data['verification_criteria']) < 20: weaknesses.append("검증 기준 불명확") # 의존성 관리 약점 if not task_data.get('dependencies'): weaknesses.append("의존성 분석 미흡") # 증거 부족 evidence_count = len(task_data.get('evidence', [])) if evidence_count < 2: weaknesses.append("증거 기반 의사결정 부족") # 실행 과정 추적 부족 log_count = len(task_data.get('execution_log', [])) if log_count < 3: weaknesses.append("실행 과정 기록 부족") # 분석 기반 약점 if analysis_data.get('technical_complexity', 0) >= 0.8: weaknesses.append("높은 기술적 복잡도") risk_factors = analysis_data.get('risk_factors', []) if len(risk_factors) >= 5: weaknesses.append("다수의 위험 요소") # 시간 관리 약점 estimated_hours = task_data.get('estimated_hours', 0) actual_hours = task_data.get('actual_hours', 0) if actual_hours > estimated_hours * 1.5: weaknesses.append("시간 예측 정확도 낮음") return weaknesses # ========================================================================= # verify_task: 엄격한 품질 검증 및 점수 산정 # ========================================================================= async def verify_task(self, task_id: str, summary: str, score: float, verification_criteria: Optional[str] = None) -> VerificationResult: """ 엄격한 품질 검증 및 점수 산정 Args: task_id: 대상 작업 ID summary: 작업 요약 score: 제안된 점수 (0-100) verification_criteria: 검증 기준 Returns: VerificationResult: 검증 결과 """ logger.info(f"🔍 작업 검증 시작: {task_id} (제안 점수: {score})") start_time = time.time() try: # 1. 작업 데이터 수집 task_data = await self._gather_task_data_for_verification(task_id) # 2. 검증 기준 분석 criteria = await self._analyze_verification_criteria( task_data, verification_criteria ) # 3. 요구사항 충족도 검증 (30%) requirements_score = await self._verify_requirements_compliance( task_data, criteria, summary ) # 4. 기술적 품질 검증 (30%) technical_score = await self._verify_technical_quality( task_data, criteria, summary ) # 5. 통합 호환성 검증 (20%) integration_score = await self._verify_integration_compatibility( task_data, criteria, summary ) # 6. 성능 확장성 검증 (20%) performance_score = await self._verify_performance_scalability( task_data, criteria, summary ) # 7. 전체 점수 계산 calculated_score = ( requirements_score * 0.3 + technical_score * 0.3 + integration_score * 0.2 + performance_score * 0.2 ) # 8. 점수 조정 (제안 점수와 계산 점수 비교) final_score = self._reconcile_scores(score, calculated_score, task_data) # 9. 통과/실패 기준 분석 passed, failed = self._analyze_pass_fail_criteria( requirements_score, technical_score, integration_score, performance_score, criteria ) # 10. 경고 및 권장사항 생성 warnings, recommendations = await self._generate_warnings_and_recommendations( task_data, requirements_score, technical_score, integration_score, performance_score ) # 11. 검증 증거 수집 evidence = await self._collect_verification_evidence( task_data, requirements_score, technical_score, integration_score, performance_score ) # 12. 검증 결과 생성 verification_time = time.time() - start_time result = VerificationResult( task_id=task_id, overall_score=final_score, requirements_compliance=requirements_score, technical_quality=technical_score, integration_compatibility=integration_score, performance_scalability=performance_score, passed_criteria=passed, failed_criteria=failed, warnings=warnings, recommendations=recommendations, verification_evidence=evidence, verification_method="comprehensive_analysis", verifier_notes=self._generate_verifier_notes(task_data, final_score), verification_time=verification_time ) # 13. 저장 self.verification_results[task_id] = result # 14. 작업 상태 업데이트 (점수 기준) await self._update_task_status_based_on_verification(task_id, result) logger.info(f"✅ 작업 검증 완료: 최종 점수={final_score:.1f} ({result.get_quality_grade().value})") return result except Exception as e: logger.error(f"❌ 작업 검증 실패: {e}") raise async def _verify_requirements_compliance(self, task_data: Dict[str, Any], criteria: Dict[str, Any], summary: str) -> float: """요구사항 충족도 검증""" logger.info("📋 요구사항 충족도 검증 중...") score = 0.0 max_score = 100.0 # 기능 완성도 (40%) completion = task_data.get('completion_percentage', 0) score += (completion / 100.0) * 40.0 # 요구사항 추적성 (25%) requirements_coverage = self._assess_requirements_coverage(task_data, summary) score += requirements_coverage * 25.0 # 제약사항 준수 (20%) constraints_adherence = self._assess_constraints_adherence(task_data, criteria) score += constraints_adherence * 20.0 # 사용자 스토리 충족 (15%) user_story_fulfillment = self._assess_user_story_fulfillment(task_data, summary) score += user_story_fulfillment * 15.0 return min(score, max_score) async def _verify_technical_quality(self, task_data: Dict[str, Any], criteria: Dict[str, Any], summary: str) -> float: """기술적 품질 검증""" logger.info("⚙️ 기술적 품질 검증 중...") score = 0.0 max_score = 100.0 # 코드 품질 (40%) code_quality = self._assess_code_quality(task_data) score += code_quality * 40.0 # 아키텍처 일관성 (25%) architecture_consistency = self._assess_architecture_consistency(task_data) score += architecture_consistency * 25.0 # 보안 고려사항 (20%) security_considerations = self._assess_security_considerations(task_data) score += security_considerations * 20.0 # 에러 처리 (15%) error_handling = self._assess_error_handling(task_data) score += error_handling * 15.0 return min(score, max_score) # ========================================================================= # 유틸리티 및 헬퍼 메서드들 # ========================================================================= def _validate_and_convert_stage(self, stage: str) -> ThoughtStage: """사고 단계 검증 및 변환""" try: return ThoughtStage(stage) except ValueError: logger.warning(f"⚠️ 알 수 없는 사고 단계: {stage}, 기본값 사용") return ThoughtStage.ANALYSIS def _analyze_complexity(self, thought: str) -> float: """사고 복잡도 분석""" complexity = 0.0 # 텍스트 길이 기반 complexity += min(len(thought) / 1000, 0.3) # 문장 구조 복잡성 sentences = thought.split('.') avg_sentence_length = sum(len(s.split()) for s in sentences) / max(len(sentences), 1) complexity += min(avg_sentence_length / 50, 0.3) # 기술 용어 밀도 tech_terms = re.findall(r'\b(implement|algorithm|architecture|system|process|analysis)\b', thought.lower()) complexity += min(len(tech_terms) / 10, 0.4) return min(complexity, 1.0) def _analyze_clarity(self, thought: str) -> float: """사고 명확성 분석""" clarity = 1.0 # 모호한 표현 감점 ambiguous_words = ['maybe', 'perhaps', 'possibly', 'might', 'could'] ambiguous_count = sum(1 for word in ambiguous_words if word in thought.lower()) clarity -= min(ambiguous_count * 0.1, 0.3) # 명확한 구조 가점 if any(marker in thought.lower() for marker in ['first', 'second', 'then', 'finally']): clarity += 0.1 # 구체적 예시 가점 if any(marker in thought.lower() for marker in ['example', 'for instance', 'such as']): clarity += 0.1 return max(min(clarity, 1.0), 0.0) def _update_cognitive_load(self, thought_process: ThoughtProcess, analysis: Dict[str, Any]): """인지 부하 업데이트""" complexity = analysis.get('complexity', 0.0) current_load = self.cognitive_load # 지수 이동 평균으로 인지 부하 업데이트 alpha = 0.3 self.cognitive_load = alpha * complexity + (1 - alpha) * current_load # 인지 부하가 높으면 경고 if self.cognitive_load > 0.8: logger.warning(f"⚠️ 높은 인지 부하 감지: {self.cognitive_load:.2f}") if __name__ == "__main__": # 기본 테스트 print("🚀 Metacognition System 테스트") async def test_metacognition(): manager = MetacognitionManager() # process_thought 테스트 thought_process = await manager.process_thought( thought="사용자 인증 시스템을 구현하기 위해서는 보안과 사용자 경험을 모두 고려해야 합니다.", thought_number=1, total_thoughts=3, next_thought_needed=True, stage="Analysis", assumptions_challenged=["단순한 구현으로 충분할 것이다"], axioms_used=["보안이 최우선이다"] ) print(f"✅ process_thought 성공: {thought_process.id}") # reflect_task 테스트 sample_task_id = "TASK-20250719-TEST001" reflection = await manager.reflect_task( task_id=sample_task_id, analysis="기술적 분석이 완료되었으며, 높은 품질의 구현이 가능함을 확인했습니다." ) print(f"✅ reflect_task 성공: 신뢰도={reflection.confidence_score:.2f}") # verify_task 테스트 verification = await manager.verify_task( task_id=sample_task_id, summary="사용자 인증 시스템이 성공적으로 구현되었으며 모든 보안 요구사항을 충족합니다.", score=85.0 ) print(f"✅ verify_task 성공: 최종 점수={verification.overall_score:.1f}") print(f"✅ 품질 등급: {verification.get_quality_grade().value}") print("🎯 모든 Metacognition 테스트 완료!") # 비동기 테스트 실행 import asyncio asyncio.run(test_metacognition())