"""GeminiCaller — gemini CLI 호출. 모든 역할이 gemini-3-flash-preview 모델을 사용하며, 역할별 thinkingBudget을 동적으로 조절합니다. MCP 서버 없이 CLI 도구를 직접 실행하는 구조입니다. """ import asyncio import json import logging import time import sys from pathlib import Path logger = logging.getLogger("variet.gemini") ROLE_PROMPTS_DIR = Path(__file__).parent.parent / "prompts" PROJECT_ROOT = Path(__file__).parent.parent GEMINI_MODEL = "gemini-3-flash-preview" # 역할별 thinkingBudget (토큰 단위) # 512=가벼운 분류/요약, 4096=계획/검수, 8192=구현/비평 ROLE_THINKING: dict[str, int] = { "unified": 512, "agent": 4096, "summarizer": 512, "planner": 4096, "coder": 8192, "reviewer": 8192, } DEFAULT_THINKING = 4096 # 동시 호출 제한 (Gemini AI Ultra 120RPM 고려) _semaphore = asyncio.Semaphore(4) # settings.json 쓰기 경합 방지 (settings 쓰기 ~ 프로세스 시작 구간 직렬화) _settings_lock = asyncio.Lock() # Windows에서 PS ExecutionPolicy 우회 _IS_WIN = sys.platform == "win32" # ~/.gemini/settings.json 경로 _SETTINGS_PATH = Path.home() / ".gemini" / "settings.json" class GeminiCallError(Exception): """Gemini CLI 호출 실패.""" pass class GeminiCaller: """Gemini CLI 호출을 관리합니다.""" def __init__(self, project_path: str = None): self.project_path = project_path self.call_count = 0 self.last_call_time = 0.0 def _build_cmd(self): """에 맞는 gemini 커맨드 빌드.""" if _IS_WIN: return ["cmd", "/c", "gemini", "--model", GEMINI_MODEL, "--approval-mode", "yolo"] return ["gemini", "--model", GEMINI_MODEL, "--approval-mode", "yolo"] def _set_thinking_budget(self, role: str): """역할별 thinkingBudget을 settings.json에 반영.""" budget = ROLE_THINKING.get(role, DEFAULT_THINKING) try: if _SETTINGS_PATH.exists(): settings = json.loads(_SETTINGS_PATH.read_text(encoding="utf-8")) else: _SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True) settings = {} # thinkingBudget configs = settings.setdefault("modelConfigs", {}) default = configs.setdefault("default", {}) thinking = default.setdefault("thinkingConfig", {}) thinking["thinkingBudget"] = budget # MCP 서버 제거 (CLI 직접 실행으로 전환) settings.pop("mcpServers", None) _SETTINGS_PATH.write_text( json.dumps(settings, indent=2, ensure_ascii=False), encoding="utf-8", ) logger.debug(f"settings.json 업데이트: role={role}, budget={budget}") except Exception as e: logger.warning(f"settings.json 업데이트 실패 (role={role}): {e}") def _clean_output(self, raw: str) -> str: """Gemini 출력에서 노이즈 라인 제거.""" lines = raw.splitlines() cleaned = [] for line in lines: if any(noise in line for noise in [ "YOLO mode", "Loaded cached", "Welcome to Gemini", "Type /help", "Gemini CLI", "MCP issues detected", ]): continue cleaned.append(line) return "\n".join(cleaned).strip() # ────────────────────────────────────────── # 텍스트 모드 (분류/리뷰/총평) # ────────────────────────────────────────── async def call(self, role: str, context: str, timeout: int = 300) -> str: """역할별 프롬프트로 텍스트 생성. 파일 접근 없이 텍스트만 주고받는 역할에 사용. (unified, reviewer, summarizer) """ async with _semaphore: return await self._call_text(role, context, timeout) async def _call_text(self, role: str, context: str, timeout: int) -> str: """텍스트 전용 호출.""" prompt_file = ROLE_PROMPTS_DIR / f"{role}.md" if prompt_file.exists(): system_prompt = prompt_file.read_text(encoding="utf-8") else: system_prompt = f"You are a {role}. Respond in Korean." full_input = ( f"=== SYSTEM INSTRUCTIONS ===\n" f"{system_prompt}\n\n" f"=== USER INPUT ===\n" f"{context}" ) try: # Lock: settings.json 쓰기 ~ 프로세스 시작 직렬화 async with _settings_lock: self._set_thinking_budget(role) proc = await asyncio.create_subprocess_exec( *self._build_cmd(), stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, stderr = await asyncio.wait_for( proc.communicate(input=full_input.encode("utf-8")), timeout=timeout, ) self.call_count += 1 self.last_call_time = time.time() result = self._clean_output(stdout.decode("utf-8", errors="replace")) if not result and stderr: err = stderr.decode("utf-8", errors="replace").strip() logger.warning(f"Gemini [{role}] 빈 응답. stderr: {err[:200]}") logger.info( f"Gemini [{role}] text #{self.call_count} " f"-- 입력 {len(full_input)}자 -> 출력 {len(result)}자" ) return result except asyncio.TimeoutError: logger.error(f"Gemini [{role}] 타임아웃 ({timeout}s) — 입력 {len(full_input)}자") raise GeminiCallError(f"Gemini 응답 시간 초과 ({timeout}초). 요청이 너무 복잡할 수 있습니다.") except FileNotFoundError: raise GeminiCallError("gemini CLI를 찾을 수 없습니다.") except Exception as e: raise GeminiCallError(f"Gemini CLI 호출 실패: {e}") # ────────────────────────────────────────── # 에이전트 모드 (코딩 — 파일 직접 읽기/쓰기) # ────────────────────────────────────────── async def call_agent( self, role: str, context: str, cwd: str, timeout: int = 600, ) -> str: """에이전트 모드 — 프로젝트 디렉토리에서 실행. Gemini가 직접 파일을 읽고/쓰고/명령을 실행합니다. cwd를 프로젝트 경로로 설정하여 파일 접근을 허용합니다. Args: role: 프롬프트 역할 (coder) context: 작업 지시 (태스크 설명) cwd: 프로젝트 루트 경로 (여기서 Gemini 실행) timeout: 타임아웃 (에이전트는 더 길게 — 기본 10분) """ async with _semaphore: return await self._call_agent_impl(role, context, cwd, timeout) async def _call_agent_impl( self, role: str, context: str, cwd: str, timeout: int, ) -> str: """에이전트 모드 구현.""" prompt_file = ROLE_PROMPTS_DIR / f"{role}.md" if prompt_file.exists(): system_prompt = prompt_file.read_text(encoding="utf-8") else: system_prompt = f"You are a {role}. Respond in Korean." # 역할에 따라 다른 지시 if role == "agent": # 범용 에이전트: 도구 사용 여부를 자율 판단 footer = ( f"프로젝트 루트: {cwd}\n" f"모든 응답은 한국어로 작성하세요.\n" f"파일 작업이 필요하면 직접 수행하고, 대화만 필요하면 바로 답변하세요." ) else: # coder 등 파일 작업 전용 역할 footer = ( f"프로젝트 루트: {cwd}\n" f"파일을 직접 생성/수정하세요. 코드블록으로 출력하지 말고, 실제 파일로 저장하세요.\n" f"작업 완료 후 변경한 파일 목록을 간단히 출력하세요.\n" f"모든 응답, 주석, 문서는 반드시 한국어로 작성하세요." ) full_input = ( f"=== SYSTEM INSTRUCTIONS ===\n" f"{system_prompt}\n\n" f"=== TASK ===\n" f"{context}\n\n" f"=== IMPORTANT ===\n" f"{footer}" ) try: # Lock: settings.json 쓰기 ~ 프로세스 시작 직렬화 async with _settings_lock: self._set_thinking_budget(role) proc = await asyncio.create_subprocess_exec( *self._build_cmd(), stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=cwd, # ★ 핵심: 프로젝트 디렉토리에서 실행 ) stdout, stderr = await asyncio.wait_for( proc.communicate(input=full_input.encode("utf-8")), timeout=timeout, ) self.call_count += 1 self.last_call_time = time.time() result = self._clean_output(stdout.decode("utf-8", errors="replace")) logger.info( f"Gemini [{role}] agent #{self.call_count} " f"-- cwd={cwd} 입력 {len(full_input)}자 -> 출력 {len(result)}자" ) return result except asyncio.TimeoutError: logger.error(f"Gemini [{role}] agent 타임아웃 ({timeout}s) — cwd={cwd}") raise GeminiCallError( f"Gemini 에이전트 응답 시간 초과 ({timeout}초). 작업이 너무 복잡할 수 있습니다." ) except FileNotFoundError: raise GeminiCallError("gemini CLI를 찾을 수 없습니다.") except Exception as e: raise GeminiCallError(f"Gemini agent 호출 실패: {e}") async def call_simple(self, prompt: str, timeout: int = 60) -> str: """시스템 프롬프트 없이 단순 호출.""" return await self.call("default", prompt, timeout)