feat: Planner 오케스트레이션 루프 구현

이중 루프 구조:
- 내부 루프 (Planner 자가검증, 최대 3회):
  계획 → 코딩 → Planner가 결과 검증 → 미달 시 추가 태스크 → 반복
  Planner가 만족할 때까지 자체적으로 보완 작업 진행
- 외부 루프 (Reviewer, 최대 2회):
  Planner 만족 → Reviewer 검토 → 반려 시 피드백으로 재계획

새 메서드:
- planner_verify(): Planner가 계획 달성도 자가 검증
- _read_recent_files(): 프로젝트 파일 읽기 공용 헬퍼
This commit is contained in:
2026-03-06 23:14:05 +09:00
parent 47a8c41a81
commit 52402065f3
2 changed files with 151 additions and 62 deletions

View File

@@ -114,31 +114,16 @@ class TaskPipeline:
return processed
# ──────────────────────────────────────────
# Batch Review
# 프로젝트 파일 읽기 (공용)
# ──────────────────────────────────────────
async def batch_review(self, tasks: list[dict], code_outputs: list[str]) -> dict:
"""에이전트가 생성/수정한 실제 파일을 리뷰.
code_outputs(에이전트 보고)와 함께 실제 프로젝트 파일 내용을 읽어서 리뷰합니다.
"""
def _read_recent_files(self, cutoff_seconds: int = 600) -> list[str]:
"""최근 변경된 프로젝트 파일 읽기."""
import os
import time as _time
# 태스크 요약
task_summaries = []
for i, task in enumerate(tasks):
title = task.get("title", task.get("description", f"Task {i+1}"))
task_summaries.append(f"### Task {i+1}: {title}")
# 에이전트 보고 요약
agent_reports = []
for i, output in enumerate(code_outputs):
agent_reports.append(f"--- Agent {i+1} 보고 ---\n{output}")
# 최근 변경된 파일 읽기 (10분 이내)
recent_files = []
cutoff = _time.time() - 600
cutoff = _time.time() - cutoff_seconds
project_root = Path(self.project_path)
skip_dirs = {".git", "__pycache__", "node_modules", ".venv", "venv"}
@@ -156,11 +141,84 @@ class TaskPipeline:
except (OSError, UnicodeDecodeError):
continue
return recent_files
# ──────────────────────────────────────────
# Planner 자가 검증 (오케스트레이션)
# ──────────────────────────────────────────
async def planner_verify(
self, user_request: str, plan: dict,
code_outputs: list[str],
) -> dict:
"""Planner가 자기 계획의 달성 여부를 검증.
실제 파일을 읽어서 계획이 충족됐는지 판단합니다.
미달이면 추가 태스크를 생성합니다.
Returns:
{
"satisfied": bool,
"feedback": "미충족 사유",
"additional_tasks": [...] (satisfied=false일 때)
}
"""
recent_files = self._read_recent_files()
files_section = "\n\n".join(recent_files) if recent_files else "(파일 없음)"
agent_reports = "\n".join(
f"--- Agent {i+1} ---\n{output}"
for i, output in enumerate(code_outputs)
)
prompt = (
f"## 원래 사용자 요청\n{user_request}\n\n"
f"## 내가 세운 계획\n{json.dumps(plan, ensure_ascii=False, indent=2)}\n\n"
f"## 에이전트 보고\n{agent_reports}\n\n"
f"## 현재 프로젝트 파일\n{files_section}\n\n"
f"## 판단 요청\n"
f"위 계획이 충족되었는지 판단하세요.\n"
f"충족되었으면 satisfied=true.\n"
f"미충족이면 satisfied=false + 부족한 부분을 해결할 추가 태스크를 생성하세요.\n\n"
f"JSON 형식:\n"
f"```json\n"
f'{{\n'
f' "satisfied": true|false,\n'
f' "feedback": "판단 근거 (한국어)",\n'
f' "additional_tasks": [\n'
f' {{"id": 1, "title": "추가 태스크", "description": "구현 내용", "type": "modify"}}\n'
f' ]\n'
f'}}\n'
f"```"
)
response = await self.gemini.call("planner", prompt, timeout=180)
self._log("planner_verify", user_request, response)
result = self._extract_json(response)
return result or {"satisfied": True, "feedback": response}
# ──────────────────────────────────────────
# Batch Review
# ──────────────────────────────────────────
async def batch_review(self, tasks: list[dict], code_outputs: list[str]) -> dict:
"""에이전트가 생성/수정한 실제 파일을 리뷰."""
task_summaries = []
for i, task in enumerate(tasks):
title = task.get("title", task.get("description", f"Task {i+1}"))
task_summaries.append(f"### Task {i+1}: {title}")
agent_reports = []
for i, output in enumerate(code_outputs):
agent_reports.append(f"--- Agent {i+1} 보고 ---\n{output}")
recent_files = self._read_recent_files()
if not recent_files:
# 변경 파일 없음 → 자동 통과 (삭제 작업 등)
return {
"passed": True,
"summary": "파일 변경 없음 또는 삭제 작업 자동 통과",
"summary": "파일 변경 없음 또는 삭제 작업 - 자동 통과",
"issues": [],
}