企业官网建设流程全解析

用 pgai Vectorizer 自动化 AI 嵌入：从文本列到语义搜索的声明式向量管理

【免费下载链接】pgaiA suite of tools to develop RAG, semantic search, and other AI applications more easily with PostgreSQL项目地址: https://gitcode.com/GitHub_Trending/pg/pgai

本文基于 pgai 仓库中的 Vectorizer 概览文档编写，完整讲解如何用一条 SQL 声明式地为 PostgreSQL 表的文本列建立、维护并查询向量嵌入（embedding）：涵盖嵌入提供商与 API Key 配置、ai.create_vectorizer各参数（loading、embedding、formatting、chunking、indexing、destination、scheduling）的实操用法、语义搜索查询写法、嵌入存储表结构与状态监控。读完后你可以直接在自己的数据库中落地一个自动同步的向量索引，并能结合仓库源码理解触发器、队列表与后台 worker 的底层实现。

一、Vectorizer 解决什么问题

向量嵌入（vector embedding）把文本压缩成富含语义的数值向量，使"意思相近但用词完全不同"的内容也能被检索到，这是传统关键词搜索做不到的。PostgreSQL 这类向量数据库擅长存储和查询这些向量，但真正的痛点在于让向量与源数据保持同步：源行插入、更新、删除时，对应的嵌入需要被生成、更新和清理，而这通常要靠开发者自己写应用层代码维护。

pgai Vectorizer 把嵌入当成一种声明式的、类似 DDL 的数据库特性来处理（类比索引，但更灵活：可以只对行中的一部分数据建立嵌入）。你在数据库里执行一条SELECT ai.create_vectorizer(...)，系统就帮你：

• 指定任意文本列（通过可定制的规则）作为嵌入来源；如果是嵌入 PDF 等二进制文档，则参考文档中的文档嵌入指南；
• 自动生成并维护可搜索的嵌入表；
• 让嵌入与源数据异步持续保持同步；
• 提供一个把基表与嵌入无缝 JOIN 起来的视图。

为了让嵌入生成高效且能扛住 LLM 端点偶发故障，嵌入生成本身由一个后台 worker执行：在托管服务（如 Timescale Cloud）的数据库中创建 Vectorizer 后，worker 自动运行并在后台创建和同步嵌入；在自托管 Postgres（或其他云厂商的 RDS 等）上，则需要你自己运行vectorizer worker 来处理各 vectorizer。

从源码结构看，整个数据流是：ai.create_vectorizer（定义在 012-vectorizer-api.sql）在源表上创建触发器和队列表，行变更事件写入队列；worker（Python 侧实现在 worker.py）轮询队列，执行"读取 → 分块 → 格式化 → 嵌入 → 写入"流水线。触发器与队列表的构建逻辑见 011-vectorizer-int.sql 中的_vectorizer_build_trigger_definition和_vectorizer_create_queue_table：INSERT/UPDATE 时把主键值插入队列表（UPDATE 时仅当"相关列"发生变化才入队），DELETE 时直接从目标嵌入表删除对应行，TRUNCATE 时同时清空目标表和队列表。这也解释了为什么嵌入与源数据能自动保持一致——同步的"记账"全部在数据库内完成，worker 只负责消耗队列。

要快速体验，可参考Vectorizer quick start（提供预置的 Docker 开发环境）；完整的技术规格见 Vectorizer API reference。

二、选择嵌入提供商并配置 API Key

Vectorizer 目前以一等公民身份支持以下嵌入提供商：

• Ollama
• Voyage AI
• OpenAI

此外，通过 LiteLLM 提供商还支持：

• Cohere
• HuggingFace Inference Endpoints
• Mistral
• Azure OpenAI
• AWS Bedrock
• Vertex AI

使用外部嵌入服务需要配置 API Key。可以存储多个 Key：给每个 Key 起一个名字，然后在 Vectorizer 配置的embedding段中引用它。默认 Key 名与提供商默认名一致：

