mozak.tech Engenharia de IA Corporativa 37%

Parte II — Sistemas que Raciocinam e Agem de Forma Autônoma

2.5 — Gerenciamento de Janela de Contexto em Sessões Longas (Context Window)

Objetivo da Aula

Ao concluir esta aula, você será capaz de:

Calcular o uso de tokens em conversas e monitorar aproximação ao limite

Implementar pruning automático: sumarizar mensagens antigas e compactar o histórico

Injetar resumos de sessão anterior ao iniciar nova janela (context stitching)

Construir o ContextManager com os 2 TODOs resolvidos

Aplicar estratégias de compressão de contexto em agentes de produção

Por que isso importa

O Claude tem 200.000 tokens de contexto. Isso parece muito, mas: - Uma conversa de 1h com um assistente de código pode facilmente usar 50.000+ tokens - Cada token adicional aumenta a latência e o custo da chamada - Contexto irrelevante dilui a atenção do modelo — qualidade degrada

A ironia: agentes são mais úteis em tarefas longas, mas tarefas longas são exatamente onde o contexto se esgota. O ContextManager resolve isso.

Além disso, o custo de tokens é real. Uma conversa longa com claude-sonnet pode custar $0.10+ em tokens de input. Pruning inteligente pode reduzir isso em 50-70% sem perda de qualidade.

Conceitos Fundamentais

Estimativa de Tokens

def estimar_tokens(texto: str) -> int:

    return len(texto) // 4  # ~4 chars/token para inglês/português

É uma aproximação. A contagem real depende do tokenizer do modelo. Para português, a razão tende a ser ligeiramente menor (mais chars por token). Para produção, use anthropic.count_tokens().

# Contagem exata via API (mais lento, mas preciso)

response = client.messages.count_tokens(

    model='claude-haiku-4-5-20251001',

    messages=mensagens,

)

print(response.input_tokens)

O Threshold de Pruning

LIMITE_TOKENS = 200_000

LIMITE_PRUNING = int(LIMITE_TOKENS * 0.80)  # 80%

80% é o threshold padrão. Razões: - Ativa pruning antes de chegar ao limite — há espaço para o resumo - Deixa 20% de buffer para a resposta do LLM (max_tokens) e overhead - Em produção, use 70% para conversas com tool calls pesadas

TODO 1: Implementar _pruning()

O pruning tem 3 passos: 1. Selecionar as mensagens mais antigas (primeira metade do histórico) 2. Resumir com o LLM 3. Substituir pelas mensagens resumidas

def _pruning(self):

    if len(self.mensagens) < 4:

        return  # muito poucas mensagens para prune

    

    # 1. Selecionar metade mais antiga

    ponto_corte = len(self.mensagens) // 2

    antigas = self.mensagens[:ponto_corte]

    recentes = self.mensagens[ponto_corte:]

    

    # 2. Resumir com o LLM

    resumo = self._summarizar(antigas)

    self.resumos.append(resumo)

    

    # 3. Substituir: uma mensagem de resumo + mensagens recentes

    mensagem_resumo = {

        'role': 'user',

        'content': f'[RESUMO DO HISTÓRICO ANTERIOR]\n{resumo}',

    }

    self.mensagens = [mensagem_resumo] + recentes

    

    tokens_atuais = estimar_tokens_mensagens(self.mensagens)

    print(f'  Pruning concluído: {tokens_atuais:,} tokens restantes, {len(self.resumos)} resumo(s)')

TODO 2: Injetar Resumos em obter_contexto()

def obter_contexto(self) -> list[dict]:

    if not self.resumos:

        return self.mensagens

    

    # Injetar resumos cumulativos no início do contexto

    resumo_cumulativo = '\n\n'.join(f'--- Resumo {i+1} ---\n{r}' for i, r in enumerate(self.resumos))

    mensagem_historico = {

        'role': 'user',

        'content': f'[CONTEXTO DE SESSÕES ANTERIORES]\n{resumo_cumulativo}',

    }

    return [mensagem_historico] + self.mensagens

Isso permite ao LLM ter consciência de toda a conversa, mesmo que partes tenham sido compactadas.

Aprofundamento Técnico

Estratégias de Compressão

Estratégia

Descrição

Custo

Qualidade

Pruning simples

Descarta mensagens antigas

Baixo

Perde contexto

