自动嵌入

向量嵌入让 Postgres 拥有强大的语义搜索能力，但将它们与内容一起管理传统上一直很复杂。本指南展示了如何使用 Supabase 的Edge Functions、pgmq、pg_net和pg_cron来自动生成和更新嵌入。

🌐 Vector embeddings enable powerful semantic search capabilities in Postgres, but managing them alongside your content has traditionally been complex. This guide demonstrates how to automate embedding generation and updates using Supabase Edge Functions, pgmq, pg_net, and pg_cron.

了解挑战 #

🌐 Understanding the challenge

在用 pgvector 实现语义搜索时，开发者通常需要：

🌐 When implementing semantic search with pgvector, developers typically need to:

1. 通过外部 API（比如 OpenAI）生成嵌入
2. 把这些嵌入和内容一起存起来
3. 当内容变化时保持嵌入同步
4. 处理嵌入生成过程中的失败和重试

虽然 Postgres 的 全文搜索 可以通过同步调用 to_tsvector 和 触发器 来内部处理这个问题，但语义搜索需要异步向像 OpenAI 这样的供应商的 API 调用来生成向量嵌入。本指南演示了如何使用触发器、队列和 Supabase Edge Functions 来弥合这个差距。

🌐 While Postgres full-text search can handle this internally through synchronous calls to to_tsvector and triggers, semantic search requires asynchronous API calls to a provider like OpenAI to generate vector embeddings. This guide demonstrates how to use triggers, queues, and Supabase Edge Functions to bridge this gap.

了解架构 #

🌐 Understanding the architecture

我们将利用以下 Postgres 和 Supabase 功能来创建自动化嵌入系统：

🌐 We'll leverage the following Postgres and Supabase features to create the automated embedding system:

1. pgvector：存储和查询向量嵌入
2. pgmq：排队等待生成嵌入的请求进行处理和重试
3. pg_net：直接从 Postgres 处理对 Edge Functions 的异步 HTTP 请求
4. pg_cron：自动处理和重试嵌入生成
5. 触发器：检测内容变化并排队生成嵌入请求
6. 边缘函数：通过像 OpenAI 这样的 API 生成嵌入（可自定义）

我们将把系统设计成这样：

🌐 We'll design the system to:

1. 尽量通用，这样可以用于任何表格和内容。这允许你在多个地方配置嵌入，每个地方都可以自定义生成嵌入时使用的输入。所有这些都会使用相同的队列基础设施和 Edge Function 来生成嵌入。
2. 优雅地处理失败，通过重试失败的任务并提供每个任务状态的详细信息。

实现 #

🌐 Implementation

我们先从搭建处理嵌入生成请求的排队和执行的基础设施开始。然后我们会创建一个示例表，当内容被插入或更新时，触发器会将这些嵌入请求加入队列。

🌐 We'll start by setting up the infrastructure needed to queue and process embedding generation requests. Then we'll create an example table with triggers to enqueue these embedding requests whenever content is inserted or updated.

步骤1：启用扩展 #

🌐 Step 1: Enable extensions

首先，启用所需的扩展：

🌐 First, enable the required extensions:

1
-- For vector operations
2
create extension if not exists vector
3
with
4
  schema extensions;
5
6
-- For queueing and processing jobs
7
-- (pgmq will create its own schema)
8
create extension if not exists pgmq;
9
10
-- For async HTTP requests
11
create extension if not exists pg_net
12
with
13
  schema extensions;
14
15
-- For scheduled processing and retries
16
-- (pg_cron will create its own schema)
17
create extension if not exists pg_cron;
18
19
-- For clearing embeddings during updates
20
create extension if not exists hstore
21
with
22
  schema extensions;

步骤 2：创建工具函数 #

🌐 Step 2: Create utility functions

在我们设置嵌入逻辑之前，我们需要先创建一些工具函数：

🌐 Before we set up our embedding logic, we need to create some utility functions:

1
-- Schema for utility functions
2
create schema util;
3
4
-- Utility function to get the Supabase project URL (required for Edge Functions)
5
create function util.project_url()
6
returns text
7
language plpgsql
8
security definer
9
as $$
10
declare
11
  secret_value text;
12
begin
13
  -- Retrieve the project URL from Vault
14
  select decrypted_secret into secret_value from vault.decrypted_secrets where name = 'project_url';
15
  return secret_value;
16
end;
17
$$;
18
19
-- Generic function to invoke any Edge Function
20
create or replace function util.invoke_edge_function(
21
  name text,
22
  body jsonb,
23
  timeout_milliseconds int = 5 * 60 * 1000  -- default 5 minute timeout
24
)
25
returns void
26
language plpgsql
27
as $$
28
declare
29
  headers_raw text;
