# Como manter embeddings sincronizados no pgvector quando os dados mudam no Laravel > Como reindexar só os chunks que mudaram, apagar vetores órfãos, pegar o que os eventos do Eloquent não veem e trocar de modelo de embedding sem derrubar o RAG. Por Lucas Roberto · IA · publicado em 2026-10-21 · 10 min de leitura Versão web: https://lucasroberto.com/blog/como-manter-embeddings-sincronizados-no-pgvector-quando-os-dados-mudam-no-laravel Para manter embeddings sincronizados no pgvector, trate o índice vetorial como uma projeção derivada dos seus dados. Cada alteração dispara um job idempotente que recorta o texto em chunks, compara o hash de cada chunk com o que já está salvo e só chama a API de embeddings para o que mudou. Uma reconciliação noturna pega o que os eventos do Eloquent deixaram passar. A troca de modelo de embedding vira uma migração com estrutura nova, nunca uma atualização no lugar. ## Por que o RAG começa a responder errado meses depois de ir para produção? O roteiro costuma ser este. Você monta o RAG, roda um comando Artisan que indexa tudo, testa, funciona, vai para produção. Meses depois o agente responde com um horário de missa que mudou, uma política de cancelamento de agendamento que não existe mais ou uma taxa antiga da central de ajuda. O modelo não alucinou. Ele leu exatamente o que a busca vetorial entregou, e a busca entregou um texto velho. No [post sobre RAG com pgvector no Laravel](/blog/como-fazer-rag-com-pgvector-no-laravel-embeddings-busca-vetorial-e-respostas-com-claude) eu foquei em montar o pipeline: gerar embeddings, buscar e responder com o Claude. Aqui o assunto é o que acontece depois do dia um, quando os dados de origem começam a mudar. Um embedding é um vetor numérico que representa o significado de um texto. Ele é gerado por uma API externa e guardado no PostgreSQL com a extensão pgvector. Como depende de uma chamada de rede, ele não se atualiza sozinho quando a linha de origem muda. Na prática aparecem três tipos de defeito: 1. **Chunk desatualizado**: o documento foi editado, mas o vetor e o texto indexados são da versão anterior. 2. **Chunk órfão**: o documento foi apagado ou despublicado, mas os chunks continuam aparecendo na busca. 3. **Vetores misturados**: parte da base foi gerada com um modelo de embedding e parte com outro. A similaridade entre eles não significa nada. Os três são silenciosos. Nenhum gera exceção, só respostas piores. ## Como modelar a tabela de chunks para sincronizar só o que mudou? Um chunk é um pedaço do documento, do tamanho certo para virar um embedding e caber no contexto do LLM. A decisão que viabiliza a sincronização incremental é guardar, junto de cada chunk, o hash SHA-256 do conteúdo e o nome do modelo que gerou o vetor. Com isso, saber se um chunk precisa de embedding novo vira uma comparação de strings, sem nenhuma chamada de API. ```php DB::statement('CREATE EXTENSION IF NOT EXISTS vector'); Schema::create('knowledge_chunks', function (Blueprint $table) { $table->id(); $table->foreignId('tenant_id')->index(); $table->string('source_type'); $table->unsignedBigInteger('source_id'); $table->unsignedInteger('position'); $table->text('content'); $table->char('content_hash', 64); $table->string('embedding_model'); $table->timestamps(); $table->index(['source_type', 'source_id']); $table->unique(['source_type', 'source_id', 'content_hash']); }); DB::statement('ALTER TABLE knowledge_chunks ADD COLUMN embedding vector(1536)'); DB::statement('CREATE INDEX knowledge_chunks_embedding_idx ON knowledge_chunks USING hnsw (embedding vector_cosine_ops)'); Schema::create('knowledge_sources', function (Blueprint $table) { $table->string('source_type'); $table->unsignedBigInteger('source_id'); $table->char('text_hash', 64); $table->string('embedding_model'); $table->timestamp('indexed_at'); $table->primary(['source_type', 'source_id']); }); ``` Alguns pontos: - `source_type` e `source_id` seguem a convenção polimórfica do Eloquent. A mesma tabela serve para artigos de ajuda, FAQs e qualquer outro model indexável. - O índice único por hash dentro do documento impede duplicata. Se o mesmo parágrafo aparece duas vezes, um chunk basta. - `knowledge_sources` guarda o hash do texto inteiro de cada documento já indexado. Ela permite pular o trabalho quando nada mudou e é a base da reconciliação noturna. - HNSW é o tipo de índice aproximado do pgvector para busca por vizinhos mais próximos. Ele aceita inserts e deletes normalmente, mas cada delete deixa tupla morta. Com muita reindexação, vale acompanhar o autovacuum dessa tabela. ## Por que o chunking precisa ser determinístico? O hash por chunk só economiza alguma coisa se uma edição pequena mudar poucos chunks. Com janelas de tamanho fixo, tipo 1000 caracteres com sobreposição, inserir uma frase no segundo parágrafo desloca todas as fronteiras dali para frente. Todos os hashes seguintes mudam e você paga o documento inteiro de novo. Por isso eu ancoro as fronteiras na estrutura do texto: cada seção (título Markdown) vira um chunk. Só as seções longas são quebradas por parágrafo. Assim o efeito cascata de uma edição fica contido dentro da seção editada. ```php final class SectionChunker { public function __construct(private int $maxChars = 1500) {} /** @return list */ public function split(string $text): array { $chunks = []; foreach (preg_split('/^(?=#{1,3} )/m', trim($text)) ?: [] as $section) { $section = trim($section); if ($section === '') { continue; } if (mb_strlen($section) <= $this->maxChars) { $chunks[] = $section; continue; } $current = ''; foreach (preg_split('/\n\s*\n/', $section) as $paragraph) { $paragraph = trim($paragraph); if ($paragraph === '') { continue; } if ($current !== '' && mb_strlen($current) + mb_strlen($paragraph) > $this->maxChars) { $chunks[] = $current; $current = ''; } $current = $current === '' ? $paragraph : $current."\n\n".$paragraph; } if ($current !== '') { $chunks[] = $current; } } return $chunks; } } ``` O trade-off é que os chunks ficam com tamanho desigual. Uma seção de duas linhas vira um chunk minúsculo. Isso pesa pouco na qualidade da busca e ajuda muito no custo de manutenção. Outra coisa: qualquer mudança no chunker (tamanho máximo, regra de corte) muda os hashes da base inteira. Trate isso como uma reindexação planejada, não como um ajuste inocente. ## Como disparar a reindexação quando o model muda no Laravel? Uma trait registra os eventos do Eloquent e despacha o job. Duas decisões aqui. A primeira é o `afterCommit()`: sem ele, o job pode rodar antes do commit e ler a versão antiga do registro. Já escrevi sobre [o bug silencioso de jobs disparados dentro de transação](/blog/jobs-disparados-dentro-de-transacao-o-bug-silencioso-do-aftercommit-no-laravel), e aqui ele aparece de forma bem sorrateira, porque o índice fica um passo atrás sem ninguém perceber. A segunda é o debounce, que significa agrupar várias edições seguidas em uma única execução. Ele sai da combinação de `delay` com job único. ```php trait Indexable { public static function bootIndexable(): void { $sync = fn (Model $model) => SyncEmbeddings::dispatch($model->getMorphClass(), $model->getKey()) ->delay(now()->addSeconds(30)) ->afterCommit(); static::saved($sync); static::deleted($sync); if (method_exists(static::class, 'restored')) { static::restored($sync); } } /** Retorna null quando o registro não deve estar no índice. */ abstract public function toIndexableText(): ?string; } class HelpArticle extends Model { use Indexable, SoftDeletes; public function toIndexableText(): ?string { if ($this->published_at === null) { return null; } return "# {$this->title}\n\n{$this->body}"; } } ``` O `toIndexableText()` concentra a regra de negócio do que entra no índice: rascunho não entra, item despublicado sai. O job não precisa saber nada sobre o model. ## Como evitar chamar a API de embeddings para texto que não mudou? O job lê o estado atual do banco no momento em que roda, não no momento em que foi despachado. Por isso é idempotente: rodar duas vezes produz o mesmo resultado. ```php final class SyncEmbeddings implements ShouldQueue, ShouldBeUniqueUntilProcessing { use Dispatchable, InteractsWithQueue, Queueable; public int $tries = 5; public array $backoff = [10, 60, 300]; public int $uniqueFor = 3600; public function __construct(public string $sourceType, public int $sourceId) {} public function uniqueId(): string { return $this->sourceType.':'.$this->sourceId; } public function middleware(): array { return [(new WithoutOverlapping($this->uniqueId()))->releaseAfter(30)]; } public function handle(SectionChunker $chunker, OpenAiEmbedder $embedder): void { $class = Relation::getMorphedModel($this->sourceType) ?? $this->sourceType; $source = $class::find($this->sourceId); $text = $source?->toIndexableText(); $key = ['source_type' => $this->sourceType, 'source_id' => $this->sourceId]; if (blank($text)) { DB::transaction(function () use ($key) { $this->chunks()->delete(); KnowledgeSource::where($key)->delete(); }); return; } $textHash = hash('sha256', $text); $indexed = KnowledgeSource::where($key)->first(); if ($indexed?->text_hash === $textHash && $indexed->embedding_model === $embedder->model()) { return; } $chunks = collect($chunker->split($text)) ->map(fn (string $content, int $i) => [ 'position' => $i, 'content' => $content, 'hash' => hash('sha256', $content), ]) ->unique('hash') ->values(); $existing = $this->chunks() ->where('embedding_model', $embedder->model()) ->pluck('content_hash') ->all(); $missing = $chunks->whereNotIn('hash', $existing)->values(); $vectors = $missing->isEmpty() ? [] : $embedder->embed($missing->pluck('content')->all()); DB::transaction(function () use ($source, $chunks, $missing, $vectors, $embedder, $key, $textHash) { $this->chunks() ->where(fn ($q) => $q ->whereNotIn('content_hash', $chunks->pluck('hash')) ->orWhere('embedding_model', '!=', $embedder->model())) ->delete(); KnowledgeChunk::insert($missing->map(fn (array $chunk, int $i) => [ 'tenant_id' => $source->tenant_id, 'source_type' => $this->sourceType, 'source_id' => $this->sourceId, 'position' => $chunk['position'], 'content' => $chunk['content'], 'content_hash' => $chunk['hash'], 'embedding' => '['.implode(',', $vectors[$i]).']', 'embedding_model' => $embedder->model(), 'created_at' => now(), 'updated_at' => now(), ])->all()); foreach ($chunks as $chunk) { $this->chunks()->where('content_hash', $chunk['hash'])->update(['position' => $chunk['position']]); } KnowledgeSource::updateOrCreate($key, [ 'text_hash' => $textHash, 'embedding_model' => $embedder->model(), 'indexed_at' => now(), ]); }); } private function chunks(): Builder { return KnowledgeChunk::query() ->where('source_type', $this->sourceType) ->where('source_id', $this->sourceId); } } ``` O que cada peça resolve: 1. **`ShouldBeUniqueUntilProcessing`** em vez de `ShouldBeUnique`: o lock de unicidade é liberado quando o job começa a rodar, não quando termina. Uma edição feita durante o processamento agenda uma nova execução em vez de ser descartada. 2. **`WithoutOverlapping`** garante que duas execuções do mesmo documento nunca rodem em paralelo. Se a segunda chegar enquanto a primeira ainda está processando, ela volta para a fila. 3. **`$uniqueFor`** evita que um lock fique preso para sempre se o worker morrer no meio. 4. **A chamada HTTP fica fora da transação.** Segurar uma transação do PostgreSQL aberta esperando uma API externa é pedir lock contention. 5. **O `position` é atualizado** porque chunks que não mudaram podem ter mudado de lugar. Isso importa se você monta o contexto em ordem ou busca chunks vizinhos. Com documentos de dezenas de chunks, um update por chunk é aceitável. Para documentos enormes, troque por um único `UPDATE ... FROM (VALUES ...)`. O embedder chama a API de embeddings da OpenAI direto com o `Http::` do Laravel. A Anthropic não oferece modelo de embeddings próprio, então mesmo quem responde com o Claude precisa de outro provedor para essa parte. A requisição é simples: ```json { "model": "text-embedding-3-small", "input": ["primeiro chunk", "segundo chunk"] } ``` ```php final class OpenAiEmbedder { public function model(): string { return config('rag.embedding_model', 'text-embedding-3-small'); } /** @param list $inputs */ public function embed(array $inputs): array { return collect($inputs) ->chunk(100) ->flatMap(function ($batch) { $response = Http::withToken(config('services.openai.key')) ->timeout(30) ->post('https://api.openai.com/v1/embeddings', [ 'model' => $this->model(), 'input' => $batch->values()->all(), ]) ->throw(); return collect($response->json('data'))->sortBy('index')->pluck('embedding'); }) ->values() ->all(); } } ``` O modelo vem de um único ponto de configuração. A busca também precisa usar esse mesmo método para gerar o embedding da pergunta. Pergunta embedada com um modelo e chunks com outro é exatamente o defeito dos vetores misturados. ## E quando o documento é apagado, despublicado ou o cliente cancela a conta? O evento `deleted` dispara tanto no soft delete quanto no `forceDelete()`. Como `find()` ignora registros na lixeira, o texto volta `null` e o job remove os chunks. Se o registro for restaurado, o evento `restored` reindexa. Não trate isso como detalhe. O chunk é uma cópia literal do conteúdo de origem. Se a origem tinha dado pessoal e foi apagada por pedido do titular, o vetor e o texto continuam no banco e continuam aparecendo para o agente. Para a LGPD, apagar a origem e esquecer a cópia é a mesma coisa que não apagar. Quando um tenant encerra a conta, a coluna `tenant_id` permite um `DELETE` direto em `knowledge_chunks`, sem depender de evento nenhum. ## Como pegar as alterações que não disparam eventos do Eloquent? Evento de model é a primeira linha de defesa, não a única. Várias coisas mudam dados sem passar por ele: - `Model::where(...)->update([...])` e `->delete()` em massa não carregam os models e não disparam `saved` nem `deleted`. - `DB::table()`, seeders, imports de CSV e SQL rodado à mão no console. - Jobs de sincronização que falharam todas as tentativas e foram parar na `failed_jobs`. A solução é uma reconciliação: comparar periodicamente o estado da origem com o que está indexado e corrigir a diferença. Com a `knowledge_sources`, ela fica barata, porque compara hashes sem chamar API nenhuma: ```php // routes/console.php Artisan::command('rag:reconcile', function () { foreach (config('rag.indexable') as $class) { $type = (new $class)->getMorphClass(); $indexed = KnowledgeSource::where('source_type', $type)->pluck('text_hash', 'source_id'); $class::query()->chunkById(500, function ($models) use ($type, $indexed) { foreach ($models as $model) { $text = $model->toIndexableText(); $current = blank($text) ? null : hash('sha256', $text); if ($current !== ($indexed[$model->getKey()] ?? null)) { SyncEmbeddings::dispatch($type, $model->getKey()); } } }); KnowledgeSource::where('source_type', $type) ->whereNotIn('source_id', $class::query()->select((new $class)->getKeyName())) ->pluck('source_id') ->each(fn ($id) => SyncEmbeddings::dispatch($type, $id)); } })->purpose('Reindexa documentos cujo hash diverge do índice vetorial'); Schedule::command('rag:reconcile')->dailyAt('03:00'); ``` O primeiro loop pega edições perdidas. O segundo pega órfãos: documentos que estão no índice mas não existem mais na origem, ou estão na lixeira. Para bases de milhares de documentos, carregar os hashes em memória funciona bem. Acima disso, faça a comparação paginada pelos dois lados. Uma métrica útil é logar quantos jobs a reconciliação despachou. Se o número sobe sem motivo, tem algum caminho de escrita ignorando os eventos. Se você usa [busca híbrida com pgvector e full-text search](/blog/como-fazer-busca-hibrida-no-laravel-com-pgvector-e-full-text-search-do-postgresql), repare na assimetria: o `tsvector` pode ser uma coluna gerada pelo próprio PostgreSQL e nunca dessincroniza. O embedding não pode, porque depende de uma API externa. Toda essa maquinaria existe só para o lado vetorial. ## Como trocar de modelo de embedding sem derrubar a busca? Trocar o modelo invalida a base inteira. Os vetores do modelo novo não são comparáveis com os do antigo, e muitas vezes a dimensão também muda, o que impede reaproveitar a coluna `vector(1536)`. O caminho que uso é o de migração com escrita dupla: 1. Criar uma estrutura nova: uma tabela `knowledge_chunks_v2` com a dimensão do modelo novo e o próprio índice HNSW. 2. Fazer o job de sincronização escrever nas duas estruturas enquanto durar a migração, para que as edições novas não fiquem para trás. 3. Rodar o backfill numa fila de baixa prioridade no Horizon, com concorrência limitada para não estourar o rate limit do provedor. 4. Comparar a qualidade das duas buscas num conjunto fixo de perguntas com os chunks esperados, antes de virar a chave. 5. Trocar a leitura por configuração, observar alguns dias e só então remover a estrutura antiga. O que não fazer: alterar `rag.embedding_model` e esperar a reconciliação reindexar tudo aos poucos. Durante horas ou dias a busca vai comparar perguntas embedadas com o modelo novo contra chunks do modelo antigo. O job acima até se corrige sozinho nesse cenário, porque apaga chunks de outro modelo, mas a busca fica degradada até o fim. ## Qual estratégia de sincronização escolher? | Estratégia | Custo de API | Atraso | Pega escrita fora do Eloquent? | Quando usar | |---|---|---|---|---| | Reindexar tudo por cron | Alto, paga a base inteira toda vez | Até a próxima execução | Sim | Bases pequenas que mudam pouco | | Reindexar o documento inteiro a cada `saved` | Médio, paga o documento inteiro | Segundos | Não | Documentos curtos, de um chunk só | | Hash por chunk + eventos + reconciliação | Baixo, paga só o que mudou | Segundos, com rede de segurança diária | Sim | Bases que mudam com frequência e alimentam agente em produção | ## Conclusão Se você já tem RAG em produção, comece pelo que é barato e pega mais defeitos. Primeiro, crie a `knowledge_sources` e rode a reconciliação uma vez só para medir. O número de documentos divergentes mostra o quanto o índice já está desatualizado hoje. Depois, mova a regra de "o que entra no índice" para um `toIndexableText()` em cada model e ligue os eventos com `afterCommit()` e debounce. Só então invista no hash por chunk e no chunking ancorado em seções, que é onde está a economia de API. E antes de qualquer troca de modelo de embedding, monte o conjunto de perguntas de comparação. Sem ele, você não tem como saber se a migração melhorou ou piorou a busca. --- Lucas Roberto é desenvolvedor back-end e fundador da Granpag. Mais textos: https://lucasroberto.com/blog