colecao-mega-skills/SamuelChien

Sistemas de Memória de Agentes

Memória é a base dos agentes inteligentes. Sem ela, cada interação começa do zero. Esta skill cobre a arquitetura da memória de agentes: curto prazo (janela de contexto), longo prazo (armazenamentos vetoriais) e as arquiteturas cognitivas que as organizam.

Insight chave: Memória não é apenas armazenamento - é recuperação. Um milhão de fatos armazenados não significam nada se você não conseguir encontrar o certo. As estratégias de fragmentação, incorporação (embedding) e recuperação determinam se seu agente lembra ou esquece.

O campo é fragmentado com terminologia inconsistente. Usamos o framework de arquitetura cognitiva CoALA: memória semântica (fatos), memória episódica (experiências) e memória procedural (conhecimento de como fazer).

Princípios

• Qualidade da memória = qualidade da recuperação, não quantidade de armazenamento
• Fragmente para recuperação, não para armazenamento
• Isolamento de contexto é o inimigo da memória
• Tipo de memória certo para a informação certa
• Decaia memórias antigas - nem tudo deve ser para sempre
• Teste a precisão da recuperação antes da produção
• Formação de memória em segundo plano é melhor que em tempo real

Capacidades

• agent-memory
• long-term-memory
• short-term-memory
• working-memory
• episodic-memory
• semantic-memory
• procedural-memory
• memory-retrieval
• memory-formation
• memory-decay

Escopo

• vector-database-operations → data-engineer
• rag-pipeline-architecture → llm-architect
• embedding-model-selection → ml-engineer
• knowledge-graph-design → knowledge-engineer

Ferramentas

Memory_frameworks

• LangMem (LangChain) - Quando: Agentes LangGraph com memória persistente Nota: Tipos de memória semântica, episódica, procedural
• MemGPT / Letta - Quando: Gerenciamento de contexto virtual, memória estilo SO Nota: Níveis hierárquicos de memória, paginação automática
• Mem0 - Quando: Camada de memória do usuário para personalização Nota: Projetado para preferências e histórico do usuário

Vector_stores

• Pinecone - Quando: Gerenciado, escala empresarial (bilhões de vetores) Nota: Melhor desempenho de consulta, maior custo
• Qdrant - Quando: Filtragem complexa de metadados, código aberto Nota: Baseado em Rust, excelente filtragem
• Weaviate - Quando: Busca híbrida, recursos de grafo de conhecimento Nota: Interface GraphQL, bom para relacionamentos
• ChromaDB - Quando: Prototipagem, aplicativos pequenos/médios Nota: Amigável para desenvolvedores, ~20ms p50 em 100K vetores
• pgvector - Quando: Já usa PostgreSQL, configuração mais simples Nota: Bom para <1M vetores, ferramentas familiares

Embedding_models

• OpenAI text-embedding-3-large - Quando: Melhor qualidade, 3072 dimensões Nota: $0.13/1M tokens
• OpenAI text-embedding-3-small - Quando: Bom equilíbrio, 1536 dimensões Nota: $0.02/1M tokens, 5x mais barato
• nomic-embed-text-v1.5 - Quando: Código aberto, implantação local Nota: 768 dimensões, boa qualidade
• all-MiniLM-L6-v2 - Quando: Leve, incorporação local rápida Nota: 384 dimensões, menor latência

Padrões

Arquitetura de Tipo de Memória

Escolhendo o tipo de memória certo para diferentes informações

Quando usar: Projetando sistema de memória de agente

ARQUITETURA DE TIPO DE MEMÓRIA (Framework CoALA):

""" Três tipos de memória para diferentes propósitos:

1. Memória Semântica: Fatos e conhecimento

   • O que você sabe sobre o mundo
   • Preferências do usuário, conhecimento do domínio
   • Armazenado em perfis (estruturados) ou coleções (não estruturados)

2. Memória Episódica: Experiências e eventos

   • O que aconteceu (eventos com carimbo de data/hora)
   • Conversas passadas, resultados de tarefas
   • Usado para aprender com a experiência

3. Memória Procedural: Como fazer as coisas

   • Regras, habilidades, fluxos de trabalho
   • Frequentemente implementado como exemplos de few-shot
   • "Como resolvi isso antes?" """

Implementação LangMem

""" from langmem import MemoryStore from langgraph.graph import StateGraph

Inicializar armazenamento de memória

memory = MemoryStore( connection_string=os.environ["POSTGRES_URL"] )

Memória semântica: perfil do usuário

await memory.semantic.upsert( namespace="user_profile", key=user_id, content={ "name": "Alice", "preferences": ["modo escuro", "respostas concisas"], "expertise_level": "developer", } )

Memória episódica: interação passada

await memory.episodic.add( namespace="conversations", content={ "timestamp": datetime.now(), "summary": "Ajudou a depurar problema de autenticação", "outcome": "resolvido", "key_insights": ["Expiração do token era a causa raiz"], }, metadata={"user_id": user_id, "topic": "debugging"} )

Memória procedural: padrão aprendido

await memory.procedural.add( namespace="skills", content={ "task_type": "debug_auth", "steps": ["Verificar expiração do token", "Verificar fluxo de renovação"], "example_interaction": few_shot_example, } ) """

