원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
Log in to leave a comment
아직 작성된 글이 없습니다
Setiap kali model baru dirilis, keterlambatan adopsi bukanlah disebabkan oleh skor benchmark. Struktur legacy yang melakukan hardcode SDK vendor tertentu tepat di tengah-tengah logika bisnis, serta tagihan API yang jumlah pastinya tidak dapat diprediksi, menjadi kendala utamanya.
Jika ditambah dengan teguran dari tim keamanan seperti “Jika data pelanggan berpindah ke wilayah luar negeri, hal itu akan melanggar Undang-Undang Perlindungan Data Pribadi,” praktisi harus bekerja lembur setiap malam hanya untuk merevisi laporan tinjauan. Mengubah endpoint adalah langkah terakhir. Sangat aman untuk mengekstrak angka menggunakan log operasional nyata guna mendapatkan persetujuan anggaran, menghapus audit melalui masking prompt, baru kemudian mengganti kodenya.
Jika Anda mengajukan rancangan hanya dengan melihat tabel harga yang dipromosikan oleh vendor, Anda akan menghadapi ledakan tagihan saat lalu lintas melonjak. Anda harus mengurai log panggilan dari server yang sedang beroperasi selama 30 hari terakhir terlebih dahulu untuk memeriksa konsumsi token aktual.
Jika Anda menggunakan CloudWatch Logs Insights, Anda dapat mengekstrak konsumsi token per permintaan dan total harian dalam 10 detik menggunakan kueri AWS CLI.
`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'
`
Model harga GPT-6 Astra adalah 1,00 per 1M token) tanpa membedakan input dan output. Simulasikan tagihan bulanan dengan mengalikan jumlah permintaan harian yang diekstrak dengan rata-rata token per permintaan.
Jika model Fable yang lama mengenakan biaya rata-rata tertimbang sebesar $5,00 per 1M token, maka unit harga saat mengadopsi Astra akan turun sebesar 80%.
| Klasifikasi Skenario | Jumlah Permintaan Harian | Rata-rata Token per Permintaan | Total Token Bulanan | Harga Satuan Astra (1M Token) | Estimasi Biaya Bulanan (USD) | Konversi Won Korea (Standar 1.350 Won) |
|---|---|---|---|---|---|---|
| Konservatif (Low) | 35.000 | 1.200 | 1.260.000.000 | $1,00 | $1.260,00 | 1.701.000 won |
| Standar (Base) | 50.000 | 1.200 | 1.800.000.000 | $1,00 | $1.800,00 | 2.430.000 won |
| Puncak (Peak) | 75.000 | 1.500 | 3.375.000.000 | $1,00 | $3.375,00 | 4.556.250 won |
Buka spreadsheet, lalu masukkan jumlah panggilan harian (50000) ke sel B1, rata-rata token input (800) ke sel B2, rata-rata token output (400) ke sel B3, dan nilai tukar standar (1350) ke sel B4. Jika Anda memasukkan =30*B1*(B2+B3) di sel B5, =(B5/10000000)*10 di sel B6, dan =B6*B4 di sel B7, tabel anggaran yang sesuai dengan fluktuasi lalu lintas akan langsung muncul.
Ketika lampiran perhitungan yang menyatakan bahwa biaya pemrosesan per 1.000 transaksi turun dari $6,00 menjadi $1,20 disertakan, persetujuan dari manajemen eksekutif akan diperoleh dalam waktu 30 menit.
Jika Anda memanggil fable.Client() atau openai.Client() langsung di dalam kode layanan, Anda harus membuka puluhan file satu per satu dan memperbaiki penanganan pengecualian setiap kali beralih ke model baru. Pekerjaan ini sangat rentan memicu gangguan layanan.
Dengan memperbaiki antarmuka menggunakan typing.Protocol Python dan menerapkan pola object adapter, Anda dapat mengganti model eksternal tanpa menyentuh logika bisnis sama sekali.
`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:
...
`
Klien Fable yang lama dan klien Astra yang baru masing-masing dibungkus dengan adapter. Kemudian, LLMFactory disediakan untuk mengembalikan objek berdasarkan nilai variabel lingkungan 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}")
`
Kecelakaan di mana format respons rusak saat mengganti model harus dicegah sebelumnya melalui pengujian unit. Ini adalah pengujian yang memvalidasi skema Pydantic dan respons 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": "Nilai hasil normal"}}
],
"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="Permintaan uji")]
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 == "Nilai hasil normal"
assert result.usage.prompt_tokens == 120
assert result.usage.completion_tokens == 45
assert result.usage.total_tokens == 165
`
Jika target panggilan dari modul titik masuk bisnis diikat ke LLMFactory.get_adapter().generate_completion(...), pekerjaan transisi tidak akan memakan waktu bahkan 1 jam. Meskipun terjadi gangguan pada model baru, Anda cukup mengembalikan variabel lingkungan ke LLM_PROVIDER=fable.
Angka pengurangan halusinasi yang tertulis dalam teks promosi pemasok harus diverifikasi secara langsung dengan data internal. 50 data regulasi pengembalian dana atau syarat pembayaran dikelompokkan ke dalam dataset gold dan dibandingkan menggunakan pustaka sumber terbuka 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())
`
Hasil pengukuran melalui fungsi scipy.stats.ttest_rel menjalani uji-t berpasangan (Paired t-test).
Disimpulkan bahwa halusinasi benar-benar berkurang hanya ketika nilai p di bawah 0,05 berdasarkan 50 sampel.
Masalah yang justru lebih berbahaya dalam layanan daripada kode galat adalah kelambatan respons yang semakin memanjang. Jika tidak ada timeout, pool koneksi backend akan langsung kering. Ini adalah kode circuit breaker yang segera memutus panggilan dan mengalihkan ke model sebelumnya jika terjadi kegagalan 5 kali berturut-turut.
`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)
`
Jika timeout koneksi melebihi 5 detik dan timeout pembacaan melebihi 90 detik, hal itu langsung dianggap sebagai kegagalan tanpa penundaan. Ketika 5 kegagalan menumpuk, sirkuit terbuka dan panggilan eksternal diblokir dalam waktu 0 detik selama 30 detik. Ini adalah garis pertahanan minimum untuk mencegah situasi di mana panggilan yang melambat menahan dan merusak seluruh server.
Jika input pengguna dikirimkan apa adanya ke API eksternal yang berada di wilayah luar negeri, hal tersebut akan melanggar Pasal 28-8 Undang-Undang Perlindungan Data Pribadi (Transfer Data Pribadi ke Luar Negeri). Masking harus diselesaikan pada tahap in-memory sebelum data keluar dari jaringan. Ekspresi reguler 13 digit yang sederhana juga akan menghapus nomor invoice atau nomor pesanan, sehingga algoritma checksum berbobot diterapkan secara bersamaan.
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
]
`
Pastikan apakah ada klausa Zero Data Retention (ZDR) dalam kontrak pemasok yang menyatakan bahwa prompt input tidak digunakan untuk pelatihan ulang. Simpanan isi prompt dicegah di CloudWatch Logs, dan hanya trace_id, waktu tundaan (latency), serta angka token yang disisakan.
Serahkan tabel alur data berikut kepada tim keamanan atau Pejabat Perlindungan Data (CPO).
| Bagian Pemrosesan | Alur Data | Data yang Ditransmisikan | Perangkat Kontrol Keamanan | Standar Kepatuhan |
|---|---|---|---|---|
| Bagian 1 | Klien $ | |||
| ightarrow$ Gateway | Prompt Asli | TLS 1.3, Otorisasi JWT | Enskripsi Bagian Komunikasi | |
| Bagian 2 | Middleware Masking In-Memory | Asli $ | ||
| ightarrow$ Masking Pengenal | Substitusi berbasis Checksum, Penghancuran Memori Instan | Prinsip Minimisasi Pengumpulan Data Pribadi | ||
| Bagian 3 | Adapter Backend $ | |||
| ightarrow$ API Astra | Payload Selesai Dimasking | Injeksi Kunci AWS Secrets Manager | Pasal 28-8 UU Perlindungan Data Pribadi | |
| Bagian 4 | API Astra $ | |||
| ightarrow$ Adapter Backend | Teks Hasil Pembuatan Model | Jaminan Zero Data Retention, TLS | Ketentuan DPA Kelas Korporat | |
| Bagian 5 | Logger Backend $ | |||
| ightarrow$ CloudWatch Logs | Angka Token dan Waktu Tundaan | Pemblokiran Pencatatan Isi, Terbatas pada Metadata | Pasal 29 UU Perlindungan Data Pribadi |
Saat mengajukan tinjauan keamanan, kumpulkan dan serahkan 1) rincian pemberitahuan entitas pemrosesan dalam kebijakan Privasi, 2) salinan kontrak DPA, 3) hasil pengujian middleware masking checksum, dan 4) dokumen pengaturan timeout serta circuit breaker secara bersamaan. Dokumen yang dilampirkan dengan kode pertahanan infrastruktur terbukti secara signifikan memperpendek periode tinjauan teknis.