源码印证了这些默认值：004-embedding.sql 中ai.embedding_openai的api_key_name参数默认即'OPENAI_API_KEY'，ai.embedding_voyageai默认'VOYAGE_API_KEY'，ai.embedding_litellm则允许通过api_key_name自定义（默认 null，由 LiteLLM 自身解析）。四种嵌入配置构造器为：

ai.embedding_openai(model, dimensions, chat_user, api_key_name, base_url) ai.embedding_ollama(model, dimensions, base_url, options, keep_alive) ai.embedding_voyageai(model, dimensions, input_type, api_key_name) -- input_type 只能是 'query' 或 'document' ai.embedding_litellm(model, dimensions, api_key_name, extra_options)

Key 的具体设置方式取决于部署形态：

Timescale Cloud

1. 在 Timescale Console 的项目设置页点击AI Model API Keys；
2. 点击Add AI Model API Keys，填入 Key 后点击Add API key。

此时 API Key 安全地存储在 Timescale Cloud 侧，而不是你的数据库里。

自托管 Postgres

设置一个与 Key 名相同的环境变量（供 vectorizer worker 进程读取），例如：

export OPENAI_API_KEY="Your OpenAI API key"

三、定义 Vectorizer

先看一个示例源表：

CREATE TABLE blog( id SERIAL PRIMARY KEY, title TEXT, authors TEXT, contents TEXT, metadata JSONB );

让系统为这张表自动生成并更新嵌入，只需一条 SQL：

SELECT ai.create_vectorizer( 'blog'::regclass, name => 'blog_embeddings', -- 可选的自定义名称，便于引用 loading => ai.loading_column('contents'), embedding => ai.embedding_ollama('nomic-embed-text', 768), destination => ai.destination_table('blog_contents_embeddings') );

该示例使用本地 Ollama 实例上的nomic-embed-text模型。其他提供商的嵌入配置写法见 API reference 的 embedding 配置一节。

各配置段的作用与底层细节（均对应 idempotent SQL 文件 中的函数定义）：

• loading（008-loading.sql）：指定嵌入数据来源。ai.loading_column('contents')从contents列读取文本；ai.loading_uri则从 S3 等本地或远程桶加载外部文档。注意loading_column还有一个retries参数，默认值为 6（加载失败重试次数）；源码中的类型校验要求列类型为text、varchar、char、bpchar或bytea，loading_uri另支持aws_role_arn参数用于以指定 IAM 角色访问 S3（校验格式必须为arn:aws:iam::%:role/%）。更完整的文档嵌入方案见文档嵌入指南。
• embedding（004-embedding.sql）：嵌入模型与维度。
• chunking（001-chunking.sql）：分块策略。默认实现为ai.chunking_character_text_splitter，参数默认值为chunk_size => 800、chunk_overlap => 400、separator => E'\n\n'；另有ai.chunking_recursive_character_text_splitter（按['\n\n', '\n', '.', '?', '!', ' ', '']逐级分隔符递归切分）和ai.chunking_none。
• indexing、scheduling、destination：见后文各节。

分块的意义：如果contents字段很长，它会被拆成多个 chunk，一篇博客文章就会产生多条嵌入。分块保证每条嵌入语义自洽——可以把它理解为"一次嵌入一个段落"。但切分也可能丢失上下文，缓解办法是用formatting参数把行数据重新注入每个 chunk（见下一节）。

命名规则（源码 010-destination.sql 的_evaluate_destination函数）：当destination_table只传一个位置参数时，系统将其作为destination，自动派生出目标存储表名<destination>_store（本例即blog_contents_embeddings_store）和同名视图blog_contents_embeddings；target_schema默认为源表所在 schema。

调度行为：在 Timescale Cloud 上，vectorizer 由 TimescaleDB 后台任务自动创建并按每 5 分钟一次的周期调度运行；自托管场景下需要手动运行 vectorizer-worker 来创建并驱动 vectorizer。