Sumarização

Resume mensagens antigas

Médio (1 extra LLM call)

Boa

Compressão seletiva

Mantém mensagens chave, descarta resto

Alto

Excelente

Sliding window

Mantém N mensagens mais recentes

Baixo

Perde contexto

O starter implementa sumarização — o melhor trade-off qualidade/custo.

_summarizar() — Detalhes da Implementação

def _summarizar(self, mensagens: list[dict]) -> str:

    historico = '\n'.join(

        f"{m['role'].upper()}: {m['content'][:200] if isinstance(m['content'], str) else '[tool call]'}"

        for m in mensagens

    )

    

    response = client.messages.create(

        model='claude-haiku-4-5-20251001',  # haiku = barato para sumarização

        max_tokens=300,

        messages=[{

            'role': 'user',

            'content': f'Resuma este histórico em 3-5 pontos chave:\n\n{historico}',

        }],

    )

    return response.content[0].text

O content[:200] trunca mensagens muito longas no histórico enviado para sumarização. O resumo de 300 tokens captura os pontos essenciais de centenas de tokens de conversa.

Context Stitching para Multi-Sessão

Quando o usuário retorna no dia seguinte, você quer carregar o contexto da sessão anterior. O padrão:

class ContextManagerPersistente(ContextManager):

    def __init__(self, session_file: str):

        super().__init__()

        self.session_file = session_file

        self._carregar_sessao()

    

    def _carregar_sessao(self):

        import json, os

        if os.path.exists(self.session_file):

            dados = json.loads(open(self.session_file).read())

            self.resumos = dados.get('resumos', [])

            # Não carrega mensagens brutas — apenas resumos

    

    def salvar_sessao(self):

        # Sumarizar conversa atual antes de salvar

        if self.mensagens:

            resumo_atual = self._summarizar(self.mensagens)

            self.resumos.append(resumo_atual)

        

        import json

        open(self.session_file, 'w').write(json.dumps({

            'resumos': self.resumos,

            'ultima_sessao': datetime.now().isoformat(),

        }, ensure_ascii=False, indent=2))

Exemplos Anotados

Exemplo 1: ContextManager Completo com TODOs Resolvidos

import os

import anthropic



client = anthropic.Anthropic(api_key=os.environ.get('ANTHROPIC_API_KEY'))



LIMITE_TOKENS = 200_000

LIMITE_PRUNING = int(LIMITE_TOKENS * 0.80)



def estimar_tokens(texto: str) -> int:

    return len(texto) // 4



def estimar_tokens_mensagens(mensagens: list[dict]) -> int:

    total = 0

    for msg in mensagens:

        content = msg.get('content', '')

        if isinstance(content, str):

            total += estimar_tokens(content)

        elif isinstance(content, list):

            for block in content:

                if isinstance(block, dict):

                    total += estimar_tokens(block.get('text', '') or block.get('content', ''))

    return total



class ContextManager:

    def __init__(self, limite_tokens: int = LIMITE_TOKENS):

        self.limite = limite_tokens

        self.mensagens: list[dict] = []

        self.resumos: list[str] = []

    

    def adicionar(self, mensagem: dict):

        self.mensagens.append(mensagem)

        tokens_atuais = estimar_tokens_mensagens(self.mensagens)

        if tokens_atuais > self.limite * 0.8:

            print(f'  ⚠ Contexto em {tokens_atuais:,}/{self.limite:,} tokens — aplicando pruning...')

            self._pruning()

    

    def _pruning(self):

        # TODO 1: implementação

        if len(self.mensagens) < 4:

            return

        

        ponto_corte = len(self.mensagens) // 2

        antigas = self.mensagens[:ponto_corte]

        recentes = self.mensagens[ponto_corte:]

        

        resumo = self._summarizar(antigas)

        self.resumos.append(resumo)

        

        self.mensagens = [

            {'role': 'user', 'content': f'[RESUMO DO HISTÓRICO ANTERIOR]\n{resumo}'}

        ] + recentes

        

        tokens_pos = estimar_tokens_mensagens(self.mensagens)

        print(f'  Pruning: {len(antigas)} mensagens → 1 resumo. Tokens: {tokens_pos:,}')

    

    def _summarizar(self, mensagens: list[dict]) -> str:

        historico = '\n'.join(

            f"{m['role'].upper()}: {m['content'][:200] if isinstance(m['content'], str) else '[tool]'}"

            for m in mensagens

        )

        response = client.messages.create(

            model='claude-haiku-4-5-20251001',

            max_tokens=300,

            messages=[{'role': 'user', 'content': f'Resuma em 3-5 pontos chave:\n{historico}'}],

        )

        return response.content[0].text

    

    def obter_contexto(self) -> list[dict]:

        # TODO 2: injetar resumos

        if not self.resumos:

            return self.mensagens

        

        resumo_cumulativo = '\n\n'.join(f'--- Resumo {i+1} ---\n{r}' for i, r in enumerate(self.resumos))

        return [

            {'role': 'user', 'content': f'[CONTEXTO ANTERIOR]\n{resumo_cumulativo}'}

        ] + self.mensagens

    

    def status(self) -> dict:

        tokens = estimar_tokens_mensagens(self.mensagens)

        return {

            'mensagens': len(self.mensagens),

            'tokens_estimados': tokens,

            'percentual_limite': f'{tokens/self.limite*100:.1f}%',

            'resumos': len(self.resumos),

        }