Recuperação de Memória em Tempo de Execução

""" async def prepare_context(user_id, query): # Obter perfil do usuário (semântico) profile = await memory.semantic.get( namespace="user_profile", key=user_id )

# Encontrar experiências passadas relevantes (episódico)
similar_experiences = await memory.episodic.search(
    namespace="conversations",
    query=query,
    filter={"user_id": user_id},
    limit=3
)

# Encontrar habilidades relevantes (procedural)
relevant_skills = await memory.procedural.search(
    namespace="skills",
    query=query,
    limit=2
)

return {
    "profile": profile,
    "past_experiences": similar_experiences,
    "relevant_skills": relevant_skills,
}

"""

Padrão de Seleção de Armazenamento Vetorial

Escolhendo o banco de dados vetorial certo para seu caso de uso

Quando usar: Configurando armazenamento de memória persistente

SELEÇÃO DE ARMAZENAMENTO VETORIAL:

""" Matriz de decisão:

	Pinecone	Qdrant	Weaviate	ChromaDB	pgvector
Escala	Bilhões	100M+	100M+	1M	1M
Gerenciado	Sim	Ambos	Ambos	Próprio	Próprio
Filtragem	Básica	Melhor	Boa	Básica	SQL
Híbrido	Não	Sim	Melhor	Não	Sim
Custo	Alto	Médio	Médio	Grátis	Grátis
Latência	5ms	7ms	10ms	20ms	15ms
"""

Pinecone (Escala Empresarial)

""" from pinecone import Pinecone

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"]) index = pc.Index("agent-memory")

Upsert com metadados

index.upsert( vectors=[ { "id": f"memory-{uuid4()}", "values": embedding, "metadata": { "user_id": user_id, "timestamp": datetime.now().isoformat(), "type": "episodic", "content": memory_text, } } ], namespace=namespace )

Consulta com filtro

results = index.query( vector=query_embedding, filter={"user_id": user_id, "type": "episodic"}, top_k=5, include_metadata=True ) """

Qdrant (Filtragem Complexa)

""" from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, Filter, FieldCondition

client = QdrantClient(url="http://localhost:6333")

Filtragem complexa com Qdrant

results = client.search( collection_name="agent_memory", query_vector=query_embedding, query_filter=Filter( must=[ FieldCondition(key="user_id", match={"value": user_id}), FieldCondition(key="type", match={"value": "semantic"}), ], should=[ FieldCondition(key="topic", match={"any": ["auth", "security"]}), ] ), limit=5 ) """

ChromaDB (Prototipagem)

""" import chromadb

client = chromadb.PersistentClient(path="./memory_db") collection = client.get_or_create_collection("agent_memory")

Simples e rápido para protótipos

collection.add( ids=[str(uuid4())], embeddings=[embedding], documents=[memory_text], metadatas=[{"user_id": user_id, "type": "episodic"}] )

results = collection.query( query_embeddings=[query_embedding], n_results=5, where={"user_id": user_id} ) """

Padrão de Estratégia de Fragmentação

Dividindo documentos em fragmentos recuperáveis

Quando usar: Processando documentos para armazenamento de memória

ESTRATÉGIAS DE FRAGMENTAÇÃO:

""" O dilema da fragmentação:

• Muito grande: O vetor perde especificidade
• Muito pequeno: Perde contexto

O tamanho ideal do fragmento depende de:

• Tipo de documento (código vs prosa vs dados)
• Padrões de consulta (factual vs exploratório)
• Modelo de incorporação (cada um tem seu ponto ideal)

Orientação geral: 256-512 tokens para a maioria dos casos de uso """

Fragmentação de Tamanho Fixo (Linha de Base)

""" from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter( chunk_size=500, # Caracteres chunk_overlap=50, # Sobreposição evita cortar frases separators=["\n\n", "\n", ". ", " ", ""] # Ordem de prioridade )

chunks = splitter.split_text(document) """

Fragmentação Semântica (Melhor Qualidade)

""" from langchain_experimental.text_splitter import SemanticChunker from langchain_openai import OpenAIEmbeddings

Divide com base na similaridade semântica

splitter = SemanticChunker( embeddings=OpenAIEmbeddings(), breakpoint_threshold_type="percentile", breakpoint_threshold_amount=95 )

chunks = splitter.split_text(document) """

Fragmentação Consciente da Estrutura (Documentos com Hierarquia)

