AI 에이전트 88%가 프로덕션에 실패하는 이유: 5계층 하네스 아키텍처가 그 답이다

# MCP 기반 도구 등록 예시 (Claude Agent SDK)
from anthropic import Anthropic
 
client = Anthropic()
 
tools = [
    {
        "name": "read_file",
        "description": "파일 내용을 읽어 반환합니다",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "읽을 파일의 절대 경로"
                }
            },
            "required": ["path"]
        }
    },
    {
        "name": "run_tests",
        "description": "지정된 테스트 스위트를 실행하고 결과를 반환합니다",
        "input_schema": {
            "type": "object",
            "properties": {
                "test_path": {
                    "type": "string",
                    "description": "테스트 파일 또는 디렉토리 경로"
                }
            },
            "required": ["test_path"]
        }
    }
]
 
response = client.messages.create(
    model="claude-opus-4-7",
    max_tokens=4096,
    tools=tools,
    messages=[{"role": "user", "content": "src/auth.py를 읽고 테스트를 실행해줘"}]
)

# PEV 패턴 구현 예시
async def pev_loop(task: str, max_retries: int = 3) -> dict:
    plan = await agent.plan(task)        # 1. 계획 수립
 
    for attempt in range(max_retries):
        result = await agent.execute(plan)   # 2. 실행
 
        # 검증은 에이전트 자신이 아닌 독립 함수에게 맡깁니다
        verification = await verifier.check(
            task=task,
            plan=plan,
            result=result
        )                                     # 3. 검증
 
        if verification.passed:
            return {"status": "success", "result": result}
 
        # 검증 실패 시 피드백을 반영해 계획을 수정하고 재시도
        plan = await agent.replan(
            original_task=task,
            failed_result=result,
            feedback=verification.feedback
        )
 
    return {"status": "failed", "last_result": result}

# 계층적 메모리 관리 예시
# mem0: 에이전트에 장기 기억을 제공하는 Python 라이브러리 (pip install mem0ai)
from mem0 import Memory
 
memory = Memory()
 
class StatefulAgent:
    def __init__(self, user_id: str):
        self.user_id = user_id
        self.session_history = []  # 단기: 세션 내에서만 유지
 
    async def process(self, message: str) -> str:
        # 장기 기억에서 현재 메시지와 관련된 맥락 검색
        long_term = memory.search(
            query=message,
            user_id=self.user_id,
            limit=5
        )
 
        # 단기 + 장기 기억을 합성해 시스템 프롬프트 구성
        context = self._build_context(
            session=self.session_history,
            long_term=long_term
        )
 
        response = await llm.complete(system=context, user=message)
 
        # 단기 기억 업데이트
        self.session_history.append({"role": "user", "content": message})
        self.session_history.append({"role": "assistant", "content": response})
 
        # 중요한 정보(프로젝트 컨벤션, 사용자 선호 등)를 장기 기억에 저장
        memory.add(
            messages=[
                {"role": "user", "content": message},
                {"role": "assistant", "content": response}
            ],
            user_id=self.user_id
        )
 
        return response
 
    def _build_context(self, session: list, long_term: list) -> str:
        # 장기 기억을 시스템 프롬프트 앞부분에 삽입
        memory_context = "\n".join(f"- {m['memory']}" for m in long_term)
        return f"과거 기억:\n{memory_context}\n\n"

# 고위험 행동 가드레일 구현
from enum import Enum
from typing import Callable
 
class RiskLevel(Enum):
    LOW      = "low"
    MEDIUM   = "medium"
    HIGH     = "high"
    CRITICAL = "critical"
 
# 도구를 등록하는 시점에 위험도와 정책을 함께 정의하는 것이 좋습니다
ACTION_POLICIES = {
    "read_file":         RiskLevel.LOW,
    "write_file":        RiskLevel.MEDIUM,
    "delete_file":       RiskLevel.CRITICAL,
    "send_http_request": RiskLevel.HIGH,
    "run_shell_command": RiskLevel.HIGH,
}
 
async def guarded_execute(
    action: str,
    params: dict,
    human_approval_fn: Callable
) -> dict:
    risk = ACTION_POLICIES.get(action, RiskLevel.HIGH)
 
    if risk == RiskLevel.CRITICAL:
        # CRITICAL 행동은 사람의 명시적 승인이 필요합니다
        approved = await human_approval_fn(
            action=action,
            params=params,
            reason=f"위험도 CRITICAL: {action} 실행 승인이 필요합니다"
        )
        if not approved:
            return {"status": "rejected", "reason": "사용자 미승인"}
 
    elif risk == RiskLevel.HIGH:
        # HIGH는 정책 검사 후 자동 승인 가능
        if not policy_check(action, params):  # 정책 검사 헬퍼
            return {"status": "blocked", "reason": "정책 위반"}
 
    return await execute_action(action, params)  # 실제 도구 실행

# LangSmith를 이용한 에이전트 트레이싱
# LangSmith: 에이전트 실행 흐름, 토큰 사용량, 실패 지점을 추적하는 관측 도구
# pip install langsmith
from langsmith import traceable
 
@traceable(name="agent-main-loop", tags=["production"])
async def agent_loop(task: str) -> str:
    """@traceable이 함수 실행 전체를 자동으로 트레이스합니다."""
    plan = await create_plan(task)
 
    results = []
    for step in plan.steps:
        result = await execute_step(step)
        results.append(result)
 
    return synthesize_results(results)
 
# 개별 단계에도 적용하면 단계별 소요 토큰과 시간이 모두 기록됩니다
@traceable(name="execute-step")
async def execute_step(step) -> dict:
    result = await run_tool(step.tool, step.params)  # 실제 도구 호출
    return {"step_type": step.type, "result": result}