Padrões e Armadilhas

Padrões

Padrão 1: Threshold de 80% com buffer generoso

LIMITE_PRUNING = int(LIMITE_TOKENS * 0.80)  # 20% de buffer

O buffer acomoda a resposta do LLM (max_tokens) e overhead de tool calls.

Padrão 2: Mensagem de resumo com marcador explícito

{'role': 'user', 'content': '[RESUMO DO HISTÓRICO ANTERIOR]\n...'}

O marcador [RESUMO...] ajuda o LLM a entender que é contexto sumarizado, não input direto do usuário.

Padrão 3: Usar modelo barato para sumarização

model='claude-haiku-4-5-20251001'  # haiku para sumarização (barato)

# modelo principal para tarefa principal (mais caro)

Armadilhas

⚠️ Armadilha 1: Pruning com mensagens em número ímpar

# Ponto de corte pode deixar uma resposta sem o request correspondente

ponto_corte = len(self.mensagens) // 2



# Garantir que o corte caia em uma mensagem 'user' (não 'assistant')

while ponto_corte < len(self.mensagens) and self.mensagens[ponto_corte]['role'] != 'user':

    ponto_corte += 1

⚠️ Armadilha 2: _summarizar() falha com tool call blocks

# Tool calls têm content como lista de dicts, não string

m['content'][:200]  # ← TypeEror se content é lista!



# Tratar corretamente:

content = m.get('content', '')

texto = content[:200] if isinstance(content, str) else '[tool call]'

⚠️ Armadilha 3: Pruning em loop quando resumo adiciona tokens

# O resumo em si adiciona tokens! Verificar após pruning

self._pruning()

tokens_pos = estimar_tokens_mensagens(self.mensagens)

if tokens_pos > self.limite * 0.8:

    # Pruning adicional necessário

    self._pruning()  # ou: raise ContextError('contexto irreduzível')
⚗ Laboratório prático — mozak.tech
Se não for realizar o laboratório, pule para o próximo capítulo.

Ponte para o Lab

O starter tem 2 TODOs:

TODO 1 em _pruning():

# Substituir: print('  [TODO 1] ...')

# Por:

if len(self.mensagens) < 4:

    return

ponto_corte = len(self.mensagens) // 2

antigas = self.mensagens[:ponto_corte]

recentes = self.mensagens[ponto_corte:]

resumo = self._summarizar(antigas)

self.resumos.append(resumo)

self.mensagens = [{'role': 'user', 'content': f'[RESUMO DO HISTÓRICO ANTERIOR]\n{resumo}'}] + recentes

TODO 2 em obter_contexto():

# Substituir: return self.mensagens

# Por:

if not self.resumos:

    return self.mensagens

resumo_cumulativo = '\n\n'.join(f'--- Resumo {i+1} ---\n{r}' for i, r in enumerate(self.resumos))

return [{'role': 'user', 'content': f'[CONTEXTO ANTERIOR]\n{resumo_cumulativo}'}] + self.mensagens

Note que o starter usa limite_tokens=5000 (baixo) para forçar o pruning durante o teste. Após implementar, rode simular_conversa_longa() e observe o pruning em ação.

Agora você está pronto para o lab.