원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.
Log in to leave a comment
아직 작성된 글이 없습니다
Cada vez que sale un nuevo modelo, la razón por la que se retrasa su adopción no es por la puntuación de los puntos de referencia (benchmarks). La estructura heredada que tiene hardcodeado el SDK de un proveedor específico en medio de la lógica de negocio, junto con una factura de API cuyo monto final es impredecible, son los factores que frenan todo.
Si a esto se le suma el señalamiento del equipo de seguridad de que “si los datos del cliente pasan a una región extranjera, se viola la ley de protección de datos personales”, el personal técnico termina haciendo horas extra todas las noches solo para corregir informes de revisión. Cambiar los puntos de enlace (endpoints) es el último paso. Lo seguro es extraer las cifras con registros de operaciones reales para obtener la aprobación del presupuesto, eliminar las auditorías mediante el enmascaramiento de prompts y luego reemplazar el código.
Si presentas una propuesta basándote únicamente en la tabla de precios que promociona el proveedor, te llevarás una sorpresa desagradable en la factura cuando el tráfico seconcentre. Primero debes analizar los registros de llamadas de los últimos 30 días del servidor en producción para verificar el consumo real de tokens.
Si utilizas CloudWatch Logs Insights, puedes extraer el consumo de tokens por solicitud y el total diario en 10 segundos con una consulta de la CLI de AWS.
`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'
`
El modelo de precios de GPT-6 Astra es de 10 dólares por cada 10 millones de tokens (1.00 dólar por 1M de tokens), sin distinción entre entrada y salida. Se simula la facturación mensual multiplicando el número de solicitudes diarias extraídas por el promedio de tokens por solicitud.
Si el modelo Fable anterior cobraba un promedio ponderado de 5.00 dólares por 1M de tokens, el precio unitario se reduce en un 80% al adoptar Astra.
| Clasificación de escenario | Número de solicitudes diarias | Promedio de tokens por solicitud | Número total de tokens mensuales | Precio unitario Astra (1M tokens) | Costo mensual estimado (USD) | Monto convertido a KRW (base 1,350 KRW) |
|---|---|---|---|---|---|---|
| Conservador (Low) | 35,000 | 1,200 | 1,260,000,000 | $1.00 | $1,260.00 | 1,701,000 KRW |
| Base (Base) | 50,000 | 1,200 | 1,800,000,000 | $1.00 | $1,800.00 | 2,430,000 KRW |
| Pico (Peak) | 75,000 | 1,500 | 3,375,000,000 | $1.00 | $3,375.00 | 4,556,250 KRW |
Abres la hoja de cálculo y colocas el número de llamadas diarias en la celda B1 (50000), los tokens de entrada promedio en B2 (800), los tokens de salida promedio en B3 (400) y el tipo de cambio base en B4 (1350). Al aplicar =30*B1*(B2+B3) en la celda B5, =(B5/10000000)*10 en B6, y =B6*B4 en B7, obtendrás inmediatamente la tabla de presupuesto según la fluctuación del tráfico.
Si se adjunta el cálculo de que el costo de procesamiento por cada 1,000 transacciones se reduce de 6.00 dólares a 1.20 dólares, la aprobación de la gerencia se consigue en tan solo 30 minutos.
Si estás llamando directamente a fable.Client() o openai.Client() dentro del código del servicio, tendrás que abrir decenas de archivos uno por uno y corregir el manejo de excepciones cada vez que cambies a un nuevo modelo. Esta tarea es ideal para provocar fallas en el servicio.
Si fijas la interfaz con typing.Protocol de Python y aplicas el patrón de diseño Adapter (Adaptador), podrás cambiar el modelo externo sin tocar en absoluto la lógica de negocio.
`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:
...
`
Envuelve el cliente Fable existente y el nuevo cliente Astra con adaptadores respectivamente. Luego, coloca un LLMFactory para que devuelva el objeto según el valor de la variable de entorno 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}")
`
Los problemas en los que el formato de respuesta se rompe al cambiar de modelo deben prevenirse de antemano mediante pruebas unitarias. Esta es una prueba que valida el esquema de Pydantic y la respuesta simulada (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
`
Si vinculas el destino de llamada del módulo de punto de entrada del negocio a LLMFactory.get_adapter().generate_completion(...), el trabajo de transición no tomará ni una hora. Incluso si ocurre un fallo en el nuevo modelo, basta con devolver la variable de entorno a LLM_PROVIDER=fable y listo.
Las cifras de reducción de alucinaciones escritas en los textos promocionales del proveedor deben verificarse directamente con los datos internos de la empresa. Se agrupan 50 datos de normativas de reembolso o términos de pago como un conjunto de datos dorados (golden dataset) y se comparan utilizando la biblioteca de código abierto 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())
`
Los resultados de la medición se someten a una prueba t para muestras emparejadas (Paired t-test) mediante la función scipy.stats.ttest_rel.
Solo cuando el valor p basado en 50 muestras desciende por debajo de 0.05 se determina que las alucinaciones realmente se han reducido.
El problema más peligroso en el servicio real no son los códigos de error, sino el retraso en la respuesta que se prolonga demasiado. Si no hay un tiempo de espera (timeout), el grupo de conexiones (connection pool) del backend se agota en un instante. Este es un código de disyuntor (circuit breaker) que corta inmediatamente la llamada tras 5 fallos consecutivos y la desvía hacia el modelo anterior.
`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)
`
Si el tiempo de espera de conexión supera los 5 segundos y el de lectura los 90 segundos, se considera un fallo sin demora. Si se acumulan 5 fallos, el circuito se abre y bloquea las llamadas externas en 0 segundos durante 30 segundos. Esta es la línea de defensa mínima para evitar que las llamadas lentas retengan el proceso y colapsen todo el servidor.
Si envías las entradas de los usuarios tal cual a una API externa ubicada en una región extranjera, infringirás el Artículo 28-8 de la Ley de Protección de Datos Personales (Transferencia internacional de datos personales). El enmascaramiento debe completarse en la fase en memoria antes de que los datos salgan de la red. Dado que una simple expresión regular de 13 dígitos también borraría los números de factura o de pedido, se aplica conjuntamente un algoritmo de suma de comprobación con ponderación (weighted checksum).
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
]
`
Verifica si el contrato con el proveedor incluye una cláusula de retención de datos cero (Zero Data Retention - ZDR) que impida el uso de los prompts de entrada para reentrenamiento. En CloudWatch Logs se evita almacenar el cuerpo del prompt, conservando únicamente el trace_id, la latencia y las métricas de tokens.
Al equipo de seguridad o al Oficial de Protección de Datos (CPO) se le entrega la siguiente tabla de flujo de datos:
| Tramo de procesamiento | Flujo de datos | Datos transmitidos | Dispositivo de control de seguridad | Criterio de cumplimiento |
|---|---|---|---|---|
| Tramo 1 | Cliente → Pasarela | Prompt original | TLS 1.3, autorización JWT | Cifrado en tránsito |
| Tramo 2 | Middleware de enmascaramiento en memoria | Original → Enmascaramiento de identificadores | Sustitución basada en checksum, destrucción inmediata en memoria | Principio de minimización de datos personales |
| Tramo 3 | Adaptador de backend → API de Astra | Payload con enmascaramiento completado | Inyección de claves mediante AWS Secrets Manager | Artículo 28-8 de la Ley de Protección de Datos Personales |
| Tramo 4 | API de Astra → Adaptador de backend | Texto generado por el modelo | Garantía de Zero Data Retention, TLS | Términos del acuerdo DPA empresarial |
| Tramo 5 | Registrador de backend → CloudWatch Logs | Métricas de tokens y latencia | Bloqueo de registro de contenido, limitado a metadatos | Artículo 29 de la Ley de Protección de Datos Personales |
Al solicitar la revisión de seguridad, se agrupan y envían de una sola vez los siguientes documentos: 1) Especificación de notificación del encargado del tratamiento dentro de la política de privacidad, 2) Copia del contrato DPA, 3) Resultados de las pruebas del middleware de enmascaramiento por checksum y 4) Documentación de configuración del tiempo de espera (timeout) y el disyuntor (circuit breaker). Un documento que incluya código de defensa de infraestructura reduce notablemente el período de revisión técnica.