import asyncio
from anthropic import Anthropic
 
client = Anthropic()
 
async def planner_agent(brief: str) -> dict:
    """짧은 브리프 → 완전한 제품 스펙 (JSON 반환)"""
    response = client.messages.create(
        model="claude-opus-4-7",
        max_tokens=8192,
        system="""당신은 제품 아키텍트입니다.
입력된 간단한 요구사항을 기술 스펙으로 확장하세요.
반드시 JSON 형식으로 반환하세요:
{"features": [...], "api_contracts": {...}, "test_criteria": [...]}""",
        messages=[{"role": "user", "content": brief}]
    )
    return parse_spec(response.content[0].text)  # JSON 파싱 헬퍼
 
async def generator_agent(spec: dict, sprint_idx: int) -> dict:
    """스펙 → 스프린트 단위 코드 구현"""
    response = client.messages.create(
        model="claude-opus-4-7",
        max_tokens=8192,
        system="당신은 시니어 엔지니어입니다. 명세에 따라 코드를 구현하세요.",
        messages=[{
            "role": "user",
            "content": f"Sprint {sprint_idx} 구현:\n{spec['features'][sprint_idx]}"
        }]
    )
    return {"code": response.content[0].text, "sprint": sprint_idx}
 
async def evaluator_agent(code: dict, test_criteria: list) -> dict:
    """테스트 실행 후 결과를 분석해 통과/실패 판정"""
    # run_tests: pytest / Playwright 등 실제 테스트 도구를 호출하는 헬퍼
    test_results = await run_tests(code["code"], test_criteria)
 
    response = client.messages.create(
        model="claude-opus-4-7",
        max_tokens=2048,
        system="테스트 결과를 분석하고 통과/실패 판정과 구체적인 피드백을 제공하세요.",
        messages=[{"role": "user", "content": f"테스트 결과:\n{test_results}"}]
    )
    return {
        "passed": "PASS" in response.content[0].text,
        "feedback": response.content[0].text
    }
 
async def three_agent_harness(brief: str) -> str:
    spec = await planner_agent(brief)      # 계획 수립
 
    final_code = []
    for sprint_idx in range(len(spec["features"])):
        retries = 0
        while retries < 3:
            code = await generator_agent(spec, sprint_idx)   # 구현
            evaluation = await evaluator_agent(              # 계층 2: 검증 루프
                code, spec["test_criteria"]
            )
 
            if evaluation["passed"]:
                final_code.append(code)
                break
 
            # 피드백을 스펙에 반영해 다음 시도에 활용
            spec["features"][sprint_idx] += f"\n피드백: {evaluation['feedback']}"
            retries += 1
 
    return assemble_final_code(final_code)  # 스프린트별 코드를 최종 결과물로 조합

// TypeScript로 5계층 통합 하네스 구성
import Anthropic from "@anthropic-ai/sdk";
import { Client as LangSmithClient } from "langsmith";
import { traceable } from "langsmith/traceable"; // 계층 5: Observability
import { Memory } from "mem0ai";                 // 계층 3: Context & Memory
 
const client = new Anthropic();
const langsmith = new LangSmithClient();
const memory = new Memory();
 
// 계층 4: Guardrails — 차단할 고위험 작업 목록
const BLOCKED_OPERATIONS = ["force_push", "delete_branch_main", "drop_table"];
 
// traceable()로 감싸면 모든 실행이 LangSmith에 자동 기록됩니다
const codingAgentHarness = traceable(
  async (task: string, repoContext: string) => {
    // 계층 3: 프로젝트 장기 메모리 로드
    const projectMemory = await memory.search({ query: task, limit: 10 });
 
    // 계층 4: 작업 사전 위험도 검사
    const isBlocked = BLOCKED_OPERATIONS.some(op => task.includes(op));
    if (isBlocked) {
      return { status: "blocked", reason: "정책 위반 작업 감지" };
    }
 
    // 계층 1: 도구 정의 (Tool Orchestration)
    const tools: Anthropic.Tool[] = [
      {
        name: "read_file",
        description: "파일 내용을 읽어 반환합니다",
        input_schema: {
          type: "object" as const,
          properties: {
            path: { type: "string", description: "읽을 파일 경로" }
          },
          required: ["path"]
        }
      },
      {
        name: "write_file",
        description: "파일에 내용을 씁니다",
        input_schema: {
          type: "object" as const,
          properties: {
            path:    { type: "string", description: "쓸 파일 경로" },
            content: { type: "string", description: "파일에 쓸 내용" }
          },
          required: ["path", "content"]
        }
      },
      {
        name: "run_tests",
        description: "테스트 스위트를 실행하고 결과를 반환합니다",
        input_schema: {
          type: "object" as const,
          properties: {
            test_path: { type: "string", description: "테스트 디렉토리 경로" }
          },
          required: ["test_path"]
        }
      }
    ];
 
    let passed = false;
    let attempts = 0;
 
    // 계층 2: PEV 검증 루프
    while (!passed && attempts < 3) {
      await client.messages.create({
        model: "claude-opus-4-7",
        max_tokens: 8192,
        tools,
        system: buildSystemPrompt(projectMemory, repoContext), // 메모리 주입
        messages: [{ role: "user", content: task }]
      });
 
      // 에이전트 자신이 아닌 독립 함수로 검증합니다
      const testResult = await runTestSuite();
      passed = testResult.passed;
      attempts++;
    }
 
    // 계층 3: 작업 결과를 장기 메모리에 저장
    await memory.add({ task, result: "completed", attempts });
 
    return { status: passed ? "success" : "failed", attempts };
  },
  { name: "coding-agent", tags: ["production"] }
);