2.4 — Memória Persistente e Capacidade de Autocorrição em Agentes (Agent Memory)

Objetivo da Aula

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

Distinguir os 3 tipos de memória em agentes (curta, longa, episódica) e quando usar cada um

Implementar busca por similaridade cosseno para recuperação de memórias relevantes

Fazer parse do resultado de reflexão e salvar como memória longa

Construir o GerenciadorMemoria completo com os 2 TODOs resolvidos

Projetar pipelines de memória para agentes de produção

Por que isso importa

LLMs não têm memória entre conversas por padrão. Cada chamada começa do zero. Um agente que “aprende” com as interações do usuário está, na verdade, armazenando informação em algum sistema externo e injetando no contexto.

A memória de agentes é o que transforma “assistente genérico” em “assistente que me conhece”. É o que permite: - “Lembrar” que o usuário prefere Python e gerar código Python automaticamente - “Aprender” que um tipo de abordagem falhou e evitar na próxima tarefa - “Contextualizar” eventos passados (“naquela reunião de junho, decidimos…”)

O starter desta unidade é um sistema de memória simplificado mas arquiteturalmente sólido. Em produção, embedding_simples seria substituído por uma chamada a anthropic.embeddings.create() ou similar, e o armazenamento em lista seria substituído por um vector database (ChromaDB, Pinecone, pgvector).

Conceitos Fundamentais

Decisão de Arquitetura: Três Tipos de Memória de Agente

Memória de agente é uma decisão de arquitetura de dados com três padrões distintos:
• Working memory (curta): o contexto ativo da sessão. FIFO — quando enche, descarta mais antigos. Custo proporcional ao tamanho; não persiste entre sessões.
• Semantic memory (longa): fatos sobre o usuário, preferências, entidades conhecidas. Upsert por chave — sobrescreve ao atualizar. Recuperada por similaridade vetorial. Persiste.
• Episodic memory (eventos): log imutável de ações com timestamp. Nunca sobrescreve. Bom para auditoria e aprendizado de padrões ao longo do tempo.
Governança crítica: o que o agente persiste sobre o usuário é dado pessoal sob LGPD — defina política de retenção e escopo antes de implementar.

Os 3 Tipos de Memória

Curta Duração (Working Memory)

self.curta: list[dict] = []  # máximo 20 entradas, FIFO

O que está ativo agora. É a janela de contexto do agente — o histórico da conversa atual. Quando o limite é atingido (20 entradas no starter), a mais antiga é descartada (pop(0)). Análogo à RAM.

Longa Duração (Semantic Memory)

self.longa: list[dict] = []  # fatos persistentes, atualização por chave

Fatos sobre o mundo e sobre o usuário que persistem entre sessões. “Usuário prefere Python”, “Empresa tem 50 funcionários”. A atualização é por chave — se a chave já existe, substitui:

self.longa = [m for m in self.longa if m['chave'] != chave]

self.longa.append(entrada)

Análogo ao SSD.

Episódica (Episodic Memory)

self.episodica: list[dict] = []  # eventos com timestamp, nunca substitui

Eventos específicos com timestamp. “Em 2024-06-10 o usuário criou um ticket sobre API”. Diferente da longa, nunca substitui — cada evento é único. Análogo ao diário.

Similaridade Cosseno

A recuperação de memórias usa similaridade semântica. O algoritmo do starter usa embeddings simplificados (bag-of-words), mas a matemática é a mesma dos sistemas de produção:

def cosseno(a: list[float], b: list[float]) -> float:

    dot = sum(x * y for x, y in zip(a, b))

    norm_a = math.sqrt(sum(x**2 for x in a))

    norm_b = math.sqrt(sum(x**2 for x in b))

    return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0

O que isso mede: o ângulo entre dois vetores no espaço de features. Valor entre -1 e 1: - 1.0: vetores idênticos (memória perfeitamente relevante) - 0.0: vetores ortogonais (sem relação) - -1.0: vetores opostos (raramente ocorre com bag-of-words)

