AI 에이전트 방어선 구축 실전 - 비용 대시보드·cron 알림·골든 세트를 코드 레벨로

들어가며

지난 AI 에이전트 운영 사고 총정리에서 방어선 5가지를 이야기했지만, "그래서 실제로 어떻게 만드냐"에 대한 코드가 없었습니다. 오늘 글은 그 빈칸을 채웁니다.

목표는 "오늘 저녁에 바로 돌릴 수 있는 실전 방어 스크립트 세트"입니다. 대기업용 풀스택 APM이 아니라 1인~소규모 팀이 주말 하루면 구축 가능한 최소 유효 방어선(Minimum Viable Defense). 구성 요소는 세 가지입니다.

• 레이어 1 - 실시간 비용 추적 (Python + SQLite + Prometheus)
• 레이어 2 - 일일 사용량 리포트 (cron + Slack webhook)
• 레이어 3 - 주간 골든 세트 회귀 테스트 (pytest 스타일)

각 레이어의 실제 코드를 Python/Bash/YAML로 제시합니다. 복붙 가능한 수준에서 시작해, 자기 환경에 맞춰 수정하면 됩니다.

1. 전체 아키텍처

세 레이어가 겹치지 않고 각자 다른 타임 스케일을 커버합니다.

[실시간 - 초~분 단위]
  API 호출 → 비용 계산 → SQLite 적재 → Prometheus 노출
  → 시간당 토큰 3배 초과 시 Slack 즉시 알림

[일일 - 매일 09:00]
  cron → 어제 사용량 집계 → 모델별/에이전트별 리포트
  → Slack 일일 요약 메시지

[주간 - 매주 금요일 10:00]
  cron → 골든 세트 20개 돌리기 → 점수 기록
  → 지난주 대비 10% 이상 하락 시 알림

기술 스택

컴포넌트	선택	이유
언어	Python 3.11+	AI SDK 생태계 최강
저장소	SQLite	1인 스택엔 충분, 설치 0 단계
메트릭	prometheus_client	표준 + Grafana 바로 연결
스케줄러	cron (또는 systemd timer)	의존성 없음
알림	Slack Incoming Webhook	무료, 즉시 연동

추가로 Grafana를 띄운다면 Prometheus + Grafana 실전 구축을 참고하면 됩니다.

2. 레이어 1 - 실시간 비용 추적

모든 LLM 호출에 래퍼를 씌워 비용과 토큰을 기록합니다.

2-1. SQLite 스키마

