Como construir um chatbot com memória de longo prazo usando vetores

1. Fundamentos da Memória de Longo Prazo em Chatbots

1.1. Diferença entre memória de curto prazo e memória de longo prazo

Chatbots tradicionais operam com memória de curto prazo, limitada ao contexto da sessão atual. Quando a conversa termina, o histórico é perdido. A memória de longo prazo, por outro lado, persiste informações entre sessões, permitindo que o chatbot lembre preferências, fatos mencionados há dias e o progresso do usuário em tarefas complexas.

Memória de curto prazo (sessão):
- Armazenada em variáveis de contexto ou cache
- Expira após término da sessão
- Geralmente limitada a 4k-8k tokens

Memória de longo prazo (persistente):
- Armazenada em banco de dados vetorial
- Persiste indefinidamente (sujeito a políticas de retenção)
- Permite recuperação semântica de informações históricas

1.2. Por que vetores de embeddings?

Embeddings convertem texto em vetores numéricos que capturam significado semântico. Isso permite que o chatbot encontre conversas relevantes mesmo quando as palavras exatas não correspondem. Por exemplo, se o usuário perguntou "Qual a previsão do tempo?" há três dias, e hoje pergunta "Como vai estar o clima?", a busca vetorial identifica a similaridade semântica.

1.3. Visão geral da arquitetura

Usuário → Consulta → Modelo de Embeddings → Busca Vetorial → Contexto Recuperado → Prompt Enriquecido → LLM → Resposta

2. Escolhendo e Configurando o Banco de Dados Vetorial

2.1. Opções populares

Ferramenta	Prós	Contras
ChromaDB	Open-source, fácil setup local	Escalabilidade limitada
Pinecone	Gerenciado, escalável	Custo por uso
Qdrant	Performance, filtros avançados	Configuração mais complexa
Weaviate	Híbrido (vetorial + texto)	Maior consumo de recursos

Para este tutorial, usaremos ChromaDB por sua simplicidade de configuração.

2.2. Instalação e configuração inicial

pip install chromadb sentence-transformers openai

2.3. Estrutura de dados para conversas

import chromadb
from chromadb.config import Settings

client = chromadb.Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="./chatbot_memory"
))

collection = client.create_collection(
    name="conversas",
    metadata={"hnsw:space": "cosine"}
)

# Estrutura de cada documento:
# id: "user_123_20240101_001"
# embedding: [0.012, -0.045, ...] (vetor de 1536 dimensões)
# metadata: {
#     "usuario": "user_123",
#     "timestamp": "2024-01-01T10:30:00",
#     "sessao": "sessao_456",
#     "tipo": "pergunta"  # ou "resposta"
# }
# document: "Qual a previsão do tempo para amanhã?"

3. Geração de Embeddings para Fragmentos de Conversa

3.1. Seleção do modelo de embeddings

OpenAI text-embedding-3-small: 1536 dimensões, alta qualidade, pago por token.

Sentence Transformers (local): Gratuito, menor latência, qualidade ligeiramente inferior.

from sentence_transformers import SentenceTransformer
modelo_embeddings = SentenceTransformer('all-MiniLM-L6-v2')

def gerar_embedding(texto):
    return modelo_embeddings.encode(texto).tolist()

3.2. Estratégia de chunking

Fragmentar conversas por turno individual é a abordagem mais comum, mas pode ser ajustada:

def fragmentar_conversa(historico):
    """
    Divide histórico em chunks significativos.
    Cada turno (pergunta+resposta) vira um fragmento.
    """
    fragmentos = []
    for i in range(0, len(historico), 2):
        pergunta = historico[i]
        resposta = historico[i+1] if i+1 < len(historico) else ""
        fragmento = f"Usuário: {pergunta}\nAssistente: {resposta}"
        fragmentos.append(fragmento)
    return fragmentos

3.3. Função para vetorizar e armazenar

def armazenar_interacao(usuario, pergunta, resposta, sessao):
    fragmento = f"Usuário: {pergunta}\nAssistente: {resposta}"
    embedding = gerar_embedding(fragmento)

    collection.add(
        embeddings=[embedding],
        documents=[fragmento],
        metadatas=[{
            "usuario": usuario,
            "timestamp": datetime.now().isoformat(),
            "sessao": sessao,
            "tipo": "interacao"
        }],
        ids=[f"{usuario}_{int(time.time())}"]
    )