Por que cosseno e não distância euclidiana? Cosseno normaliza pelo tamanho dos vetores — um texto longo e um curto sobre o mesmo assunto terão cosseno alto, mas distância euclidiana alta. Para relevância semântica, queremos normalização.

TODO 1: Implementar recuperar() com Similaridade Cosseno

def recuperar(self, query: str, top_k: int = 3) -> list[dict]:

    query_emb = embedding_simples(query)

    todas = self.curta + self.longa + self.episodica



    # Calcular similaridade para cada memória

    com_sim = []

    for m in todas:

        sim = cosseno(query_emb, m['embedding'])

        com_sim.append({**m, 'similaridade': sim})



    # Ordenar por similaridade decrescente e retornar top_k

    com_sim.sort(key=lambda x: x['similaridade'], reverse=True)

    return com_sim[:top_k]

O campo 'similaridade' adicionado ao resultado permite ao chamador entender quão relevante é cada memória — útil para filtrar memórias com sim < 0.1 (completamente irrelevantes).

TODO 2: Fazer Parse da Reflexão e Salvar em Memória Longa

def reflection_loop(self, tarefa_recente: str) -> str:

    memorias_contexto = self.recuperar(tarefa_recente)

    contexto = '\n'.join(f'- {m["chave"]}: {m["valor"]}' for m in memorias_contexto)



    response = client.messages.create(

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

        max_tokens=300,

        messages=[{

            'role': 'user',

            'content': f'''Reflita sobre a tarefa e gere aprendizados.

TAREFA: {tarefa_recente}

CONTEXTO: {contexto}



Retorne JSON:

{{"aprendizado": "...", "proxima_vez": "...", "padrão_identificado": "..."}}'''

        }],

    )



    reflexao_texto = response.content[0].text



    # TODO 2: parse JSON e salvar em memória longa

    try:

        # Extrair JSON mesmo se houver texto ao redor

        inicio = reflexao_texto.find('{')

        fim = reflexao_texto.rfind('}') + 1

        if inicio >= 0 and fim > inicio:

            reflexao = json.loads(reflexao_texto[inicio:fim])

            # Salvar aprendizado como memória longa

            self.salvar('longa', 'ultimo_aprendizado', reflexao.get('aprendizado', ''))

            if reflexao.get('padrão_identificado'):

                self.salvar('longa', 'padrao_identificado', reflexao['padrão_identificado'])

    except (json.JSONDecodeError, ValueError):

        pass  # se parse falhar, ainda retornamos o texto



    return reflexao_texto

Aprofundamento Técnico

Embedding Simplificado vs Embeddings Reais

O embedding_simples() do starter usa bag-of-words com 16 palavras-chave fixas:

vocab = ['usuario', 'preferencia', 'python', 'projeto', 'data', 'api', 'erro', 'custo',

         'prazo', 'reuniao', 'cliente', 'deploy', 'bug', 'feature', 'ticket', 'sprint']

return [1.0 if w in palavras else 0.0 for w in vocab]

Limitações: 16 dimensões, só palavras exatas, sem semântica. Em produção:

import anthropic



def embedding_real(texto: str) -> list[float]:

    # Usar API de embeddings real

    # Nota: Anthropic não tem API de embeddings pública ainda

    # Alternativas: OpenAI text-embedding-3-small, Cohere, ou modelos locais

    from openai import OpenAI

    client_emb = OpenAI()

    response = client_emb.embeddings.create(

        model='text-embedding-3-small',

        input=texto,

    )

    return response.data[0].embedding  # 1536 dimensões

Com embeddings reais, “Python” e “programação em Python” teriam alta similaridade.

Memória com Vetor Database

Para agentes de produção com milhares de memórias, listas em memória não escalam. O padrão é:

# ChromaDB (vector DB local)

import chromadb