30
  auth_header text;
31
begin
32
  -- If we're in a PostgREST session, reuse the request headers for authorization
33
  headers_raw := current_setting('request.headers', true);
34
35
  -- Only try to parse if headers are present
36
  auth_header := case
37
    when headers_raw is not null then
38
      (headers_raw::json->>'authorization')
39
    else
40
      null
41
  end;
42
43
  -- Perform async HTTP request to the edge function
44
  perform net.http_post(
45
    url => util.project_url() || '/functions/v1/' || name,
46
    headers => jsonb_build_object(
47
      'Content-Type', 'application/json',
48
      'Authorization', auth_header
49
    ),
50
    body => body,
51
    timeout_milliseconds => timeout_milliseconds
52
  );
53
end;
54
$$;
55
56
-- Generic trigger function to clear a column on update
57
create or replace function util.clear_column()
58
returns trigger
59
language plpgsql as $$
60
declare
61
    clear_column text := TG_ARGV[0];
62
begin
63
    NEW := NEW #= hstore(clear_column, NULL);
64
    return NEW;
65
end;
66
$$;

我们在这里创建：

🌐 Here we create:

• 一个用于存储实用函数的模式 util。
• 一个从 Vault 获取 Supabase 项目 URL 的函数。接下来我们会添加这个秘密。
• 一个通用函数，用于调用任何带有指定名称和请求体的边缘函数。
• 一个通用的触发器函数，用于在更新时清除某一列。这个函数接收列名作为参数，并将 NEW 记录中的该列设置为 NULL。我们稍后会讲解如何使用这个函数。

每个项目都有一个独特的 API URL，用于调用 Edge 功能。根据你的环境将项目 URL 密钥添加到 Vault 中。

🌐 Every project has a unique API URL that is required to invoke Edge Functions. Add the project URL secret to Vault depending on your environment.

在使用本地 Supabase 堆栈时，在你的 supabase/seed.sql 文件中添加以下内容：

🌐 When working with a local Supabase stack, add the following to your supabase/seed.sql file:

1
select
2
  vault.create_secret('http://api.supabase.internal:8000', 'project_url');

在部署到云平台时，打开 SQL 编辑器 并运行以下命令，将 <project-url> 替换为你的 项目 API URL ：

🌐 When deploying to the cloud platform, open the SQL editor and run the following, replacing <project-url> with your project's API URL:

1
select
2
  vault.create_secret('<project-url>', 'project_url');

第3步：创建队列和触发器 #

🌐 Step 3: Create queue and triggers

我们的目标是在表中插入或更新内容时自动生成嵌入。我们可以使用触发器和队列来实现这一点。我们的方法是在表中插入或更新记录时自动将嵌入任务加入队列，然后使用定时任务异步处理它们。如果任务失败，它会留在队列中，并在下一次计划任务时重试。

🌐 Our goal is to automatically generate embeddings whenever content is inserted or updated within a table. We can use triggers and queues to achieve this. Our approach is to automatically queue embedding jobs whenever records are inserted or updated in a table, then process them asynchronously using a cron job. If a job fails, it will remain in the queue and be retried in the next scheduled task.

首先我们创建一个 pgmq 队列来处理嵌入请求：

🌐 First we create a pgmq queue for processing embedding requests:

1
-- Queue for processing embedding jobs
2
select pgmq.create('embedding_jobs');

接下来我们创建一个触发器函数来排队嵌入任务。我们将使用这个函数来处理插入和更新事件：

🌐 Next we create a trigger function to queue embedding jobs. We'll use this function to handle both insert and update events:

1
-- Generic trigger function to queue embedding jobs
2
create or replace function util.queue_embeddings()
3
returns trigger
4
language plpgsql
5
security definer
6
set search_path = ''
7
as $$
8
declare
9
  content_function text = TG_ARGV[0];
10
  embedding_column text = TG_ARGV[1];
11
begin
12
  perform pgmq.send(
13
    queue_name => 'embedding_jobs',
14
    msg => jsonb_build_object(
15
      'id', NEW.id,
16
      'schema', TG_TABLE_SCHEMA,
17
      'table', TG_TABLE_NAME,
18
      'contentFunction', content_function,
19
      'embeddingColumn', embedding_column
20
    )
21
  );
22
  return NEW;
23
end;
24
$$;

我们的 util.queue_embeddings 触发器函数是通用的，可以用于任何表和内容函数。它接受两个参数：

🌐 Our util.queue_embeddings trigger function is generic and can be used with any table and content function. It accepts two arguments:

1. content_function：返回要嵌入的文本内容的函数名称。该函数应该接受一行作为输入并返回文本（参见 embedding_input 示例）。

   这让你可以自定义传给嵌入模型的文本输入——比如，你可以把多列拼接在一起，比如 title 和 content，然后把结果用作输入。

2. embedding_column：存放嵌入的目标列名称。

请注意，util.queue_embeddings 触发器函数需要一个 for each row 子句才能正常工作。查看 使用方法 了解如何在你的表中使用这个触发器函数的示例。

🌐 Note that the util.queue_embeddings trigger function requires a for each row clause to work correctly. See Usage for an example of how to use this trigger function with your table.

接下来我们将创建一个函数来处理嵌入任务。这个函数会从队列中读取任务，将它们分批处理，然后调用 Edge 函数来生成嵌入。我们会使用 pg_cron 来安排这个函数每隔 10 秒运行一次。

🌐 Next we'll create a function to process the embedding jobs. This function will read jobs from the queue, group them into batches, and invoke the Edge Function to generate embeddings. We'll use pg_cron to schedule this function to run every 10 seconds.

1
-- Function to process embedding jobs from the queue
2
create or replace function util.process_embeddings(
3
  batch_size int = 10,
4
  max_requests int = 10,
5
  timeout_milliseconds int = 5 * 60 * 1000 -- default 5 minute timeout
6
)
7
returns void
8
language plpgsql
9
as $$
10
declare
11
  job_batches jsonb[];
12
  batch jsonb;
13
begin
14
  with
15
    -- First get jobs and assign batch numbers
16
    numbered_jobs as (
17
      select
18
        message || jsonb_build_object('jobId', msg_id) as job_info,
19
        (row_number() over (order by 1) - 1) / batch_size as batch_num
20
      from pgmq.read(
21
        queue_name => 'embedding_jobs',
22
        vt => timeout_milliseconds / 1000,
23
        qty => max_requests * batch_size
24
      )
25
    ),
26
    -- Then group jobs into batches
27
    batched_jobs as (
28
      select
29
        jsonb_agg(job_info) as batch_array,
30
        batch_num
31
      from numbered_jobs
32
      group by batch_num
33
    )
34
  -- Finally aggregate all batches into array
35
  select coalesce(array_agg(batch_array), array[]::jsonb[])
36
  from batched_jobs
37
  into job_batches;
38
39
  -- Invoke the embed edge function for each batch
40
  foreach batch in array job_batches loop
41
    perform util.invoke_edge_function(
42
      name => 'embed',
43
      body => batch,
44
      timeout_milliseconds => timeout_milliseconds
45
    );
46
  end loop;
47
end;
48
$$;
49
50
-- Schedule the embedding processing
51
select
52
  cron.schedule(
53
    'process-embeddings',
54
    '10 seconds',
55
    $$
56
    select util.process_embeddings();
57
    $$
58
  );

关于这种方法的常见问题：

🌐 Common questions about this approach:

为什么不在一次 Edge Function 请求中生成所有嵌入呢？ #

🌐 Why not generate all embeddings in a single Edge Function request?

虽然这是可能的，但这可能导致处理时间很长并可能超时。批处理让我们可以同时处理多个嵌入，并更有效地应对失败。

🌐 While this is possible, it can lead to long processing times and potential timeouts. Batching allows us to process multiple embeddings concurrently and handle failures more effectively.

为什么不每行只放一个请求？ #

🌐 Why not one request per row?

这种方法可能会导致 API 限速和性能问题。批量处理在效率和可靠性之间提供了一种平衡。

🌐 This approach can lead to API rate limiting and performance issues. Batching provides a balance between efficiency and reliability.

为什么要排队请求而不是立即处理它们？ #

🌐 Why queue requests instead of processing them immediately?

排队让我们能够优雅地处理失败、重试请求，并更有效地管理并发。具体来说，我们使用 pgmq 的可见性超时来确保失败的请求会被重试。

🌐 Queuing allows us to handle failures gracefully, retry requests, and manage concurrency more effectively. Specifically we are using pgmq's visibility timeouts to ensure that failed requests are retried.

可见性超时是怎么工作的？ #

🌐 How do visibility timeouts work?

每次我们从队列中读取一条消息时，都会设置一个可见性超时，这会告诉 pgmq 在一定时间内将消息对其他读取者隐藏。如果 Edge Function 在这段时间内处理消息失败，消息会再次变为可见，并由下一个计划任务重试。

🌐 Every time we read a message from the queue, we set a visibility timeout which tells pgmq to hide the message from other readers for a certain period. If the Edge Function fails to process the message within this period, the message becomes visible again and will be retried by the next scheduled task.

我们怎么处理重试？ #

