TuBrief
Subscribed Channels
Videos
Community

레거시 백엔드에 GPT-6 Astra 붙일 때 예산 승인과 보안 통과를 먼저 끝내는 법이 있습니다

TuBrief Editorial
September 13, 2026
0
컴퓨터/소프트웨어

Written with AI assistance from the source video. The video is the authority.

한국어EnglishEspañolالعربيةDeutsch中文Françaisहिन्दीPortuguêsРусский日本語Bahasa Indonesia

Related Video

GPT 6 Astra 출시 (Fable 5.1보다 더 뛰어날까?)7:34

GPT 6 Astra 출시 (Fable 5.1보다 더 뛰어날까?)

Chase AI

More from the community

사내 시스템에 llm api 붙일 때 마주하는 현실적인 한계와 대응법

September 13, 2026

에이전트끼리 대화하다 6천만 원 청구서가 나오는 이유

September 13, 2026

사내 RAG 벡터 검색에 Okta 권한 필터를 직접 거는 방법

September 13, 2026

브라우저 에이전트에게 내 구글 계정을 통째로 넘기면 안 되는 이유

September 12, 2026

Apple Won the AI Race

September 12, 2026

노코드 구독료로 월 20만 원 나가던 1인 창업자가 한 달 7천 원짜리 서버로 갈아탄 과정

September 12, 2026

Comments (0)

Log in to leave a comment

No posts yet

© 2026 . All rights reserved.

TuBrief
Subscribed Channels
Videos
Community
Log in

레거시 백엔드에 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) 단일 요율입니다. 수식은 단순합니다.

Ttotal=30×Nreq×(Tin+Tout)T_{\text{total}} = 30 \times N_{\text{req}} \times (T_{\text{in}} + T_{\text{out}})Ttotal​=30×Nreq​×(Tin​+Tout​)Cmonthly(USD)=Ttotal1,000,000×1.00C_{\text{monthly}} (\text{USD}) = \frac{T_{\text{total}}}{1,000,000} \times 1.00Cmonthly​(USD)=1,000,000Ttotal​​×1.00

로그에서 뽑은 수치를 아래 스프레드시트 구조에 대입하면 보고용 표가 나옵니다.

시나리오 일일 요청 건수 (NreqN_{\text{req}}Nreq​) 건당 토큰 합 (Tin+ToutT_{\text{in}}+T_{\text{out}}Tin​+Tout​) 월간 총 토큰 (TtotalT_{\text{total}}Ttotal​) 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를 걸어둡니다. 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가 환경변수 하나로 처리합니다.

# 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](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](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) 응답을 검증하는 테스트입니다.

# tests/test_llm_contract.py
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로 비교합니다.

# evaluate_models.py
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](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)을 거칩니다.

t=dˉsd/nt = \frac{\bar{d}}{s_d / \sqrt{n}}t=sd​/n​dˉ​

표본 50개 기준 p-value가 0.05 아래로 내려갈 때만 실제 환각이 줄어들었다고 판단합니다.

정작 서비스에서 더 위험한 문제는 에러 코드보다 늘어지는 응답 지연입니다. 타임아웃이 없으면 백엔드 커넥션 풀이 순식간에 말라버립니다. 5회 연속 실패 시 즉시 호출을 끊고 이전 모델로 우회하는 서킷 브레이커 코드입니다.

# core/resilient_client.py
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자리 정규식은 송장번호나 주문번호까지 지워버리므로 가중치 체크섬 알고리즘을 함께 태웁니다.

Checksum=11−((∑i=112di×wi) mod 11)\text{Checksum} = 11 - \left( \left( \sum_{i=1}^{12} d_i \times w_i \right) \bmod 11 \right)Checksum=11−((i=1∑12​di​×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) 타임아웃 및 서킷 브레이커 설정 문서를 한 번에 묶어서 제출합니다. 인프라 방어 코드가 첨부된 문서는 기술 검토 기간을 눈에 띄게 줄여줍니다.