class GerenciadorMemoriaVetorial:

    def __init__(self):

        self.client = chromadb.Client()

        self.collection = self.client.create_collection('memorias')

    

    def salvar(self, tipo: str, chave: str, valor: str):

        emb = embedding_real(f'{chave} {valor}')

        self.collection.upsert(

            ids=[f'{tipo}:{chave}'],

            embeddings=[emb],

            documents=[valor],

            metadatas=[{'tipo': tipo, 'chave': chave}],

        )

    

    def recuperar(self, query: str, top_k: int = 3) -> list[dict]:

        emb = embedding_real(query)

        resultados = self.collection.query(

            query_embeddings=[emb],

            n_results=top_k,

        )

        return [

            {'chave': m['chave'], 'valor': d, 'similaridade': 1 - dist}

            for m, d, dist in zip(

                resultados['metadatas'][0],

                resultados['documents'][0],

                resultados['distances'][0]

            )

        ]

Reflexão como Meta-Aprendizado

O reflection_loop é uma implementação de meta-raciocínio: o agente pensa sobre suas próprias ações. O padrão é:

Executar a tarefa

Refletir sobre o que funcionou e o que não funcionou

Armazenar o aprendizado na memória longa

Aplicar na próxima vez (via recuperação de memória relevante)

Isso cria um loop de aprendizado contínuo — o agente fica melhor com o tempo.

Exemplos Anotados

Exemplo 1: Sistema de Memória Completo com Recuperação

import os

import json

import math

import datetime

import anthropic



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



def embedding_simples(texto: str) -> list[float]:

    palavras = texto.lower().split()

    vocab = ['usuario', 'preferencia', 'python', 'projeto', 'data', 'api', 'erro', 'custo',

             'prazo', 'reuniao', 'cliente', 'deploy', 'bug', 'feature', 'ticket', 'sprint']

    return [1.0 if w in palavras else 0.0 for w in vocab]



def cosseno(a: list[float], b: list[float]) -> float:

    dot = sum(x * y for x, y in zip(a, b))

    norm_a = math.sqrt(sum(x**2 for x in a))

    norm_b = math.sqrt(sum(x**2 for x in b))

    return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0



class GerenciadorMemoria:

    def __init__(self):

        self.curta: list[dict] = []

        self.longa: list[dict] = []

        self.episodica: list[dict] = []

    

    def salvar(self, tipo: str, chave: str, valor: str):

        entrada = {

            'chave': chave,

            'valor': valor,

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

            'embedding': embedding_simples(f'{chave} {valor}'),

        }

        if tipo == 'curta':

            self.curta.append(entrada)

            if len(self.curta) > 20:

                self.curta.pop(0)  # FIFO

        elif tipo == 'longa':

            # Upsert por chave

            self.longa = [m for m in self.longa if m['chave'] != chave]

            self.longa.append(entrada)

        elif tipo == 'episodica':

            self.episodica.append(entrada)  # nunca substitui

    

    def recuperar(self, query: str, top_k: int = 3) -> list[dict]:

        # TODO 1: implementação completa

        query_emb = embedding_simples(query)

        todas = self.curta + self.longa + self.episodica

        

        com_sim = [{**m, 'similaridade': cosseno(query_emb, m['embedding'])} for m in todas]

        com_sim.sort(key=lambda x: x['similaridade'], reverse=True)

        return com_sim[:top_k]

    

    def reflection_loop(self, tarefa_recente: str) -> str:

        memorias = self.recuperar(tarefa_recente)

        contexto = '\n'.join(f'- {m["chave"]}: {m["valor"]}' for m in memorias)

        

        response = client.messages.create(

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

            max_tokens=300,

            messages=[{

                'role': 'user',

                'content': f'Reflita:\nTAREFA: {tarefa_recente}\nCONTEXTO: {contexto}\n'

                           f'Retorne JSON: {{"aprendizado":"...","proxima_vez":"...","padrão_identificado":"..."}}'

            }],

        )

        

        texto = response.content[0].text

        

        # TODO 2: parse e salvar

        try:

            i, f = texto.find('{'), texto.rfind('}') + 1

            if i >= 0 and f > i:

                r = json.loads(texto[i:f])

                if r.get('aprendizado'):

                    self.salvar('longa', 'aprendizado_recente', r['aprendizado'])

                if r.get('padrão_identificado'):

                    self.salvar('longa', 'padrao', r['padrão_identificado'])

        except (json.JSONDecodeError, ValueError):

            pass

        

        return texto