🌐 How do we handle retries?

我们使用 pg_cron 来安排一个任务，该任务从队列中读取消息并处理它们。如果 Edge Function 处理消息失败，消息将在超时后再次可见，并可以被下一个计划任务重试。

🌐 We use pg_cron to schedule a task that reads messages from the queue and processes them. If the Edge Function fails to process a message, it becomes visible again after a timeout and can be retried by the next scheduled task.

处理用10秒的间隔合适吗？ #

🌐 Is 10 seconds a good interval for processing?

这个间隔是一个不错的起点，但你可能需要根据你的工作量和生成嵌入所需的时间进行调整。你可以调整 batch_size、max_requests 和 timeout_milliseconds 参数来优化性能。

🌐 This interval is a good starting point, but you may need to adjust it based on your workload and the time it takes to generate embeddings. You can adjust the batch_size, max_requests, and timeout_milliseconds parameters to optimize performance.

第4步：创建边缘函数 #

🌐 Step 4: Create the Edge Function

最后我们将创建 Edge Function 来生成嵌入。在这个例子中，我们会使用 OpenAI 的 API，但你也可以用其他任何嵌入生成服务替代它。

🌐 Finally we'll create the Edge Function to generate embeddings. We'll use OpenAI's API in this example, but you can replace it with any other embedding generation service.

使用 Supabase CLI 创建一个新的 Edge 函数：

🌐 Use the Supabase CLI to create a new Edge Function:

1
supabase functions new embed

这将创建一个名为 supabase/functions/embed 的新目录，其中有一个 index.ts 文件。用以下内容替换这个文件的内容：

🌐 This will create a new directory supabase/functions/embed with an index.ts file. Replace the contents of this file with the following:

supabase/functions/embed/index.ts:

1
// Setup type definitions for built-in Supabase Runtime APIs
2
import 'jsr:@supabase/functions-js/edge-runtime.d.ts'
3
4
// We'll make a direct Postgres connection to update the document
5
import postgres from 'https://deno.land/x/postgresjs@v3.4.5/mod.js'
6
// We'll use the OpenAI API to generate embeddings
7
import OpenAI from 'jsr:@openai/openai'
8
import { z } from 'npm:zod'
9
10
// Initialize OpenAI client
11
const openai = new OpenAI({
12
  // We'll need to manually set the `OPENAI_API_KEY` environment variable
13
  apiKey: Deno.env.get('OPENAI_API_KEY'),
14
})
15
16
// Initialize Postgres client
17
const sql = postgres(
18
  // `SUPABASE_DB_URL` is a built-in environment variable
19
  Deno.env.get('SUPABASE_DB_URL')!
20
)
21
22
const jobSchema = z.object({
23
  jobId: z.number(),
24
  id: z.number(),
25
  schema: z.string(),
26
  table: z.string(),
27
  contentFunction: z.string(),
28
  embeddingColumn: z.string(),
29
})
30
31
const failedJobSchema = jobSchema.extend({
32
  error: z.string(),
33
})
34
35
type Job = z.infer<typeof jobSchema>
36
type FailedJob = z.infer<typeof failedJobSchema>
37
38
type Row = {
39
  id: string
40
  content: unknown
41
}
42
43
const QUEUE_NAME = 'embedding_jobs'
44
45
// Listen for HTTP requests
46
Deno.serve(async (req) => {
47
  if (req.method !== 'POST') {
48
    return new Response('expected POST request', { status: 405 })
49
  }
50
51
  if (req.headers.get('content-type') !== 'application/json') {
52
    return new Response('expected json body', { status: 400 })
53
  }
54
55
  // Use Zod to parse and validate the request body
56
  const parseResult = z.array(jobSchema).safeParse(await req.json())
57
58
  if (parseResult.error) {
59
    return new Response(`invalid request body: ${parseResult.error.message}`, {
60
      status: 400,
61
    })
62
  }
63
64
  const pendingJobs = parseResult.data
65
66
  // Track jobs that completed successfully
67
  const completedJobs: Job[] = []
68
69
  // Track jobs that failed due to an error
70
  const failedJobs: FailedJob[] = []
71
72
  async function processJobs() {
73
    let currentJob: Job | undefined
74
75
    while ((currentJob = pendingJobs.shift()) !== undefined) {
76
      try {
77
        await processJob(currentJob)
78
        completedJobs.push(currentJob)
79
      } catch (error) {
80
        failedJobs.push({
81
          ...currentJob,
82
          error: error instanceof Error ? error.message : JSON.stringify(error),
83
        })
84
      }
85
    }
86
  }
87
88
  try {
89
    // Process jobs while listening for worker termination
90
    await Promise.race([processJobs(), catchUnload()])
91
  } catch (error) {
92
    // If the worker is terminating (e.g. wall clock limit reached),
93
    // add pending jobs to fail list with termination reason
94
    failedJobs.push(
95
      ...pendingJobs.map((job) => ({
96
        ...job,
97
        error: error instanceof Error ? error.message : JSON.stringify(error),
98
      }))
99
    )
100
  }
101
102
  // Log completed and failed jobs for traceability
103
  console.log('finished processing jobs:', {
104
    completedJobs: completedJobs.length,
105
    failedJobs: failedJobs.length,
106
  })
107
108
  return new Response(
109
    JSON.stringify({
110
      completedJobs,
111
      failedJobs,
112
    }),
113
    {
114
      // 200 OK response
115
      status: 200,
116
117
      // Custom headers to report job status
118
      headers: {
119
        'content-type': 'application/json',
120
        'x-completed-jobs': completedJobs.length.toString(),
121
        'x-failed-jobs': failedJobs.length.toString(),
122
      },
123
    }
124
  )
125
})
126
127
/**
128
 * Generates an embedding for the given text.
129
 */
