Lab 08: 플래닝 에이전트 구현

고급 마감: 2026-04-29

목표

• 실제 코드베이스를 분석하는 CodebaseAnalyzer 구현
• 분석 결과를 바탕으로 PlannerAgent가 구체적인 spec.md 생성
• 태스크 의존성 그래프 관리 및 우선순위 정렬

PlannerAgent의 역할

플래닝 에이전트는 파이프라인의 두뇌다. 좋은 플래너는 다음 세 가지를 수행한다.

1. 코드베이스 현황 파악 — 어떤 파일이 존재하고 어떤 함수가 어디에 있는가
2. 목표 분해 — 큰 목표를 독립적이고 검증 가능한 서브태스크로 쪼개기
3. 사양서 생성 — 코더가 모호함 없이 작업할 수 있는 spec.md 작성

구현 요구사항

1. codebase_analyzer.py — 코드베이스 분석기

codebase_analyzer.py

import ast
import subprocess
from pathlib import Path
from dataclasses import dataclass, field

@dataclass
class FileInfo:
    path: str
    size_lines: int
    functions: list[str]
    classes: list[str]
    imports: list[str]
    has_tests: bool

@dataclass
class CodebaseSnapshot:
    root: str
    total_files: int
    total_lines: int
    files: list[FileInfo]
    test_files: list[str]
    entry_points: list[str]

    def to_summary(self, max_files: int = 20) -> str:
        """LLM에게 전달할 간결한 요약을 생성한다."""
        lines = [
            f"코드베이스: {self.root}",
            f"총 파일: {self.total_files}개 | 총 라인: {self.total_lines:,}줄",
            f"테스트 파일: {len(self.test_files)}개",
            "",
            "주요 파일:"
        ]
        for fi in sorted(self.files, key=lambda x: x.size_lines, reverse=True)[:max_files]:
            funcs = ", ".join(fi.functions[:5])
            lines.append(f"  {fi.path} ({fi.size_lines}줄) — {funcs}")
        return "\n".join(lines)


class CodebaseAnalyzer:
    IGNORE_DIRS = {".git", "__pycache__", ".venv", "venv", "node_modules", ".mypy_cache"}

    def __init__(self, root: str):
        self.root = Path(root)

    def analyze(self) -> CodebaseSnapshot:
        files: list[FileInfo] = []
        for py_file in self.root.rglob("*.py"):
            if any(d in py_file.parts for d in self.IGNORE_DIRS):
                continue
            info = self._analyze_file(py_file)
            files.append(info)

        test_files = [f.path for f in files if f.has_tests]
        entry_points = self._find_entry_points(files)

        return CodebaseSnapshot(
            root=str(self.root),
            total_files=len(files),
            total_lines=sum(f.size_lines for f in files),
            files=files,
            test_files=test_files,
            entry_points=entry_points
        )

    def _analyze_file(self, path: Path) -> FileInfo:
        source = path.read_text(errors="replace")
        lines = source.splitlines()

        functions, classes, imports = [], [], []
        try:
            tree = ast.parse(source)
            for node in ast.walk(tree):
                if isinstance(node, ast.FunctionDef):
                    functions.append(node.name)
                elif isinstance(node, ast.ClassDef):
                    classes.append(node.name)
                elif isinstance(node, (ast.Import, ast.ImportFrom)):
                    if isinstance(node, ast.Import):
                        imports.extend(alias.name for alias in node.names)
                    else:
                        imports.append(node.module or "")
        except SyntaxError:
            pass

        rel_path = str(path.relative_to(self.root))
        return FileInfo(
            path=rel_path,
            size_lines=len(lines),
            functions=functions,
            classes=classes,
            imports=[i for i in imports if i],
            has_tests=rel_path.startswith("test") or "/test" in rel_path
        )

    def _find_entry_points(self, files: list[FileInfo]) -> list[str]:
        """main() 함수가 있는 파일을 진입점으로 추정한다."""
        return [f.path for f in files if "main" in f.functions]

2. spec_generator.py — 사양서 생성기

spec_generator.py

import anthropic
from datetime import datetime
from pathlib import Path
from codebase_analyzer import CodebaseSnapshot

SPEC_SYSTEM = """
You are a senior software architect creating implementation specifications.
Given a codebase analysis and an objective, produce a detailed spec.md.

The spec must include:
1. Objective (one sentence)
2. Scope (which files will change)
3. Implementation plan (numbered steps, each verifiable)
4. Test strategy (how to verify each step)
5. Risk assessment (what could go wrong)

Be concrete. Reference actual file names and function names from the codebase.
"""

class SpecGenerator:
    def __init__(self):
        self.client = anthropic.Anthropic()

    def generate(self, snapshot: CodebaseSnapshot, objective: str) -> str:
        prompt = f"""
Codebase Analysis:
{snapshot.to_summary()}

Objective: {objective}

Generate a detailed spec.md for this task.
"""
        response = self.client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=SPEC_SYSTEM,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text

    def save(self, content: str, path: str = "spec.md"):
        header = f"<!-- Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')} -->\n\n"
        Path(path).write_text(header + content)
        print(f"[SpecGenerator] spec.md 저장 완료: {path}")

3. dependency_graph.py — 태스크 의존성 관리

dependency_graph.py

from collections import defaultdict, deque
from dataclasses import dataclass

@dataclass
class Task:
    id: str
    description: str
    assignee: str
    depends_on: list[str]
    priority: int = 0

