컨텐츠로 건너뛰기
AI 엔지니어링 2026

Lab 07: 멀티에이전트 파이프라인 설계

고급 마감: 2026-04-22

목표

  • 멀티에이전트 시스템의 설계 원칙 적용
  • 5단계 파이프라인(Planner → Researcher → Coder → QA → Reviewer) 아키텍처 설계
  • 에이전트 간 통신을 위한 JSON 스키마 정의
  • Planner→Coder 2단계 파이프라인 실제 구현

멀티에이전트 파이프라인 아키텍처

단일 에이전트는 복잡한 작업에서 한계를 보인다. 역할 분리를 통해 각 에이전트가 좁고 명확한 책임을 갖도록 하면 품질과 추적성이 높아진다.

사용자 요청
    │
    ▼
[Planner]  ──spec.md──▶  [Researcher]  ──context.json──▶  [Coder]
                                                               │
                                                         코드 변경
                                                               │
                                          [QA] ◀──────────────┘
                                           │
                                      테스트 결과
                                           │
                                    [Reviewer] ──▶ 최종 승인/거부

구현 요구사항

1. 에이전트 간 통신 JSON 스키마

모든 에이전트는 정해진 스키마로 데이터를 주고받는다.

schemas/planner_output.json
{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "PlannerOutput",
  "type": "object",
  "required": ["task_id", "objective", "subtasks", "constraints"],
  "properties": {
    "task_id": { "type": "string", "pattern": "^task-[0-9]{4}$" },
    "objective": { "type": "string", "minLength": 10 },
    "subtasks": {
      "type": "array",
      "minItems": 1,
      "items": {
        "type": "object",
        "required": ["id", "description", "assignee", "depends_on"],
        "properties": {
          "id": { "type": "string" },
          "description": { "type": "string" },
          "assignee": {
            "type": "string",
            "enum": ["researcher", "coder", "qa", "reviewer"]
          },
          "depends_on": {
            "type": "array",
            "items": { "type": "string" }
          }
        }
      }
    },
    "constraints": {
      "type": "object",
      "properties": {
        "max_iterations": { "type": "integer", "minimum": 1, "maximum": 20 },
        "forbidden_packages": { "type": "array", "items": { "type": "string" } },
        "target_files": { "type": "array", "items": { "type": "string" } }
      }
    }
  }
}
schemas/coder_output.json
{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "CoderOutput",
  "type": "object",
  "required": ["task_id", "changes", "test_command", "status"],
  "properties": {
    "task_id": { "type": "string" },
    "changes": {
      "type": "array",
      "items": {
        "type": "object",
        "required": ["file", "action", "description"],
        "properties": {
          "file": { "type": "string" },
          "action": { "enum": ["create", "modify", "delete"] },
          "description": { "type": "string" }
        }
      }
    },
    "test_command": { "type": "string" },
    "status": { "enum": ["complete", "partial", "blocked"] },
    "blocker": { "type": "string" }
  }
}

2. base_agent.py — 공통 에이전트 기반 클래스

base_agent.py
import anthropic
import json
from abc import ABC, abstractmethod
from pathlib import Path
from jsonschema import validate, ValidationError


class BaseAgent(ABC):
    """모든 파이프라인 에이전트의 공통 기반 클래스."""


    def __init__(self, name: str, schema_path: str | None = None):
        self.name = name
        self.client = anthropic.Anthropic()
        self.schema = self._load_schema(schema_path) if schema_path else None
        self.messages: list[dict] = []


    def _load_schema(self, path: str) -> dict:
        return json.loads(Path(path).read_text())


    def _validate_output(self, data: dict) -> bool:
        if self.schema is None:
            return True
        try:
            validate(instance=data, schema=self.schema)
            return True
        except ValidationError as e:
            print(f"[{self.name}] 스키마 검증 실패: {e.message}")
            return False


    def _call(self, system: str, user: str) -> str:
        self.messages.append({"role": "user", "content": user})
        response = self.client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system=system,
            messages=self.messages,
        )
        text = response.content[0].text
        self.messages.append({"role": "assistant", "content": text})
        return text


    def _extract_json(self, text: str) -> dict:
        """응답 텍스트에서 JSON 블록을 추출한다."""
        import re
        match = re.search(r"```json\s*([\s\S]+?)\s*```", text)
        if match:
            return json.loads(match.group(1))
        # 코드 블록 없이 순수 JSON인 경우
        return json.loads(text.strip())


    @abstractmethod
    def run(self, input_data: dict) -> dict:
        pass

3. planner_agent.py — 플래너 에이전트

planner_agent.py
from base_agent import BaseAgent
import uuid


