레거시 백엔드에 GPT-6 Astra 붙일 때 예산 승인과 보안 통과를 먼저 끝내는 법이 있습니다
새 모델이 나올 때마다 도입이 늦어지는 건 벤치마크 점수 때문이 아닙니다. 비즈니스 로직 한가운데 특정 벤더 SDK를 하드코딩해 둔 레거시 구조, 그리고 대체 얼마가 나올지 가늠이 안 되는 API 청구서가 발목을 잡습니다.
여기에 "해외 리전으로 고객 데이터가 넘어가면 개인정보보호법에 걸린다"는 보안팀 지적까지 더해지면 실무자는 매일 밤 야근하며 검토 보고서만 고치게 됩니다. 엔드포인트를 바꾸는 일은 마지막 단계입니다. 실제 운영 로그로 숫자를 뽑아 예산 승인을 따내고, 프롬프트 마스킹으로 감사를 털어낸 뒤 코드를 갈아 끼워야 안전합니다.
30일 치 CloudWatch 로그로 계산하는 Astra API 예산
벤더가 홍보하는 단가표만 보고 기안을 올리면 트래픽이 몰릴 때 청구서 폭탄을 맞습니다. 운영 중인 서버의 최근 30일 호출 로그를 파싱해 실제 토큰 소비량을 먼저 확인해야 합니다.
CloudWatch Logs Insights를 쓴다면 AWS CLI 쿼리로 10초 안에 건당 토큰 소비량과 일일 총량을 뽑을 수 있습니다.
#!/usr/bin/env bash
set -euo pipefail
LOG_GROUP_NAME="/aws/backend/legacy-llm-service"
START_TIME=$(date -v-30d +%s 2>/dev/null || date -d "30 days ago" +%s)
END_TIME=$(date +%s)
QUERY_STRING='fields @timestamp, usage.prompt_tokens as p_tok, usage.completion_tokens as c_tok
| filter ispresent(p_tok) and ispresent(c_tok)
| stats count(@timestamp) as total_requests,
sum(p_tok) as sum_prompt_tokens,
sum(c_tok) as sum_completion_tokens,
avg(p_tok) as avg_prompt_tokens,
avg(c_tok) as avg_completion_tokens,
percentile(p_tok + c_tok, 95) as p95_total_tokens'
QUERY_ID=$(aws logs start-query \
--log-group-name "$LOG_GROUP_NAME" \
--start-time "$START_TIME" \
--end-time "$END_TIME" \
--query-string "$QUERY_STRING" \
--output text --query 'queryId')
sleep 5
aws logs get-query-results --query-id "$QUERY_ID" \
--output json | jq -r '.results[] | map({(.field): .value}) | add'
GPT-6 Astra의 가격 모델은 입력과 출력 구분 없이 1천만 토큰당 10달러(1M 토큰당 1.00달러)입니다. 뽑아낸 일일 요청 건수와 건당 평균 토큰을 곱해 월간 청구액을 시뮬레이션합니다.
기존 Fable 모델이 1M 토큰당 가중 평균 5.00달러를 청구하고 있었다면, Astra 도입 시 단가는 80% 내려갑니다.
시나리오 구분
일일 요청 건수
건당 평균 토큰
월간 총 토큰 수
Astra 단가 (1M 토큰)
월 예상 비용 (USD)
원화 환산액 (1,350원 기준)
보수적 (Low)
35,000
1,200
1,260,000,000
$1.00
$1,260.00
1,701,000원
기준 (Base)
50,000
1,200
1,800,000,000
$1.00
$1,800.00
2,430,000원
피크 (Peak)
75,000
1,500
3,375,000,000
$1.00
$3,375.00
4,556,250원
스프레드시트를 열고 B1 셀에 일일 호출 건수(50000), B2 셀에 평균 입력 토큰(800), B3 셀에 평균 출력 토큰(400), B4 셀에 기준 환율(1350)을 넣습니다. B5 셀에 =30*B1*(B2+B3), B6 셀에 =(B5/10000000)*10, B7 셀에 =B6*B4를 걸어두면 트래픽 변동에 따른 예산표가 바로 나옵니다.
트랜잭션 1,000건당 처리 비용이 6.00달러에서 1.20달러로 줄어든다는 계산서가 첨부되면 경영진 승인은 30분 만에 떨어집니다.
환경변수만 바꿔서 모델을 전환하는 어댑터 패턴
서비스 코드 안에서 fable.Client()나 openai.Client()를 직접 호출하고 있다면, 새 모델로 넘어갈 때마다 수십 개 파일을 일일이 열어 예외 처리를 고쳐야 합니다. 이 작업은 서비스 장애로 이어지기 딱 좋습니다.
파이썬의 typing.Protocol로 인터페이스를 고정하고 객체 어댑터 패턴을 적용하면 비즈니스 로직을 전혀 건드리지 않고 외부 모델을 바꿀 수 있습니다.
# core/llm_protocol.py
from typing import Protocol, List, Optional
from pydantic import BaseModel, Field
class LLMMessage(BaseModel):
role: str = Field(..., description="system, user, assistant")
content: str
class LLMUsage(BaseModel):
prompt_tokens: int
completion_tokens: int
total_tokens: int
class LLMResponse(BaseModel):
content: str
usage: LLMUsage
model_name: str
provider: str
class LLMClientAdapter(Protocol):
async def generate_completion(
self,
messages: List[LLMMessage],
temperature: float = 0.2,
max_tokens: Optional[int] = None
) -> LLMResponse:
...
기존 Fable 클라이언트와 새 Astra 클라이언트를 각각 어댑터로 감쌉니다. 그리고 LLMFactory를 두어 환경변수 LLM_PROVIDER 값에 따라 객체를 반환하도록 만듭니다.
# core/adapters.py
import httpx
import os
from typing import List, Optional
from core.llm_protocol import LLMClientAdapter, LLMMessage, LLMResponse, LLMUsage
class AstraAdapter:
def __init__(self, api_key: str, base_url: str = "https://api.astra.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.client = httpx.AsyncClient(
base_url=self.base_url,
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=httpx.Timeout(connect=5.0, read=90.0, write=10.0, pool=5.0)
)
async def generate_completion(
self,
messages: List[LLMMessage],
temperature: float = 0.2,
max_tokens: Optional[int] = None
) -> LLMResponse:
payload = {
"model": "gpt-6-astra",
"messages": [msg.model_dump() for msg in messages],
"temperature": temperature,
}
if max_tokens:
payload["max_tokens"] = max_tokens
resp = await self.client.post("/chat/completions", json=payload)
resp.raise_for_status()
data = resp.json()
return LLMResponse(
content=data["choices"][0]["message"]["content"],
usage=LLMUsage(
prompt_tokens=data["usage"]["prompt_tokens"],
completion_tokens=data["usage"]["completion_tokens"],
total_tokens=data["usage"]["total_tokens"]
),
model_name=data["model"],
provider="astra"
)
class FableAdapter:
def __init__(self, api_key: str, endpoint_url: str):
self.api_key = api_key
self.endpoint_url = endpoint_url
self.client = httpx.AsyncClient(
headers={"X-Fable-Key": self.api_key},
timeout=httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0)
)
async def generate_completion(
self,
messages: List[LLMMessage],
temperature: float = 0.2,
max_tokens: Optional[int] = None
) -> LLMResponse:
fable_payload = {
"prompt_sequence": [{"speaker": m.role, "text": m.content} for m in messages],
"gen_params": {"temp": temperature, "limit": max_tokens or 1024}
}
resp = await self.client.post(self.endpoint_url, json=fable_payload)
resp.raise_for_status()
data = resp.json()
return LLMResponse(
content=data["result"]["generated_text"],
usage=LLMUsage(
prompt_tokens=data["meta"]["tokens_in"],
completion_tokens=data["meta"]["tokens_out"],
total_tokens=data["meta"]["tokens_in"] + data["meta"]["tokens_out"]
# GPT-6 Astra 도입 전에 사내 로그 30일치부터 뽑아봐야 하는 이유
새 모델 벤치마크 점수가 아무리 높아도 레거시 서버에 바로 꽂을 수는 없습니다. 서비스 코드 곳곳에 특정 AI 공급자 SDK가 엉켜 있으면 엔드포인트 하나 바꾸려다 야근만 늘어납니다. 게다가 재무팀은 예산안을 요구하고, 보안팀은 해외 서버로 나가는 데이터를 증빙하라며 배포를 막아섭니다.
이 정체를 뚫는 일은 모델 성능을 검증하는 작업이 아닙니다. 트래픽 비용을 숫자로 계산하고, 외부 의존성을 격리하며, 민감 정보를 걸러내는 일부터 시작해야 합니다.
## 실제 호출 로그로 뽑는 30분짜리 API 예산안
영업용 단가표만 보고 예산을 올리면 첫 달 청구서에서 틀어집니다. 실제 운영 환경에서 찍힌 30일치 호출 로그를 직접 파싱해야 오차가 없습니다. AWS 환경이라면 CloudWatch Logs Insights 쿼리로 호출당 토큰 소비량을 바로 긁어올 수 있습니다.
```bash
#!/usr/bin/env bash
set -euo pipefail
LOG_GROUP_NAME="/aws/backend/legacy-llm-service"
START_TIME=$(date -v-30d +%s 2>/dev/null || date -d "30 days ago" +%s)
END_TIME=$(date +%s)
QUERY_STRING='fields @timestamp, usage.prompt_tokens as p_tok, usage.completion_tokens as c_tok
| filter ispresent(p_tok) and ispresent(c_tok)
| stats count(@timestamp) as total_requests,
sum(p_tok) as sum_prompt_tokens,
sum(c_tok) as sum_completion_tokens,
avg(p_tok) as avg_prompt_tokens,
avg(c_tok) as avg_completion_tokens,
percentile(p_tok + c_tok, 95) as p95_total_tokens'
QUERY_ID=$(aws logs start-query \
--log-group-name "$LOG_GROUP_NAME" \
--start-time "$START_TIME" \
--end-time "$END_TIME" \
--query-string "$QUERY_STRING" \
--output text --query 'queryId')
sleep 5
aws logs get-query-results --query-id "$QUERY_ID" \
--output json | jq -r '.results[] | map({(.field): .value}) | add'
GPT-6 Astra의 가격 정책은 입출력 구분 없이 1천만 토큰당 10달러($1.00/1M tokens) 단일 요율입니다. 수식은 단순합니다.
스프레드시트를 만들 때 B1에 일일 호출 건수(50000), B2에 평균 입력 토큰(800), B3에 평균 출력 토큰(400), B4에 환율(1350)을 넣습니다. B5 셀에 =30*B1*(B2+B3)를 넣고, B6에 =(B5/10000000)*10, B7에 =B6*B4를 걸어둡니다. 1M 토큰당 5달러를 청구하던 기존 공급자 모델과 비교하면 비용이 80% 줄어듭니다. 기준 시나리오에서 월 7,200달러(약 972만 원)를 아끼는 셈이니, 예산 승인 단계에서 반려당할 이유가 사라집니다.
배포 없이 환경변수로 모델을 바꾸는 구조
비즈니스 로직 안에서 특정 모델 SDK를 직접 부르는 구조는 피해야 합니다. 공급자가 자체 정의한 에러 클래스가 서비스 전체로 번지기 때문입니다. 파이썬 typing.Protocol로 인터페이스를 먼저 고정하고, 구현체를 어댑터로 감싸면 호출부 코드를 건드릴 필요가 없습니다.
# core/llm_protocol.py
from typing import Protocol, List, Optional
from pydantic import BaseModel, Field
class LLMMessage(BaseModel):
role: str = Field(..., description="system, user, assistant")
content: str
class LLMUsage(BaseModel):
prompt_tokens: int
completion_tokens: int
total_tokens: int
class LLMResponse(BaseModel):
content: str
usage: LLMUsage
model_name: str
provider: str
class LLMClientAdapter(Protocol):
async def generate_completion(
self,
messages: List[LLMMessage],
temperature: float = 0.2,
max_tokens: Optional[int] = None
) -> LLMResponse:
...
기존 Fable 엔드포인트와 신규 Astra 엔드포인트를 각각 어댑터 클래스로 격리합니다. 런타임 전환은 LLMFactory가 환경변수 하나로 처리합니다.
Connect 타임아웃 5초, Read 타임아웃 90초를 넘어가면 지체 없이 실패로 간주합니다. 실패 5회가 쌓이면 서킷이 열리고 30초 동안 외부 호출을 0초 만에 차단합니다. 늦어지는 호출을 붙잡고 서버 전체를 죽이는 일을 막는 최소한의 방어선입니다.
보안 감사를 통과하는 마스킹 파이프라인
해외 리전에 있는 외부 API로 사용자 입력을 그대로 보내면 개인정보보호법 제28조의8(개인정보의 국외 이전)에 걸립니다. 데이터가 네트워크 밖으로 나가기 전 인메모리 단계에서 마스킹을 끝내야 합니다. 단순 13자리 정규식은 송장번호나 주문번호까지 지워버리므로 가중치 체크섬 알고리즘을 함께 태웁니다.
Checksum=11−((i=1∑12di×wi)mod11)
# security/masking_middleware.py
import re
from typing import List
from core.llm_protocol import LLMMessage
class PIIMaskingPipeline:
EMAIL_REGEX = re.compile(r'[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+')
INTERNAL_IP_REGEX = re.compile(
r'\b(?:10\.\d{1,3}\.\d{1,3}\.\d{1,3}|'
r'172\.(?:1[6-9]|2\d|3[0-1])\.\d{1,3}\.\d{1,3}|'
r'192\.168\.\d{1,3}\.\d{1,3})\b'
)
RRN_CANDIDATE_REGEX = re.compile(r'\b(\d{6})[- ]?(\d{7})\b')
@classmethod
def is_valid_rrn(cls, front: str, back: str) -> bool:
full = front + back
if len(full) != 13 or not full.isdigit():
return False
weights = [2, 3, 4, 5, 6, 7, 8, 9, 2, 3, 4, 5]
s = sum(int(full[i]) * weights[i] for i in range(12))
remainder = (11 - (s % 11)) % 10
return remainder == int(full[12])
@classmethod
def mask_rrn(cls, text: str) -> str:
def replace(match):
front, back = match.group(1), match.group(2)
if cls.is_valid_rrn(front, back):
return "[RESIDENT_ID_MASKED]"
return match.group(0)
return cls.RRN_CANDIDATE_REGEX.sub(replace, text)
@classmethod
def sanitize(cls, text: str) -> str:
text = cls.mask_rrn(text)
text = cls.EMAIL_REGEX.sub("[EMAIL_MASKED]", text)
text = cls.INTERNAL_IP_REGEX.sub("[INTERNAL_IP_MASKED]", text)
return text
@classmethod
def sanitize_messages(cls, messages: List[LLMMessage]) -> List[LLMMessage]:
return [
LLMMessage(role=m.role, content=cls.sanitize(m.content))
for m in messages
]
공급사 계약서에 입력 프롬프트를 재학습에 쓰지 않는다는 Zero Data Retention(ZDR) 조항이 있는지 확인합니다. CloudWatch Logs에는 프롬프트 본문 저장을 막고 trace_id와 지연 시간, 토큰 수치만 남깁니다.
보안팀이나 개인정보보호책임자(CPO)에게는 아래 데이터 흐름표를 넘겨줍니다.
처리 구간
데이터 흐름
전송 데이터
보안 통제 장치
컴플라이언스 기준
1구간
클라이언트 → 게이트웨이
원본 프롬프트
TLS 1.3, JWT 인가
통신 구간 암호화
2구간
인메모리 마스킹 미들웨어
원본 → 식별자 마스킹
체크섬 기반 치환, 메모리 즉시 파기
개인정보 최소 수집 원칙
3구간
백엔드 어댑터 → Astra API
마스킹 완료 페이로드
AWS Secrets Manager 키 주입
개인정보보호법 제28조의8
4구간
Astra API → 백엔드 어댑터
모델 생성 텍스트
Zero Data Retention 보장, TLS
기업용 DPA 약관
5구간
백엔드 로거 → CloudWatch Logs
토큰 수치 및 지연 시간
본문 기록 차단, 메타데이터 한정
개인정보보호법 제29조
보안 검토를 신청할 때 1) 개인정보 처리방침 내 수탁사 고지 명세, 2) DPA 계약서 사본, 3) 체크섬 마스킹 미들웨어 테스트 결과, 4) 타임아웃 및 서킷 브레이커 설정 문서를 한 번에 묶어서 제출합니다. 인프라 방어 코드가 첨부된 문서는 기술 검토 기간을 눈에 띄게 줄여줍니다.