四、向 Vectorizer chunk 注入上下文

分块会丢失上下文，比如你可能希望每个 chunk 都带上文章的标题和作者。formatting参数使用Python 模板字符串实现这一点：模板可以访问源行的所有列，外加一个包含 chunk 文本的特殊变量$chunk。

SELECT ai.create_vectorizer( 'blog'::regclass, loading => ai.loading_column('contents'), embedding => ai.embedding_ollama('nomic-embed-text', 768), formatting => ai.formatting_python_template('$title - by $author - $chunk'), destination => ai.destination_table('blog_contents_embeddings') );

默认格式字符串就是'$chunk'。注意：如果注入了标题等前缀，格式化后的文本变长，可能需要调小chunk_size以确保格式化后的文本仍在 token 限制内。

源码层面（002-formatting.sql）有两处值得知道的硬校验：

1. 模板必须包含$chunk占位符，否则create_vectorizer直接抛异常；
2. 源表不能存在名为chunk的列，因为视图会把源表的chunk列与嵌入侧的chunk字段混淆。

五、查询嵌入：语义搜索

create_vectorizer会生成一个与view_name同名的视图（本例为blog_contents_embeddings），其中包含 blog 表的全部嵌入。由于通常每条源文档对应多个 chunk，因此每篇博客在视图里会有多行。

视图包含 blog 表的所有列，另加以下列：

找与查询最接近的嵌入，用这条标准 SQL：

SELECT chunk, embedding <=> <query embedding> as distance FROM blog_contents_embeddings ORDER BY distance LIMIT 10;

<=>运算符计算查询向量与每行嵌入向量的距离，这是最简的语义搜索写法。也可以在数据库内部为查询文本生成嵌入：pgai 的 PostgreSQL extension 提供了ai.ollama_embed等函数。

结合元数据过滤：视图携带源表所有列，任意加 WHERE 条件即可。例如按部门过滤（假设metadata为 JSONB）：

SELECT chunk, embedding <=> <query embedding> as distance FROM blog_contents_embeddings WHERE metadata->>'department' = 'finance' ORDER BY distance LIMIT 10;

按作者搜索同理：

SELECT chunk, embedding <=> <query embedding> as distance, author FROM blog_contents_embeddings WHERE author = 'Bulgakov' ORDER BY distance LIMIT 10;

SQLAlchemy 用法：如果应用侧用 SQLAlchemy 映射模型，可以用vectorizer_relationship声明嵌入关系，例如：

class Wiki(Base): __tablename__ = "wiki" id: Mapped[int] = mapped_column(primary_key=True) url: Mapped[str] title: Mapped[str] text: Mapped[str] # 为 text 字段添加向量嵌入关系 text_embeddings = vectorizer_relationship( target_table='wiki_embeddings', dimensions=384 )

然后按余弦距离排序做语义检索：

async def _find_relevant_chunks(client: ollama.AsyncClient, query: str, limit: int = 2) -> WikiSearchResult: response = await client.embed(model="all-minilm", input=query) embedding = response.embeddings[0] with Session(engine) as session: result = session.query( Wiki, Wiki.text_embeddings.embedding.cosine_distance(embedding).label('distance') ).join(Wiki.text_embeddings).order_by( 'distance' ).limit(limit).all() return result

查询上还可以叠加任意其他过滤条件。

六、提升查询性能：为嵌入列建向量索引

在嵌入列上建向量索引可显著改善查询性能。在 Timescale Cloud 上，当向量数据达到 10 万行后会自动创建 vectorscale 索引；该行为可配置，也可以显式指定其他索引类型。例如使用 HNSW 索引：

SELECT ai.create_vectorizer( 'blog'::regclass, loading => ai.loading_column('contents'), embedding => ai.embedding_ollama('nomic-embed-text', 768), formatting => ai.formatting_python_template('$title - by $author - $chunk'), indexing => ai.indexing_hnsw(min_rows => 100000, opclass => 'vector_l2_ops'), destination => ai.destination_table('blog_contents_embeddings') );