130
async function generateEmbedding(text: string) {
131
  const response = await openai.embeddings.create({
132
    model: 'text-embedding-3-small',
133
    input: text,
134
  })
135
  const [data] = response.data
136
137
  if (!data) {
138
    throw new Error('failed to generate embedding')
139
  }
140
141
  return data.embedding
142
}
143
144
/**
145
 * Processes an embedding job.
146
 */
147
async function processJob(job: Job) {
148
  const { jobId, id, schema, table, contentFunction, embeddingColumn } = job
149
150
  // Fetch content for the schema/table/row combination
151
  const [row]: [Row] = await sql`
152
    select
153
      id,
154
      ${sql(contentFunction)}(t) as content
155
    from
156
      ${sql(schema)}.${sql(table)} t
157
    where
158
      id = ${id}
159
  `
160
161
  if (!row) {
162
    throw new Error(`row not found: ${schema}.${table}/${id}`)
163
  }
164
165
  if (typeof row.content !== 'string') {
166
    throw new Error(`invalid content - expected string: ${schema}.${table}/${id}`)
167
  }
168
169
  const embedding = await generateEmbedding(row.content)
170
171
  await sql`
172
    update
173
      ${sql(schema)}.${sql(table)}
174
    set
175
      ${sql(embeddingColumn)} = ${JSON.stringify(embedding)}
176
    where
177
      id = ${id}
178
  `
179
180
  await sql`
181
    select pgmq.delete(${QUEUE_NAME}, ${jobId}::bigint)
182
  `
183
}
184
185
/**
186
 * Returns a promise that rejects if the worker is terminating.
187
 */
188
function catchUnload() {
189
  return new Promise((reject) => {
190
    addEventListener('beforeunload', (ev: any) => {
191
      reject(new Error(ev.detail?.reason))
192
    })
193
  })
194
}

Edge 功能监听来自 pg_net 的 HTTP 请求，并处理每个嵌入任务。它是一个通用的工作器，可以处理任何表和列的嵌入任务。它使用 OpenAI 的 API 生成嵌入，并更新数据库中对应的行。任务处理完成后，它还会从队列中删除该任务。

🌐 The Edge Function listens for incoming HTTP requests from pg_net and processes each embedding job. It is a generic worker that can handle embedding jobs for any table and column. It uses OpenAI's API to generate embeddings and updates the corresponding row in the database. It also deletes the job from the queue once it has been processed.

这个功能被设计用来独立处理多个任务。如果一个任务失败，不会影响其他任务的处理。该功能会返回一个 200 OK 响应，其中包含已完成和失败的任务列表。我们可以利用这些信息来诊断失败的任务。更多详情请参见 故障排除。

🌐 The function is designed to process multiple jobs independently. If one job fails, it will not affect the processing of other jobs. The function returns a 200 OK response with a list of completed and failed jobs. We can use this information to diagnose failed jobs. See Troubleshooting for more details.

你需要设置 OPENAI_API_KEY 环境变量来进行 OpenAI 身份验证。在本地运行 Edge Function 时，你可以把它添加到一个 .env 文件中：

🌐 You will need to set the OPENAI_API_KEY environment variable to authenticate with OpenAI. When running the Edge Function locally, you can add it to a .env file:

.env:

1
OPENAI_API_KEY=your-api-key

当你准备好部署 Edge Function 时，可以使用 Supabase CLI 设置环境变量：

🌐 When you're ready to deploy the Edge Function, set can set the environment variable using the Supabase CLI:

1
supabase secrets set --env-file .env

