Objetivo da Aula
Ao concluir esta aula, você será capaz de:
Implementar um bot com histórico de conversa multi-turn usando a API Claude
Construir um pipeline OCR → extração estruturada → bot de queries sobre documentos
Gerenciar histórico de conversa sem explodir o context window
Debugar problemas de continuidade de contexto em conversas longas
Aplicar o padrão de “documento carregado + bot especializado” para qualquer tipo de documento
Por que isso importa
A maioria das aplicações de LLM em produção não são one-shot — são conversações. Um chatbot de suporte ao cliente precisa lembrar o que foi dito no início da conversa. Um assistente de análise de documentos precisa manter contexto entre várias perguntas sobre o mesmo documento. Um agente de análise de NF-e precisa responder “e o CNPJ do emitente?” depois de já ter respondido “o valor total é R$ 1.250,00”.
Fundamento: API Stateless e Gestão de Estado de Conversa
A API do LLM não tem sessão — cada requisição é independente. A "memória de conversa" é uma ilusão criada pela sua aplicação: a cada requisição, você envia o histórico completo de mensagens desde o início. Consequência direta: o custo cresce linearmente com o número de turnos. Com 20 turnos de 500 tokens cada, a 21ª requisição envia 10k tokens só de histórico. Em arquitetura escalável, o histórico deve ser armazenado por usuário em banco de dados — não em memória de instância de servidor, que não persiste entre pods e não escala horizontalmente. Estratégias de compressão: janela deslizante (descarta os turnos mais antigos) ou sumarização incremental (substitui os turnos antigos por um resumo).
A API da Anthropic (e a maioria dos LLMs) é stateless — cada request é independente. Para criar a ilusão de memória, você mantém o histórico de conversa no cliente e envia tudo a cada request. Isso cria um trade-off fundamental: mais contexto = melhor continuidade, mas mais tokens = mais custo e eventualmente chega no limite da context window.
O segundo aspecto desta unidade — o pipeline OCR + bot — é um pattern extremamente prático para o mercado brasileiro. Empresas têm pilhas de documentos fiscais (NF-e), contratos, relatórios, laudos médicos. Processar esses documentos via Vision API e depois permitir queries em linguagem natural sobre eles é um produto de alto valor e baixo custo técnico de implementação.
Conceitos Fundamentais
Multi-Turn Conversations: O Protocolo
A API da Anthropic usa um array de mensagens onde cada mensagem tem role (user ou assistant) e content. Para criar uma conversa multi-turn, você adiciona cada turno ao array:
# Turno 1
messages = [{'role': 'user', 'content': 'Olá, qual seu nome?'}]
response1 = client.messages.create(model=..., messages=messages)
resposta1 = response1.content[0].text
# Adiciona a resposta do assistente ao histórico
messages.append({'role': 'assistant', 'content': resposta1})
# Turno 2 — o assistente "lembra" do turno anterior
messages.append({'role': 'user', 'content': 'Você pode me ajudar com Python?'})
response2 = client.messages.create(model=..., messages=messages)
resposta2 = response2.content[0].text
# messages agora tem 4 itens:
# [user: "Olá...", assistant: "...", user: "Python?", assistant: "..."]Regra crucial: o array deve sempre alternar user → assistant → user → assistant. Nunca dois user consecutivos ou dois assistant consecutivos — a API retorna erro. Se você precisa adicionar contexto, adicione como parte do conteúdo do próximo user message.
Context Window: O Limite
Context window é o limite máximo de tokens que o modelo processa em uma única chamada. Para Claude, é 200k tokens. Isso soa enorme, mas:
Histórico de 100 turnos × 500 tokens por turno = 50k tokens
+ System prompt (500 tokens)
+ Documento carregado (2000 tokens)
= 52.500 tokens de input a cada query
Claude Sonnet: 52.500 × $3/1M = $0.16 por query
1000 queries/dia = $160/dia = $4800/mêsEm contextos de produção com muitos usuários, você precisa de estratégias para limitar o crescimento do histórico:
Truncamento por janela deslizante:
MAX_TURNOS = 10
if len(historico) > MAX_TURNOS * 2: # × 2 porque cada turno tem user + assistant
historico = historico[-(MAX_TURNOS * 2):] # mantém apenas os últimos N turnosSumarização de histórico antigo:
if len(historico) > 20:
# Sumariza os primeiros 15 turnos em um parágrafo
resumo = sumarizar_historico(historico[:15])
historico = [
{'role': 'user', 'content': f'[Contexto anterior resumido]: {resumo}'},
{'role': 'assistant', 'content': 'Entendido. Continuando a partir do resumo.'},
] + historico[15:]O Pattern “Documento + Bot”
Este pattern é extremamente versátil:
Fase 1 (uma vez por documento): carrega o documento (via OCR, extração de PDF, etc.) em um objeto estruturado
Fase 2 (por session): cria um bot especializado que conhece aquele documento específico
Fase 3 (múltiplos turnos): usuário faz perguntas, bot responde usando o documento como grounding
class BotDocumento:
def __init__(self):
self.documento = None
self.historico = []
def carregar(self, doc):
self.documento = doc
self.historico = [] # reset histórico ao carregar novo documento
def perguntar(self, pergunta: str) -> str:
if not self.documento:
return "Nenhum documento carregado."
# System prompt com contexto do documento
system = f"""Você é um assistente especializado em análise de documentos.
Dados do documento atual:
{json.dumps(self.documento.campos, ensure_ascii=False, indent=2)}
Responda APENAS com base nos dados do documento acima.
Se a informação não estiver nos dados, diga claramente."""
# Adiciona nova pergunta ao histórico
self.historico.append({'role': 'user', 'content': pergunta})
response = client.messages.create(
model='claude-haiku-4-5-20251001',
max_tokens=200,
system=system,
messages=self.historico,
)
resposta = response.content[0].text
# Adiciona resposta do assistente ao histórico para manter contexto
self.historico.append({'role': 'assistant', 'content': resposta})
return respostaPor que o sistema funciona? O system com os dados do documento é enviado a cada request (com Prompt Caching ativo, isso fica barato após o primeiro request). O self.historico acumula os turnos da conversa. O modelo vê tudo: os dados do documento + o histórico de perguntas e respostas — e pode responder “e o emitente?” corretamente porque viu a resposta anterior sobre o valor.
Grounding de Bot em Documentos Estruturados
O system prompt do bot é a chave para comportamento correto. Um bom system prompt para bot de documento:
system = f"""Você é um assistente especializado em responder perguntas sobre este documento.
TIPO DE DOCUMENTO: {documento.tipo}
CONFIANÇA DA EXTRAÇÃO: {documento.confianca:.0%}
DADOS EXTRAÍDOS:
{json.dumps(documento.campos, ensure_ascii=False, indent=2)}
REGRAS:
1. Responda APENAS com base nos dados acima
2. Se um campo não estiver nos dados ou for null, diga "esta informação não consta no documento"
3. Não invente dados que não estão nos campos extraídos
4. Seja conciso — 1-2 frases por resposta
5. Para campos de valor monetário, use formato "R$ X.XXX,XX"
"""Regras explícitas no system prompt reduzem alucinação drasticamente. O modelo sem essas regras pode usar seu conhecimento de treino para “completar” informações ausentes — o que parece útil mas é perigoso em documentos financeiros ou jurídicos.
Aprofundamento Técnico
Gerenciamento de Histórico com Limite de Tokens
Em vez de truncar por número de turnos (que é impreciso), a abordagem mais robusta é truncar por número de tokens:
def estimar_tokens(texto: str) -> int:
"""Estimativa grosseira: ~4 chars por token para PT-BR."""
return len(texto) // 4
def truncar_historico(
historico: list[dict],
max_tokens_historico: int = 10_000,
) -> list[dict]:
"""Mantém os turnos mais recentes que cabem no limite de tokens."""
total = sum(estimar_tokens(m['content']) for m in historico)
while total > max_tokens_historico and len(historico) > 2:
# Remove o par mais antigo (user + assistant)
removido_user = historico.pop(0)
removido_assistant = historico.pop(0)
total -= estimar_tokens(removido_user['content'])
total -= estimar_tokens(removido_assistant['content'])
return historicoAlternativa com summarização:
def comprimir_historico(historico: list[dict]) -> list[dict]:
"""Comprime histórico longo em resumo + turnos recentes."""
if len(historico) <= 10:
return historico
# Pega os primeiros N turnos para sumarizar
antigos = historico[:-6]
recentes = historico[-6:]
texto_para_resumir = '\n'.join(
f"{'Usuário' if m['role'] == 'user' else 'Assistente'}: {m['content']}"
for m in antigos
)
response = client.messages.create(
model='claude-haiku-4-5-20251001',
max_tokens=200,
messages=[{
'role': 'user',
'content': f'Resuma em 3-5 frases os pontos principais desta conversa:\n\n{texto_para_resumir}'
}]
)
resumo = response.content[0].text
return [
{'role': 'user', 'content': f'[Resumo da conversa anterior: {resumo}]'},
{'role': 'assistant', 'content': 'Entendido.'},
] + recentesIntegração com Pipeline OCR
O pipeline completo da unidade:
Imagem → Vision API → DocumentoExtraido → BotDocumento → Histórico de Queries
import json
import base64
import anthropic
from dataclasses import dataclass
from typing import Optional
client = anthropic.Anthropic()
@dataclass
class DocumentoExtraido:
tipo: str
campos: dict
confianca: float
texto_bruto: str
def extrair_documento(image_data: str, media_type: str = 'image/png') -> DocumentoExtraido:
"""Fase 1: Vision API → dados estruturados."""
response = client.messages.create(
model='claude-opus-4-8',
max_tokens=600,
messages=[{
'role': 'user',
'content': [
{
'type': 'image',
'source': {'type': 'base64', 'media_type': media_type, 'data': image_data},
},
{'type': 'text', 'text': """Extraia os dados deste documento. JSON:
{
"tipo": "recibo|nota_fiscal|contrato|formulario|outro|nao_documento",
"campos": {
"data": "DD/MM/AAAA ou null",
"valor_total": número ou null,
"emitente": "string ou null",
"numero_documento": "string ou null"
},
"confianca": 0.0-1.0,
"observacoes": "campos ilegíveis ou notas"
}"""}
]
}]
)
texto = response.content[0].text
try:
dados = json.loads(texto)
except json.JSONDecodeError:
import re
match = re.search(r'\{[\s\S]+\}', texto)
dados = json.loads(match.group(0)) if match else {
'tipo': 'erro', 'campos': {}, 'confianca': 0
}
return DocumentoExtraido(
tipo=dados.get('tipo', 'desconhecido'),
campos=dados.get('campos', {}),
confianca=float(dados.get('confianca', 0.5)),
texto_bruto=texto,
)
class BotDocumento:
"""Fase 2: Bot com histórico para queries sobre o documento extraído."""
def __init__(self):
self.documento: Optional[DocumentoExtraido] = None
self.historico: list[dict] = []
def carregar(self, doc: DocumentoExtraido) -> None:
self.documento = doc
self.historico = [] # reset para novo documento
print(f"Documento carregado: {doc.tipo} (confiança: {doc.confianca:.0%})")
def perguntar(self, pergunta: str) -> str:
if not self.documento:
return "Nenhum documento carregado. Use carregar() primeiro."
# Contexto do documento como JSON formatado no system
contexto = json.dumps(self.documento.campos, ensure_ascii=False, indent=2)
# TODO: Implemente o bot com histórico
# 1. Adicione a pergunta ao self.historico
# 2. Chame client.messages.create com:
# - system: prompt com os dados do documento (contexto acima)
# - messages: self.historico (que agora inclui a pergunta)
# 3. Extraia a resposta
# 4. Adicione a resposta ao self.historico (para manter contexto)
# 5. Retorne o texto da resposta
# Versão atual sem histórico (substitua com TODO acima):
response = client.messages.create(
model='claude-haiku-4-5-20251001',
max_tokens=200,
system=f"Você responde perguntas sobre documentos. Dados:\n{contexto}",
messages=self.historico + [{'role': 'user', 'content': pergunta}],
)
resposta = response.content[0].text
self.historico.extend([
{'role': 'user', 'content': pergunta},
{'role': 'assistant', 'content': resposta},
])
return resposta
def limpar_historico(self) -> None:
"""Reseta a conversa sem descarregar o documento."""
self.historico = []
print("Histórico limpo. Documento ainda carregado.")Por Que o Histórico Deve Ser No Cliente, Não No Servidor
Uma decisão de arquitetura importante: em produção, o histórico de conversa deve ficar no cliente (browser, app mobile) ou em banco de dados associado ao usuário — não em memória no servidor.
Problema com histórico em memória no servidor:
Servidor A: guarda historico de user-123
Servidor B: não tem historico de user-123
→ Load balancer rota request do user-123 para Servidor B
→ Bot "esquece" a conversaSolução: histórico persistido por usuário:
# No banco de dados (PostgreSQL, DynamoDB, etc.)
class RepositorioHistorico:
def carregar(self, user_id: str, doc_id: str) -> list[dict]:
return db.query(
'SELECT role, content FROM historico WHERE user_id=? AND doc_id=? ORDER BY created_at',
[user_id, doc_id]
)
def salvar_turno(self, user_id: str, doc_id: str, role: str, content: str) -> None:
db.execute(
'INSERT INTO historico (user_id, doc_id, role, content) VALUES (?, ?, ?, ?)',
[user_id, doc_id, role, content]
)Para o lab (ambiente local), histórico em memória na classe BotDocumento é suficiente.
Exemplos Anotados
Exemplo 1: BotDocumento Completo com Histórico
import json
import anthropic
from dataclasses import dataclass
client = anthropic.Anthropic()
@dataclass
class DocumentoExtraido:
tipo: str
campos: dict
confianca: float
texto_bruto: str
class BotDocumento:
"""Bot multi-turn para queries sobre documentos extraídos."""
def __init__(self):
self.documento = None
self.historico = []
def carregar(self, doc: DocumentoExtraido) -> None:
self.documento = doc
self.historico = [] # reset ao carregar novo doc
print(f"Documento {doc.tipo} carregado (confiança: {doc.confianca:.0%})")
def perguntar(self, pergunta: str) -> str:
if not self.documento:
return "Nenhum documento carregado."
# Dados do documento como JSON formatado
contexto_doc = json.dumps(self.documento.campos, ensure_ascii=False, indent=2)
# System prompt com os dados do documento
# → enviado a cada request (candidato a Prompt Caching)
system = f"""Você é um assistente especializado em análise de documentos.
DOCUMENTO ATUAL:
Tipo: {self.documento.tipo}
Confiança da extração: {self.documento.confianca:.0%}
Campos extraídos:
{contexto_doc}
REGRAS:
- Responda APENAS com base nos campos acima
- Se um campo for null ou não constar, diga "esta informação não está no documento"
- Seja direto e objetivo (1-2 frases)
- Para valores monetários: formato "R$ X.XXX,XX"
"""
# Adiciona a pergunta ao histórico ANTES de chamar a API
self.historico.append({'role': 'user', 'content': pergunta})
response = client.messages.create(
model='claude-haiku-4-5-20251001',
max_tokens=200,
system=system,
messages=self.historico, # histórico completo incluindo a nova pergunta
)
resposta = response.content[0].text
# Adiciona resposta ao histórico para manter contexto nos próximos turnos
self.historico.append({'role': 'assistant', 'content': resposta})
return resposta
def resumo_conversa(self) -> str:
if not self.historico:
return "Nenhuma pergunta feita ainda."
turnos = len(self.historico) // 2
return f"{turnos} pergunta(s) feita(s) nesta sessão."
# Simulação com documento mock (em produção vem do Vision API)
doc_mock = DocumentoExtraido(
tipo='recibo',
campos={
'data': '15/03/2025',
'valor_total': 1250.00,
'emitente': 'Tech Solutions Ltda',
'numero_documento': 'RC-2025-0342',
'outros_campos': {
'forma_pagamento': 'cartão de crédito',
'parcelas': 3,
}
},
confianca=0.92,
texto_bruto='...',
)
bot = BotDocumento()
bot.carregar(doc_mock)
# Conversa multi-turn — note como "e o emitente?" funciona
perguntas = [
'Qual o valor total deste recibo?',
'E o emitente?', # precisa do contexto da pergunta anterior para fazer sentido
'Foi parcelado?',
'Quando foi emitido?',
'Qual o número do documento?',
]
for p in perguntas:
print(f'\nP: {p}')
r = bot.perguntar(p)
print(f'R: {r}')
print(f'\n{bot.resumo_conversa()}')
print(f'Turnos no histórico: {len(bot.historico)}')Exemplo 2: Pipeline Completo OCR + Bot
def pipeline_completo_demo():
"""
Demonstra o pipeline: imagem → extração → bot → queries
Usa imagem sintética simples (1x1 pixel) pois não temos imagem real.
"""
# Fase 1: Imagem sintética (em produção: foto real de documento)
# PNG 1x1 pixel branco em base64
imagem_mock = 'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=='
print('=== FASE 1: Extração via Vision API ===')
doc = extrair_documento(imagem_mock)
print(f'Tipo: {doc.tipo}')
print(f'Campos: {json.dumps(doc.campos, ensure_ascii=False, indent=2)}')
print(f'Confiança: {doc.confianca:.0%}')
# Para imagem sintética, campos serão vazios/null
# Em produção com imagem real de NF-e, campos seriam preenchidos
print('\n=== FASE 2: Bot de Queries ===')
bot = BotDocumento()
bot.carregar(doc)
# Demonstração com 3 perguntas encadeadas
demo_perguntas = [
'Qual o tipo de documento?',
'Qual o valor total?',
'E a data de emissão?',
]
for p in demo_perguntas:
print(f'\nP: {p}')
r = bot.perguntar(p)
print(f'R: {r}')
print(f'\nHistórico: {len(bot.historico)} mensagens')
pipeline_completo_demo()Padrões e Armadilhas
Padrões
Padrão 1: Reset de histórico ao carregar novo documento
def carregar(self, doc):
self.documento = doc
self.historico = [] # SEMPRE reseta — histórico do doc anterior não faz sentido aquiSem o reset, o bot vai misturar contexto de documentos diferentes.
Padrão 2: Adicionar turno do usuário ao histórico ANTES de chamar a API
# CORRETO: adiciona antes
self.historico.append({'role': 'user', 'content': pergunta})
response = client.messages.create(..., messages=self.historico)
resposta = response.content[0].text
self.historico.append({'role': 'assistant', 'content': resposta})
# ERRADO: passa pergunta separada (não é adicionada ao histórico)
response = client.messages.create(
messages=self.historico + [{'role': 'user', 'content': pergunta}]
)
# histórico não foi atualizado — próxima pergunta perde contextoPadrão 3: Dados do documento no system, não nas messages
# CORRETO: documento no system (cacheável, não polui o histórico de conversa)
system = f"Dados do documento:\n{contexto_doc}"
response = client.messages.create(system=system, messages=self.historico)
# MENOS EFICIENTE: documento na primeira message do histórico
# → fica no histórico de conversa, consome tokens em cada request
# mas pode ser necessário se o histórico precisar mostrar o documentoArmadilhas
⚠️ Armadilha 1: Dois user ou dois assistant consecutivos
# ERRADO: dois user seguidos → API retorna erro
historico = [
{'role': 'user', 'content': 'Primeira pergunta'},
{'role': 'user', 'content': 'Segunda pergunta'}, # ← INVÁLIDO
]
# CORRETO: sempre alternar
historico = [
{'role': 'user', 'content': 'Primeira pergunta'},
{'role': 'assistant', 'content': 'Resposta'},
{'role': 'user', 'content': 'Segunda pergunta'},
]⚠️ Armadilha 2: Histórico crescendo sem limite
# PROBLEMA: após 500 turnos, cada request envia 500 × 500 tokens = 250k tokens
# → pode ultrapassar o context window do modelo
# → custo explode
# SOLUÇÃO: limite explícito
MAX_HISTORICO = 20 # turnos
if len(self.historico) > MAX_HISTORICO * 2:
self.historico = self.historico[-(MAX_HISTORICO * 2):]⚠️ Armadilha 3: Não extrair content[0].text de forma segura
# PROBLEMA: se stop_reason for 'max_tokens', pode haver múltiplos content blocks
resposta = response.content[0].text # ok para stop_reason 'end_turn'
# MAIS SEGURO: sempre verificar se content existe e tem tipo 'text'
texto_blocks = [b.text for b in response.content if hasattr(b, 'text')]
resposta = texto_blocks[0] if texto_blocks else ''⚠️ Armadilha 4: Deixar confiança baixa passar sem flag
# PROBLEMA: extração com 20% de confiança vai gerar respostas erradas do bot
# SOLUÇÃO: threshold de confiança
if doc.confianca < 0.5:
print(f"⚠️ Confiança baixa ({doc.confianca:.0%}). Resultado pode ser impreciso.")
# Opcionalmente: pede nova tentativa ou solicita imagem de melhor qualidadeSe não for realizar o laboratório, pule para o próximo capítulo.
Ponte para o Lab
O starter tem 1 TODO central:
TODO — Em BotDocumento.perguntar(): o código atual faz a chamada de API corretamente, mas não mantém histórico adequadamente. O self.historico está sendo usado corretamente na chamada: messages=self.historico + [{'role': 'user', 'content': pergunta}]. O que falta: após receber a resposta, os dois itens (pergunta do usuário + resposta do assistente) precisam ser persistidos em self.historico. Use self.historico.extend([...]) para adicionar ambos de uma vez. A estrutura correta: [{'role': 'user', 'content': pergunta}, {'role': 'assistant', 'content': resposta}].
Depois de implementar, o main() do starter faz 3 perguntas encadeadas: 1. “Qual o valor total?” — resposta com valor 2. “Quem emitiu?” — resposta com emitente 3. “Quando foi emitido?” — resposta com data
Com histórico funcionando, o bot mantém contexto e pode responder perguntas de acompanhamento como “e o emitente?” fazendo referência ao que foi discutido antes. Sem o histórico, cada pergunta seria respondida de forma isolada.
Experimento opcional: teste com self.historico = [] no perguntar() (desabilitando acumulação) e observe como o bot “esquece” o contexto entre perguntas.
Agora você está pronto para o lab.