源码层面（005-indexing.sql）可用的索引配置函数有：

• ai.indexing_none()：不建向量索引；
• ai.indexing_default()：默认策略，由 GUC 参数ai.indexing_default解析为 diskann 或 hnsw；
• ai.indexing_hnsw(min_rows => 100000, opclass => 'vector_cosine_ops', m, ef_construction, create_when_queue_empty => true)：min_rows表示目标表达到该行数才建索引（默认 100000）；opclass默认vector_cosine_ops。需要注意，当前源码中opclass的校验白名单为vector_ip_ops、vector_cosine_ops、vector_l1_ops，选用其他 opclass（如vector_l2_ops）时请以你所安装版本的支持情况为准；
• ai.indexing_diskann(min_rows, storage_layout, num_neighbors, search_list_size, ...)：storage_layout只接受memory_optimized或plain。

索引由后台任务周期性触发创建，具体决策逻辑在 011-vectorizer-int.sql 的_vectorizer_should_create_vector_index：先检查索引是否已存在，再看create_when_queue_empty（默认 true）——队列非空则跳过，避免在数据还在大量写入时建索引；随后统计目标表行数（用LIMIT min_rows的高效计数），达到min_rows才执行建索引。真正建索引的_vectorizer_create_vector_index还会先抢一个针对目标表 OID 的事务级 advisory lock，防止并发进程重复建索引。

注意：索引依赖周期性运行的后台任务，因此若调度被禁用（自托管安装的默认状态），索引不会自动创建，需要手动建立。

七、控制 Vectorizer 的运行时机

在 Timescale Cloud 上，用**调度（scheduling）**控制 vectorizer 何时运行：一个计划任务定期检查是否有待处理工作，有则触发云函数去嵌入数据。默认使用 TimescaleDB 后台任务，周期为 5 分钟。当表足够大后，调度还会顺带处理嵌入列的索引创建。

自托管时，vectorizer worker 用轮询机制检查是否有工作，因此调度既不需要也被默认禁用。源码印证：003-scheduling.sql 提供ai.scheduling_none()、ai.scheduling_default()（由 GUCai.scheduling_default解析，默认落到 none）和ai.scheduling_timescaledb(schedule_interval => '5m', initial_start, fixed_schedule, timezone)。

再次强调：调度禁用时索引不会自动创建，需要手动建。

八、嵌入存储表

视图基于一张存储嵌入的表，上例中名为blog_contents_embeddings_store。直接查询这张表（而不是视图）在某些场景下更高效。表结构为：

CREATE TABLE blog_contents_embeddings_store( embedding_uuid UUID NOT NULL PRIMARY KEY DEFAULT gen_random_uuid(), id INT, -- 引用 blog 表的主键 chunk_seq INT NOT NULL, chunk TEXT NOT NULL, embedding VECTOR(768) NOT NULL, UNIQUE (id, chunk_seq), FOREIGN KEY (id) REFERENCES public.blog(id) ON DELETE CASCADE );

这与源码_vectorizer_create_target_table（011-vectorizer-int.sql）动态生成的 DDL 一致：embedding_uuid主键（默认gen_random_uuid()）、源表主键列（本例为id，支持复合主键）、chunk_seq、chunk、embedding vector(维度)，以及(主键列, chunk_seq)的唯一约束——它保证了同一源行的同一 chunk 不会产生重复嵌入（重放队列也不会出错）。源行删除由触发器从目标表级联清除。

两种嵌入存储目的地

Vectorizer 支持两种方式存储嵌入，选择依据是：

• 因为分块需要每行多个嵌入（最常见）→ 选table目的地；
• 只需要每行一个嵌入（源文本很短，或上游已完成分块、源表存的就是 chunk）→ 选column目的地。

1. Table 目的地（默认）

