CD
01bf8aa258
- 모델: gemini-3-flash-preview (2.0 Flash → 3 Flash 업그레이드) - 역할별 thinkingBudget 동적 조절: unified/summarizer = 512 (가벼운 분류/요약) planner = 4096 (계획/검수) coder/reviewer = 8192 (구현/비평) - ~/.gemini/settings.json에 호출 직전 반영 - Docker에서도 settings.json 마운트로 동작
239 lines
8.8 KiB
Python
239 lines
8.8 KiB
Python
"""GeminiCaller — gemini CLI 호출.
|
|
|
|
모든 역할이 gemini-3-flash-preview 모델을 사용하며,
|
|
역할별 thinkingBudget을 동적으로 조절합니다.
|
|
"""
|
|
|
|
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"
|
|
GEMINI_MODEL = "gemini-3-flash-preview"
|
|
|
|
# 역할별 thinkingBudget (토큰 단위)
|
|
# 512=가벼운 분류/요약, 4096=계획/검수, 8192=구현/비평
|
|
ROLE_THINKING: dict[str, int] = {
|
|
"unified": 512,
|
|
"summarizer": 512,
|
|
"planner": 4096,
|
|
"coder": 8192,
|
|
"reviewer": 8192,
|
|
}
|
|
DEFAULT_THINKING = 4096
|
|
|
|
# 동시 호출 제한 (Gemini AI Ultra 120RPM 고려)
|
|
_semaphore = asyncio.Semaphore(4)
|
|
|
|
# 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 = {}
|
|
|
|
# modelConfigs.default.thinkingConfig.thinkingBudget 설정
|
|
configs = settings.setdefault("modelConfigs", {})
|
|
default = configs.setdefault("default", {})
|
|
thinking = default.setdefault("thinkingConfig", {})
|
|
thinking["thinkingBudget"] = budget
|
|
|
|
_SETTINGS_PATH.write_text(
|
|
json.dumps(settings, indent=2, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
logger.debug(f"thinkingBudget={budget} for role={role}")
|
|
except Exception as e:
|
|
logger.warning(f"thinkingBudget 설정 실패 (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",
|
|
]):
|
|
continue
|
|
cleaned.append(line)
|
|
return "\n".join(cleaned).strip()
|
|
|
|
# ──────────────────────────────────────────
|
|
# 텍스트 모드 (분류/리뷰/총평)
|
|
# ──────────────────────────────────────────
|
|
|
|
async def call(self, role: str, context: str, timeout: int = 120) -> 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:
|
|
"""텍스트 전용 호출."""
|
|
self._set_thinking_budget(role)
|
|
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:
|
|
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:
|
|
raise GeminiCallError(f"Gemini timeout ({timeout}s) -- role={role}")
|
|
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 = 300,
|
|
) -> str:
|
|
"""에이전트 모드 — 프로젝트 디렉토리에서 실행.
|
|
|
|
Gemini가 직접 파일을 읽고/쓰고/명령을 실행합니다.
|
|
cwd를 프로젝트 경로로 설정하여 파일 접근을 허용합니다.
|
|
|
|
Args:
|
|
role: 프롬프트 역할 (coder)
|
|
context: 작업 지시 (태스크 설명)
|
|
cwd: 프로젝트 루트 경로 (여기서 Gemini 실행)
|
|
timeout: 타임아웃 (에이전트는 더 길게 — 기본 5분)
|
|
"""
|
|
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:
|
|
"""에이전트 모드 구현."""
|
|
self._set_thinking_budget(role)
|
|
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"=== TASK ===\n"
|
|
f"{context}\n\n"
|
|
f"=== IMPORTANT ===\n"
|
|
f"프로젝트 루트: {cwd}\n"
|
|
f"파일을 직접 생성/수정하세요. 코드블록으로 출력하지 말고, 실제 파일로 저장하세요.\n"
|
|
f"작업 완료 후 변경한 파일 목록을 간단히 출력하세요."
|
|
)
|
|
|
|
try:
|
|
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:
|
|
raise GeminiCallError(
|
|
f"Gemini agent timeout ({timeout}s) -- role={role}, cwd={cwd}"
|
|
)
|
|
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)
|