class DependencyGraph:
    """태스크 의존성을 DAG로 관리하고 위상 정렬을 수행한다."""

    def __init__(self, tasks: list[Task]):
        self.tasks = {t.id: t for t in tasks}
        self._validate()

    def _validate(self):
        """의존성 사이클 여부를 검사한다."""
        visited, in_stack = set(), set()

        def dfs(task_id: str):
            visited.add(task_id)
            in_stack.add(task_id)
            for dep in self.tasks[task_id].depends_on:
                if dep not in self.tasks:
                    raise ValueError(f"Unknown dependency: {dep}")
                if dep in in_stack:
                    raise ValueError(f"Cycle detected: {task_id} -> {dep}")
                if dep not in visited:
                    dfs(dep)
            in_stack.remove(task_id)

        for tid in self.tasks:
            if tid not in visited:
                dfs(tid)

    def topological_sort(self) -> list[Task]:
        """의존성 순서로 정렬된 태스크 리스트를 반환한다."""
        in_degree = defaultdict(int)
        for task in self.tasks.values():
            for dep in task.depends_on:
                in_degree[task.id] += 1

        queue = deque([
            t for t in self.tasks.values() if in_degree[t.id] == 0
        ])
        result = []

        while queue:
            task = queue.popleft()
            result.append(task)
            for other in self.tasks.values():
                if task.id in other.depends_on:
                    in_degree[other.id] -= 1
                    if in_degree[other.id] == 0:
                        queue.append(other)

        if len(result) != len(self.tasks):
            raise ValueError("사이클이 존재해 위상 정렬 불가")
        return result

    def get_ready_tasks(self, completed: set[str]) -> list[Task]:
        """완료된 태스크 집합을 기준으로 즉시 실행 가능한 태스크를 반환한다."""
        return [
            t for t in self.tasks.values()
            if t.id not in completed
            and all(dep in completed for dep in t.depends_on)
        ]

4. planner_agent.py — 완성된 플래너

# planner_agent.py (Lab 07에서 확장)
from codebase_analyzer import CodebaseAnalyzer
from spec_generator import SpecGenerator
from dependency_graph import DependencyGraph, Task
import json
from pathlib import Path

class PlannerAgent:
    def __init__(self, codebase_root: str):
        self.analyzer = CodebaseAnalyzer(codebase_root)
        self.spec_gen = SpecGenerator()

    def plan(self, objective: str, output_dir: str = ".") -> dict:
        print("[Planner] 코드베이스 분석 중...")
        snapshot = self.analyzer.analyze()

        print("[Planner] spec.md 생성 중...")
        spec_content = self.spec_gen.generate(snapshot, objective)
        self.spec_gen.save(spec_content, f"{output_dir}/spec.md")

        print("[Planner] 태스크 분해 중...")
        tasks = self._decompose(spec_content, objective)

        graph = DependencyGraph(tasks)
        ordered = graph.topological_sort()

        plan = {
            "objective": objective,
            "codebase_summary": snapshot.to_summary(max_files=10),
            "spec_path": f"{output_dir}/spec.md",
            "tasks": [
                {"id": t.id, "description": t.description,
                 "assignee": t.assignee, "depends_on": t.depends_on}
                for t in ordered
            ]
        }
        Path(f"{output_dir}/plan.json").write_text(
            json.dumps(plan, indent=2, ensure_ascii=False)
        )
        print(f"[Planner] 완료: {len(tasks)}개 태스크, plan.json 저장")
        return plan

    def _decompose(self, spec: str, objective: str) -> list[Task]:
        """spec.md를 분석해 Task 리스트를 생성한다. 실제로는 LLM 호출."""
        # 실습에서 직접 구현 — spec 내용을 파싱하거나 LLM에 재요청
        return [
            Task("t-01", "코드베이스 탐색 및 관련 파일 파악", "researcher", []),
            Task("t-02", "핵심 기능 구현", "coder", ["t-01"]),
            Task("t-03", "단위 테스트 작성 및 실행", "qa", ["t-02"]),
            Task("t-04", "코드 리뷰", "reviewer", ["t-03"]),
        ]

1. codebase_analyzer.py 구현 및 자체 코드베이스 분석 테스트
2. spec_generator.py 구현 — 실제 spec.md 생성 확인
3. dependency_graph.py 구현 — 사이클 감지 테스트 포함
4. PlannerAgent.plan() 실행: python -c "from planner_agent import PlannerAgent; PlannerAgent('.').plan('모든 테스트를 통과하도록 calculator.py를 수정해줘')"
5. 생성된 spec.md와 plan.json 검토

제출물

assignments/lab-08/[학번]/에 PR:

• codebase_analyzer.py — FileInfo, CodebaseSnapshot, CodebaseAnalyzer
• spec_generator.py — LLM 기반 spec.md 생성
• dependency_graph.py — 위상 정렬 및 사이클 감지
• planner_agent.py — 위 세 모듈을 통합한 완성 버전
• spec.md — 실제 생성된 사양서 예시
• plan.json — 실제 생성된 계획 JSON
• README.md — 코드베이스 분석 결과 요약 및 플래너 동작 설명

spec.md 품질

spec.md의 품질이 전체 파이프라인 성능을 결정한다. 좋은 spec은 코더가 어떤 파일의 어떤 함수를 어떻게 수정해야 하는지 명확히 알 수 있어야 한다.