Objetivo da Aula
Ao concluir esta aula, você será capaz de:
Gerar dataset sintético no tom AssAI via Claude (TODO 1)
Executar pipeline de fine-tuning com upload e polling (TODO 2)
Implementar harness de avaliação no golden set (TODO 3)
Criar AssAIClient wrapper com métricas de latência e custo (TODO 4)
Monitorar qualidade em produção com log JSONL e métricas P50/P99 (TODO 5)
Por que isso importa
Esta unidade é o projeto integrador de fine-tuning. Cada TODO mapeia para uma fase real de um projeto de modelo customizado:
TODO 1: Geração de dados → TODO 2: Treinamento → TODO 3: Avaliação → TODO 4: Deploy → TODO 5: MonitoramentoEmpresas que constroem modelos customizados sem monitoramento de produção não conseguem detectar regressões. Empresas que não avaliam antes de deploy colocam modelos de qualidade inferior em produção.
Conceitos Fundamentais
TODO 1: gerar_dataset_assai(n_exemplos)
def gerar_dataset_assai(n_exemplos: int = 50) -> list[dict]:
TOPICOS = [
"RAG e recuperação de informação",
"Fine-tuning e quando usar",
"Arquiteturas de agentes",
"Custo e otimização de LLMs",
"Segurança e governança de IA",
"DevOps com IA",
"Avaliação de modelos",
]
dataset = []
exemplos_por_topico = max(1, n_exemplos // len(TOPICOS))
for topico in TOPICOS:
if len(dataset) >= n_exemplos:
break
# Gerar perguntas diversas sobre o tópico
response_perguntas = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=300,
messages=[{
"role": "user",
"content": f"Gere {exemplos_por_topico} perguntas técnicas diversas sobre '{topico}' para consultores de IA. Retorne apenas as perguntas, uma por linha.",
}],
)
perguntas = [p.strip() for p in response_perguntas.content[0].text.split('\n') if p.strip() and len(p) > 10][:exemplos_por_topico]
for pergunta in perguntas:
# Gerar resposta no tom AssAI
response_resp = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=300,
system=SISTEMA_ASSAI,
messages=[{"role": "user", "content": pergunta}],
)
resposta = response_resp.content[0].text
dataset.append({
"messages": [
{"role": "system", "content": SISTEMA_ASSAI},
{"role": "user", "content": pergunta},
{"role": "assistant", "content": resposta},
]
})
# Salvar dataset
import json
with open("dataset_assai.jsonl", "w", encoding="utf-8") as f:
for ex in dataset:
f.write(json.dumps(ex, ensure_ascii=False) + "\n")
print(f" Dataset gerado: {len(dataset)} exemplos → dataset_assai.jsonl")
return datasetTODO 2: executar_finetuning(dataset_path)
def executar_finetuning(dataset_path: str) -> str:
if DRY_RUN:
print(f" [DRY-RUN] Fine-tuning de {dataset_path}")
return "ft:gpt-4o-mini:assai:suporte-v1:DRY"
from openai import OpenAI
oai = OpenAI()
# 1. Upload
with open(dataset_path, "rb") as f:
file_resp = oai.files.create(file=f, purpose="fine-tune")
print(f" [UPLOAD] file_id: {file_resp.id}")
# 2. Criar job
job = oai.fine_tuning.jobs.create(
training_file=file_resp.id,
model="gpt-4o-mini-2024-07-18",
hyperparameters={"n_epochs": 3, "batch_size": "auto"},
suffix="assai-suporte-v1",
)
print(f" [JOB] job_id: {job.id}")
# 3. Polling
import time
for _ in range(120): # 120 × 30s = 60 min
j = oai.fine_tuning.jobs.retrieve(job.id)
if j.status == "succeeded":
print(f" [OK] modelo: {j.fine_tuned_model}")
return j.fine_tuned_model
if j.status in ("failed", "cancelled"):
raise RuntimeError(f"Fine-tuning {j.status}: {j.error}")
time.sleep(30)
raise TimeoutError("Fine-tuning não concluiu em 60 minutos")TODO 3: avaliar_modelo(modelo_id)
def avaliar_modelo(modelo_id: str) -> dict:
from openai import OpenAI
oai = OpenAI()
scores = []
criterios_ok = 0
for ex in GOLDEN_SET_ASSAI:
if DRY_RUN:
resp = f"[DRY] Resposta simulada para: {ex['pergunta'][:30]}"
else:
r = oai.chat.completions.create(
model=modelo_id,
messages=[
{"role": "system", "content": SISTEMA_ASSAI},
{"role": "user", "content": ex["pergunta"]},
],
max_tokens=300,
)
resp = r.choices[0].message.content
# Critério 1: resposta ≤ 200 palavras
palavras = len(resp.split())
dentro_limite = palavras <= 200
# Critério 2: não usa "AIs" ou "inteligências artificiais"
sem_ai_termo = "AIs" not in resp and "inteligências artificiais" not in resp
# Critério 3: inclui recomendação prática
tem_recomendacao = any(p in resp.lower() for p in ["recomend", "sugiro", "use ", "utilize"])
if dentro_limite and sem_ai_termo and tem_recomendacao:
criterios_ok += 1
# Score por similaridade com esperado
palavras_esperado = set(ex["esperado"].lower().split())
palavras_resp = set(resp.lower().split())
score = len(palavras_esperado & palavras_resp) / max(len(palavras_esperado), 1)
scores.append(score)
return {
"score_medio": round(sum(scores) / len(scores), 3),
"criterios_aprovados": criterios_ok,
"total": len(GOLDEN_SET_ASSAI),
}TODO 4: AssAIClient.responder(pergunta)
class AssAIClient:
def __init__(self, modelo_ft: str):
self.modelo = modelo_ft
self.oai = None if DRY_RUN else __import__("openai").OpenAI()
def responder(self, pergunta: str) -> dict:
import time
inicio = time.time()
if DRY_RUN:
resposta = f"[DRY] Resposta AssAI para '{pergunta[:30]}': modelos de linguagem são..."
else:
r = self.oai.chat.completions.create(
model=self.modelo,
messages=[
{"role": "system", "content": SISTEMA_ASSAI},
{"role": "user", "content": pergunta},
],
max_tokens=300,
)
resposta = r.choices[0].message.content
latencia_ms = int((time.time() - inicio) * 1000)
palavras = len(resposta.split())
# Estimar custo (aproximação)
tokens_input = len(pergunta.split()) * 1.3 + len(SISTEMA_ASSAI.split()) * 1.3
tokens_output = palavras * 1.3
custo_usd = (tokens_input * 0.000001 + tokens_output * 0.000002) # gpt-4o-mini FT
return {
"resposta": resposta,
"latencia_ms": latencia_ms,
"custo_usd": round(custo_usd, 6),
"palavras": palavras,
"dentro_limite": palavras <= 200,
}TODO 5: monitorar_qualidade(perguntas, cliente)
import json
import time
from datetime import datetime
def monitorar_qualidade(perguntas: list[str], cliente: AssAIClient):
log_path = "producao_monitor.jsonl"
resultados = []
for pergunta in perguntas:
r = cliente.responder(pergunta)
entrada = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"pergunta": pergunta[:100],
"palavras": r["palavras"],
"latencia_ms": r["latencia_ms"],
"custo_usd": r["custo_usd"],
"dentro_limite": r["dentro_limite"],
}
# Salvar em JSONL
with open(log_path, "a", encoding="utf-8") as f:
f.write(json.dumps(entrada, ensure_ascii=False) + "\n")
resultados.append(entrada)
# Calcular métricas
latencias = sorted(r["latencia_ms"] for r in resultados)
p50 = latencias[len(latencias) // 2]
p99 = latencias[int(len(latencias) * 0.99)] if len(latencias) > 1 else latencias[-1]
custo_medio = sum(r["custo_usd"] for r in resultados) / len(resultados)
pct_dentro_limite = sum(r["dentro_limite"] for r in resultados) / len(resultados) * 100
print(f"\n 📊 Monitor de produção ({len(perguntas)} perguntas):")
print(f" Latência P50: {p50}ms | P99: {p99}ms")
print(f" Custo médio: ${custo_medio:.6f}")
print(f" % dentro de 200 palavras: {pct_dentro_limite:.0f}%")
print(f" Log salvo em: {log_path}")Aprofundamento Técnico
P50 e P99 de Latência
P50 (mediana): metade das requests são mais rápidas, metade são mais lentas
P99 (99o percentil): apenas 1% das requests são mais lentas que esse valor
Por que P99 importa mais que P50:
- P50=200ms pode parecer ótimo
- P99=5000ms significa que 1 em cada 100 usuários espera 5s
- Em 10.000 usuários/dia: 100 experiências ruins por diaPadrões e Armadilhas
Padrões
Padrão 1: Dry-run padrão, live explícito
DRY_RUN = "--live" not in sys.argvPadrão 2: Log append-only para monitoramento
with open(log_path, "a", encoding="utf-8") as f:
f.write(json.dumps(entrada) + "\n")Armadilhas
⚠️ Armadilha 1: Dataset gerado com sistema prompt diferente do produção
# Se SISTEMA_ASSAI muda entre geração e inferência → modelo treinado em distribuição errada
# Usar exatamente o mesmo SISTEMA_ASSAI em TODO 1 e TODO 4⚠️ Armadilha 2: Golden set com perguntas do dataset de treino
# Separar golden set ANTES de gerar o dataset
# Nunca incluir perguntas do golden set no dataset de treino⚠️ Armadilha 3: P99 calculado com amostra pequena
# Com 5 perguntas, P99 = a mais lenta = P100 = não tem significado estatístico
# Mínimo de 100 requests para P99 significativo⚗ Laboratório prático — mozak.tech
Se não for realizar o laboratório, pule para o próximo capítulo.
Se não for realizar o laboratório, pule para o próximo capítulo.
Ponte para o Lab
5 TODOs sequenciais:
gerar_dataset_assai(n=50): gerar perguntas por tópico, responder com SISTEMA_ASSAI, salvar JSONL
executar_finetuning(dataset_path): upload → criar job (suffix=assai-suporte-v1) → polling
avaliar_modelo(modelo_id): checar ≤200 palavras, sem “AIs”, tem recomendação; calcular score
AssAIClient.responder(pergunta): chamar modelo FT, medir latência, calcular custo, contar palavras
monitorar_qualidade(perguntas, cliente): salvar JSONL append, calcular P50/P99/custo médioRode com python starter.py (dry-run) primeiro, depois python starter.py --live.
Agora você está pronto para o lab.