""" from langchain.text_splitter import MarkdownHeaderTextSplitter

Respeita a estrutura do documento

splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[ ("#", "Cabeçalho 1"), ("##", "Cabeçalho 2"), ("###", "Cabeçalho 3"), ] )

chunks = splitter.split_text(markdown_doc)

Cada fragmento tem metadados de cabeçalho para contexto

"""

Fragmentação Contextual (Abordagem da Anthropic)

"""

Adiciona contexto a cada fragmento antes de incorporar

Reduz falhas de recuperação em 35%

def add_context_to_chunk(chunk, document_summary): context_prompt = f''' Resumo do documento: {document_summary}

O seguinte é um fragmento deste documento:
{chunk}
'''
return context_prompt

Incorpore o fragmento contextualizado, não o fragmento bruto

for chunk in chunks: contextualized = add_context_to_chunk(chunk, summary) embedding = embed(contextualized) store(chunk, embedding) # Armazena original, incorpora contextualizado """

Fragmentação Específica para Código

""" from langchain.text_splitter import Language, RecursiveCharacterTextSplitter

Divisão ciente da linguagem

python_splitter = RecursiveCharacterTextSplitter.from_language( language=Language.PYTHON, chunk_size=1000, chunk_overlap=200 )

Respeita limites de função/classe

chunks = python_splitter.split_text(python_code) """

Formação de Memória em Segundo Plano

Processando memórias de forma assíncrona para melhor qualidade

Quando usar: Você quer maior recall sem desacelerar interações

FORMAÇÃO DE MEMÓRIA EM SEGUNDO PLANO:

""" A extração de memória em tempo real desacelera conversas e adiciona complexidade às chamadas de ferramentas do agente. O processamento em segundo plano após as conversas produz memórias de maior qualidade.

Padrão: Formação de memória subconsciente """

Processamento em Segundo Plano com LangGraph

""" from langgraph.graph import StateGraph from langgraph.checkpoint.postgres import PostgresSaver

async def background_memory_processor(thread_id: str): # Executa após a conversa terminar ou ficar ociosa conversation = await load_conversation(thread_id)

# Extrai insights sem pressão de tempo
insights = await llm.invoke('''
    Analise esta conversa e extraia:
    1. Fatos chave aprendidos sobre o usuário
    2. Preferências do usuário reveladas
    3. Tarefas concluídas ou pendentes
    4. Padrões no comportamento do usuário

    Seja minucioso - isso roda em segundo plano.

    Conversa:
    {conversation}
''')

# Armazena na memória de longo prazo
for insight in insights:
    await memory.semantic.upsert(
        namespace="user_insights",
        key=generate_key(insight),
        content=insight,
        metadata={"source_thread": thread_id}
    )

Aciona no final da conversa ou timeout de inatividade

@on_conversation_idle(timeout_minutes=5) async def process_conversation(thread_id): await background_memory_processor(thread_id) """

Consolidação de Memória (Como o Sono)

"""

Consolida e desduplica memórias periodicamente

async def consolidate_memories(user_id: str): # Obtém todas as memórias do usuário memories = await memory.semantic.list( namespace="user_insights", filter={"user_id": user_id} )

# Encontra memórias semelhantes (potenciais duplicatas)
clusters = cluster_by_similarity(memories, threshold=0.9)

# Mescla memórias semelhantes
for cluster in clusters:
    if len(cluster) > 1:
        merged = await llm.invoke(f'''
            Consolide essas memórias relacionadas em uma:
            {cluster}

            Preserve todas as informações importantes.
        ''')
        await memory.semantic.upsert(
            namespace="user_insights",
            key=generate_key(merged),
            content=merged
        )
        # Deleta originais
        for old in cluster:
            await memory.semantic.delete(old.id)

"""

Padrão de Decaimento de Memória

Esquecendo memórias antigas e irrelevantes

Quando usar: A memória cresce muito, a recuperação fica lenta

DECAIMENTO DE MEMÓRIA:

""" Nem todas as memórias devem viver para sempre:

• Preferências antigas podem estar desatualizadas
• Detalhes de tarefas perdem relevância
• Memórias conflitantes confundem a recuperação

Implemente decaimento inteligente baseado em:

• Relevância (quando foi criada/acessada?)
• Frequência (com que frequência é recuperada?)
• Importância (é um fato central ou um detalhe?) """

Decaimento Baseado em Tempo

""" from datetime import datetime, timedelta

async def decay_old_memories(namespace: str, max_age_days: int): cutoff = datetime.now() - timedelta(days=max_age_days)

old_memories = await memory.episodic.list(
    namespace=namespace,
    filter={"last_accessed": {"$lt": cutoff.isoformat()}}
)

for mem in old_memories:
    # Exclusão suave (marca como arquivado)
    await memory.episodic.update(
        id=mem.id,
        metadata={"archived": True, "archived_at": datetime.now().isoformat()}
    )

print(f"Arquivou {len(old_memories)} memórias antigas")

"""

… [truncado para tradução]