or

1
supabase secrets set OPENAI_API_KEY=<your-api-key>

或者，你可以用自己的嵌入生成逻辑来替换 generateEmbedding 函数。

🌐 Alternatively, you can replace the generateEmbedding function with your own embedding generation logic.

查看 部署到生产环境 以获取有关如何部署 Edge Function 的更多信息。

🌐 See Deploy to Production for more information on how to deploy the Edge Function.

用法 #

🌐 Usage

基础设施搭建好之后，按照这个示例可以自动为一张文档表生成向量。你也可以用这种方法处理多张表，并根据需要为每次向量生成自定义输入。

🌐 With the infrastructure in place, follow this example to automatically generate embeddings for a table of documents. You can use this approach with multiple tables and customize the input for each embedding generation as needed.

1. 创建一个用来存储带有嵌入的文档的表 #

🌐 1. Create table to store documents with embeddings

我们将建立一个新的 documents 表来存储我们的内容和嵌入：

🌐 We'll set up a new documents table that will store our content and embeddings:

1
-- Table to store documents with embeddings
2
create table documents (
3
  id integer primary key generated always as identity,
4
  title text not null,
5
  content text not null,
6
  embedding halfvec(1536),
7
  created_at timestamp with time zone default now()
8
);
9
10
-- Index for vector search over document embeddings
11
create index on documents using hnsw (embedding halfvec_cosine_ops);

我们的 documents 表存储每个文档的标题和内容以及它的向量嵌入。我们使用 halfvec(1536) 列来存储这些嵌入。

🌐 Our documents table stores the title and content of each document along with its vector embedding. We use a halfvec(1536) column to store the embeddings.

halfvec 是一种 pgvector 数据类型，用于以半精度（16 位）存储浮点值以节省空间。我们的 Edge Function 使用了 OpenAI 的 text-embedding-3-small 模型，该模型生成 1536 维的嵌入，因此我们这里使用相同的维度。根据你的嵌入模型生成的维度数量调整即可。

我们在向量列上使用了一个 HNSW 索引。注意我们选择了 halfvec_cosine_ops 作为索引方法，这意味着我们将来的查询需要使用余弦距离（<=>）来查找相似的嵌入。此外，HNSW 索引支持的 halfvec 向量最大维度为 4000，所以在选择嵌入模型时要记得这一点。如果你的模型生成的嵌入维度超过 4000，则需要在索引之前降低维度。可以参考 Matryoshka 嵌入 来寻找一种缩短维度的解决方案。

🌐 We use an HNSW index on the vector column. Note that we are choosing halfvec_cosine_ops as the index method, which means our future queries will need to use cosine distance (<=>) to find similar embeddings. Also note that HNSW indexes support a maximum of 4000 dimensions for halfvec vectors, so keep this in mind when choosing an embedding model. If your model generates embeddings with more than 4000 dimensions, you will need to reduce the dimensionality before indexing them. See Matryoshka embeddings for a potential solution to shortening dimensions.

另外请注意，表必须有一个名为 id 的主键列，这样我们的触发器才能正确配合 util.queue_embeddings 函数工作，并且我们的 Edge Function 才能更新正确的行。

🌐 Also note that the table must have a primary key column named id for our triggers to work correctly with the util.queue_embeddings function and for our Edge Function to update the correct row.

2. 创建触发器来排队嵌入任务 #

🌐 2. Create triggers to enqueue embedding jobs

现在我们将设置触发器，在内容被插入或更新时排队嵌入任务：

🌐 Now we'll set up the triggers to enqueue embedding jobs whenever content is inserted or updated:

1
-- Customize the input for embedding generation
2
-- e.g. Concatenate title and content with a markdown header
3
create or replace function embedding_input(doc documents)
4
returns text
5
language plpgsql
6
immutable
7
as $$
8
begin
9
  return '# ' || doc.title || E'\n\n' || doc.content;
10
end;
11
$$;
12
13
-- Trigger for insert events
14
create trigger embed_documents_on_insert
15
  after insert
16
  on documents
17
  for each row
18
  execute function util.queue_embeddings('embedding_input', 'embedding');
19
20
-- Trigger for update events
21
create trigger embed_documents_on_update
22
  after update of title, content -- must match the columns in embedding_input()
23
  on documents
24
  for each row
25
  execute function util.queue_embeddings('embedding_input', 'embedding');

我们创建两个触发器：

🌐 We create 2 triggers:

1. embed_documents_on_insert：每当在 documents 表中插入新行时，就会将嵌入任务排入队列。
2. embed_documents_on_update：每当 documents 表中的 title 或 content 列被更新时，就会将嵌入任务加入队列。

