Tratamento de erros em cadeias de solicitações
Valide resultados intermediários e recupere-se de falhas na cadeia.
Tratamento de erros em cadeias de solicitações é uma aula grátis de AI Prompt Engineering no CoddyKit. Esta é a aula 4 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de AI Prompt Engineering, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de AI Prompt Engineering inclui 4 aulas no total.
Por que as cadeias falham
As cadeias de instruções introduzem novos modos de falha que sistemas com uma única instrução não apresentam. Cada etapa pode falhar à sua maneira, e as falhas se acumulam — uma saída incorreta da Etapa 2 corrompe todas as etapas seguintes.
Modos de falha comuns:
- O modelo retorna JSON malformado que não pode ser analisado
- O modelo interpreta a tarefa de forma equivocada e produz uma saída semanticamente incorreta
- Limites de taxa ou tempos limite da API causam falhas nas etapas
- O limite da janela de contexto é excedido em uma cadeia longa
- O modelo inventa dados que as etapas seguintes tratam como fatos
Validando a saída após cada etapa
A primeira linha de defesa é validar a saída imediatamente após cada etapa, antes de passá-la para a próxima. Nunca presuma que o modelo retornou o que você solicitou.
import json
def validate_json_output(raw_text, required_fields):
'Parse and validate that required fields are present in model output.'
try:
data = json.loads(raw_text.strip())
except json.JSONDecodeError as e:
raise ValueError(f'Invalid JSON: {e}. Raw: {raw_text[:200]}')
missing = [f for f in required_fields if f not in data]
if missing:
raise ValueError(f'Missing required fields: {missing}. Got: {list(data.keys())}')
return data
# Usage after a chain step
raw = '{"sentiment": "positive", "priority": "high"}'
validated = validate_json_output(raw, required_fields=['sentiment', 'priority'])
print('Valid:', validated)Lógica de novas tentativas para falhas transitórias
Falhas da API (limites de taxa, tempos limite e erros do servidor) são transitórias. Implemente uma lógica de novas tentativas com espera exponencial para falhas no nível da rede:
import anthropic, time
client = anthropic.Anthropic(api_key='YOUR_API_KEY')
def call_with_retry(prompt, max_retries=3, base_delay=1.0):
last_error = None
for attempt in range(max_retries):
try:
r = client.messages.create(
model='claude-opus-4-5', max_tokens=500,
messages=[{'role': 'user', 'content': prompt}]
)
return r.content[0].text
except anthropic.RateLimitError as e:
wait = base_delay * (2 ** attempt)
print(f'Rate limited. Waiting {wait}s before retry {attempt+1}/{max_retries}...')
time.sleep(wait)
last_error = e
except anthropic.APIError as e:
last_error = e
if attempt < max_retries - 1:
time.sleep(base_delay)
raise RuntimeError(f'All retries exhausted: {last_error}')Validação semântica
Algumas falhas são estruturalmente válidas, mas semanticamente incorretas — o modelo retorna JSON válido, porém com valores incorretos. Use uma etapa leve de validação para verificar a correção semântica:
def semantic_validate(data, schema_rules):
'Apply semantic validation rules to parsed output.'
errors = []
for field, rules in schema_rules.items():
value = data.get(field)
if rules.get('required') and value is None:
errors.append(f'{field} is required but missing')
continue
if 'allowed_values' in rules and value not in rules['allowed_values']:
errors.append(f'{field} must be one of {rules["allowed_values"]}, got: {value}')
if 'min_length' in rules and isinstance(value, list) and len(value) < rules['min_length']:
errors.append(f'{field} must have at least {rules["min_length"]} items, got {len(value)}')
if errors:
raise ValueError('Semantic validation failed: ' + '; '.join(errors))
return data
rules = {'sentiment': {'allowed_values': ['positive', 'negative', 'mixed']}, 'issues': {'min_length': 1}}
data = {'sentiment': 'positive', 'issues': ['login bug']}
print(semantic_validate(data, rules))Instruções alternativas
Quando uma etapa falhar na validação após as novas tentativas, uma instrução alternativa poderá produzir uma saída mais simples, mas utilizável, em vez de fazer toda a cadeia falhar:
import json
def call_with_fallback(primary_prompt, fallback_prompt, required_fields):
# Try primary prompt
try:
raw = call_with_retry(primary_prompt)
return validate_json_output(raw, required_fields)
except (ValueError, RuntimeError) as e:
print(f'Primary prompt failed: {e}. Trying fallback...')
# Try simpler fallback prompt
try:
raw = call_with_retry(fallback_prompt)
return validate_json_output(raw, required_fields)
except (ValueError, RuntimeError) as e:
print(f'Fallback also failed: {e}. Returning safe default.')
# Return safe default — chain continues with minimal data
return {field: None for field in required_fields}
# Usage
primary = 'Analyze this review. Return JSON with 10 fields: {...}'
fallback = 'Classify this review. Return JSON: {"sentiment": "positive|negative|neutral"}'
result = call_with_fallback(primary, fallback, ['sentiment'])
print(result)Disjuntores de circuito
Um disjuntor de circuito impede que uma cadeia com falhas desperdice chamadas à API. Após N falhas consecutivas, ele abre o circuito e retorna um erro imediatamente, sem fazer novas chamadas à API:
class CircuitBreaker:
def __init__(self, failure_threshold=3, recovery_timeout=60):
self.failure_count = 0
self.threshold = failure_threshold
self.state = 'closed' # closed = normal, open = blocking
self.opened_at = None
def call(self, fn, *args, **kwargs):
import time
if self.state == 'open':
elapsed = time.time() - self.opened_at
if elapsed > 60: # recovery_timeout
self.state = 'half-open'
else:
raise RuntimeError('Circuit open — skipping API call')
try:
result = fn(*args, **kwargs)
self.failure_count = 0
self.state = 'closed'
return result
except Exception as e:
self.failure_count += 1
if self.failure_count >= self.threshold:
self.state = 'open'
self.opened_at = time.time()
print(f'Circuit opened after {self.failure_count} failures.')
raise e
cb = CircuitBreaker(failure_threshold=3)
print('Circuit breaker initialized.')Criando pontos de verificação para cadeias longas
Para cadeias com muitas etapas ou etapas dispendiosas, use pontos de verificação para salvar resultados intermediários. Se uma etapa final falhar, retome a partir do ponto de verificação em vez de reiniciar pela Etapa 1:
import json, os
CHECKPOINT_DIR = '/tmp/chain_checkpoints'
os.makedirs(CHECKPOINT_DIR, exist_ok=True)
def save_checkpoint(run_id, step_id, data):
path = os.path.join(CHECKPOINT_DIR, f'{run_id}_step{step_id}.json')
with open(path, 'w') as f:
json.dump(data, f)
print(f'Checkpoint saved: step {step_id}')
def load_checkpoint(run_id, step_id):
path = os.path.join(CHECKPOINT_DIR, f'{run_id}_step{step_id}.json')
if os.path.exists(path):
with open(path) as f:
return json.load(f)
return None
def run_with_checkpoints(run_id, input_data):
step1 = load_checkpoint(run_id, 1) or json.loads(call(f'Step 1 processing: {input_data}'))
save_checkpoint(run_id, 1, step1)
step2 = load_checkpoint(run_id, 2) or json.loads(call(f'Step 2 processing: {step1}'))
save_checkpoint(run_id, 2, step2)
return step2
print('Checkpointing system defined.')Degradação controlada
Quando uma etapa da cadeia falha e não pode ser recuperada, a degradação controlada permite que a cadeia continue com dados parciais, em vez de falhar completamente:
def process_with_degradation(tickets):
results = []
for ticket in tickets:
try:
# Full chain: extract -> classify -> respond
extracted = json.loads(call(f'Extract issue from ticket. Return JSON: {{"issue": str}}\n\n{ticket}'))
classified = json.loads(call(f'Classify priority. Return JSON: {{"priority": str}}\n\n{extracted["issue"]}'))
response = call(f'Draft response for {classified["priority"]} priority: {extracted["issue"]}')
results.append({'ticket': ticket, 'response': response, 'degraded': False})
except Exception as e:
print(f'Chain failed for ticket, using fallback: {e}')
# Fallback: simple direct response without classification
simple_response = call(f'Respond to this support ticket:\n{ticket}')
results.append({'ticket': ticket, 'response': simple_response, 'degraded': True})
return results
print('Graceful degradation pipeline defined.')Registro estruturado de erros
Registre os erros com contexto suficiente para diagnosticar qual etapa falhou, qual era a entrada e o que o modelo retornou:
import logging, traceback
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('chain')
def logged_step(step_name, prompt, validator=None):
start = datetime.utcnow()
try:
raw = call_with_retry(prompt)
result = validator(raw) if validator else raw
logger.info(f'[{step_name}] SUCCESS in {(datetime.utcnow()-start).total_seconds():.2f}s')
return result
except Exception as e:
logger.error(f'[{step_name}] FAILED after {(datetime.utcnow()-start).total_seconds():.2f}s')
logger.error(f'[{step_name}] PROMPT: {prompt[:200]}')
logger.error(f'[{step_name}] ERROR: {traceback.format_exc()}')
raise
print('Structured error logging defined.')Testando cenários de erro
Teste explicitamente seu tratamento de erros injetando falhas. Use objetos simulados para simular erros da API e saídas malformadas:
from unittest.mock import patch, MagicMock
def test_fallback_on_json_error():
with patch('__main__.call') as mock_call:
# First call returns malformed JSON, fallback returns valid JSON
mock_call.side_effect = [
'This is not JSON at all',
'{"sentiment": "positive"}'
]
result = call_with_fallback(
primary_prompt='Analyze review with 10 fields',
fallback_prompt='Just classify sentiment as JSON',
required_fields=['sentiment']
)
assert result['sentiment'] == 'positive'
print('PASS: fallback activated correctly on JSON parse error')
def test_circuit_breaker_opens():
cb = CircuitBreaker(failure_threshold=2)
for i in range(2):
try:
cb.call(lambda: (_ for _ in ()).throw(RuntimeError('API fail')))
except RuntimeError:
pass
assert cb.state == 'open'
print('PASS: circuit breaker opened after 2 failures')
print('Error handling tests defined.')Monitorando a saúde da cadeia em produção
Em produção, acompanhe as métricas de saúde da cadeia para detectar degradação antes que os usuários a percebam:
- Taxa de sucesso da etapa: Porcentagem de execuções em que cada etapa é concluída na primeira tentativa
- Taxa de ativação da alternativa: Com que frequência a instrução alternativa é usada?
- Taxa de degradação: Qual fração das execuções da cadeia é concluída em modo degradado?
- Latência da etapa: Acompanhe a latência p50/p95 por etapa — uma etapa lenta indica problemas de complexidade da instrução
- Taxa de falha da validação: Uma taxa alta indica que a instrução precisa ser refinada
Verificação rápida
Qual é a finalidade de um disjuntor de circuito em uma cadeia de instruções?
Tratamento de erros em cadeias — principais conclusões
Um tratamento de erros robusto é o que diferencia cadeias de protótipo de sistemas de produção:
- Valide a saída de cada etapa antes de passá-la às etapas seguintes — nunca presuma que o modelo retornou dados corretos
- Faça novas tentativas para falhas transitórias da API usando espera exponencial — limites de taxa e tempos limite podem ser recuperados
- Use instruções alternativas para obter uma saída mais simples quando a instrução principal falhar na validação semântica
- Os disjuntores de circuito impedem o desperdício de chamadas à API após falhas repetidas
- Crie pontos de verificação para etapas dispendiosas, permitindo que cadeias longas sejam retomadas após uma falha em uma etapa final
- A degradação controlada mantém o fluxo em execução com dados parciais, em vez de fazê-lo falhar
- Acompanhe em produção as taxas de sucesso das etapas, de uso de alternativas e de degradação
Perguntas Frequentes
A aula “Tratamento de erros em cadeias de solicitações” é grátis?
Sim — o texto completo de “Tratamento de erros em cadeias de solicitações” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de AI Prompt Engineering, atualize para CoddyKit PRO. O curso de AI Prompt Engineering inclui 4 aulas no total.
O que vou aprender em “Tratamento de erros em cadeias de solicitações”?
Valide resultados intermediários e recupere-se de falhas na cadeia. Você pratica AI Prompt Engineering com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.
Preciso ter experiência prévia para começar AI Prompt Engineering?
Nenhuma experiência prévia é necessária. AI Prompt Engineering no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 4 de 4.
Quanto tempo leva a aula “Tratamento de erros em cadeias de solicitações”?
A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.
Posso escrever e executar código nesta aula de AI Prompt Engineering?
Sim. Cada aula de AI Prompt Engineering inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.
Todas as aulas deste curso
- O que é encadeamento de solicitações?
- Padrões de resultado para entrada
- Cadeias de transformação sequencial
- Tratamento de erros em cadeias de solicitações