Written with AI assistance from the source video. The video is the authority.
Log in to leave a comment
No posts yet
새 모델이 나올 때마다 도입이 늦어지는 건 벤치마크 점수 때문이 아닙니다. 비즈니스 로직 한가운데 특정 벤더 SDK를 하드코딩해 둔 레거시 구조, 그리고 대체 얼마가 나올지 가늠이 안 되는 API 청구서가 발목을 잡습니다.
여기에 "해외 리전으로 고객 데이터가 넘어가면 개인정보보호법에 걸린다"는 보안팀 지적까지 더해지면 실무자는 매일 밤 야근하며 검토 보고서만 고치게 됩니다. 엔드포인트를 바꾸는 일은 마지막 단계입니다. 실제 운영 로그로 숫자를 뽑아 예산 승인을 따내고, 프롬프트 마스킹으로 감사를 털어낸 뒤 코드를 갈아 끼워야 안전합니다.
벤더가 홍보하는 단가표만 보고 기안을 올리면 트래픽이 몰릴 때 청구서 폭탄을 맞습니다. 운영 중인 서버의 최근 30일 호출 로그를 파싱해 실제 토큰 소비량을 먼저 확인해야 합니다.
CloudWatch Logs Insights를 쓴다면 AWS CLI 쿼리로 10초 안에 건당 토큰 소비량과 일일 총량을 뽑을 수 있습니다.
`bash
#!/usr/bin/env bash
set -euo pipefail
LOG_GROUP_NAME="/aws/backend/legacy-llm-service"
START_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로 인터페이스를 고정하고 객체 어댑터 패턴을 적용하면 비즈니스 로직을 전혀 건드리지 않고 외부 모델을 바꿀 수 있습니다.
`python
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 값에 따라 객체를 반환하도록 만듭니다.
`python
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"]
),
model_name="fable-legacy",
provider="fable"
)
class LLMFactory:
@staticmethod
def get_adapter() -> LLMClientAdapter:
provider = os.getenv("LLM_PROVIDER", "astra").lower()
if provider == "astra":
return AstraAdapter(
api_key=os.environ["ASTRA_API_KEY"],
base_url=os.getenv("ASTRA_BASE_URL", "https://api.astra.ai/v1")
)
elif provider == "fable":
return FableAdapter(
api_key=os.environ["FABLE_API_KEY"],
endpoint_url=os.environ["FABLE_ENDPOINT_URL"]
)
raise ValueError(f"Unsupported LLM provider: {provider}")
`
모델을 바꿀 때 응답 형식이 깨지는 사고는 단위 테스트로 미리 잡아야 합니다. Pydantic 스키마와 목(mock) 응답을 검증하는 테스트입니다.
`python
import pytest
from core.llm_protocol import LLMMessage, LLMResponse
from core.adapters import AstraAdapter
@pytest.mark.asyncio
async def test_astra_response_contract_compliance(monkeypatch):
mock_payload = {
"id": "chatcmpl-astra-001",
"model": "gpt-6-astra",
"choices": [
{"message": {"role": "assistant", "content": "정상적인 결괏값"}}
],
"usage": {
"prompt_tokens": 120,
"completion_tokens": 45,
"total_tokens": 165
}
}
class MockResponse:
def raise_for_status(self): pass
def json(self): return mock_payload
async def mock_post(*args, **kwargs):
return MockResponse()
adapter = AstraAdapter(api_key="test-key")
monkeypatch.setattr(adapter.client, "post", mock_post)
messages = [LLMMessage(role="user", content="테스트 요청")]
result: LLMResponse = await adapter.generate_completion(messages)
assert isinstance(result, LLMResponse)
assert result.provider == "astra"
assert result.model_name == "gpt-6-astra"
assert result.content == "정상적인 결괏값"
assert result.usage.prompt_tokens == 120
assert result.usage.completion_tokens == 45
assert result.usage.total_tokens == 165
`
비즈니스 진입점 모듈의 호출 대상을 LLMFactory.get_adapter().generate_completion(...)으로 묶어두면 전환 작업은 1시간도 걸리지 않습니다. 새 모델에 장애가 나더라도 LLM_PROVIDER=fable로 환경변수를 되돌리면 끝납니다.
공급사 홍보 문구에 적힌 환각률 감소 수치는 사내 데이터로 직접 검증해야 합니다. 환불 규정이나 결제 약관 데이터 50건을 골든 데이터셋으로 묶어 오픈소스 라이브러리 DeepEval로 비교합니다.
`python
import json
import asyncio
from deepeval.test_case import LLMTestCase
from deepeval.metrics import HallucinationMetric
from core.adapters import AstraAdapter, FableAdapter
from core.llm_protocol import LLMMessage
async def run_batch_evaluation():
with open("eval_dataset_50.json", "r", encoding="utf-8") as f:
cases = json.load(f)
astra = AstraAdapter(api_key="astra-key")
fable = FableAdapter(api_key="fable-key", endpoint_url="https://api.fable.internal/v1")
hallucination_metric = HallucinationMetric(threshold=0.1)
astra_scores, fable_scores = [], []
for case in cases:
msg = [LLMMessage(role="user", content=case["input"])]
astra_res = await astra.generate_completion(msg)
tc_astra = LLMTestCase(input=case["input"], actual_output=astra_res.content, retrieval_context=case["retrieval_context"])
hallucination_metric.measure(tc_astra)
astra_scores.append(hallucination_metric.score)
fable_res = await fable.generate_completion(msg)
tc_fable = LLMTestCase(input=case["input"], actual_output=fable_res.content, retrieval_context=case["retrieval_context"])
hallucination_metric.measure(tc_fable)
fable_scores.append(hallucination_metric.score)
avg_fable = sum(fable_scores) / len(fable_scores)
avg_astra = sum(astra_scores) / len(astra_scores)
delta = ((avg_fable - avg_astra) / avg_fable) * 100
print(f"Empirical Hallucination Reduction: {delta:.2f}%")
if name == "main":
asyncio.run(run_batch_evaluation())
`
측정 결과는 scipy.stats.ttest_rel 함수로 대응표본 t-검정(Paired t-test)을 거칩니다.
표본 50개 기준 p-value가 0.05 아래로 내려갈 때만 실제 환각이 줄어들었다고 판단합니다.
정작 서비스에서 더 위험한 문제는 에러 코드보다 늘어지는 응답 지연입니다. 타임아웃이 없으면 백엔드 커넥션 풀이 순식간에 말라버립니다. 5회 연속 실패 시 즉시 호출을 끊고 이전 모델로 우회하는 서킷 브레이커 코드입니다.
`python
import logging
from pybreaker import CircuitBreaker, CircuitBreakerError
from core.adapters import AstraAdapter, FableAdapter
from core.llm_protocol import LLMMessage, LLMResponse
logger = logging.getLogger(name)
astra_breaker = CircuitBreaker(fail_max=5, reset_timeout=30)
class ResilientLLMClient:
def init(self, primary: AstraAdapter, fallback: FableAdapter):
self.primary = primary
self.fallback = fallback
async def execute(self, messages: list[LLMMessage]) -> LLMResponse:
try:
return await self._call_primary(messages)
except (CircuitBreakerError, Exception) as exc:
logger.warning("Primary degraded [%s]. Executing fallback.", str(exc))
return await self._call_fallback(messages)
async def _call_primary(self, messages: list[LLMMessage]) -> LLMResponse:
if astra_breaker.current_state == "open":
raise CircuitBreakerError("Circuit breaker OPEN")
try:
response = await self.primary.generate_completion(messages)
astra_breaker.success()
return response
except Exception as err:
astra_breaker.fail()
raise err
async def _call_fallback(self, messages: list[LLMMessage]) -> LLMResponse:
return await self.fallback.generate_completion(messages)
`
Connect 타임아웃 5초, Read 타임아웃 90초를 넘어가면 지체 없이 실패로 간주합니다. 실패 5회가 쌓이면 서킷이 열리고 30초 동안 외부 호출을 0초 만에 차단합니다. 늦어지는 호출을 붙잡고 서버 전체를 죽이는 일을 막는 최소한의 방어선입니다.
해외 리전에 있는 외부 API로 사용자 입력을 그대로 보내면 개인정보보호법 제28조의8(개인정보의 국외 이전)에 걸립니다. 데이터가 네트워크 밖으로 나가기 전 인메모리 단계에서 마스킹을 끝내야 합니다. 단순 13자리 정규식은 송장번호나 주문번호까지 지워버리므로 가중치 체크섬 알고리즘을 함께 태웁니다.
ext{Checksum} = 11 - left( left( sum_{i=1}^{12} d_i imes w_i ight) mod 11 ight)`python
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) 타임아웃 및 서킷 브레이커 설정 문서를 한 번에 묶어서 제출합니다. 인프라 방어 코드가 첨부된 문서는 기술 검토 기간을 눈에 띄게 줄여줍니다.