MCP Localization Project

#!/usr/bin/env python3 """ Shrimp Task Manager - Execution & Control System 로컬 AI의 실행 및 제어 시스템: 지능형 작업 실행과 연구 이 모듈은 계획된 작업의 실제 실행을 담당합니다: - execute_task: 지능형 작업 실행 및 실시간 모니터링 - research_mode: 체계적 연구 및 심층 조사 실행 시스템은 AI가 계획한 작업을 체계적으로 수행하고, 진행 과정을 모니터링하며 필요에 따라 적응적으로 조정하는 능력을 제공합니다. """ import os import re import json import time import asyncio import logging import subprocess from typing import Dict, List, Optional, Any, Tuple, Set, Callable from datetime import datetime, timedelta from dataclasses import dataclass, field from enum import Enum from pathlib import Path # 로컬 모듈 임포트 from .shrimp_base import ( Task, TaskStatus, TaskPriority, TaskCategory, RelatedFile, FileRelationType, Evidence, generate_task_id ) # 로깅 설정 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # ============================================================================= # 실행 관련 데이터 클래스 (Execution Data Classes) # ============================================================================= class ExecutionStatus(Enum): """실행 상태""" INITIALIZING = "initializing" # 초기화 중 PLANNING = "planning" # 계획 수립 중 EXECUTING = "executing" # 실행 중 MONITORING = "monitoring" # 모니터링 중 ADAPTING = "adapting" # 적응 중 COMPLETING = "completing" # 완료 중 SUCCEEDED = "succeeded" # 성공 FAILED = "failed" # 실패 PAUSED = "paused" # 일시정지 CANCELLED = "cancelled" # 취소 class ResearchPhase(Enum): """연구 단계""" TOPIC_ANALYSIS = "topic_analysis" # 주제 분석 INFORMATION_GATHERING = "information_gathering" # 정보 수집 SOURCE_VALIDATION = "source_validation" # 출처 검증 PATTERN_IDENTIFICATION = "pattern_identification" # 패턴 식별 SYNTHESIS = "synthesis" # 종합 CONCLUSION = "conclusion" # 결론 @dataclass class ExecutionStep: """실행 단계""" step_id: str # 단계 ID name: str # 단계 이름 description: str # 단계 설명 command: str # 실행할 명령 expected_outcome: str # 예상 결과 prerequisites: List[str] = field(default_factory=list) # 전제조건 success_criteria: List[str] = field(default_factory=list) # 성공 기준 failure_indicators: List[str] = field(default_factory=list) # 실패 지표 timeout_seconds: float = 300.0 # 타임아웃 (초) retry_count: int = 0 # 재시도 횟수 max_retries: int = 3 # 최대 재시도 # 실행 결과 status: ExecutionStatus = ExecutionStatus.INITIALIZING start_time: Optional[datetime] = None end_time: Optional[datetime] = None result: Optional[str] = None error_message: Optional[str] = None output_files: List[str] = field(default_factory=list) def to_dict(self) -> Dict[str, Any]: """딕셔너리로 변환""" return { 'step_id': self.step_id, 'name': self.name, 'description': self.description, 'command': self.command, 'expected_outcome': self.expected_outcome, 'prerequisites': self.prerequisites, 'success_criteria': self.success_criteria, 'failure_indicators': self.failure_indicators, 'timeout_seconds': self.timeout_seconds, 'retry_count': self.retry_count, 'max_retries': self.max_retries, 'status': self.status.value, 'start_time': self.start_time.isoformat() if self.start_time else None, 'end_time': self.end_time.isoformat() if self.end_time else None, 'result': self.result, 'error_message': self.error_message, 'output_files': self.output_files } @dataclass class ExecutionPlan: """실행 계획""" task_id: str # 작업 ID execution_strategy: str # 실행 전략 steps: List[ExecutionStep] # 실행 단계들 dependencies: List[str] = field(default_factory=list) # 의존성 resource_requirements: Dict[str, Any] = field(default_factory=dict) # 리소스 요구사항 risk_mitigation: List[str] = field(default_factory=list) # 위험 완화 방안 monitoring_points: List[str] = field(default_factory=list) # 모니터링 지점 # 실행 상태 overall_status: ExecutionStatus = ExecutionStatus.INITIALIZING current_step: int = 0 success_rate: float = 0.0 estimated_completion: Optional[datetime] = None def to_dict(self) -> Dict[str, Any]: """딕셔너리로 변환""" return { 'task_id': self.task_id, 'execution_strategy': self.execution_strategy, 'steps': [step.to_dict() for step in self.steps], 'dependencies': self.dependencies, 'resource_requirements': self.resource_requirements, 'risk_mitigation': self.risk_mitigation, 'monitoring_points': self.monitoring_points, 'overall_status': self.overall_status.value, 'current_step': self.current_step, 'success_rate': self.success_rate, 'estimated_completion': self.estimated_completion.isoformat() if self.estimated_completion else None } @dataclass class ResearchResult: """연구 결과""" topic: str # 연구 주제 current_state: str # 현재 상태 next_steps: str # 다음 단계 previous_state: str = "" # 이전 상태 # 연구 내용 findings: List[str] = field(default_factory=list) # 발견사항 sources: List[str] = field(default_factory=list) # 정보 출처 evidence: List[Evidence] = field(default_factory=list) # 수집된 증거 patterns: List[str] = field(default_factory=list) # 식별된 패턴 insights: List[str] = field(default_factory=list) # 통찰 recommendations: List[str] = field(default_factory=list) # 권장사항 # 메타데이터 research_phase: ResearchPhase = ResearchPhase.TOPIC_ANALYSIS confidence_level: float = 0.0 # 신뢰도 (0.0-1.0) completeness: float = 0.0 # 완성도 (0.0-1.0) created_at: datetime = field(default_factory=datetime.now) def to_dict(self) -> Dict[str, Any]: """딕셔너리로 변환""" return { 'topic': self.topic, 'current_state': self.current_state, 'next_steps': self.next_steps, 'previous_state': self.previous_state, 'findings': self.findings, 'sources': self.sources, 'evidence': [ev.to_dict() for ev in self.evidence], 'patterns': self.patterns, 'insights': self.insights, 'recommendations': self.recommendations, 'research_phase': self.research_phase.value, 'confidence_level': self.confidence_level, 'completeness': self.completeness, 'created_at': self.created_at.isoformat() } # ============================================================================= # Execution & Control Manager - 핵심 클래스 # ============================================================================= class ExecutionControlManager: """실행 및 제어 관리자""" def __init__(self, working_directory: str = "/home/skyki/qwen2.5"): """초기화""" self.working_directory = working_directory self.execution_plans: Dict[str, ExecutionPlan] = {} self.research_results: Dict[str, ResearchResult] = {} self.active_executions: Dict[str, asyncio.Task] = {} # 실행 모니터링 self.execution_metrics = { 'total_executions': 0, 'successful_executions': 0, 'failed_executions': 0, 'average_execution_time': 0.0, 'resource_usage': {} } # 도구 인터페이스 (실제 구현에서는 MCP 도구들과 연결) self.tool_interfaces = { 'terminal': self._mock_terminal_interface, 'filesystem': self._mock_filesystem_interface, 'editor': self._mock_editor_interface, 'browser': self._mock_browser_interface, 'debugger': self._mock_debugger_interface } logger.info("ExecutionControlManager 초기화 완료") # ========================================================================= # execute_task: 지능형 작업 실행 # ========================================================================= async def execute_task(self, task_id: str) -> Tuple[bool, str]: """ 지능형 작업 실행 및 실시간 모니터링 Args: task_id: 실행할 작업 ID Returns: Tuple[bool, str]: (성공여부, 실행 결과 메시지) """ logger.info(f"🚀 작업 실행 시작: {task_id}") try: # 1. 작업 데이터 로드 task_data = await self._load_task_data(task_id) if not task_data: raise ValueError(f"작업 {task_id}를 찾을 수 없습니다") # 2. 실행 계획 수립 execution_plan = await self._create_execution_plan(task_data) self.execution_plans[task_id] = execution_plan # 3. 실행 전 검증 validation_result = await self._validate_execution_prerequisites(execution_plan) if not validation_result['valid']: return False, f"실행 전 검증 실패: {validation_result['error']}" # 4. 실행 시작 execution_plan.overall_status = ExecutionStatus.EXECUTING execution_task = asyncio.create_task( self._execute_plan_with_monitoring(execution_plan) ) self.active_executions[task_id] = execution_task # 5. 실행 완료 대기 execution_result = await execution_task # 6. 실행 후 처리 await self._post_execution_processing(task_id, execution_plan, execution_result) # 7. 메트릭 업데이트 self._update_execution_metrics(execution_plan, execution_result) # 8. 결과 반환 if execution_result['success']: logger.info(f"✅ 작업 실행 성공: {task_id}") return True, execution_result['message'] else: logger.error(f"❌ 작업 실행 실패: {task_id}") return False, execution_result['message'] except Exception as e: logger.error(f"❌ 작업 실행 중 오류 발생: {e}") return False, f"실행 중 오류: {e}" finally: # 정리 if task_id in self.active_executions: del self.active_executions[task_id] async def _create_execution_plan(self, task_data: Dict[str, Any]) -> ExecutionPlan: """실행 계획 수립""" logger.info("📋 실행 계획 수립 중...") task_id = task_data['id'] implementation_guide = task_data.get('implementation_guide', '') related_files = task_data.get('related_files', []) # 실행 전략 결정 strategy = self._determine_execution_strategy(task_data) # 실행 단계 생성 steps = await self._generate_execution_steps(task_data, strategy) # 의존성 분석 dependencies = self._analyze_execution_dependencies(task_data, steps) # 리소스 요구사항 계산 resources = self._calculate_resource_requirements(steps) # 위험 완화 방안 수립 risk_mitigation = self._develop_risk_mitigation_strategies(task_data, steps) # 모니터링 지점 설정 monitoring_points = self._setup_monitoring_points(steps) return ExecutionPlan( task_id=task_id, execution_strategy=strategy, steps=steps, dependencies=dependencies, resource_requirements=resources, risk_mitigation=risk_mitigation, monitoring_points=monitoring_points ) async def _execute_plan_with_monitoring(self, plan: ExecutionPlan) -> Dict[str, Any]: """모니터링과 함께 계획 실행""" logger.info("⚡ 계획 실행 및 모니터링 시작...") result = { 'success': False, 'message': '', 'completed_steps': 0, 'failed_steps': 0, 'execution_time': 0.0, 'outputs': [] } start_time = time.time() try: for i, step in enumerate(plan.steps): plan.current_step = i logger.info(f"📌 단계 {i+1}/{len(plan.steps)} 실행: {step.name}") # 단계 실행 step_result = await self._execute_step_with_retry(step) if step_result['success']: result['completed_steps'] += 1 result['outputs'].extend(step_result.get('outputs', [])) # 진행률 업데이트 plan.success_rate = result['completed_steps'] / len(plan.steps) * 100 # 모니터링 체크포인트 await self._check_monitoring_points(plan, i) else: result['failed_steps'] += 1 result['message'] = f"단계 {i+1} 실패: {step_result['error']}" # 실패 처리 전략 if not await self._handle_step_failure(plan, step, step_result): break # 전체 성공 여부 결정 if result['completed_steps'] == len(plan.steps): result['success'] = True result['message'] = "모든 단계 성공적으로 완료" plan.overall_status = ExecutionStatus.SUCCEEDED else: plan.overall_status = ExecutionStatus.FAILED if not result['message']: result['message'] = f"{result['failed_steps']}개 단계 실패" except Exception as e: result['message'] = f"실행 중 예외 발생: {e}" plan.overall_status = ExecutionStatus.FAILED logger.error(f"❌ 계획 실행 중 예외: {e}") finally: result['execution_time'] = time.time() - start_time logger.info(f"📊 실행 완료: {result['completed_steps']}/{len(plan.steps)} 단계, " f"{result['execution_time']:.2f}초") return result async def _execute_step_with_retry(self, step: ExecutionStep) -> Dict[str, Any]: """재시도와 함께 단계 실행""" result = {'success': False, 'error': '', 'outputs': []} step.start_time = datetime.now() step.status = ExecutionStatus.EXECUTING for attempt in range(step.max_retries + 1): try: # 전제조건 확인 if not await self._check_step_prerequisites(step): result['error'] = "전제조건 미충족" continue # 명령 실행 execution_result = await self._execute_command(step.command, step.timeout_seconds) # 성공 기준 확인 if await self._check_success_criteria(step, execution_result): step.status = ExecutionStatus.SUCCEEDED step.result = execution_result['output'] result['success'] = True result['outputs'] = execution_result.get('output_files', []) break else: result['error'] = f"성공 기준 미충족: {execution_result.get('error', '')}" except asyncio.TimeoutError: result['error'] = f"타임아웃 ({step.timeout_seconds}초)" except Exception as e: result['error'] = f"실행 오류: {e}" # 재시도 로직 if attempt < step.max_retries: step.retry_count += 1 wait_time = min(2 ** attempt, 60) # 지수 백오프 logger.warning(f"⚠️ 단계 재시도 {attempt + 1}/{step.max_retries}: {wait_time}초 대기") await asyncio.sleep(wait_time) if not result['success']: step.status = ExecutionStatus.FAILED step.error_message = result['error'] step.end_time = datetime.now() return result # ========================================================================= # research_mode: 체계적 연구 및 심층 조사 # ========================================================================= async def research_mode(self, topic: str, current_state: str, next_steps: str, previous_state: str = "") -> ResearchResult: """ 체계적 연구 및 심층 조사 Args: topic: 연구 주제 current_state: 현재 상태 next_steps: 다음 단계 previous_state: 이전 상태 Returns: ResearchResult: 연구 결과 """ logger.info(f"🔬 연구 모드 시작: {topic[:50]}...") try: # 1. 연구 결과 객체 초기화 research = ResearchResult( topic=topic, current_state=current_state, next_steps=next_steps, previous_state=previous_state ) # 2. 주제 분석 research.research_phase = ResearchPhase.TOPIC_ANALYSIS topic_analysis = await self._analyze_research_topic(topic, current_state) research.findings.extend(topic_analysis['key_areas']) # 3. 정보 수집 research.research_phase = ResearchPhase.INFORMATION_GATHERING information = await self._gather_research_information(topic, topic_analysis) research.sources.extend(information['sources']) research.evidence.extend(information['evidence']) # 4. 출처 검증 research.research_phase = ResearchPhase.SOURCE_VALIDATION validated_info = await self._validate_research_sources(research.sources, research.evidence) research.confidence_level = validated_info['overall_confidence'] # 5. 패턴 식별 research.research_phase = ResearchPhase.PATTERN_IDENTIFICATION patterns = await self._identify_research_patterns(research.findings, research.evidence) research.patterns.extend(patterns['patterns']) research.insights.extend(patterns['insights']) # 6. 종합 및 결론 research.research_phase = ResearchPhase.SYNTHESIS synthesis = await self._synthesize_research_findings(research) research.recommendations.extend(synthesis['recommendations']) # 7. 결론 도출 research.research_phase = ResearchPhase.CONCLUSION research.completeness = self._calculate_research_completeness(research) # 8. 이전 연구와 통합 (있는 경우) if previous_state: research = await self._integrate_with_previous_research(research, previous_state) # 9. 저장 research_id = f"RESEARCH-{datetime.now().strftime('%Y%m%d-%H%M%S')}" self.research_results[research_id] = research logger.info(f"✅ 연구 완료: 신뢰도={research.confidence_level:.2f}, " f"완성도={research.completeness:.2f}") return research except Exception as e: logger.error(f"❌ 연구 실패: {e}") raise async def _analyze_research_topic(self, topic: str, current_state: str) -> Dict[str, Any]: """연구 주제 분석""" logger.info("📊 연구 주제 분석 중...") analysis = { 'key_areas': [], 'research_questions': [], 'scope': '', 'complexity': 0.0, 'required_resources': [] } # 주제 키워드 추출 keywords = self._extract_research_keywords(topic) # 핵심 영역 식별 key_areas = self._identify_key_research_areas(topic, keywords) analysis['key_areas'] = key_areas # 연구 질문 생성 research_questions = self._generate_research_questions(topic, current_state, key_areas) analysis['research_questions'] = research_questions # 범위 정의 analysis['scope'] = self._define_research_scope(topic, key_areas) # 복잡도 평가 analysis['complexity'] = self._assess_research_complexity(topic, key_areas, research_questions) # 필요 리소스 식별 analysis['required_resources'] = self._identify_required_resources(topic, analysis['complexity']) return analysis async def _gather_research_information(self, topic: str, analysis: Dict[str, Any]) -> Dict[str, Any]: """연구 정보 수집""" logger.info("📚 연구 정보 수집 중...") information = { 'sources': [], 'evidence': [], 'data_points': [], 'expert_opinions': [], 'case_studies': [] } # 코드베이스 조사 codebase_info = await self._research_codebase(topic, analysis) information['sources'].extend(codebase_info['sources']) information['evidence'].extend(codebase_info['evidence']) # 문서 조사 documentation_info = await self._research_documentation(topic, analysis) information['sources'].extend(documentation_info['sources']) information['data_points'].extend(documentation_info['data_points']) # 외부 리소스 조사 (모의) external_info = await self._research_external_resources(topic, analysis) information['expert_opinions'].extend(external_info['opinions']) information['case_studies'].extend(external_info['cases']) # 증거 객체 생성 for source in information['sources']: evidence = Evidence( source=source['name'], content=source['content'], reliability=source.get('reliability', 0.7), timestamp=datetime.now(), evidence_type='research_data' ) information['evidence'].append(evidence) return information # ========================================================================= # 모의 도구 인터페이스 (실제 구현에서는 MCP 도구와 연결) # ========================================================================= async def _mock_terminal_interface(self, command: str, timeout: float = 30.0) -> Dict[str, Any]: """모의 터미널 인터페이스""" logger.info(f"🔧 터미널 명령 실행: {command}") # 안전한 명령들만 시뮬레이션 safe_commands = ['ls', 'pwd', 'echo', 'cat', 'grep', 'find', 'wc'] command_parts = command.split() base_command = command_parts[0] if command_parts else '' if base_command in safe_commands: # 시뮬레이션된 결과 await asyncio.sleep(0.1) # 시뮬레이션 지연 return { 'success': True, 'output': f"모의 실행 결과: {command}", 'error': '', 'exit_code': 0, 'output_files': [] } else: return { 'success': False, 'output': '', 'error': f"안전하지 않은 명령: {command}", 'exit_code': 1, 'output_files': [] } async def _mock_filesystem_interface(self, operation: str, **kwargs) -> Dict[str, Any]: """모의 파일시스템 인터페이스""" logger.info(f"📁 파일시스템 작업: {operation}") await asyncio.sleep(0.05) # 시뮬레이션 지연 return { 'success': True, 'result': f"모의 {operation} 작업 완료", 'files_affected': kwargs.get('files', []) } async def _mock_editor_interface(self, file_path: str, changes: List[Dict]) -> Dict[str, Any]: """모의 에디터 인터페이스""" logger.info(f"✏️ 파일 편집: {file_path}") await asyncio.sleep(0.1) # 시뮬레이션 지연 return { 'success': True, 'changes_applied': len(changes), 'backup_created': True } async def _mock_browser_interface(self, url: str, action: str) -> Dict[str, Any]: """모의 브라우저 인터페이스""" logger.info(f"🌐 브라우저 작업: {action} on {url}") await asyncio.sleep(0.2) # 시뮬레이션 지연 return { 'success': True, 'page_title': "모의 페이지", 'content_length': 1024, 'links_found': 5 } async def _mock_debugger_interface(self, language: str, file_path: str) -> Dict[str, Any]: """모의 디버거 인터페이스""" logger.info(f"🐛 디버깅: {language} - {file_path}") await asyncio.sleep(0.3) # 시뮬레이션 지연 return { 'success': True, 'breakpoints_set': 2, 'variables_inspected': 10, 'issues_found': 0 } # ========================================================================= # 유틸리티 및 헬퍼 메서드들 # ========================================================================= async def _load_task_data(self, task_id: str) -> Optional[Dict[str, Any]]: """작업 데이터 로드""" # 실제 구현에서는 데이터베이스나 파일에서 로드 # 여기서는 모의 데이터 반환 return { 'id': task_id, 'name': f"작업 {task_id}", 'description': "테스트 작업입니다", 'implementation_guide': "1. 파일 생성\n2. 코드 작성\n3. 테스트 실행", 'related_files': [ {'path': '/test/file.py', 'type': 'TO_MODIFY'} ], 'dependencies': [], 'category': 'development' } def _determine_execution_strategy(self, task_data: Dict[str, Any]) -> str: """실행 전략 결정""" category = task_data.get('category', 'development') complexity = len(task_data.get('implementation_guide', '')) if complexity > 500: return "incremental_with_checkpoints" elif category in ['testing', 'validation']: return "test_driven_execution" elif category == 'research': return "exploratory_execution" else: return "standard_sequential" async def _generate_execution_steps(self, task_data: Dict[str, Any], strategy: str) -> List[ExecutionStep]: """실행 단계 생성""" steps = [] implementation_guide = task_data.get('implementation_guide', '') # 구현 가이드에서 단계 추출 lines = implementation_guide.split('\n') step_lines = [line.strip() for line in lines if line.strip() and (line.strip().startswith(('1.', '2.', '3.', '-', '*')) or any(word in line.lower() for word in ['step', '단계', 'then', 'next']))] for i, step_line in enumerate(step_lines): step = ExecutionStep( step_id=f"STEP-{i+1:03d}", name=f"단계 {i+1}", description=step_line, command=self._convert_description_to_command(step_line), expected_outcome=f"단계 {i+1} 완료", timeout_seconds=300.0 ) steps.append(step) # 기본 단계들 (구현 가이드가 비어있는 경우) if not steps: steps = [ ExecutionStep("STEP-001", "초기화", "작업 환경 설정", "echo '초기화'", "환경 준비 완료"), ExecutionStep("STEP-002", "구현", "핵심 기능 구현", "echo '구현'", "기능 구현 완료"), ExecutionStep("STEP-003", "검증", "결과 검증", "echo '검증'", "검증 완료") ] return steps def _convert_description_to_command(self, description: str) -> str: """설명을 명령으로 변환""" description_lower = description.lower() if any(word in description_lower for word in ['파일', 'file', '생성', 'create']): return "echo '파일 생성 시뮬레이션'" elif any(word in description_lower for word in ['코드', 'code', '작성', 'write']): return "echo '코드 작성 시뮬레이션'" elif any(word in description_lower for word in ['테스트', 'test', '검증', 'verify']): return "echo '테스트 실행 시뮬레이션'" elif any(word in description_lower for word in ['설치', 'install', '설정', 'setup']): return "echo '설치/설정 시뮬레이션'" else: return f"echo '{description} 시뮬레이션'" async def _execute_command(self, command: str, timeout: float) -> Dict[str, Any]: """명령 실행""" try: # 명령 타입에 따라 적절한 인터페이스 선택 if command.startswith('echo') or any(cmd in command for cmd in ['ls', 'pwd', 'cat']): return await self.tool_interfaces['terminal'](command, timeout) elif 'file' in command.lower(): return await self.tool_interfaces['filesystem']('read', files=[command]) else: return await self.tool_interfaces['terminal'](command, timeout) except Exception as e: return { 'success': False, 'output': '', 'error': str(e), 'exit_code': 1, 'output_files': [] } if __name__ == "__main__": # 기본 테스트 print("🚀 Execution & Control System 테스트") async def test_execution(): manager = ExecutionControlManager() # execute_task 테스트 success, message = await manager.execute_task("TASK-20250719-TEST001") print(f"✅ execute_task 결과: {success} - {message}") # research_mode 테스트 research = await manager.research_mode( topic="Python 비동기 프로그래밍 최적화", current_state="asyncio 기본 사용법 조사 완료", next_steps="성능 최적화 방안 연구" ) print(f"✅ research_mode 성공: 신뢰도={research.confidence_level:.2f}") print(f"✅ 발견사항 {len(research.findings)}개, 권장사항 {len(research.recommendations)}개") print("🎯 모든 Execution & Control 테스트 완료!") # 비동기 테스트 실행 import asyncio asyncio.run(test_execution())