MCP Localization Project

#!/usr/bin/env python3 """ Shrimp Task Manager - 메인 통합 클래스 로컬 AI의 완전한 사고 구조 시스템 이 모듈은 모든 Shrimp Task Manager 기능을 통합하는 메인 클래스입니다: - 15개 핵심 기능의 통합 인터페이스 - MCP (Model Context Protocol) 서버 구현 - 지능형 작업 관리 및 메타인지 시스템 - 무한 컨텍스트 활용 최적화 ShrimpTaskManager는 로컬 AI의 치명적 단점인 문맥 파악 실패와 작업 방향성 상실을 해결하는 핵심 시스템입니다. """ import os import sys import json import sqlite3 import asyncio import logging from typing import Dict, List, Optional, Any, Tuple, Set, Union from datetime import datetime, timedelta from dataclasses import dataclass, field from pathlib import Path # 로컬 모듈 임포트 from .shrimp_base import ( Task, TaskStatus, TaskPriority, TaskCategory, RelatedFile, Evidence, TaskDependency, ThoughtProcess, ProjectRules, FileRelationType, ThoughtStage, QualityScore, generate_task_id ) # 기능별 모듈 임포트 from .shrimp_data import DataManagementManager, SearchScope, SortOrder from .shrimp_execution import ExecutionControlManager, ExecutionStatus, ResearchPhase from .shrimp_governance import ProjectGovernanceManager, ChangeImpact, GovernanceAction from .shrimp_lifecycle import TaskLifecycleManager from .shrimp_metacognition import MetacognitionManager, ThinkingMode # 로깅 설정 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # ============================================================================= # 메인 통합 클래스 (Main Integration Class) # ============================================================================= class ShrimpTaskManager: """ Shrimp Task Manager 메인 클래스 로컬 AI의 완전한 사고 구조를 구현하는 통합 시스템. 무한 컨텍스트를 전략적으로 활용하여 GPT-4o를 뛰어넘는 능력 제공. 주요 특징: - 15개 핵심 기능의 완전한 통합 - 증거 기반 의사결정 시스템 - 지능형 작업 분해 및 관리 - 메타인지 및 품질 관리 시스템 - MCP 프로토콜 완전 지원 """ def __init__(self, db_path: str = "shrimp_task_manager.db"): """ ShrimpTaskManager 초기화 Args: db_path: SQLite 데이터베이스 파일 경로 """ self.db_path = db_path self.db_connection = None # 핵심 시스템 컴포넌트 초기화 self.data_manager = DataManagementManager(db_path) self.execution_engine = ExecutionControlManager(db_path) self.governance = ProjectGovernanceManager(db_path) self.lifecycle = TaskLifecycleManager(db_path) self.metacognition = MetacognitionManager(db_path) # 세션 정보 self.session_id = generate_task_id("SESSION") self.start_time = datetime.now() logger.info(f"🎯 ShrimpTaskManager 초기화 완료 - 세션: {self.session_id}") async def initialize(self) -> Dict[str, Any]: """ 비동기 시스템 초기화 Returns: 초기화 결과 및 시스템 상태 """ try: # 데이터베이스 초기화 await self._init_database() # 각 컴포넌트 초기화 await self.data_manager.initialize() await self.execution_engine.initialize() await self.governance.initialize() await self.lifecycle.initialize() await self.metacognition.initialize() logger.info("🚀 모든 시스템 컴포넌트 초기화 완료") return { "success": True, "session_id": self.session_id, "start_time": self.start_time.isoformat(), "components": [ "DataManager", "ExecutionEngine", "ProjectGovernance", "TaskLifecycle", "MetacognitionEngine" ], "features": [ "plan_task", "analyze_task", "reflect_task", "split_tasks", "list_tasks", "execute_task", "verify_task", "delete_task", "clear_all_tasks", "update_task", "query_task", "get_task_detail", "process_thought", "init_project_rules", "research_mode" ] } except Exception as e: logger.error(f"❌ 시스템 초기화 실패: {e}") return { "success": False, "error": str(e), "session_id": self.session_id } async def _init_database(self) -> None: """데이터베이스 초기화 및 테이블 생성""" self.db_connection = sqlite3.connect(self.db_path) self.db_connection.execute("PRAGMA foreign_keys = ON") # 공통 테이블 생성 await self._create_tables() logger.info(f"📊 데이터베이스 초기화 완료: {self.db_path}") async def _create_tables(self) -> None: """필요한 모든 테이블 생성""" cursor = self.db_connection.cursor() # 세션 테이블 cursor.execute(""" CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, start_time TEXT NOT NULL, end_time TEXT, status TEXT DEFAULT 'active' ) """) # 세션 등록 cursor.execute(""" INSERT OR REPLACE INTO sessions (session_id, start_time, status) VALUES (?, ?, 'active') """, (self.session_id, self.start_time.isoformat())) self.db_connection.commit() # ========================================================================= # A. 작업 생명주기 관리 (Task Lifecycle Management) # ========================================================================= async def plan_task( self, description: str, requirements: Optional[str] = None, existing_tasks_reference: bool = False ) -> Dict[str, Any]: """ 지능형 작업 계획 사용자 요구사항을 분석하여 실행 가능한 작업 단위로 분해하고 프로젝트 전체 맥락에서 최적화된 계획을 수립합니다. Args: description: 완전한 작업 문제 설명 requirements: 특정 기술 요구사항 또는 제약조건 existing_tasks_reference: 기존 작업 참조 여부 Returns: 작업 계획 결과 및 생성된 작업 정보 """ logger.info(f"📋 작업 계획 시작: {description[:50]}...") try: # 메타인지 엔진을 통한 계획 수립 planning_result = await self.metacognition.plan_with_context( description=description, requirements=requirements, existing_tasks_reference=existing_tasks_reference ) # 라이프사이클 이벤트 기록 await self.lifecycle.record_event( event_type=LifecycleEvent.PLANNING_STARTED, description=f"작업 계획 수립: {description[:100]}" ) return planning_result except Exception as e: logger.error(f"❌ 작업 계획 실패: {e}") return { "success": False, "error": str(e), "description": description } async def analyze_task( self, summary: str, initial_concept: str, previous_analysis: Optional[str] = None ) -> Dict[str, Any]: """ 심층 기술 분석 작업의 기술적 타당성과 구현 방법을 상세 분석하고 코드베이스 전체 맥락에서 최적 구현 방법을 도출합니다. Args: summary: 구조화된 작업 요약 initial_concept: 초기 해답 구상 (50자 이상) previous_analysis: 이전 분석 결과 (재분석 시) Returns: 기술 분석 결과 및 구현 가이드 """ logger.info(f"🔍 기술 분석 시작: {summary[:50]}...") try: # 실행 엔진을 통한 분석 analysis_result = await self.execution_engine.analyze_requirements( summary=summary, initial_concept=initial_concept, previous_analysis=previous_analysis ) # 라이프사이클 이벤트 기록 await self.lifecycle.record_event( event_type=LifecycleEvent.ANALYSIS_COMPLETED, description=f"기술 분석 완료: {summary[:100]}" ) return analysis_result except Exception as e: logger.error(f"❌ 기술 분석 실패: {e}") return { "success": False, "error": str(e), "summary": summary } async def process_thought( self, thought: str, thought_number: int, total_thoughts: int, next_thought_needed: bool, stage: str, tags: Optional[List[str]] = None, axioms_used: Optional[List[str]] = None, assumptions_challenged: Optional[List[str]] = None ) -> Dict[str, Any]: """ 구조화된 사고 처리 복잡한 문제에 대한 체계적 사고 과정을 관리하고 다각도 정보 수집 및 종합적 판단을 수행합니다. Args: thought: 사고 내용 thought_number: 현재 사고 번호 total_thoughts: 예상 총 사고 수량 next_thought_needed: 다음 사고 필요 여부 stage: 사고 단계 tags: 사고 태그 axioms_used: 사용된 공리 assumptions_challenged: 도전된 가정 Returns: 사고 처리 결과 및 다음 단계 가이드 """ logger.info(f"💭 사고 처리 시작: 단계 {thought_number}/{total_thoughts} - {stage}") try: # 메타인지 엔진을 통한 사고 처리 thought_result = await self.metacognition.process_structured_thought( thought=thought, thought_number=thought_number, total_thoughts=total_thoughts, next_thought_needed=next_thought_needed, stage=stage, tags=tags or [], axioms_used=axioms_used or [], assumptions_challenged=assumptions_challenged or [] ) return thought_result except Exception as e: logger.error(f"❌ 사고 처리 실패: {e}") return { "success": False, "error": str(e), "thought_number": thought_number } async def split_tasks( self, global_analysis_result: str, tasks_raw: str, update_mode: str = "clearAllTasks" ) -> Dict[str, Any]: """ 지능형 작업 분할 복잡한 작업을 최적 단위로 분할하고 의존성을 관리합니다. 각 작업은 1-2일 내에 완료 가능한 단위로 설계됩니다. Args: global_analysis_result: 전역 분석 결과 tasks_raw: 구조화된 작업 목록 (JSON 형식) update_mode: 작업 업데이트 모드 Returns: 작업 분할 결과 및 생성된 작업들 """ logger.info("✂️ 지능형 작업 분할 시작...") try: # 거버넌스 시스템을 통한 작업 분할 split_result = await self.governance.split_and_organize_tasks( global_analysis_result=global_analysis_result, tasks_raw=tasks_raw, update_mode=update_mode ) # 라이프사이클 이벤트 기록 await self.lifecycle.record_event( event_type=LifecycleEvent.TASK_CREATED, description=f"작업 분할 완료: {len(split_result.get('tasks', []))}개 작업 생성" ) return split_result except Exception as e: logger.error(f"❌ 작업 분할 실패: {e}") return { "success": False, "error": str(e), "update_mode": update_mode } # ========================================================================= # B. 메타인지 및 품질 관리 (Metacognition & Quality) # ========================================================================= async def reflect_task( self, summary: str, analysis: str ) -> Dict[str, Any]: """ 비판적 성찰 완료된 분석이나 작업에 대한 비판적 검토를 수행하고 베스트 프랙티스와 비교하여 품질 향상 방안을 제시합니다. Args: summary: 구조화된 작업 요약 analysis: 완전한 기술 분석 결과 Returns: 성찰 결과 및 개선 방안 """ logger.info(f"🤔 비판적 성찰 시작: {summary[:50]}...") try: # 메타인지 엔진을 통한 성찰 reflection_result = await self.metacognition.reflect_on_analysis( summary=summary, analysis=analysis, reflection_type=ReflectionType.CRITICAL_REVIEW ) # 라이프사이클 이벤트 기록 await self.lifecycle.record_event( event_type=LifecycleEvent.REVIEW_COMPLETED, description=f"비판적 성찰 완료: {summary[:100]}" ) return reflection_result except Exception as e: logger.error(f"❌ 비판적 성찰 실패: {e}") return { "success": False, "error": str(e), "summary": summary }