Files
variet-agent/core/gemini_caller.py

80 lines
2.5 KiB
Python

"""GeminiCaller — gemini headless 호출.
stdin으로 시스템 프롬프트 + 컨텍스트를 직접 전달합니다.
cmd /c 래핑으로 PowerShell 실행 정책 우회.
"""
import asyncio
import time
from pathlib import Path
ROLE_PROMPTS_DIR = Path(__file__).parent.parent / "prompts"
class GeminiCaller:
"""Gemini CLI headless 호출을 관리합니다."""
def __init__(self, project_path: str = None):
self.project_path = project_path
self.call_count = 0
self.last_call_time = 0.0
async def call(self, role: str, context: str, timeout: int = 120) -> str:
"""역할별 프롬프트로 gemini 호출.
시스템 프롬프트와 컨텍스트를 하나로 합쳐 stdin으로 전달.
"""
# 시스템 프롬프트 로드
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(
"cmd", "/c", "gemini --approval-mode yolo",
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()
output = stdout.decode("utf-8", errors="replace").strip()
# YOLO 모드 메시지 제거
lines = output.splitlines()
cleaned = []
for line in lines:
if "YOLO mode" in line or "Loaded cached" in line:
continue
cleaned.append(line)
return "\n".join(cleaned).strip()
except asyncio.TimeoutError:
return f"[ERROR] Gemini CLI timeout after {timeout}s"
except Exception as e:
return f"[ERROR] Gemini CLI call failed: {e}"
async def call_simple(self, prompt: str, timeout: int = 60) -> str:
"""시스템 프롬프트 없이 단순 호출."""
return await self.call("default", prompt, timeout)