-- schema.sql
CREATE TABLE IF NOT EXISTS llm_calls (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    trace_id TEXT NOT NULL,
    session_id TEXT,
    agent_name TEXT NOT NULL,
    model TEXT NOT NULL,
    prompt_tokens INTEGER NOT NULL,
    completion_tokens INTEGER NOT NULL,
    latency_ms INTEGER NOT NULL,
    cost_usd REAL NOT NULL,
    finish_reason TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_llm_agent_time ON llm_calls(agent_name, created_at);
CREATE INDEX idx_llm_trace ON llm_calls(trace_id);

2-2. 비용 계산 상수

# pricing.py
# 2026년 4월 기준 가격 (per 1M tokens)
PRICING = {
    "claude-opus-4-7":      {"input": 5.00, "output": 25.00},
    "claude-sonnet-4-6":    {"input": 3.00, "output": 15.00},
    "claude-haiku-4-5":     {"input": 0.80, "output": 4.00},
    "gpt-5-4":              {"input": 2.50, "output": 15.00},
    "gpt-5-4-pro":          {"input": 30.00, "output": 180.00},
    "gpt-5-4-mini":         {"input": 0.75, "output": 3.00},
    "gpt-5-4-nano":         {"input": 0.20, "output": 0.80},
}

def compute_cost(model: str, p_tokens: int, c_tokens: int) -> float:
    if model not in PRICING:
        raise ValueError(f"Unknown model: {model}")
    rate = PRICING[model]
    return (p_tokens * rate["input"] + c_tokens * rate["output"]) / 1_000_000

2-3. 호출 래퍼 (Anthropic SDK 예시)

# wrapper.py
import time, sqlite3, uuid
from anthropic import Anthropic
from pricing import compute_cost
from prometheus_client import Counter, Histogram

tokens_counter = Counter(
    "ai_tokens_total", "Total tokens",
    ["model", "type", "agent"]
)
cost_counter = Counter(
    "ai_cost_usd_total", "Total cost USD",
    ["model", "agent"]
)
latency_hist = Histogram(
    "ai_call_latency_ms", "Call latency",
    ["model", "agent"]
)

client = Anthropic()
DB = "ai_usage.db"

def call_claude(agent: str, model: str, messages: list,
                session_id: str = "", **kw):
    trace_id = str(uuid.uuid4())
    t0 = time.time()
    resp = client.messages.create(model=model, messages=messages, **kw)
    latency_ms = int((time.time() - t0) * 1000)

    p_tok = resp.usage.input_tokens
    c_tok = resp.usage.output_tokens
    cost = compute_cost(model, p_tok, c_tok)

    # Prometheus 메트릭
    tokens_counter.labels(model, "input", agent).inc(p_tok)
    tokens_counter.labels(model, "output", agent).inc(c_tok)
    cost_counter.labels(model, agent).inc(cost)
    latency_hist.labels(model, agent).observe(latency_ms)

    # SQLite 적재
    with sqlite3.connect(DB) as db:
        db.execute("""
            INSERT INTO llm_calls
            (trace_id, session_id, agent_name, model, prompt_tokens,
             completion_tokens, latency_ms, cost_usd, finish_reason)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (trace_id, session_id, agent, model,
              p_tok, c_tok, latency_ms, cost, resp.stop_reason))

    return resp, {"trace_id": trace_id, "cost": cost,
                  "tokens": p_tok + c_tok}

2-4. 시간당 이상치 탐지 (실시간)

# anomaly_watch.py
import sqlite3, requests, os
from datetime import datetime, timedelta

SLACK_WEBHOOK = os.environ["SLACK_WEBHOOK"]
BASELINE_HOURS = 24 * 7  # 지난 일주일

def check_spike():
    with sqlite3.connect("ai_usage.db") as db:
        # 최근 1시간
        recent = db.execute("""
            SELECT SUM(prompt_tokens + completion_tokens)
            FROM llm_calls
            WHERE created_at >= datetime('now', '-1 hour')
        """).fetchone()[0] or 0

        # 지난 7일의 시간당 평균
        avg = db.execute("""
            SELECT AVG(hourly) FROM (
                SELECT SUM(prompt_tokens + completion_tokens) AS hourly
                FROM llm_calls
                WHERE created_at >= datetime('now', '-7 days')
                GROUP BY strftime('%Y-%m-%d %H', created_at)
            )
        """).fetchone()[0] or 1

        ratio = recent / avg if avg > 0 else 0
        if ratio > 3.0:
            alert(f"🚨 토큰 급증 감지: 지난 1시간 {recent:,} "
                  f"(평소의 {ratio:.1f}배)")

def alert(msg: str):
    requests.post(SLACK_WEBHOOK, json={"text": msg}, timeout=5)

if __name__ == "__main__":
    check_spike()

이 스크립트를 10분마다 돌립니다.

# crontab -e
*/10 * * * * cd /opt/ai-defense && /usr/bin/python3 anomaly_watch.py

3. 레이어 2 - 일일 사용량 리포트

매일 아침 어제 사용량을 Slack에 요약해서 받습니다.

# daily_report.py
import sqlite3, requests, os
from datetime import date, timedelta

SLACK_WEBHOOK = os.environ["SLACK_WEBHOOK"]
DAILY_BUDGET = float(os.environ.get("DAILY_BUDGET_USD", "20.0"))

def build_report():
    yesterday = (date.today() - timedelta(days=1)).isoformat()

    with sqlite3.connect("ai_usage.db") as db:
        rows = db.execute("""
            SELECT agent_name, model,
                   COUNT(*) AS calls,
                   SUM(prompt_tokens) AS p_tok,
                   SUM(completion_tokens) AS c_tok,
                   SUM(cost_usd) AS cost
            FROM llm_calls
            WHERE DATE(created_at) = ?
            GROUP BY agent_name, model
            ORDER BY cost DESC
        """, (yesterday,)).fetchall()

    if not rows:
        return "어제 사용 내역 없음"

    total = sum(r[5] for r in rows)
    budget_pct = (total / DAILY_BUDGET) * 100
    status = "🟢" if budget_pct < 80 else "🟡" if budget_pct < 100 else "🔴"

    msg = [f"*📊 AI 일일 리포트 ({yesterday})*",
           f"{status} 총 비용: ${total:.2f} / 예산 ${DAILY_BUDGET:.2f} "
           f"({budget_pct:.0f}%)",
           "", "| 에이전트 | 모델 | 호출 | 비용 |",
           "|---------|------|-----|------|"]
    for r in rows[:10]:
        msg.append(f"| {r[0]} | {r[1]} | {r[2]:,} | ${r[5]:.2f} |")

    return "\n".join(msg)

def main():
    report = build_report()
    requests.post(SLACK_WEBHOOK, json={"text": report}, timeout=10)

if __name__ == "__main__":
    main()

# cron - 매일 09:00
0 9 * * * cd /opt/ai-defense && /usr/bin/python3 daily_report.py

Slack 메시지 샘플

📊 AI 일일 리포트 (2026-04-20)
🟡 총 비용: $18.43 / 예산 $20.00 (92%)

| 에이전트            | 모델              | 호출  | 비용    |
|--------------------|-------------------|-------|---------|
| backend-agent      | claude-opus-4-7   | 142   | $8.30   |
| frontend-agent     | claude-sonnet-4-6 | 89    | $3.10   |
| pm-agent           | gpt-5-4           | 47    | $2.85   |
| qa-agent           | claude-haiku-4-5  | 203   | $2.40   |
| marketer-agent     | gpt-5-4-mini      | 38    | $1.78   |

이 리포트 하나만 매일 보면 비용 이상·특정 에이전트 폭주·예산 소진 속도가 한눈에 들어옵니다.

4. 레이어 3 - 주간 골든 세트 회귀 테스트

품질 드리프트 감지의 유일한 객관적 수단. "공급자가 어제와 같다"는 주장을 수치로 검증합니다.

4-1. 골든 세트 형식

# golden_set.yaml
cases:
  - id: py-func-001
    agent: backend-agent
    input: |
      Python 함수를 작성하세요: 두 정렬된 리스트를 머지해서
      중복 없이 정렬된 결과를 반환. 시간 복잡도 O(n+m) 이내.
    rubric:
      - returns_sorted_list
      - handles_duplicates
      - time_complexity_ok
      - includes_type_hints
    expected_outputs:
      - "set"  # 중복 제거에 set 또는 명시적 체크 필요
      - "def "
      - "List"

  - id: api-design-001
    agent: backend-agent
    input: |
      Spring Boot REST 엔드포인트 설계:
      - 고객 주문 조회 (GET)
      - 주문 취소 (POST /cancel)
      - RFC 7807 ProblemDetail 에러 포맷
    rubric:
      - uses_problem_detail
      - correct_http_methods
      - path_versioning_present
      - has_example_responses
    expected_outputs:
      - "ProblemDetail"
      - "/api/v"
      - "@RestController"

  # ... 총 20개 케이스

4-2. 평가 러너

# golden_eval.py
import yaml, sqlite3, requests, os
from datetime import datetime
from wrapper import call_claude

def load_golden():
    with open("golden_set.yaml") as f:
        return yaml.safe_load(f)["cases"]

def score_case(output: str, expected: list, rubric: list) -> float:
    # 단순 휴리스틱 스코어: expected 단어 매칭 + rubric 체크
    hits = sum(1 for e in expected if e.lower() in output.lower())
    ratio = hits / len(expected) if expected else 1.0
    return ratio

def run():
    cases = load_golden()
    results = []

    with sqlite3.connect("ai_usage.db") as db:
        db.execute("""
            CREATE TABLE IF NOT EXISTS golden_scores (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                run_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                case_id TEXT, agent TEXT, score REAL
            )
        """)

        for c in cases:
            resp, meta = call_claude(
                agent=c["agent"],
                model="claude-opus-4-7",
                messages=[{"role": "user", "content": c["input"]}],
                max_tokens=2000,
            )
            output = resp.content[0].text
            score = score_case(output, c["expected_outputs"], c["rubric"])
            results.append((c["id"], c["agent"], score))
            db.execute("INSERT INTO golden_scores (case_id, agent, score) "
                       "VALUES (?, ?, ?)", (c["id"], c["agent"], score))

    return results

def check_drift(results):
    with sqlite3.connect("ai_usage.db") as db:
        # 지난주 평균
        prev = db.execute("""
            SELECT case_id, AVG(score) FROM golden_scores
            WHERE run_at >= datetime('now', '-14 days')
              AND run_at < datetime('now', '-7 days')
            GROUP BY case_id
        """).fetchall()
        prev_map = dict(prev)

    drifts = []
    for cid, agent, score in results:
        baseline = prev_map.get(cid)
        if baseline and score < baseline * 0.9:
            drifts.append((cid, agent, baseline, score))
    return drifts

def main():
    results = run()
    drifts = check_drift(results)
    avg = sum(r[2] for r in results) / len(results)

    msg = [f"*🧪 주간 골든 세트 결과*",
           f"평균 점수: {avg:.2f} (케이스 {len(results)}개)"]
    if drifts:
        msg.append("\n🔻 드리프트 감지:")
        for cid, agent, b, s in drifts:
            pct = ((s - b) / b) * 100
            msg.append(f"  - {cid} ({agent}): {b:.2f} → {s:.2f} ({pct:+.0f}%)")
    else:
        msg.append("✅ 드리프트 없음")

    requests.post(os.environ["SLACK_WEBHOOK"],
                  json={"text": "\n".join(msg)})

if __name__ == "__main__":
    main()

# 매주 금요일 10시
0 10 * * 5 cd /opt/ai-defense && /usr/bin/python3 golden_eval.py

Slack 출력 샘플

🧪 주간 골든 세트 결과
평균 점수: 0.87 (케이스 20개)

🔻 드리프트 감지:
  - api-design-001 (backend-agent): 0.95 → 0.80 (-16%)
  - py-func-003 (backend-agent): 0.90 → 0.75 (-17%)

이 알림이 울리는 순간 복구 플레이북 유형 B가 발동됩니다. 프롬프트·모델 alias·공급자 changelog를 순서대로 점검.

5. Grafana 대시보드 연결

Prometheus 메트릭을 Grafana에 연결하면 실시간 시각화가 끝납니다.

5-1. docker-compose.yml

services:
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports: ["9090:9090"]

  grafana:
    image: grafana/grafana:latest
    ports: ["3000:3000"]
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=change_me

5-2. prometheus.yml

scrape_configs:
  - job_name: ai-defense
    static_configs:
      - targets: ["host.docker.internal:8000"]
    scrape_interval: 15s

5-3. 대시보드 핵심 4개 패널 (JSON 쿼리)

패널	쿼리
오늘 누적 비용	sum(increase(ai_cost_usd_total[24h]))
시간당 토큰 사용량	sum(rate(ai_tokens_total[1h])) by (agent)
에이전트별 평균 지연	histogram_quantile(0.5, rate(ai_call_latency_ms_bucket[5m]))
에러·타임아웃 카운트	increase(ai_errors_total[1h])

Grafana에서 "Import Dashboard" 후 위 쿼리들을 패널에 꽂으면 10분 안에 실시간 대시보드가 완성됩니다.

6. Claude Code / Cursor 로컬 훅

API 직접 호출만 잡으면 IDE 내부 호출은 놓칩니다. 로컬 훅으로 보강합니다.

6-1. Claude Code 후크 (settings.json)

{
  "hooks": {
    "postToolUse": {
      "command": "python3 /opt/ai-defense/log_tool_use.py",
      "env": {
        "AI_DEFENSE_DB": "/opt/ai-defense/ai_usage.db"
      }
    }
  }
}

6-2. log_tool_use.py

# Claude Code tool 호출 후 실행
import sys, json, sqlite3, os
from datetime import datetime

data = json.loads(sys.stdin.read())
db_path = os.environ["AI_DEFENSE_DB"]

with sqlite3.connect(db_path) as db:
    db.execute("""
        CREATE TABLE IF NOT EXISTS tool_uses (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            tool_name TEXT, input_size INTEGER,
            session_id TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    db.execute("INSERT INTO tool_uses (tool_name, input_size, session_id) "
               "VALUES (?, ?, ?)",
               (data.get("tool_name"),
                len(json.dumps(data.get("input", ""))),
                data.get("session_id", "")))

6-3. Cursor는 어떻게?

Cursor는 공식 hook API가 제한적입니다. 대안은 네트워크 레벨에서 가로채기. 로컬에서 mitmproxy를 띄워 API 트래픽을 로깅하거나, Anthropic/OpenAI API 키를 프록시 키로 대체해 프록시 서버에서 비용을 기록하는 구조가 현실적입니다.

# minimal API proxy (flask)
from flask import Flask, request, Response
import requests, sqlite3, uuid, json

app = Flask(__name__)

@app.route("/v1/<path:p>", methods=["POST"])
def proxy(p):
    trace_id = str(uuid.uuid4())
    body = request.get_json()
    # 실제 Anthropic으로 프록시
    r = requests.post(f"https://api.anthropic.com/v1/{p}",
                      headers={k: v for k, v in request.headers
                               if k.lower() != "host"},
                      json=body, timeout=120)
    # 비용 로깅
    if r.status_code == 200:
        log_usage(r.json(), trace_id, "cursor")
    return Response(r.content, status=r.status_code,
                    headers=dict(r.headers))

이 프록시로 CURSOR_API_BASE=http://localhost:8080/v1 라우팅하면 Cursor 호출도 전부 기록됩니다.

7. 비용 이상치 탐지 고도화

단순히 "3배 초과" 룰은 거짓 양성이 많습니다. 조금 정교하게 가져갑시다.

요일/시간 보정

# 월요일 아침 피크는 "평소"이니까 거짓 경보 X
SELECT AVG(hourly) FROM (
  SELECT SUM(prompt_tokens + completion_tokens) AS hourly
  FROM llm_calls
  WHERE created_at >= datetime('now', '-28 days')
    AND strftime('%w %H', created_at) = strftime('%w %H', 'now')
  GROUP BY strftime('%Y-%m-%d %H', created_at)
)

같은 요일·시간대만 기준선으로 쓰면 주간 패턴을 반영합니다.

이중 임계치

• Warning: 평소의 2배 → Slack 알림만
• Critical: 평소의 5배 or 분당 5만 토큰 → 자동 에이전트 일시 정지

if rate_per_minute > 50_000:
    os.system("systemctl stop backend-agent.service")
    alert("🛑 에이전트 자동 정지 (루프 의심)")
elif ratio > 2.0:
    alert(f"⚠️ 경고: 평소의 {ratio:.1f}배")

8. 배포와 운영

디렉토리 구조

/opt/ai-defense/
├── schema.sql
├── pricing.py
├── wrapper.py
├── anomaly_watch.py
├── daily_report.py
├── golden_eval.py
├── golden_set.yaml
├── log_tool_use.py
├── docker-compose.yml
├── prometheus.yml
├── ai_usage.db         # SQLite 파일
└── .env                # SLACK_WEBHOOK, DAILY_BUDGET_USD 등

systemd 단위 예시

# /etc/systemd/system/ai-metrics.service
[Unit]
Description=AI Defense metrics exporter
After=network.target

[Service]
Type=simple
EnvironmentFile=/opt/ai-defense/.env
WorkingDirectory=/opt/ai-defense
ExecStart=/usr/bin/python3 /opt/ai-defense/metrics_server.py
Restart=on-failure

[Install]
WantedBy=multi-user.target

백업

ai_usage.db는 운영 데이터입니다. 매주 별도 볼륨으로 백업하세요.

# 매주 일요일 03:00
0 3 * * 0 sqlite3 /opt/ai-defense/ai_usage.db ".backup /backup/ai_usage_$(date +%%Y%%m%%d).db"

9. 첫 주 운영 체크리스트

구축 완료 후 첫 주에 확인할 것들.

항목	체크
wrapper.py로 모든 호출이 기록되는가	SQLite 레코드 카운트로 확인
anomaly_watch.py cron이 돌고 있는가	/var/log/cron 또는 systemctl status
daily_report.py Slack 출력 포맷이 읽을 만한가	하루 받아보고 조정
golden_set 케이스 20개가 모두 실패 없이 도는가	첫 실행 에러 디버깅
Grafana 대시보드 4개 패널 모두 데이터가 들어오는가	시각화 확인
Claude Code 로컬 훅이 적재되는가	tool_uses 테이블 확인
Cursor 프록시를 쓴다면 정상 라우팅 확인	샘플 호출 후 SQLite 확인
Critical 자동 정지 시나리오 수동 테스트	가짜 트래픽으로 발화 확인

10. 확장 로드맵

첫 버전이 돌아가면 다음 단계로 확장해보세요.

• PII 스캐너: 프롬프트·응답에 민감정보가 섞였는지 자동 체크
• LLM-as-Judge: 별도 판사 모델로 5% 샘플 자동 채점
• 다중 공급자 통합: OpenAI·Gemini·Claude 사용량을 한 대시보드에
• 팀 SSO 연동: 개발자별·팀별 예산 분리
• 비용 할당 리포트: 기능·제품별 비용 자동 분배 (financeOps)
• 경고 피로 감소: 유사 경보 자동 그룹핑

특히 사내 MCP 서버를 쓰고 있다면 MCP 호출까지 이 파이프라인에 통합하는 게 다음 큰 단계입니다.

마치며

AI 방어선 실전 구축의 핵심 포인트를 정리합니다.

• 3-레이어 구조가 커버리지의 핵심. 실시간(분 단위)·일일·주간 — 서로 다른 타임 스케일을 겹치게 해야 사고가 새어 나가지 않습니다. 한 가지 주기만 쓰면 반드시 구멍이 생깁니다. 실시간은 루프·폭주용, 일일은 예산·추세용, 주간은 품질 드리프트용.
• SQLite + Prometheus + Slack만으로 충분. APM 비용 월 $200씩 낼 필요 없습니다. 위 코드 전체가 1인 스택 관점에서 한 주말이면 구축 가능하고, 운영비는 서버 값 외 $0입니다. 조직 규모가 커지면 그때 별도 스택 도입 검토해도 늦지 않습니다.
• Claude Code hook + API proxy 조합으로 사각지대 없애기. 단순 API 래핑만 하면 IDE 내부 호출을 놓칩니다. Claude Code는 hook, Cursor는 프록시로 우회해서 "모든 경로의 호출"을 한 DB에 모아야 정확한 그림이 나옵니다.
• 골든 세트가 공급자 투명성 하락에 대한 유일한 방어. Transparency Index 40 시대에 "공급자가 어제와 같다"는 주장을 검증하는 수단은 우리 조직의 회귀 테스트뿐입니다. 20개 케이스로 시작해 매주 조금씩 늘려가세요.
• 이중 임계치와 자동 정지가 진짜 예방이다. 알림만 보내는 시스템은 새벽에 터지면 못 막습니다. Critical 임계에 systemctl stop까지 자동으로 걸어두면 "자는 동안 $500 소진" 같은 사고를 구조적으로 차단합니다.

코드 자체는 공개된 조각들이지만 이걸 "실제로 돌리고 있는 팀"은 의외로 드뭅니다. 대부분 "나중에 하자"로 미루다가 사고가 터진 뒤 부랴부랴 구축합니다. Uber가 $3.4B 태워서 배운 교훈을 무료로 가져갈 수 있는 구간이 지금입니다. 다음 포스트에서는 이 방어 스택 위에 LLM-as-Judge 기반 품질 자동 채점 시스템을 얹는 방법을 다뤄볼 예정입니다.