4. Arquitetura do Pipeline de Memória

4.1. Fluxo de interação completo

def buscar_contexto_relevante(consulta, usuario, top_k=5):
    embedding_consulta = gerar_embedding(consulta)

    resultados = collection.query(
        query_embeddings=[embedding_consulta],
        n_results=top_k,
        where={"usuario": usuario},
        include=["documents", "metadatas", "distances"]
    )

    return resultados

def gerar_resposta_com_memoria(consulta, usuario, sessao_atual):
    # 1. Buscar memória de curto prazo (cache de sessão)
    contexto_sessao = cache_sessao.get(sessao_atual, "")

    # 2. Buscar memória de longo prazo (vetores)
    memorias = buscar_contexto_relevante(consulta, usuario)
    contexto_historico = "\n".join(memorias['documents'][0])

    # 3. Montar prompt enriquecido
    prompt = f"""Contexto da sessão atual:
{contexto_sessao}

Histórico relevante do usuário:
{contexto_historico}

Pergunta atual: {consulta}

Responda considerando todo o contexto acima."""

    # 4. Chamar LLM
    resposta = chamar_llm(prompt)

    # 5. Armazenar nova interação
    armazenar_interacao(usuario, consulta, resposta, sessao_atual)

    return resposta

4.2. Mecanismo de recuperação com filtros

def buscar_por_usuario_e_tempo(consulta, usuario, dias_max=30):
    data_limite = (datetime.now() - timedelta(days=dias_max)).isoformat()

    resultados = collection.query(
        query_embeddings=[gerar_embedding(consulta)],
        n_results=10,
        where={
            "$and": [
                {"usuario": {"$eq": usuario}},
                {"timestamp": {"$gte": data_limite}}
            ]
        }
    )
    return resultados

5. Gerenciamento do Ciclo de Vida da Memória

5.1. Políticas de retenção

def limpar_memorias_antigas(dias_retencao=90):
    data_limite = (datetime.now() - timedelta(days=dias_retencao)).isoformat()

    # Identificar memórias antigas
    antigas = collection.get(
        where={"timestamp": {"$lt": data_limite}}
    )

    # Opção 1: Remover completamente
    if antigas['ids']:
        collection.delete(ids=antigas['ids'])

    # Opção 2: Sumarizar antes de remover
    for usuario in set(m['usuario'] for m in antigas['metadatas']):
        memorias_usuario = [
            doc for doc, meta in zip(antigas['documents'], antigas['metadatas'])
            if meta['usuario'] == usuario
        ]
        resumo = sumarizar_memorias(memorias_usuario)
        armazenar_resumo(usuario, resumo)

5.2. Atualização incremental

def adicionar_interacao_incremental(usuario, pergunta, resposta):
    # Adiciona apenas a nova interação sem reindexar
    armazenar_interacao(usuario, pergunta, resposta, sessao_atual)
    # O ChromaDB gerencia o índice HNSW automaticamente

5.3. Rotina de sumarização automática

def sumarizar_memorias(memorias):
    prompt = f"""Resuma as seguintes interações do usuário em 3-5 pontos principais:
{chr(10).join(memorias)}

Resumo conciso:"""

    resposta = chamar_llm(prompt)
    return resposta

def armazenar_resumo(usuario, resumo):
    embedding = gerar_embedding(resumo)
    collection.add(
        embeddings=[embedding],
        documents=[f"RESUMO: {resumo}"],
        metadatas=[{
            "usuario": usuario,
            "tipo": "resumo",
            "timestamp": datetime.now().isoformat()
        }],
        ids=[f"resumo_{usuario}_{int(time.time())}"]
    )

6. Tratamento de Casos Complexos e Otimizações

6.1. Memória hierárquica

class ChatbotComMemoria:
    def __init__(self):
        self.cache_sessao = {}  # Memória de curto prazo
        self.banco_vetorial = collection  # Memória de longo prazo

    def obter_contexto_completo(self, usuario, sessao, consulta):
        # Cache de sessão (últimos 5 turnos)
        contexto_curto = self.cache_sessao.get(sessao, [])[-5:]

        # Memória de longo prazo (busca semântica)
        contexto_longo = self.banco_vetorial.query(
            query_embeddings=[gerar_embedding(consulta)],
            n_results=3,
            where={"usuario": usuario}
        )

        return {
            "curto_prazo": contexto_curto,
            "longo_prazo": contexto_longo
        }