这两个触发器都使用相同的 util.queue_embeddings 函数来将嵌入任务排队处理。它们接受两个参数：

🌐 Both of these triggers use the same util.queue_embeddings function that will queue the embedding jobs for processing. They accept 2 arguments:

1. embedding_input：生成嵌入输入的函数的名称。这个函数允许你自定义传给嵌入模型的文本输入（例如，将标题和内容拼接在一起）。函数应该接收单行作为输入，并返回文本。
2. embedding：存放嵌入的目标列名称。

注意，更新触发器只有在 title 或 content 列被更新时才会触发。这是为了避免在其他列更新时对嵌入列进行不必要的更新。确保这些列与 embedding_input 函数中使用的列匹配。

🌐 Note that the update trigger only fires when the title or content columns are updated. This is to avoid unnecessary updates to the embedding column when other columns are updated. Make sure that these columns match the columns used in the embedding_input function.

(可选) 在更新时清除嵌入 #

🌐 (Optional) Clearing embeddings on update

请注意，我们的触发器会在内容更新时排队新的嵌入任务，但不会清除已有的嵌入。这意味着在新的嵌入生成并更新之前，嵌入可能会暂时与内容不同步。

🌐 Note that our trigger will enqueue new embedding jobs when content is updated, but it will not clear any existing embeddings. This means that an embedding can be temporarily out of sync with the content until the new embedding is generated and updated.

如果比起有 任何 嵌入，更重要的是要有 准确的 嵌入，你可以添加另一个触发器来清除现有的嵌入，直到生成新的嵌入为止：

🌐 If it is more important to have accurate embeddings than any embedding, you can add another trigger to clear the existing embedding until the new one is generated:

1
-- Trigger to clear the embedding column on update
2
create trigger clear_document_embedding_on_update
3
  before update of title, content -- must match the columns in embedding_input()
4
  on documents
5
  for each row
6
  execute function util.clear_column('embedding');

util.clear_column 是我们之前创建的一个通用触发器函数，可以用来清除表中的任何列。

• 它接受列名作为参数。这个列必须可以为空。
• 它需要一个带有 for each row 条款的 before 触发器。
• 它需要我们之前创建的 hstore 扩展。

这个例子会在 title 或 content 列更新时清空 embedding 列（注意 of title, content 条款）。这样可以确保 embedding 始终与标题和内容保持同步，但在生成新的 embedding 之前，搜索结果可能会暂时出现空缺。

🌐 This example will clear the embedding column whenever the title or content columns are updated (note the of title, content clause). This ensures that the embedding is always in sync with the title and content, but it will result in temporary gaps in search results until the new embedding is generated.

我们刻意使用 before 触发器，因为它允许我们在记录写入磁盘之前进行修改，从而避免了如果使用 after 触发器时需要的额外 update 语句。

🌐 We intentionally use a before trigger because it allows us to modify the record before it's written to disk, avoiding an extra update statement that would be needed with an after trigger.

3. 插入和更新文档 #

🌐 3. Insert and update documents

插入一个新文档并更新其内容，看看嵌入生成的效果：

🌐 Insert a new document and update its content to see the embedding generation in action:

1
-- Insert a new document
2
insert into documents (title, content)
3
values
4
  ('Understanding Vector Databases', 'Vector databases are specialized...');
5
6
-- Immediately check the embedding column
7
select id, embedding
8
from documents
9
where title = 'Understanding Vector Databases';

你应该注意到，在插入文档后，embedding 列最初是 null。这是因为嵌入生成是异步的，会在下一个计划任务中由 Edge Function 处理。

🌐 You should observe that the embedding column is initially null after inserting the document. This is because the embedding generation is asynchronous and will be processed by the Edge Function in the next scheduled task.

等待最多 10 秒让下一个任务运行，然后再次检查 embedding 列：

🌐 Wait up to 10 seconds for the next task to run, then check the embedding column again:

1
select id, embedding
2
from documents
3
where title = 'Understanding Vector Databases';

你应该看看为这个文档生成的嵌入。

🌐 You should see the generated embedding for the document.

接下来，更新文档的内容：

🌐 Next, update the content of the document:

1
-- Update the content of the document
2
update documents
3
set content = 'Vector databases allow you to query...'
4
where title = 'Understanding Vector Databases';
5
6
-- Immediately check the embedding column
7
select id, embedding
8
from documents
9
where title = 'Understanding Vector Databases';

你应该注意到，更新内容后 embedding 列会被重置为 null。这是因为我们添加了一个触发器，每当内容更新时都会清除现有的嵌入。嵌入将在下一个计划任务中由 Edge Function 重新生成。