PLANNER_SYSTEM = """
You are a software planning agent. Your job is to decompose a user request
into a structured plan that other agents can execute.


Output ONLY a valid JSON object matching the PlannerOutput schema.
Do not include any explanation outside the JSON block.
"""


class PlannerAgent(BaseAgent):
    def __init__(self):
        super().__init__(
            name="Planner",
            schema_path="schemas/planner_output.json"
        )


    def run(self, input_data: dict) -> dict:
        objective = input_data["objective"]
        codebase_summary = input_data.get("codebase_summary", "")


        user_prompt = f"""
Objective: {objective}


Codebase context:
{codebase_summary}


Create a task plan. Use task_id format "task-XXXX".
Assign subtasks to: researcher, coder, qa, reviewer.
"""
        response_text = self._call(PLANNER_SYSTEM, user_prompt)


        try:
            plan = self._extract_json(response_text)
        except Exception as e:
            print(f"[Planner] JSON 파싱 실패: {e}")
            # 폴백 플랜
            plan = {
                "task_id": f"task-{uuid.uuid4().hex[:4]}",
                "objective": objective,
                "subtasks": [
                    {
                        "id": "st-01",
                        "description": objective,
                        "assignee": "coder",
                        "depends_on": []
                    }
                ],
                "constraints": {"max_iterations": 5, "forbidden_packages": [], "target_files": []}
            }


        if self._validate_output(plan):
            print(f"[Planner] 계획 생성 완료: {len(plan['subtasks'])}개 서브태스크")
        return plan

4. coder_agent.py — 코더 에이전트

coder_agent.py
import subprocess
from base_agent import BaseAgent


CODER_SYSTEM = """
You are a coding agent. You receive a task plan and implement the code changes.
After making changes, run the specified test command to verify.


Output a JSON object matching the CoderOutput schema.
"""


class CoderAgent(BaseAgent):
    def __init__(self):
        super().__init__(
            name="Coder",
            schema_path="schemas/coder_output.json"
        )


    def run(self, input_data: dict) -> dict:
        plan = input_data["plan"]
        coder_tasks = [
            t for t in plan["subtasks"] if t["assignee"] == "coder"
        ]


        user_prompt = f"""
Task ID: {plan['task_id']}
Objective: {plan['objective']}


Your subtasks:
{chr(10).join(f"- {t['description']}" for t in coder_tasks)}


Constraints:
- Max iterations: {plan['constraints'].get('max_iterations', 5)}
- Forbidden packages: {plan['constraints'].get('forbidden_packages', [])}


Implement the changes and report what you did.
"""
        response_text = self._call(CODER_SYSTEM, user_prompt)


        # Claude Code를 헤드리스로 실행해 실제 코드 변경 수행
        result = subprocess.run(
            ["claude", "--print", "--no-color",
             "--dangerously-skip-permissions", response_text],
            capture_output=True, text=True, timeout=300
        )


        return {
            "task_id": plan["task_id"],
            "changes": [],        # 실제로는 git diff 파싱
            "test_command": "pytest tests/ -q",
            "status": "complete" if result.returncode == 0 else "partial",
            "claude_output": result.stdout[:500]
        }
  1. schemas/ 디렉터리에 두 스키마 파일 생성
  2. base_agent.py 구현 (pip install jsonschema)
  3. planner_agent.py 구현 및 단독 테스트
  4. coder_agent.py 구현
  5. 두 에이전트를 연결하는 pipeline.py 작성 및 실행

5단계 파이프라인 설계 문서

구현한 2단계를 바탕으로 나머지 3단계의 설계를 문서화한다.

pipeline_design.md
## Researcher (미구현)
- 입력: PlannerOutput
- 역할: 코드베이스 탐색, 관련 파일 목록화, 의존성 파악
- 출력: context.json (관련 파일 + 핵심 함수 요약)


## QA (Lab 09에서 구현)
- 입력: CoderOutput
- 역할: 테스트 실행, 코드 커버리지 측정, 회귀 감지
- 출력: qa_report.json


## Reviewer (미구현)
- 입력: QA 보고서 + CoderOutput
- 역할: 코드 품질 검토, 승인/거부 결정
- 출력: review_decision.json

제출물

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

  • schemas/planner_output.json, schemas/coder_output.json
  • base_agent.py — 공통 기반 클래스
  • planner_agent.py — 완전히 동작하는 플래너
  • coder_agent.py — 완전히 동작하는 코더
  • pipeline.py — Planner→Coder 연결 실행 스크립트
  • pipeline_design.md — 5단계 전체 설계 문서
  • README.md — 실행 결과 및 파이프라인 확장 시 예상 과제