6.2. Otimizações de escalabilidade

# Batch de embeddings
def processar_lote(interacoes):
    textos = [f"{i['pergunta']} {i['resposta']}" for i in interacoes]
    embeddings = modelo_embeddings.encode(textos, batch_size=32)

    # Adicionar em lote
    collection.add(
        embeddings=embeddings.tolist(),
        documents=textos,
        metadatas=[{
            "usuario": i['usuario'],
            "timestamp": i['timestamp']
        } for i in interacoes],
        ids=[f"batch_{i}" for i in range(len(interacoes))]
    )

6.3. Privacidade e segurança

def anonimizar_dados(texto):
    # Remove emails
    texto = re.sub(r'\S+@\S+', '[EMAIL]', texto)
    # Remove números de telefone
    texto = re.sub(r'\(\d{2}\)\s*\d{4,5}-\d{4}', '[TELEFONE]', texto)
    # Remove nomes próprios (usando NER)
    from spacy import load
    nlp = load('pt_core_news_sm')
    doc = nlp(texto)
    for ent in doc.ents:
        if ent.label_ == "PER":
            texto = texto.replace(ent.text, "[NOME]")
    return texto

7. Testes, Monitoramento e Iteração

7.1. Métricas de qualidade

def avaliar_recuperacao(consultas_teste, relevantes_esperados):
    acertos = 0
    for consulta, esperado in zip(consultas_teste, relevantes_esperados):
        resultados = buscar_contexto_relevante(consulta, "test_user")
        recuperados = resultados['documents'][0]
        if esperado in recuperados:
            acertos += 1
    return acertos / len(consultas_teste)  # recall@k

7.2. Monitoramento

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("chatbot_memoria")

def buscar_com_log(consulta, usuario):
    inicio = time.time()
    resultados = buscar_contexto_relevante(consulta, usuario)
    latencia = time.time() - inicio

    logger.info(f"Consulta: {consulta[:50]}... | Usuário: {usuario} | "
                f"Resultados: {len(resultados['ids'][0])} | "
                f"Latência: {latencia:.3f}s | "
                f"Tokens embedding: {len(consulta.split())}")

    return resultados

7.3. Estratégias de melhoria contínua

def ajustar_chunk_size(historico_feedback):
    # Analisar feedback do usuário para ajustar tamanho dos chunks
    chunk_sizes = [128, 256, 512]
    melhor_size = 256
    melhor_precisao = 0

    for size in chunk_sizes:
        precisao = simular_com_chunk_size(size, historico_feedback)
        if precisao > melhor_precisao:
            melhor_precisao = precisao
            melhor_size = size

    return melhor_size

---

Referências

• ChromaDB Documentation — Documentação oficial do banco vetorial ChromaDB, incluindo guias de instalação, API de consulta e estratégias de indexação para chatbots.

• OpenAI Embeddings Guide — Guia oficial da OpenAI sobre embeddings, incluindo o modelo text-embedding-3-small, casos de uso para busca semântica e boas práticas de implementação.

• Sentence Transformers Documentation — Documentação da biblioteca Sentence Transformers para geração de embeddings locais, com modelos pré-treinados e tutoriais de fine-tuning.

• Pinecone Vector Database Tutorial — Tutorial completo sobre bancos de dados vetoriais, incluindo conceitos de similaridade coseno, indexação HNSW e estratégias de recuperação para chatbots.

• LangChain Memory Systems — Guia da LangChain sobre implementação de diferentes tipos de memória para LLMs, incluindo buffer de conversação, memória vetorial e sumarização.

• Qdrant Vector Search Engine — Documentação do Qdrant com exemplos de filtros avançados, busca por payload e otimizações de performance para sistemas de recomendação e chatbots.

• Weaviate Hybrid Search — Documentação do Weaviate sobre busca híbrida (vetorial + palavra-chave), ideal para chatbots que precisam combinar precisão semântica com recall de termos exatos.