# Demo

mem = GerenciadorMemoria()

mem.salvar('longa', 'preferencia_linguagem', 'Usuário prefere Python')

mem.salvar('episodica', 'reuniao_jun', 'Reunião sobre adoção de LLMs')

mem.salvar('curta', 'contexto', 'Trabalhando em pipeline RAG')



resultados = mem.recuperar('python deploy api', top_k=3)

for r in resultados:

    print(f'  sim={r["similaridade"]:.2f} | {r["chave"]}: {r["valor"][:50]}')



reflexao = mem.reflection_loop('Implementei RAG em Python com sucesso')

print(f'\nReflexão: {reflexao[:200]}')

Padrões e Armadilhas

Padrões

Padrão 1: Upsert por chave na memória longa

# Atualizar em vez de duplicar

self.longa = [m for m in self.longa if m['chave'] != chave]

self.longa.append(nova_entrada)

Garante que “preferencia_linguagem” tem sempre o valor mais recente.

Padrão 2: Incluir similaridade no resultado

return [{**m, 'similaridade': sim} for m, sim in zip(todas, sims)]

Permite ao agente filtrar memórias com sim < 0.1 (irrelevantes).

Padrão 3: Extrair JSON com indexação explícita

inicio = texto.find('{')

fim = texto.rfind('}') + 1

json.loads(texto[inicio:fim])

Funciona mesmo se o LLM adicionar texto antes/depois do JSON.

Armadilhas

⚠️ Armadilha 1: Norma zero em cosseno → divisão por zero

# Problema: vetor de zeros → divisão por zero

return dot / (norm_a * norm_b)  # ZeroDivisionError se texto for só stopwords



# Solução: verificar antes

return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0

⚠️ Armadilha 2: Memória curta cresce sem limite

# RUIM: nunca limpa

self.curta.append(entrada)



# BOM: manter janela de tamanho fixo

self.curta.append(entrada)

if len(self.curta) > 20:

    self.curta.pop(0)

⚠️ Armadilha 3: Salvar memória episódica com chave única quebraria o evento

# ERRADO: salvar episodica com tipo='longa' vai fazer upsert (perde histórico)

self.salvar('longa', 'reuniao', 'Reunião 1')

self.salvar('longa', 'reuniao', 'Reunião 2')  # apaga a primeira!



# CORRETO: usar tipo='episodica' ou chave única por evento

self.salvar('episodica', f'reuniao_{datetime.now().isoformat()}', 'Reunião sobre IA')

⚗ 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 recuperar():

# Substituir: return todas[:top_k]

# Por:

query_emb = embedding_simples(query)

com_sim = [{**m, 'similaridade': cosseno(query_emb, m['embedding'])} for m in todas]

com_sim.sort(key=lambda x: x['similaridade'], reverse=True)

return com_sim[:top_k]

TODO 2 em reflection_loop():

# Após: reflexao_texto = response.content[0].text

# Adicionar:

try:

    i = reflexao_texto.find('{')

    f = reflexao_texto.rfind('}') + 1

    if i >= 0 and f > i:

        r = json.loads(reflexao_texto[i:f])

        if r.get('aprendizado'):

            self.salvar('longa', 'aprendizado_recente', r['aprendizado'])

except (json.JSONDecodeError, ValueError):

    pass

Após implementar, rode main() e verifique que recuperar('python deploy api') retorna “preferencia_linguagem” com similaridade > 0.

Agora você está pronto para o lab.