Written with AI assistance from the source video. The video is the authority.
Log in to leave a comment
No posts yet
A adoção de um novo modelo costuma atrasar não por causa das pontuações de benchmark. O que realmente emperra o processo é a estrutura legada, com o SDK de um fornecedor específico hardcoded bem no meio da lógica de negócios, além de uma fatura de API cujos custos são imprevisíveis.
Quando a isso se soma o apontamento da equipe de segurança de que “enviar dados de clientes para uma região estrangeira viola a lei de proteção de dados”, os profissionais acabam passando noites em claro apenas revisando relatórios. Alterar os endpoints é a última etapa. O ideal e mais seguro é extrair números de logs operacionais reais para obter a aprovação do orçamento, passar pela auditoria com mascaramento de prompts e só então substituir o código.
Submeter uma proposta baseada apenas na tabela de preços divulgada pelo fornecedor pode gerar uma conta exorbitante quando o tráfego atinge o pico. É necessário primeiro analisar os logs de chamadas dos últimos 30 dias do servidor em produção para verificar o consumo real de tokens.
Se você utiliza o CloudWatch Logs Insights, é possível extrair o consumo de tokens por requisição e o total diário em 10 segundos usando uma query do 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'
`
O modelo de preços do GPT-6 Astra custa 10 dólares por 10 milhões de tokens, sem distinção entre entrada e saída ($1,00 por 1M de tokens). Multiplicando a quantidade diária de requisições extraída e a média de tokens por requisição, simulamos a fatura mensal.
Se o modelo Fable anterior cobrava uma média ponderada de $5,00 por 1M de tokens, o custo unitário com o Astra cai 80%.
| Cenário | Requisições diárias () | Total de tokens por requisição () | Total mensal de tokens () | Preço unitário do Astra (1M de tokens) | Custo mensal estimado (USD) | Conversão em 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 |
Ao abrir a planilha, insira o número de chamadas diárias (50000) na célula B1, a média de tokens de entrada (800) na B2, a média de tokens de saída (400) na B3 e a taxa de câmbio de referência (1350) na B4. Insira =30*B1*(B2+B3) na célula B5, =(B5/10000000)*10 na B6 e =B6*B4 na B7 para obter imediatamente a tabela de orçamento de acordo com a variação de tráfego. No cenário base, isso representa uma economia de 7.200 dólares mensais (cerca de 9,72 milhões de wons), eliminando qualquer motivo para rejeição na etapa de aprovação orçamentária.
Evite invocar diretamente o SDK de um modelo específico dentro da lógica de negócios, pois classes de erro customizadas do fornecedor podem se espalhar por todo o serviço. Usando o typing.Protocol do Python para fixar a interface antecipadamente e encapsulando as implementações com o padrão Object Adapter, torna-se desnecessário alterar o código de chamada.
`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:
...
`
Isola-se o endpoint legado Fable e o novo endpoint Astra em classes adaptadoras separadas. A alternância em tempo de execução é tratada por um LLMFactory baseado em uma única variável de ambiente.
`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}")
`
Erros de formatação de resposta ao trocar de modelo devem ser prevenidos com testes unitários. O teste a seguir valida o esquema Pydantic e a resposta 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": "Resultado 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="Teste de requisição")]
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 == "Resultado normal"
assert result.usage.prompt_tokens == 120
assert result.usage.completion_tokens == 45
assert result.usage.total_tokens == 165
`
Ao vincular o destino de chamada do módulo de ponto de entrada de negócios a LLMFactory.get_adapter().generate_completion(...), a tarefa de transição leva menos de uma hora. Se houver falhas no novo modelo, basta reverter a variável de ambiente para LLM_PROVIDER=fable.
A redução na taxa de alucinação declarada no material de marketing do fornecedor deve ser verificada diretamente com dados internos da empresa. Um dataset golden de 50 entradas contendo regras de reembolso ou termos de pagamento é comparado utilizando a biblioteca open-source 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())
`
Os resultados das medições passam por um teste t pareado (Paired t-test) usando a função scipy.stats.ttest_rel.
Considera-se que houve redução efetiva de alucinações apenas quando o p-value com base em 50 amostras fica abaixo de 0,05.
Um problema ainda mais perigoso em serviços do que códigos de erro são os atrasos excessivos de resposta. Sem um timeout, o pool de conexões do backend esgota-se num piscar de olhos. Segue o código de um circuit breaker que interrompe imediatamente as chamadas após 5 falhas consecutivas e redireciona para o 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)
`
Se o timeout de conexão ultrapassar 5 segundos ou o de leitura 90 segundos, a requisição é tratada como falha sem demora. Quando se acumulam 5 falhas, o circuito abre e bloqueia chamadas externas instantaneamente (0 segundos) pelos próximos 30 segundos. Esta é a linha de defesa mínima para evitar que chamadas lentas derrubem o servidor inteiro.
Enviar entradas de usuários diretamente para uma API externa localizada em uma região estrangeira viola o Artigo 28-8 da Lei de Proteção de Informações Pessoais (Transferência Internacional de Dados Pessoais). O mascaramento deve ser concluído na fase em memória antes que os dados saiam da rede. Expressões regulares simples de 13 dígitos acabam apagando números de fatura ou de pedido, por isso aplica-se paralelamente um algoritmo de checksum ponderado.
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
]
`
Verifique se o contrato com o fornecedor inclui uma cláusula de Zero Data Retention (ZDR) que garanta que os prompts de entrada não serão usados para re-treinamento. No CloudWatch Logs, evite salvar o corpo dos prompts, retendo apenas o trace_id, a latência e a contagem de tokens.
Para a equipe de segurança ou o encarregado de proteção de dados (DPO), encaminhe a seguinte tabela de fluxo de dados:
| Seção de processamento | Fluxo de dados | Dados transmitidos | Controles de segurança | Critérios de conformidade |
|---|---|---|---|---|
| Seção 1 | Cliente → Gateway | Prompt original | TLS 1.3, Autorização JWT | Criptografia em trânsito |
| Seção 2 | Middleware de mascaramento em memória | Original → Mascaramento de identificadores | Substituição baseada em checksum, descarte imediato da memória | Princípio da minimização de dados pessoais |
| Seção 3 | Adaptador de backend → API Astra | Payload mascarado | Injeção de chave via AWS Secrets Manager | Artigo 28-8 da Lei de Proteção de Dados |
| Seção 4 | API Astra → Adaptador de backend | Texto gerado pelo modelo | Garantia de Zero Data Retention, TLS | Termos DPA corporativos |
| Seção 5 | Registrador de backend → CloudWatch Logs | Contagem de tokens e latência | Bloqueio de gravação do corpo, restrito a metadados | Artigo 29 da Lei de Proteção de Dados |
Ao solicitar a revisão de segurança, envie simultaneamente: 1) o aviso de subcontratantes na política de privacidade, 2) uma cópia do contrato DPA, 3) os resultados dos testes do middleware de mascaramento com checksum e 4) a documentação de configuração de timeout e circuit breaker. Documentos acompanhados de código de defesa de infraestrutura reduzem visivelmente o período de revisão técnica.