自动嵌入

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
-- 实用函数的模式create schema util;-- 获取Supabase项目URL的实用函数（边缘函数所需）create function util.project_url()returns textlanguage plpgsqlsecurity defineras $$declare  secret_value text;begin  -- 从Vault中检索项目URL  select decrypted_secret into secret_value from vault.decrypted_secrets where name = 'project_url';  return secret_value;end;$$;-- 调用任何边缘函数的通用函数create or replace function util.invoke_edge_function(  name text,  body jsonb,  timeout_milliseconds int = 5 * 60 * 1000  -- 默认5分钟超时)returns voidlanguage plpgsqlas $$declare  headers_raw text;  auth_header text;begin  -- 如果我们处于PostgREST会话中，重用请求头进行授权  headers_raw := current_setting('request.headers', true);  -- 仅当存在头信息时才尝试解析  auth_header := case    when headers_raw is not null then      (headers_raw::json->>'authorization')    else      null  end;  -- 对边缘函数执行异步HTTP请求  perform net.http_post(    url => util.project_url() || '/functions/v1/' || name,    headers => jsonb_build_object(      'Content-Type', 'application/json',      'Authorization', auth_header    ),    body => body,    timeout_milliseconds => timeout_milliseconds  );end;$$;-- 更新时清除列的通用触发器函数create or replace function util.clear_column()returns triggerlanguage plpgsql as $$declare    clear_column text := TG_ARGV[0];begin    NEW := NEW #= hstore(clear_column, NULL);    return NEW;end;$$;

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

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

1
2
-- 用于处理嵌入任务的队列select pgmq.create('embedding_jobs');

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
-- 用于将嵌入任务排入队列的通用触发器函数create or replace function util.queue_embeddings()returns triggerlanguage plpgsqlas $$declare  content_function text = TG_ARGV[0];  embedding_column text = TG_ARGV[1];begin  perform pgmq.send(    queue_name => 'embedding_jobs',    msg => jsonb_build_object(      'id', NEW.id,      'schema', TG_TABLE_SCHEMA,      'table', TG_TABLE_NAME,      'contentFunction', content_function,      'embeddingColumn', embedding_column    )  );  return NEW;end;$$;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
-- 用于处理队列中嵌入任务的函数create or replace function util.process_embeddings(  batch_size int = 10,  max_requests int = 10,  timeout_milliseconds int = 5 * 60 * 1000 -- 默认5分钟超时)returns voidlanguage plpgsqlas $$declare  job_batches jsonb[];  batch jsonb;begin  with    -- 首先获取任务并分配批次编号    numbered_jobs as (      select        message || jsonb_build_object('jobId', msg_id) as job_info,        (row_number() over (order by 1) - 1) / batch_size as batch_num      from pgmq.read(        queue_name => 'embedding_jobs',        vt => timeout_milliseconds / 1000,        qty => max_requests * batch_size      )    ),    -- 然后将任务分组为批次    batched_jobs as (      select        jsonb_agg(job_info) as batch_array,        batch_num      from numbered_jobs      group by batch_num    )  -- 最后将所有批次聚合为数组  select array_agg(batch_array)  from batched_jobs  into job_batches;  -- 为每个批次调用嵌入边缘函数  foreach batch in array job_batches loop    perform util.invoke_edge_function(      name => 'embed',      body => batch,      timeout_milliseconds => timeout_milliseconds    );  end loop;end;$$;-- 安排嵌入处理select  cron.schedule(    'process-embeddings',    '10 seconds',    $$    select util.process_embeddings();    $$  );