创建独立的嵌入表 + 与源表 JOIN 的视图：

SELECT ai.create_vectorizer( 'blog'::regclass, name => 'blog_vectorizer', loading => ai.loading_column('contents'), embedding => ai.embedding_ollama('nomic-embed-text', 768), destination => ai.destination_table( target_schema => 'public', target_table => 'blog_embeddings_store', view_name => 'blog_embeddings' ), );

适用场景：

• 每行需要多个嵌入（分块）；
• 需要拆分的大文本字段；
• 正在向量化文档（文档通常必须分块）。

2. Column 目的地

直接把嵌入列加到源表上。这只适用于 vectorizer 不做分块的场景——要求源数据与嵌入一一对应，适合源文本较短（例如上游管道已分好块）的情况。工作流程：应用往表里插入数据时嵌入列为 NULL，vectorizer 读行、生成嵌入后回填该列。

SELECT ai.create_vectorizer( 'product_descriptions'::regclass, name => 'product_descriptions_vectorizer', loading => ai.loading_column('description'), embedding => ai.embedding_openai('text-embedding-3-small', 768), chunking => ai.chunking_none(), -- column 目的地必须 destination => ai.destination_column('description_embedding') );

适用场景：

• 严格需要每行恰好一个嵌入；
• 不需要分块的短文本；
• 应用入库前已完成分块；
• 不想创建额外的数据库对象。

注意：column 目的地必须配合ai.chunking_none()。源码 010-destination.sql 的_validate_destination函数会强制这一约束，否则抛出chunking must be none for column destination；且_vectorizer_add_embedding_column会先检查列是否已存在，已存在则只发 NOTICE 跳过，重复执行create_vectorizer不会报错。

九、监控 Vectorizer

由于嵌入是异步创建的，在写入源数据到嵌入可用之间存在延迟。用vectorizer_status视图查看状态：

SELECT * FROM ai.vectorizer_status;

示例输出：

pending_items表示仍等待生成嵌入的条目数。为避免在大积压时逐行计数拖慢查询，当积压超过 10,000 时，该值直接返回 bigint 最大值（9223372036854775807）而不是精确计数。源码印证（012-vectorizer-api.sql 的ai.vectorizer_queue_pending）：非精确模式下用SELECT count(*) FROM (SELECT 1 FROM 队列表 LIMIT 10001)，结果若为 10001 即替换为9223372036854775807；另外视图对pending_items做了权限保护——当前用户无队列表 SELECT 权限时该列显示为 NULL。

如需精确值，调用ai.vectorizer_queue_pending函数并传exact_count => true（默认false）：

select ai.vectorizer_queue_pending(1, exact_count=>true);

该函数还有按 vectorizer 名称的重载（exact_count参数同上）。另外，worker 进程的运行状态（心跳计数、成功/错误计数、最后错误信息等）记录在ai.vectorizer_worker_process与ai.vectorizer_worker_progress表中，相关函数定义见 013-worker-tracking.sql，可用于排查 worker 层面的故障。

十、小结与延伸阅读

pgai Vectorizer 的核心价值在于把"嵌入与源数据同步"这一工程负担从应用层收进了数据库内：声明一条create_vectorizer之后，触发器负责记账（INSERT/UPDATE/DELETE/TRUNCATE 全覆盖）、队列负责解耦、worker 负责调用嵌入模型，你只需面对一张 JOIN 视图写语义搜索 SQL。进一步学习建议按以下顺序：

• Vectorizer quick start：Docker 环境下的快速上手；
• Vectorizer API reference：全部配置段与函数的完整规格；
• vectorizer worker：自托管部署下 worker 的安装与配置；
• 文档嵌入指南：PDF 等二进制文档的嵌入流程（配合ai.loading_uri）。

【免费下载链接】pgaiA suite of tools to develop RAG, semantic search, and other AI applications more easily with PostgreSQL项目地址: https://gitcode.com/GitHub_Trending/pg/pgai