🌐 You should observe that the embedding column is reset to null after updating the content. This is because of the trigger we added to clear existing embeddings whenever the content is updated. The embedding will be regenerated by the Edge Function in the next scheduled task.

等待最多 10 秒让下一个任务运行，然后再次检查 embedding 列：

🌐 Wait up to 10 seconds for the next task to run, then check the embedding column again:

1
select id, embedding
2
from documents
3
where title = 'Understanding Vector Databases';

你应该看看这个文档的更新嵌入。

🌐 You should see the updated embedding for the document.

最后我们来更新一下文档的标题：

🌐 Finally we'll update the title of the document:

1
-- Update the title of the document
2
update documents
3
set title = 'Understanding Vector Databases with Supabase'
4
where title = 'Understanding Vector Databases';

你应该注意到，在更新标题后，embedding 列再次被重置为 null。这是因为我们添加的触发器会在 content 或 title 列被更新时清除现有的嵌入。嵌入将在下一个计划任务中由 Edge Function 重新生成。

🌐 You should observe that the embedding column is once again reset to null after updating the title. This is because the trigger we added to clear existing embeddings fires when either the content or title columns are updated. The embedding will be regenerated by the Edge Function in the next scheduled task.

等待最多 10 秒让下一个任务运行，然后再次检查 embedding 列：

🌐 Wait up to 10 seconds for the next task to run, then check the embedding column again:

1
select id, embedding
2
from documents
3
where title = 'Understanding Vector Databases with Supabase';

你应该看看这个文档的更新嵌入。

🌐 You should see the updated embedding for the document.

故障排除 #

🌐 Troubleshooting

embed Edge Function 会处理一批嵌入任务，并返回一个包含已完成和失败任务列表的 200 OK 响应。例如：

🌐 The embed Edge Function processes a batch of embedding jobs and returns a 200 OK response with a list of completed and failed jobs in the body. For example:

1
{
2
  "completedJobs": [
3
    {
4
      "jobId": "1",
5
      "id": "1",
6
      "schema": "public",
7
      "table": "documents",
8
      "contentFunction": "embedding_input",
9
      "embeddingColumn": "embedding"
10
    }
11
  ],
12
  "failedJobs": [
13
    {
14
      "jobId": "2",
15
      "id": "2",
16
      "schema": "public",
17
      "table": "documents",
18
      "contentFunction": "embedding_input",
19
      "embeddingColumn": "embedding",
20
      "error": "error connecting to openai api"
21
    }
22
  ]
23
}

它还会在响应头中返回已完成和失败的任务数量。例如：

🌐 It also returns the number of completed and failed jobs in the response headers. For example:

1
x-completed-jobs: 1
2
x-failed-jobs: 1

你也可以使用 x-deno-execution-id 头来跟踪 Edge Function 在 dashboard 日志中的执行情况。

🌐 You can also use the x-deno-execution-id header to trace the execution of the Edge Function within the dashboard logs.

每个失败的工作都会包含一个带有失败描述的 error 字段。工作失败的原因可能包括：

🌐 Each failed job includes an error field with a description of the failure. Reasons for a job failing could include:

• 通过外部 API 生成嵌入时出错
• 连接数据库时出错
• 边缘函数被终止（例如，由于墙钟限制）
• 处理过程中抛出的任何其他错误

pg_net 会把 HTTP 响应存储在 net._http_response 表中，可以查询这个表来诊断嵌入生成过程中的问题。

1
select
2
  *
3
from
4
  net._http_response
5
where
6
  (headers->>'x-failed-jobs')::int > 0;

结论 #

🌐 Conclusion

在 Postgres 中自动生成和更新嵌入向量，让你可以轻松构建强大的语义搜索功能，而无需手动管理嵌入向量的复杂操作。

🌐 Automating embedding generation and updates in Postgres allow you to build powerful semantic search capabilities without the complexity of managing embeddings manually.

通过将 Postgres 的触发器、队列以及其他扩展功能与 Supabase Edge Functions 结合起来，我们可以创建一个强大的系统，异步处理嵌入生成，并自动重试失败的任务。

🌐 By combining Postgres features like triggers, queues, and other extensions with Supabase Edge Functions, we can create a robust system that handles embedding generation asynchronously and retries failed jobs automatically.

这个系统可以自定义，以配合任何内容和嵌入生成服务，提供一个用于 Postgres 语义搜索的灵活且可扩展的解决方案。

🌐 This system can be customized to work with any content and embedding generation service, providing a flexible and scalable solution for semantic search in Postgres.

另请参阅 #

🌐 See also

• 嵌入是什么？
• 语义搜索
• 向量索引
• Supabase 边缘函数