1
supabase functions new embed

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
// 为内置的Supabase运行时API设置类型定义import 'jsr:@supabase/functions-js/edge-runtime.d.ts'// 我们将使用OpenAI API生成嵌入向量import OpenAI from 'jsr:@openai/openai'import { z } from 'npm:zod'// 我们将直接连接Postgres以更新文档import postgres from 'https://deno.land/x/postgresjs@v3.4.5/mod.js'// 初始化OpenAI客户端const openai = new OpenAI({  // 我们需要手动设置 `OPENAI_API_KEY` 环境变量  apiKey: Deno.env.get('OPENAI_API_KEY'),})// 初始化Postgres客户端const sql = postgres(  // `SUPABASE_DB_URL` 是一个内置的环境变量  Deno.env.get('SUPABASE_DB_URL')!)const jobSchema = z.object({  jobId: z.number(),  id: z.number(),  schema: z.string(),  table: z.string(),  contentFunction: z.string(),  embeddingColumn: z.string(),})const failedJobSchema = jobSchema.extend({  error: z.string(),})type Job = z.infer<typeof jobSchema>type FailedJob = z.infer<typeof failedJobSchema>type Row = {  id: string  content: unknown}const QUEUE_NAME = 'embedding_jobs'// 监听HTTP请求Deno.serve(async (req) => {  if (req.method !== 'POST') {    return new Response('expected POST request', { status: 405 })  }  if (req.headers.get('content-type')!== 'application/json') {    return new Response('expected json body', { status: 400 })  }  // 使用Zod解析并验证请求体  const parseResult = z.array(jobSchema).safeParse(await req.json())  if (parseResult.error) {    return new Response(`invalid request body: ${parseResult.error.message}`, {      status: 400,    })  }  const pendingJobs = parseResult.data  // 跟踪成功完成的作业  const completedJobs: Job[] = []  // 跟踪因错误而失败的作业  const failedJobs: FailedJob[] = []  async function processJobs() {    let currentJob: Job | undefined    while ((currentJob = pendingJobs.shift())!== undefined) {      try {        await processJob(currentJob)        completedJobs.push(currentJob)      } catch (error) {        failedJobs.push({          ...currentJob,          error: error instanceof Error? error.message : JSON.stringify(error),        })      }    }  }  try {    // 在监听工作进程终止的同时处理作业    await Promise.race([processJobs(), catchUnload()])  } catch (error) {    // 如果工作进程正在终止（例如达到时间限制），    // 将未处理的作业添加到失败列表中，并附上终止原因    failedJobs.push(      ...pendingJobs.map((job) => ({        ...job,        error: error instanceof Error? error.message : JSON.stringify(error),      }))    )  }  // 记录已完成和失败的作业，以便进行追溯  console.log('finished processing jobs:', {    completedJobs: completedJobs.length,    failedJobs: failedJobs.length,  })  return new Response(    JSON.stringify({      completedJobs,      failedJobs,    }),    {      // 200 OK响应      status: 200,      // 用于报告作业状态的自定义标头      headers: {        'content-type': 'application/json',        'x-completed-jobs': completedJobs.length.toString(),        'x-failed-jobs': failedJobs.length.toString(),      },    }  )})/** * 为给定文本生成嵌入向量。 */async function generateEmbedding(text: string) {  const response = await openai.embeddings.create({    model: 'text-embedding-3-small',    input: text,  })  const [data] = response.data  if (!data) {    throw new Error('failed to generate embedding')  }  return data.embedding}/** * 处理嵌入作业。 */async function processJob(job: Job) {  const { jobId, id, schema, table, contentFunction, embeddingColumn } = job  // 获取模式/表/行组合的内容  const [row]: [Row] = await sql`    select      id,      ${sql(contentFunction)}(t) as content    from      ${sql(schema)}.${sql(table)} t    where      id = ${id}  `  if (!row) {    throw new Error(`row not found: ${schema}.${table}/${id}`)  }  if (typeof row.content!== 'string') {    throw new Error(`invalid content - expected string: ${schema}.${table}/${id}`)  }  const embedding = await generateEmbedding(row.content)  await sql`    update      ${sql(schema)}.${sql(table)}    set      ${sql(embeddingColumn)} = ${JSON.stringify(embedding)}    where      id = ${id}  `  await sql`    select pgmq.delete(${QUEUE_NAME}, ${jobId}::bigint)  `}/** * 返回一个Promise，如果工作进程正在终止，则该Promise会被拒绝。 */function catchUnload() {  return new Promise((reject) => {    addEventListener('beforeunload', (ev: any) => {      reject(new Error(ev.detail?.reason))    })  })}

1
OPENAI_API_KEY=your-api-key

1
supabase secrets set --env-file .env

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

1
2
3
4
5
6
7
8
9
10
11
-- 用于存储带嵌入向量文档的表create table documents (  id integer primary key generated always as identity,  title text not null,  content text not null,  embedding halfvec(1536),  created_at timestamp with time zone default now());-- 用于对文档嵌入向量进行向量搜索的索引create index on documents using hnsw (embedding halfvec_cosine_ops);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
-- 自定义嵌入生成的输入-- 例如，使用 Markdown 标题连接标题和内容create or replace function embedding_input(doc documents)returns textlanguage plpgsqlimmutableas $$begin  return '# ' || doc.title || E'\n\n' || doc.content;end;$$;-- 插入事件的触发器create trigger embed_documents_on_insert  after insert  on documents  for each row  execute function util.queue_embeddings('embedding_input', 'embedding');-- 更新事件的触发器create trigger embed_documents_on_update  after update of title, content -- 必须与 embedding_input() 中的列匹配  on documents  for each row  execute function util.queue_embeddings('embedding_input', 'embedding');

1
2
3
4
5
6
-- 更新时清除嵌入向量列的触发器create trigger clear_document_embedding_on_update  before update of title, content -- 必须与 embedding_input() 中的列匹配  on documents  for each row  execute function util.clear_column('embedding');

1
2
3
4
5
6
7
8
9
-- 插入一个新文档insert into documents (title, content)values  ('理解向量数据库', '向量数据库是专门用于……');-- 立即检查嵌入向量列select id, embeddingfrom documentswhere title = '理解向量数据库';

1
2
3
select id, embeddingfrom documentswhere title = '理解向量数据库';

1
2
3
4
5
6
7
8
9
-- 更新文档的内容update documentsset content = '向量数据库允许你查询……'where title = '理解向量数据库';-- 立即检查嵌入向量列select id, embeddingfrom documentswhere title = '理解向量数据库';

1
2
3
select id, embeddingfrom documentswhere title = '理解向量数据库';

1
2
3
4
-- 更新文档的标题update documentsset title = '使用 Supabase 理解向量数据库'where title = '理解向量数据库';

1
2
3
select id, embeddingfrom documentswhere title = '使用 Supabase 理解向量数据库';

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

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

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