Arquitetando um sistema robusto de criação de incorporação de vetores para dados PostgreSQL

Incorporar dados armazenados em uma tabela é sem dúvida útil – com aplicações que vão desde pesquisa semântica e sistemas de recomendação até aplicações generativas de IA e geração aumentada de recuperação. Mas criar e gerenciar embeddings para dados em tabelas PostgreSQL pode ser complicado, com muitas considerações e casos extremos a serem considerados, como manter os embeddings atualizados com atualizações e exclusões de tabelas, garantir resiliência contra falhas e impacto nos sistemas existentes dependentes de a mesa. PostgreSQL para dados residentes no PostgreSQL usando – nosso sistema de criação de incorporação simples e resiliente para dados residentes no PostgreSQL. Usando um aplicativo de blog com dados armazenados em um banco de dados PostgreSQL como exemplo, abordamos como criar e manter embeddings de vetores atualizados usando Python, LangChain e Timescale Vector. Em uma postagem anterior do blog, detalhamos um guia passo a passo sobre o processo de criação e gerenciamento de embeddings PgVetorizador Nesta postagem do blog, discutiremos as decisões técnicas de design e as compensações que fizemos ao construir para garantir simplicidade, resiliência e alto desempenho. Também discutiremos designs alternativos se você quiser criar o seu próprio. o PgVectorizer Vamos começar. Projeto de um vetorizador de alto desempenho para dados PostgreSQL (PgVectorizer) Primeiro, vamos descrever como funcionará o sistema que estamos construindo. Sinta-se à vontade para pular esta seção se você já leu o . Postagem do PgVectorizer Visão geral do sistema Como exemplo ilustrativo, usaremos uma aplicação de blog simples que armazena dados no PostgreSQL usando uma tabela definida da seguinte forma: CREATE TABLE blog ( id SERIAL PRIMARY KEY NOT NULL, title TEXT NOT NULL, author TEXT NOT NULL, contents TEXT NOT NULL, category TEXT NOT NULL, published_time TIMESTAMPTZ NULL --NULL if not yet published ); Queremos criar embeddings no conteúdo da postagem do blog para que possamos usá-lo posteriormente para pesquisa semântica e geração aumentada de recuperação de energia. Os embeddings só devem existir e ser pesquisáveis em blogs que foram publicados (onde é ). published_time NOT NULL Ao construir este sistema de embeddings, conseguimos identificar uma série de objetivos que qualquer sistema simples e resiliente que crie embeddings deveria ter: Isso permite que sistemas e aplicações que já utilizam esta tabela não sejam impactados por alterações no sistema de incorporação. Isto é especialmente importante para sistemas legados. Nenhuma modificação na tabela original. Ter que modificar o código que altera a tabela pode não ser possível para sistemas legados. Também é um design de software ruim porque combina sistemas que não usam embeddings com código que gera a incorporação. Nenhuma modificação nos aplicativos que interagem com a tabela. quando as linhas da tabela de origem forem alteradas (neste caso, a tabela do blog). Isso diminui a carga de manutenção e contribui para um software livre de preocupações. Ao mesmo tempo, esta atualização não precisa ser instantânea ou dentro do mesmo commit. Para a maioria dos sistemas, “consistência eventual” é suficiente. Atualize automaticamente os embeddings a maioria dos sistemas gera incorporações por meio de uma chamada para um sistema externo, como a API OpenAI. Em cenários em que o sistema externo está inoperante ou ocorre um mau funcionamento da rede, é imperativo que o restante do seu sistema de banco de dados continue funcionando. Garanta resiliência contra falhas de rede e de serviço: Essas diretrizes foram a base de uma arquitetura robusta que implementamos usando o , uma biblioteca para trabalhar com dados vetoriais usando PostgreSQL. Para concluir o trabalho com êxito, foram adicionadas novas funcionalidades a esta biblioteca— —para tornar a incorporação de dados PostgreSQL o mais simples possível. Biblioteca Python de vetor de escala de tempo PgVetorizador Aqui está a arquitetura que escolhemos: Neste design, primeiro adicionamos um gatilho à tabela do blog que monitora alterações e, ao ver uma modificação, insere um trabalho na tabela blog_work_queue que indica que uma linha na tabela do blog está desatualizada com sua incorporação. Em uma programação fixa, um trabalho de criador de embeddings pesquisará a tabela blog_work_queue e, se encontrar trabalho para fazer, fará o seguinte em um loop: Ler e bloquear uma linha na tabela blog_work_queue Leia a linha correspondente na tabela do blog Crie uma incorporação para os dados na linha do blog Escreva a incorporação na tabela blog_embedding Exclua a linha bloqueada na tabela blog_work_queue Para ver este sistema em ação, veja um exemplo de uso para . criar e manter embeddings em uma tabela PostgreSQL usando OpenAI, LangChain e Timescale Vector nesta postagem do blog Voltando ao exemplo da nossa tabela de aplicação de blog, em alto nível, tem que fazer duas coisas: o PgVectorizer Acompanhe as alterações nas linhas do blog para saber quais linhas foram alteradas. Forneça um método para processar as alterações para criar embeddings. Ambos devem ser altamente simultâneos e de alto desempenho. Vamos ver como isso funciona. Acompanhe as alterações na tabela do blog com a tabela blog_work_queue Você pode criar uma tabela de fila de trabalho simples com o seguinte: CREATE TABLE blog_embedding_work_queue ( id INT ); CREATE INDEX ON blog_embedding_work_queue(id); Esta é uma tabela muito simples, mas há um item digno de nota: esta tabela não possui chave única. Isso foi feito para evitar problemas de bloqueio ao processar a fila, mas significa que podemos ter duplicatas. Discutiremos a compensação posteriormente na Alternativa 1 abaixo. Em seguida, você cria um gatilho para rastrear quaisquer alterações feitas no : blog CREATE OR REPLACE FUNCTION blog_wq_for_embedding() RETURNS TRIGGER LANGUAGE PLPGSQL AS $$ BEGIN IF (TG_OP = 'DELETE') THEN INSERT INTO blog_embedding_work_queue VALUES (OLD.id); ELSE INSERT INTO blog_embedding_work_queue VALUES (NEW.id); END IF; RETURN NULL; END; $$; CREATE TRIGGER track_changes_for_embedding AFTER INSERT OR UPDATE OR DELETE ON blog FOR EACH ROW EXECUTE PROCEDURE blog_wq_for_embedding(); INSERT INTO blog_embedding_work_queue SELECT id FROM blog WHERE published_time is NOT NULL; O gatilho insere o ID do blog que mudou para blog_work_queue. Instalamos o gatilho e, em seguida, inserimos todos os blogs existentes no work_queue. Essa ordem é importante para garantir que nenhum ID seja descartado. Agora, vamos descrever alguns designs alternativos e por que os rejeitamos. Alternativa 1: implemente uma chave primária ou exclusiva para a tabela blog_work_queue. A introdução desta chave eliminaria o problema de entradas duplicadas. No entanto, não está isento de desafios, principalmente porque tal chave nos forçaria a usar a cláusula para inserir novos IDs na tabela, e essa cláusula bloqueia o ID na árvore B. INSERT…ON CONFLICT DO NOTHING Eis o dilema: durante a fase de processamento, é necessário deletar as linhas que estão sendo trabalhadas para evitar o processamento simultâneo. No entanto, a confirmação desta exclusão só pode ser feita após a incorporação correspondente ter sido colocada em blog_embeddings. Isso garante que nenhum ID seja perdido se houver uma interrupção no meio do caminho, por exemplo, se a criação da incorporação falhar após a exclusão, mas antes da incorporação ser gravada. Agora, se criarmos uma chave única ou primária, a transação que supervisiona a exclusão permanece aberta. Conseqüentemente, isso atua como um bloqueio nesses IDs específicos, impedindo sua inserção de volta no blog_work_queue durante toda a duração do trabalho de criação de incorporação. Dado que leva mais tempo para criar embeddings do que uma transação típica de banco de dados, isso significa problemas. O bloqueio paralisaria o gatilho da tabela principal do 'blog', levando a uma queda no desempenho do aplicativo principal. Para piorar as coisas, se processar várias linhas em um lote, os impasses também se tornam um problema potencial. No entanto, os potenciais problemas decorrentes de entradas duplicadas ocasionais podem ser geridos durante a fase de processamento, conforme ilustrado mais adiante. Uma duplicata esporádica aqui e ali não é um problema, pois aumenta apenas marginalmente a quantidade de trabalho que o trabalho de incorporação executa. Isto é certamente mais palatável do que enfrentar os desafios de bloqueio mencionados acima. Alternativa 2: Acompanhe o trabalho que precisa ser feito adicionando uma coluna à tabela do blog para rastrear se existe uma incorporação atualizada. Por exemplo, poderíamos adicionar uma coluna booleana definida como falsa na modificação e invertida para verdadeira quando a incorporação for criada. Existem três razões para rejeitar este design: embedded Não queremos modificar a tabela pelos motivos já mencionados acima. blog A obtenção eficiente de uma lista de blogs não incorporados exigiria um índice adicional (ou índice parcial) na tabela de blogs. Isso desaceleraria outras operações. Isso aumenta a rotatividade na tabela porque cada modificação agora seria gravada duas vezes (uma vez com embedding=false e uma vez com embedding=true) devido à natureza MVCC do PostgreSQL. Um work_queue_table separado resolve esses problemas. Alternativa 3: Crie os embeddings diretamente no trigger. Essa abordagem tem vários problemas: Se o serviço de incorporação estiver inativo, o gatilho precisará falhar (abortando sua transação) ou você precisará criar um caminho de código de backup que... armazene os IDs que não puderam ser incorporados em uma fila. A última solução nos leva de volta ao design proposto, mas com mais complexidade acrescentada. Esse gatilho provavelmente será muito mais lento que o restante das operações do banco de dados devido à latência necessária para entrar em contato com um serviço externo. Isso desacelerará o restante das operações do banco de dados na tabela. Força o usuário a escrever o código de incorporação de criação diretamente no banco de dados. Dado que a língua franca da IA é o Python e que a criação de incorporação geralmente requer muitas outras bibliotecas, isso nem sempre é fácil ou mesmo possível (especialmente se for executado em um ambiente de nuvem PostgreSQL hospedado). É muito melhor ter um design onde você tenha a opção de criar embeddings dentro ou fora do banco de dados. Agora que temos uma lista de blogs que precisam ser incorporados, vamos processar a lista! Crie os embeddings Existem muitas maneiras de criar incorporações. Recomendamos usar um script Python externo. Este script verificará a fila de trabalhos e as postagens de blog relacionadas, invocará um serviço externo para criar os embeddings e, em seguida, armazenará esses embeddings de volta no banco de dados. Nosso raciocínio para esta estratégia é o seguinte: : Recomendamos porque oferece um ecossistema rico e incomparável para tarefas de dados de IA, destacado pelo poderoso desenvolvimento de LLM e bibliotecas de dados como e LlamaIndex. Escolha de Python Python LangChain : queríamos que os usuários tivessem controle sobre como incorporar seus dados. No entanto, ao mesmo tempo, muitos provedores de nuvem Postgres não permitem a execução de código Python arbitrário dentro do banco de dados devido a questões de segurança. Portanto, para permitir que os usuários tenham flexibilidade tanto na incorporação de scripts quanto no local onde hospedam seu banco de dados, optamos por um design que usava scripts Python externos. Optando por um script externo em vez de PL/Python Os trabalhos devem ter desempenho e segurança de simultaneidade. A simultaneidade garante que, se os jobs começarem a ficar atrasados, os agendadores poderão iniciar mais jobs para ajudar o sistema a acompanhar e lidar com a carga. Veremos como configurar cada um desses métodos posteriormente, mas primeiro, vamos ver como seria o script Python. Fundamentalmente, o roteiro tem três partes: Leia a fila de trabalho e a postagem do blog Crie uma incorporação para a postagem do blog Escreva a incorporação na tabela blog_embedding As etapas 2 e 3 são executadas por um retorno de chamada que definimos no . Então, vamos examinar mais profundamente como processamos a fila de trabalho. embed_and_write Postagem do blog PgVectorizer Processar a fila de trabalho Primeiro mostraremos o código e depois destacaremos os principais elementos em jogo: def process_queue(embed_and_write_cb, batch_size:int=10): with psycopg2.connect(TIMESCALE_SERVICE_URL) as conn: with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(f""" SELECT to_regclass('blog_embedding_work_queue')::oid; """) table_oid = cursor.fetchone()[0] cursor.execute(f""" WITH selected_rows AS ( SELECT id FROM blog_embedding_work_queue LIMIT {int(batch_size)} FOR UPDATE SKIP LOCKED ), locked_items AS ( SELECT id, pg_try_advisory_xact_lock( {int(table_oid)}, id) AS locked FROM ( SELECT DISTINCT id FROM selected_rows ORDER BY id ) as ids ), deleted_rows AS ( DELETE FROM blog_embedding_work_queue WHERE id IN ( SELECT id FROM locked_items WHERE locked = true ORDER BY id ) ) SELECT locked_items.id as locked_id, {self.table_name}.* FROM locked_items LEFT JOIN blog ON blog.id = locked_items.id WHERE locked = true ORDER BY locked_items.id """) res = cursor.fetchall() if len(res) > 0: embed_and_write_cb(res) return len(res) process_queue(embed_and_write) O código SQL no trecho acima é sutil porque foi projetado para ter desempenho e segurança de simultaneidade, então vamos analisá-lo: : Inicialmente, o sistema recupera um número especificado de entradas da fila de trabalho, determinado pelo parâmetro de tamanho da fila em lote. Um bloqueio FOR UPDATE é usado para garantir que os scripts em execução simultânea não tentem processar os mesmos itens da fila. A diretiva SKIP LOCKED garante que se alguma entrada estiver sendo tratada por outro script, o sistema irá ignorá-la em vez de esperar, evitando atrasos desnecessários. Retirando itens da fila de trabalho : devido à possibilidade de entradas duplicadas para o mesmo blog_id na tabela de fila de trabalho, simplesmente bloquear essa tabela é insuficiente. O processamento simultâneo do mesmo ID por trabalhos diferentes seria prejudicial. Considere a seguinte condição de corrida potencial: Bloqueio de IDs de blog A tarefa 1 inicia e acessa um blog, recuperando a versão 1. Ocorre uma atualização externa no blog. Posteriormente, começa o Job 2, obtendo a versão 2. Ambos os trabalhos iniciam o processo de geração de incorporação. O trabalho 2 é concluído, armazenando o embedding correspondente à versão 2 do blog. A tarefa 1, após a conclusão, substitui erroneamente a incorporação da versão 2 pela versão 1 desatualizada. Embora seja possível combater esse problema introduzindo o rastreamento explícito de versões, ele introduz uma complexidade considerável sem benefício de desempenho. A estratégia que optamos não apenas atenua esse problema, mas também evita operações redundantes e desperdício de trabalho com a execução simultânea de scripts. Um bloqueio de aconselhamento Postgres, prefixado com o identificador da tabela para evitar possíveis sobreposições com outros bloqueios, é empregado. A variante , análoga à aplicação anterior de SKIP LOCKED, garante que o sistema evite esperar por bloqueios. A inclusão da cláusula ORDER BY blog_id ajuda a evitar possíveis impasses. Abordaremos algumas alternativas abaixo. try : o script exclui todos os itens da fila de trabalho dos blogs que bloqueamos com sucesso. Se esses itens da fila estiverem visíveis por meio do MVCC (Controle de simultaneidade multiversão), suas atualizações serão manifestadas na linha do blog recuperada. Observe que excluímos todos os itens com o ID do blog fornecido, não apenas os itens lidos ao selecionar as linhas: isso lida efetivamente com entradas duplicadas para o mesmo ID do blog. É crucial observar que esta exclusão só é confirmada após invocar a função embed_and_write() e o subsequente armazenamento da incorporação atualizada. Essa sequência garante que não perderemos nenhuma atualização, mesmo se o script falhar durante a fase de geração de incorporação. Limpando a fila de trabalho Na última etapa, buscamos os blogs para processar. Observe o uso da junção esquerda: que nos permite recuperar os IDs do blog para itens excluídos que não terão uma linha de blog. Precisamos rastrear esses itens para excluir suas incorporações. No retorno de chamada , usamos public_time sendo NULL como sentinela para o blog que está sendo excluído (ou não publicado, caso em que também queremos excluir a incorporação). Fazendo com que os blogs sejam processados: embed_and_write Alternativa 4: Evite usar bloqueios consultivos usando outra tabela. Se o sistema já utiliza bloqueios consultivos e você está preocupado com colisões, é possível usar uma tabela com um ID de blog como chave primária e bloquear as linhas. Na verdade, esta pode ser a própria tabela do blog se você tiver certeza de que esses bloqueios não deixarão nenhum outro sistema lento (lembre-se, esses bloqueios devem ser mantidos durante todo o processo de incorporação, o que pode demorar um pouco). Alternativamente, você pode ter uma tabela blog_embedding_locks apenas para essa finalidade. Não sugerimos a criação dessa tabela porque achamos que ela pode ser um grande desperdício em termos de espaço, e o uso de bloqueios consultivos evita essa sobrecarga. Conclusão e próximos passos e mantê-los automaticamente atualizados. Essa arquitetura garante que os embeddings permaneçam sincronizados com os dados em constante evolução, respondendo perfeitamente a inserções, modificações e exclusões. Apresentamos o PgVectorizer e delineamos um sistema especializado na geração de embeddings vetoriais a partir de dados armazenados no PostgreSQL Nesta postagem do blog, mostramos os bastidores de como criamos um sistema que possui resiliência, lidando com eficácia com possíveis tempos de inatividade do serviço de geração de incorporação. Seu design é adequado para gerenciar uma alta taxa de modificações de dados e pode usar perfeitamente processos de geração de incorporação simultâneos para acomodar cargas elevadas. Além disso, o paradigma de submeter dados ao PostgreSQL e usar o banco de dados para gerenciar a geração de incorporação em segundo plano surge como um mecanismo fácil para supervisionar a manutenção da incorporação em meio a modificações de dados. Uma infinidade de demonstrações e tutoriais no espaço de IA concentram-se exclusivamente na criação inicial de dados a partir de documentos, ignorando as intrincadas nuances associadas à preservação da sincronização de dados à medida que ela evolui. No entanto, em ambientes de produção reais, os dados mudam invariavelmente, e lidar com as complexidades de rastrear e sincronizar essas mudanças não é uma tarefa trivial. Mas é para isso que um banco de dados foi projetado! Por que não apenas usá-lo? : Aqui estão alguns recursos para continuar sua jornada de aprendizado : armazene os embeddings gerados a partir de seus dados PostgreSQL em um banco de dados vetorial rápido e escalonável construído em PostgreSQL. Saber mais e . Experimente o Timescale Vector gratuitamente por 90 dias sobre o vetor de escala de tempo como funciona : Saiba mais sobre o PgVectorizer e como usá-lo através do . Leia os documentos biblioteca Python timescale_vector : siga este tutorial passo a passo sobre como criar, incorporar e armazenar dados de postagem de blog a partir de uma tabela PostgreSQL usando os métodos discutidos nesta postagem de blog. Tutorial – Incorporando dados de blog no PostgreSQL Usamos LangChain para ilustrar a análise de documentos e a criação de incorporação em nossos exemplos acima. Saiba mais sobre como usá-lo com o Timescale Vector para armazenamento de vetores, pesquisa por similaridade e pesquisa híbrida. LangChain e vetor de escala de tempo : Escrito por Matvey Arye. Também publicado aqui.