Supabase向量数据库与AI应用完全教程
从零开始掌握 Supabase pgvector 扩展，构建生产级向量检索与 RAG 系统
目录
- 为什么选择 Supabase 作为向量数据库
- 环境准备与 pgvector 扩展启用
- 向量表设计与索引创建
- Embedding 存储与检索实战
- RLS 权限控制与多租户隔离
- 实时订阅向量数据变更
- Edge Functions 集成 Embedding 生成
- 构建完整 RAG 系统
- 与 Pinecone / Chroma / Weaviate 对比
- 企业级部署与性能优化
- 最佳实践清单
1. 为什么选择 Supabase 作为向量数据库
Supabase 是基于 PostgreSQL 的开源 Firebase 替代方案。自 2023 年起，它深度集成了 pgvector 扩展，使得开发者可以在一个统一的平台中同时管理关系型数据和向量数据。
核心优势
| 特性 | 说明 |
|---|---|
| 统一数据层 | 结构化数据与向量数据共存于同一 PostgreSQL 实例，无需维护独立向量数据库 |
| RLS 原生安全 | 利用 PostgreSQL 行级安全策略实现细粒度权限控制 |
| Edge Functions | 基于 Deno 的服务端函数，可在边缘节点执行 Embedding 生成 |
| 实时订阅 | 通过 WebSocket 监听向量数据变更，实现实时 RAG 更新 |
| 托管运维 | 自动备份、扩容、监控，降低运维成本 |
| 开源生态 | 社区活跃，支持自托管部署 |
典型应用场景
- 语义搜索：基于文本含义而非关键词的搜索系统
- RAG（检索增强生成）：为 LLM 提供外部知识上下文
- 推荐系统：基于用户/物品 Embedding 的相似度推荐
- 异常检测：通过向量距离发现异常数据点
- 多模态检索：图文跨模态搜索
2. 环境准备与 pgvector 扩展启用
2.1 创建 Supabase 项目
前往 supabase.com 创建新项目，或使用 CLI 本地部署：
# 安装 Supabase CLI
npm install -g supabase
# 初始化本地项目
supabase init
# 启动本地开发环境
supabase start
2.2 启用 pgvector 扩展
在 Supabase 控制台的 SQL Editor 中执行：
-- 启用 pgvector 扩展
CREATE EXTENSION IF NOT EXISTS vector WITH SCHEMA extensions;
-- 验证安装
SELECT extname, extversion FROM pg_extension WHERE extname = 'vector';
如果使用本地 CLI，也可以在迁移文件中启用：
# 创建迁移文件
supabase migration new enable_vector_extension
在生成的迁移文件中写入：
-- supabase/migrations/xxxx_enable_vector_extension.sql
CREATE EXTENSION IF NOT EXISTS vector WITH SCHEMA extensions;
2.3 安装客户端依赖
# JavaScript/TypeScript 项目
npm install @supabase/supabase-js
# Python 项目
pip install supabase
2.4 初始化客户端
// src/supabase.ts
import { createClient } from '@supabase/supabase-js'
const supabaseUrl = process.env.SUPABASE_URL!
const supabaseKey = process.env.SUPABASE_ANON_KEY!
export const supabase = createClient(supabaseUrl, supabaseKey)
# supabase_client.py
import os
from supabase import create_client, Client
url: str = os.environ.get("SUPABASE_URL")
key: str = os.environ.get("SUPABASE_ANON_KEY")
supabase: Client = create_client(url, key)
3. 向量表设计与索引创建
3.1 设计向量数据表
-- 创建文档表，包含向量列
CREATE TABLE documents (
id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
content TEXT NOT NULL, -- 原始文本内容
metadata JSONB DEFAULT '{}'::jsonb, -- 元数据（来源、标签等）
embedding VECTOR(1536), -- OpenAI text-embedding-3-small 维度
token_count INTEGER, -- token 数量统计
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
-- 为元数据创建 GIN 索引以支持 JSONB 查询
CREATE INDEX idx_documents_metadata ON documents USING GIN (metadata);
-- 为常见查询字段添加索引
CREATE INDEX idx_documents_created_at ON documents (created_at DESC);
3.2 向量索引类型选择
pgvector 支持两种索引类型，选择取决于数据规模和精度需求：
| 索引类型 | 算法 | 适用场景 | 构建速度 | 查询速度 | 精度 |
|---|---|---|---|---|---|
| HNSW | 分层可导航小世界图 | 百万级数据，高精度需求 | 慢 | 极快 | 高 |
| IVFFlat | 倒排文件 + 平面量化 | 十万级数据，快速构建 | 快 | 快 | 中 |
-- HNSW 索引（推荐用于生产环境）
-- 使用余弦距离
CREATE INDEX idx_documents_embedding_cosine
ON documents USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
-- 使用 L2 距离
CREATE INDEX idx_documents_embedding_l2
ON documents USING hnsw (embedding vector_l2_ops)
WITH (m = 16, ef_construction = 200);
-- IVFFlat 索引（适用于数据量较小或快速原型）
-- 先分析数据分布确定 lists 数量
CREATE INDEX idx_documents_embedding_ivf
ON documents USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);
3.3 HNSW 参数调优
-- 创建表时预设参数
ALTER TABLE documents SET (
autovacuum_vacuum_scale_factor = 0.01,
autovacuum_analyze_scale_factor = 0.005
);
-- 查询时调整搜索参数（会话级别）
SET hnsw.ef_search = 100; -- 增加搜索精度，默认 40
参数说明：
m：每个节点的最大连接数，越大越精确但内存消耗越多（推荐 16-64）ef_construction：构建时的搜索范围，越大索引质量越高（推荐 100-300）ef_search：查询时的搜索范围，越大精度越高但速度越慢
3.4 多维度向量支持
某些场景需要存储不同维度的 Embedding：
-- 使用通用向量列（支持任意维度，但无法建索引）
CREATE TABLE multi_embeddings (
id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
content TEXT,
embedding_384 VECTOR(384), -- all-MiniLM-L6-v2
embedding_768 VECTOR(768), -- text-embedding-004
embedding_1536 VECTOR(1536) -- text-embedding-3-small
);
-- 为不同维度分别建索引
CREATE INDEX idx_multi_emb_384 ON multi_embeddings USING hnsw (embedding_384 vector_cosine_ops);
CREATE INDEX idx_multi_emb_768 ON multi_embeddings USING hnsw (embedding_768 vector_cosine_ops);
CREATE INDEX idx_multi_emb_1536 ON multi_embeddings USING hnsw (embedding_1536 vector_cosine_ops);
4. Embedding 存储与检索实战
4.1 生成并存储 Embedding
import { supabase } from './supabase'
interface DocumentInput {
content: string
metadata?: Record<string, any>
}
// 使用 OpenAI 生成 Embedding 并存入 Supabase
async function embedAndStore(doc: DocumentInput) {
// 1. 生成 Embedding
const embeddingResponse = await fetch('https://api.openai.com/v1/embeddings', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`
},
body: JSON.stringify({
model: 'text-embedding-3-small',
input: doc.content
})
})
const { data } = await embeddingResponse.json()
const embedding = data[0].embedding
// 2. 存入 Supabase
const { data: inserted, error } = await supabase
.from('documents')
.insert({
content: doc.content,
metadata: doc.metadata || {},
embedding: JSON.stringify(embedding),
token_count: doc.content.split(/\s+/).length // 简化统计
})
.select()
.single()
if (error) throw error
return inserted
}
import openai
from supabase_client import supabase
def embed_and_store(content: str, metadata: dict = None):
"""生成 Embedding 并存储到 Supabase"""
# 1. 生成 Embedding
response = openai.embeddings.create(
model="text-embedding-3-small",
input=content
)
embedding = response.data[0].embedding
# 2. 存入 Supabase
result = supabase.table("documents").insert({
"content": content,
"metadata": metadata or {},
"embedding": embedding,
"token_count": len(content.split())
}).execute()
return result.data[0]
4.2 批量插入优化
async function batchEmbedAndStore(docs: DocumentInput[], batchSize = 100) {
for (let i = 0; i < docs.length; i += batchSize) {
const batch = docs.slice(i, i + batchSize)
// 批量生成 Embedding（OpenAI 支持批量请求）
const embeddingResponse = await fetch('https://api.openai.com/v1/embeddings', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`
},
body: JSON.stringify({
model: 'text-embedding-3-small',
input: batch.map(d => d.content)
})
})
const { data: embeddings } = await embeddingResponse.json()
// 批量插入
const rows = batch.map((doc, idx) => ({
content: doc.content,
metadata: doc.metadata || {},
embedding: JSON.stringify(embeddings[idx].embedding)
}))
const { error } = await supabase.from('documents').insert(rows)
if (error) throw error
console.log(`Inserted batch ${Math.floor(i / batchSize) + 1}`)
}
}
4.3 语义搜索查询
-- 基本语义搜索（余弦距离）
SELECT
id,
content,
metadata,
1 - (embedding <=> $1::vector) AS similarity
FROM documents
ORDER BY embedding <=> $1::vector
LIMIT 10;
-- 带阈值过滤的语义搜索
SELECT
id,
content,
metadata,
1 - (embedding <=> $1::vector) AS similarity
FROM documents
WHERE 1 - (embedding <=> $1::vector) > 0.75
ORDER BY embedding <=> $1::vector
LIMIT 20;
-- 结合元数据过滤的混合搜索
SELECT
id,
content,
metadata,
1 - (embedding <=> $1::vector) AS similarity
FROM documents
WHERE
metadata->>'category' = '技术文档'
AND created_at > NOW() - INTERVAL '30 days'
ORDER BY embedding <=> $1::vector
LIMIT 10;
async function semanticSearch(query: string, options?: {
matchThreshold?: number
matchCount?: number
filter?: Record<string, any>
}) {
const {
matchThreshold = 0.5,
matchCount = 10,
filter = {}
} = options || {}
// 生成查询 Embedding
const embeddingResponse = await fetch('https://api.openai.com/v1/embeddings', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`
},
body: JSON.stringify({
model: 'text-embedding-3-small',
input: query
})
})
const { data } = await embeddingResponse.json()
const queryEmbedding = data[0].embedding
// 使用 RPC 函数进行搜索（推荐封装为数据库函数）
const { data: results, error } = await supabase.rpc('match_documents', {
query_embedding: JSON.stringify(queryEmbedding),
match_threshold: matchThreshold,
match_count: matchCount,
filter_metadata: filter
})
if (error) throw error
return results
}
4.4 封装数据库函数（推荐）
-- 创建通用的语义搜索函数
CREATE OR REPLACE FUNCTION match_documents(
query_embedding VECTOR(1536),
match_threshold FLOAT DEFAULT 0.5,
match_count INT DEFAULT 10,
filter_metadata JSONB DEFAULT '{}'::jsonb
)
RETURNS TABLE (
id UUID,
content TEXT,
metadata JSONB,
similarity FLOAT
)
LANGUAGE plpgsql
AS $$
BEGIN
RETURN QUERY
SELECT
documents.id,
documents.content,
documents.metadata,
1 - (documents.embedding <=> query_embedding) AS similarity
FROM documents
WHERE
1 - (documents.embedding <=> query_embedding) > match_threshold
AND documents.metadata @> filter_metadata
ORDER BY documents.embedding <=> query_embedding
LIMIT match_count;
END;
$$;
-- 创建全文搜索 + 向量搜索的混合函数
CREATE OR REPLACE FUNCTION hybrid_search(
query_text TEXT,
query_embedding VECTOR(1536),
match_count INT DEFAULT 10,
full_text_weight FLOAT DEFAULT 0.3,
semantic_weight FLOAT DEFAULT 0.7
)
RETURNS TABLE (
id UUID,
content TEXT,
metadata JSONB,
combined_score FLOAT
)
LANGUAGE plpgsql
AS $$
BEGIN
RETURN QUERY
WITH full_text AS (
SELECT
documents.id,
ts_rank(documents.content_tsvector, plainto_tsquery('chinese', query_text)) AS rank
FROM documents
WHERE documents.content_tsvector @@ plainto_tsquery('chinese', query_text)
),
semantic AS (
SELECT
documents.id,
1 - (documents.embedding <=> query_embedding) AS rank
FROM documents
ORDER BY documents.embedding <=> query_embedding
LIMIT match_count * 2
),
combined AS (
SELECT
COALESCE(full_text.id, semantic.id) AS id,
COALESCE(full_text.rank, 0) * full_text_weight
+ COALESCE(semantic.rank, 0) * semantic_weight AS score
FROM full_text
FULL OUTER JOIN semantic ON full_text.id = semantic.id
)
SELECT
documents.id,
documents.content,
documents.metadata,
combined.score AS combined_score
FROM combined
JOIN documents ON documents.id = combined.id
ORDER BY combined.score DESC
LIMIT match_count;
END;
$$;
5. RLS 权限控制与多租户隔离
5.1 基于用户的行级安全
-- 启用 RLS
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;
-- 策略：用户只能访问自己的文档
CREATE POLICY "Users can view own documents" ON documents
FOR SELECT USING (auth.uid() = user_id);
CREATE POLICY "Users can insert own documents" ON documents
FOR INSERT WITH CHECK (auth.uid() = user_id);
CREATE POLICY "Users can update own documents" ON documents
FOR UPDATE USING (auth.uid() = user_id);
CREATE POLICY "Users can delete own documents" ON documents
FOR DELETE USING (auth.uid() = user_id);
需要在表中添加 user_id 列：
ALTER TABLE documents ADD COLUMN user_id UUID REFERENCES auth.users(id);
CREATE INDEX idx_documents_user_id ON documents (user_id);
5.2 基于组织的多租户隔离
-- 创建组织成员关系表
CREATE TABLE organization_members (
id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
user_id UUID REFERENCES auth.users(id),
org_id UUID NOT NULL,
role TEXT DEFAULT 'member' CHECK (role IN ('owner', 'admin', 'member', 'viewer')),
created_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(user_id, org_id)
);
-- 文档表添加组织归属
ALTER TABLE documents ADD COLUMN org_id UUID;
-- 组织级 RLS 策略
CREATE POLICY "Org members can view org documents" ON documents
FOR SELECT USING (
EXISTS (
SELECT 1 FROM organization_members
WHERE organization_members.user_id = auth.uid()
AND organization_members.org_id = documents.org_id
)
);
CREATE POLICY "Org admins can manage org documents" ON documents
FOR ALL USING (
EXISTS (
SELECT 1 FROM organization_members
WHERE organization_members.user_id = auth.uid()
AND organization_members.org_id = documents.org_id
AND organization_members.role IN ('owner', 'admin')
)
);
5.3 搜索函数中的权限集成
-- 在搜索函数中自动应用 RLS
CREATE OR REPLACE FUNCTION match_documents_secure(
query_embedding VECTOR(1536),
match_threshold FLOAT DEFAULT 0.5,
match_count INT DEFAULT 10
)
RETURNS TABLE (
id UUID,
content TEXT,
metadata JSONB,
similarity FLOAT
)
LANGUAGE plpgsql
SECURITY DEFINER -- 使用函数定义者权限执行
SET search_path = public
AS $$
BEGIN
-- 检查调用者是否已认证
IF auth.uid() IS NULL THEN
RAISE EXCEPTION 'Authentication required';
END IF;
RETURN QUERY
SELECT
d.id,
d.content,
d.metadata,
1 - (d.embedding <=> query_embedding) AS similarity
FROM documents d
WHERE
-- 自动过滤用户有权访问的文档
(
d.user_id = auth.uid()
OR EXISTS (
SELECT 1 FROM organization_members om
WHERE om.user_id = auth.uid()
AND om.org_id = d.org_id
)
)
AND 1 - (d.embedding <=> query_embedding) > match_threshold
ORDER BY d.embedding <=> query_embedding
LIMIT match_count;
END;
$$;
6. 实时订阅向量数据变更
6.1 监听文档变更
// 订阅新文档插入
const channel = supabase
.channel('documents-changes')
.on(
'postgres_changes',
{
event: 'INSERT',
schema: 'public',
table: 'documents',
filter: `org_id=eq.${orgId}`
},
(payload) => {
console.log('新文档插入:', payload.new)
// 触发下游处理，如更新索引、通知用户
}
)
.on(
'postgres_changes',
{
event: 'UPDATE',
schema: 'public',
table: 'documents'
},
(payload) => {
console.log('文档更新:', payload.new)
}
)
.subscribe()
// 清理订阅
// supabase.removeChannel(channel)
6.2 实时 RAG 更新管道
class RealtimeRAGPipeline {
private channel: any
constructor(private supabase: any) {}
start(orgId: string) {
this.channel = this.supabase
.channel('rag-updates')
.on(
'postgres_changes',
{
event: '*',
schema: 'public',
table: 'documents',
filter: `org_id=eq.${orgId}`
},
async (payload: any) => {
switch (payload.eventType) {
case 'INSERT':
await this.handleNewDocument(payload.new)
break
case 'UPDATE':
await this.handleDocumentUpdate(payload.new, payload.old)
break
case 'DELETE':
await this.handleDocumentDelete(payload.old)
break
}
}
)
.subscribe()
}
private async handleNewDocument(doc: any) {
// 文档已有 embedding，无需额外处理
console.log(`[RAG] 新知识入库: ${doc.id}`)
}
private async handleDocumentUpdate(newDoc: any, oldDoc: any) {
if (newDoc.content !== oldDoc.content) {
console.log(`[RAG] 内容变更，需重新生成 Embedding: ${newDoc.id}`)
// 触发重新 Embedding 流程
}
}
private async handleDocumentDelete(oldDoc: any) {
console.log(`[RAG] 知识移除: ${oldDoc.id}`)
}
stop() {
if (this.channel) {
this.supabase.removeChannel(this.channel)
}
}
}
7. Edge Functions 集成 Embedding 生成
7.1 创建 Embedding Edge Function
// supabase/functions/embed/index.ts
import { serve } from 'https://deno.land/std@0.168.0/http/server.ts'
import { createClient } from 'https://esm.sh/@supabase/supabase-js@2'
const OPENAI_API_KEY = Deno.env.get('OPENAI_API_KEY')!
const SUPABASE_URL = Deno.env.get('SUPABASE_URL')!
const SUPABASE_SERVICE_KEY = Deno.env.get('SUPABASE_SERVICE_ROLE_KEY')!
serve(async (req) => {
try {
const { content, metadata, user_id } = await req.json()
if (!content) {
return new Response(
JSON.stringify({ error: 'content is required' }),
{ status: 400, headers: { 'Content-Type': 'application/json' } }
)
}
// 1. 调用 OpenAI 生成 Embedding
const embeddingResponse = await fetch('https://api.openai.com/v1/embeddings', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${OPENAI_API_KEY}`
},
body: JSON.stringify({
model: 'text-embedding-3-small',
input: content
})
})
const embeddingData = await embeddingResponse.json()
const embedding = embeddingData.data[0].embedding
// 2. 使用 Service Role 写入数据库（绕过 RLS）
const supabase = createClient(SUPABASE_URL, SUPABASE_SERVICE_KEY)
const { data, error } = await supabase
.from('documents')
.insert({
content,
metadata: metadata || {},
embedding: JSON.stringify(embedding),
user_id
})
.select()
.single()
if (error) throw error
return new Response(
JSON.stringify({ success: true, document: data }),
{ headers: { 'Content-Type': 'application/json' } }
)
} catch (error) {
return new Response(
JSON.stringify({ error: error.message }),
{ status: 500, headers: { 'Content-Type': 'application/json' } }
)
}
})
7.2 搜索 Edge Function
// supabase/functions/search/index.ts
import { serve } from 'https://deno.land/std@0.168.0/http/server.ts'
import { createClient } from 'https://esm.sh/@supabase/supabase-js@2'
serve(async (req) => {
const { query, match_threshold = 0.5, match_count = 10 } = await req.json()
// 验证用户身份
const authHeader = req.headers.get('Authorization')!
const supabase = createClient(
Deno.env.get('SUPABASE_URL')!,
Deno.env.get('SUPABASE_ANON_KEY')!,
{ global: { headers: { Authorization: authHeader } } }
)
const { data: { user } } = await supabase.auth.getUser()
if (!user) {
return new Response('Unauthorized', { status: 401 })
}
// 生成查询 Embedding
const embeddingResponse = await fetch('https://api.openai.com/v1/embeddings', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${Deno.env.get('OPENAI_API_KEY')}`
},
body: JSON.stringify({
model: 'text-embedding-3-small',
input: query
})
})
const embeddingData = await embeddingResponse.json()
const queryEmbedding = embeddingData.data[0].embedding
// 执行语义搜索（RLS 自动过滤）
const { data: results, error } = await supabase.rpc('match_documents_secure', {
query_embedding: JSON.stringify(queryEmbedding),
match_threshold,
match_count
})
return new Response(
JSON.stringify({ results }),
{ headers: { 'Content-Type': 'application/json' } }
)
})
7.3 部署 Edge Function
# 部署函数
supabase functions deploy embed
supabase functions deploy search
# 设置环境变量
supabase secrets set OPENAI_API_KEY=sk-your-key-here
# 本地测试
supabase functions serve embed --env-file ./supabase/.env.local
8. 构建完整 RAG 系统
8.1 系统架构
用户提问
↓
[查询理解] → 生成查询 Embedding
↓
[向量检索] → pgvector 语义搜索
↓
[上下文组装] → 拼接 Top-K 文档片段
↓
[LLM 生成] → 基于上下文生成回答
↓
[引用溯源] → 返回来源文档引用
8.2 完整 RAG 实现
// rag.ts
import { createClient } from '@supabase/supabase-js'
const supabase = createClient(
process.env.SUPABASE_URL!,
process.env.SUPABASE_SERVICE_ROLE_KEY!
)
interface RAGResponse {
answer: string
sources: Array<{ id: string; content: string; similarity: number }>
usage: { prompt_tokens: number; completion_tokens: number }
}
async function ragQuery(question: string): Promise<RAGResponse> {
// Step 1: 生成问题的 Embedding
const embeddingRes = await fetch('https://api.openai.com/v1/embeddings', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`
},
body: JSON.stringify({
model: 'text-embedding-3-small',
input: question
})
})
const { data: [{ embedding: queryEmbedding }] } = await embeddingRes.json()
// Step 2: 向量检索相关文档
const { data: documents } = await supabase.rpc('match_documents', {
query_embedding: JSON.stringify(queryEmbedding),
match_threshold: 0.6,
match_count: 5
})
if (!documents || documents.length === 0) {
return {
answer: '抱歉，未找到与您问题相关的知识内容。请尝试换个方式提问。',
sources: [],
usage: { prompt_tokens: 0, completion_tokens: 0 }
}
}
// Step 3: 组装上下文
const context = documents
.map((doc, i) => `[${i + 1}] ${doc.content}`)
.join('\n\n')
// Step 4: 调用 LLM 生成回答
const chatRes = await fetch('https://api.openai.com/v1/chat/completions', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`
},
body: JSON.stringify({
model: 'gpt-4o-mini',
messages: [
{
role: 'system',
content: `你是一个基于知识库的问答助手。根据以下参考资料回答用户问题。
如果参考资料中没有相关信息，请明确说明你无法根据现有知识回答。
回答时请引用来源编号，格式为 [1]、[2] 等。
参考资料：
${context}`
},
{ role: 'user', content: question }
],
temperature: 0.3,
max_tokens: 1000
})
})
const chatData = await chatRes.json()
const answer = chatData.choices[0].message.content
const usage = chatData.usage
return {
answer,
sources: documents.map(doc => ({
id: doc.id,
content: doc.content.substring(0, 200),
similarity: doc.similarity
})),
usage: {
prompt_tokens: usage.prompt_tokens,
completion_tokens: usage.completion_tokens
}
}
}
8.3 文档分块策略
// chunking.ts
interface ChunkOptions {
maxChunkSize: number // 最大 chunk 字符数
overlapSize: number // 重叠字符数
separator: string // 分隔符优先级
}
function chunkDocument(text: string, options: ChunkOptions): string[] {
const { maxChunkSize = 1000, overlapSize = 200, separator = '\n' } = options
// 先按段落分割
const paragraphs = text.split(separator).filter(p => p.trim())
const chunks: string[] = []
let currentChunk = ''
for (const paragraph of paragraphs) {
if (currentChunk.length + paragraph.length > maxChunkSize && currentChunk.length > 0) {
chunks.push(currentChunk.trim())
// 保留尾部重叠
const overlap = currentChunk.slice(-overlapSize)
currentChunk = overlap + separator + paragraph
} else {
currentChunk += (currentChunk ? separator : '') + paragraph
}
}
if (currentChunk.trim()) {
chunks.push(currentChunk.trim())
}
return chunks
}
// 按固定 token 数分块（更精确）
function chunkByTokens(text: string, maxTokens = 500, overlapTokens = 50): string[] {
const words = text.split(/\s+/)
const chunks: string[] = []
let i = 0
while (i < words.length) {
const chunkWords = words.slice(i, i + maxTokens)
chunks.push(chunkWords.join(' '))
i += maxTokens - overlapTokens
}
return chunks
}
9. 与 Pinecone / Chroma / Weaviate 对比
功能对比表
| 特性 | Supabase pgvector | Pinecone | Chroma | Weaviate |
|---|---|---|---|---|
| 部署方式 | 自托管 / Supabase Cloud | 纯 SaaS | 自托管 / 嵌入式 | 自托管 / Cloud |
| 数据存储 | PostgreSQL | 专用存储 | SQLite / DuckDB | 专用存储 |
| 关系数据 | ✅ 原生支持 | ❌ | ❌ | ❌ |
| 全文搜索 | ✅ tsvector | ❌ | ✅ 基础 | ✅ BM25 |
| 元数据过滤 | ✅ SQL 全能力 | ✅ 有限 | ✅ 有限 | ✅ GraphQL |
| 实时订阅 | ✅ WebSocket | ❌ | ❌ | ✅ GraphQL Sub |
| RBAC/RLS | ✅ 行级安全 | ✅ API Key | ❌ | ✅ OIDC |
| ACID 事务 | ✅ | ❌ | ❌ | ❌ |
| 最大向量维度 | 16,000 | 20,000 | 无限制 | 65,535 |
| 索引算法 | HNSW / IVFFlat | 专利算法 | HNSW | HNSW |
| 免费层 | 500MB 数据库 | 100K 向量 | 开源免费 | 开源免费 |
| 学习曲线 | 低（SQL 熟悉） | 中 | 低 | 高 |
选择建议
选择 Supabase pgvector 当：
├── 已有 PostgreSQL 经验
├── 需要关系数据 + 向量数据统一管理
├── 需要行级安全和多租户隔离
├── 项目需要 ACID 事务保证
└── 团队希望减少基础设施组件
选择 Pinecone 当：
├── 纯向量搜索场景，不需要关系数据
├── 追求零运维的全托管体验
├── 需要超大规模向量索引（十亿级）
└── 预算充足
选择 Chroma 当：
├── 快速原型和实验
├── 嵌入式应用（无需独立服务器）
├── 本地开发和测试
└── Python 生态项目
选择 Weaviate 当：
├── 需要高级向量搜索功能（多租户、命名空间）
├── GraphQL API 偏好
├── 需要内置的向量化模块
└── 大规模生产部署
从 Pinecone 迁移到 Supabase
-- 创建兼容 Pinecone 数据结构的表
CREATE TABLE pinecone_migrated (
id TEXT PRIMARY KEY, -- Pinecone 使用字符串 ID
content TEXT,
metadata JSONB,
embedding VECTOR(1536)
);
-- 迁移脚本（Node.js）
/*
async function migrateFromPinecone() {
const pinecone = await Pinecone.connect({ apiKey: '...' })
const index = pinecone.index('my-index')
// 分批读取 Pinecone 数据
const iterator = await index.listPaginated({ limit: 1000 })
for await (const batch of iterator) {
const vectors = await index.fetch(batch.vectors.map(v => v.id))
const rows = Object.values(vectors).map(v => ({
id: v.id,
metadata: v.metadata,
embedding: JSON.stringify(v.values)
}))
await supabase.from('pinecone_migrated').upsert(rows)
}
}
*/
10. 企业级部署与性能优化
10.1 连接池配置
// 使用 Supabase 连接池（推荐生产环境）
import { createClient } from '@supabase/supabase-js'
const supabase = createClient(
process.env.SUPABASE_URL!,
process.env.SUPABASE_ANON_KEY!,
{
db: {
schema: 'public'
},
// 连接池配置
global: {
headers: { 'x-my-custom-header': 'my-app-name' }
}
}
)
// 使用 PostgreSQL 连接池（直接连接场景）
import { Pool } from 'pg'
const pool = new Pool({
connectionString: process.env.DATABASE_URL,
max: 20, // 最大连接数
idleTimeoutMillis: 30000, // 空闲连接超时
connectionTimeoutMillis: 2000 // 连接超时
})
10.2 查询性能优化
-- 1. 预热向量缓存
SET hnsw.ef_search = 100;
-- 2. 使用 EXPLAIN ANALYZE 分析查询计划
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT id, content, 1 - (embedding <=> '[0.1,0.2,...]'::vector) AS similarity
FROM documents
ORDER BY embedding <=> '[0.1,0.2,...]'::vector
LIMIT 10;
-- 3. 分区表策略（超大数据集）
CREATE TABLE documents_partitioned (
LIKE documents INCLUDING ALL
) PARTITION BY RANGE (created_at);
CREATE TABLE documents_2024 PARTITION OF documents_partitioned
FOR VALUES FROM ('2024-01-01') TO ('2025-01-01');
CREATE TABLE documents_2025 PARTITION OF documents_partitioned
FOR VALUES FROM ('2025-01-01') TO ('2026-01-01');
-- 4. 定期维护统计信息
ANALYZE documents;
10.3 监控指标
-- 监控向量索引大小
SELECT
schemaname,
tablename,
indexname,
pg_size_pretty(pg_relation_size(indexrelid)) AS index_size
FROM pg_stat_user_indexes
WHERE indexname LIKE '%embedding%';
-- 监控查询性能
SELECT
query,
calls,
mean_exec_time,
total_exec_time,
rows
FROM pg_stat_statements
WHERE query LIKE '%embedding%'
ORDER BY mean_exec_time DESC
LIMIT 10;
-- 监控缓存命中率
SELECT
relname,
heap_blks_read,
heap_blks_hit,
CASE WHEN heap_blks_hit + heap_blks_read > 0
THEN round(heap_blks_hit * 100.0 / (heap_blks_hit + heap_blks_read), 2)
ELSE 0
END AS cache_hit_ratio
FROM pg_statio_user_tables
WHERE relname = 'documents';
10.4 备份与恢复策略
# Supabase 自动备份（Pro 计划以上）
# 每日自动备份，保留 7 天
# 手动备份向量数据
pg_dump -h db.your-project.supabase.co \
-U postgres \
-d postgres \
-t documents \
-F c \
-f documents_backup.dump
# 恢复
pg_restore -h db.your-project.supabase.co \
-U postgres \
-d postgres \
-F c \
documents_backup.dump
11. 最佳实践清单
设计阶段
- 根据 Embedding 模型确定向量维度（OpenAI: 1536, Cohere: 1024, 本地模型: 384/768）
- 选择合适的距离度量（语义搜索用余弦，精确匹配用 L2）
- 设计元数据结构，规划 JSONB 字段索引
- 规划 RLS 策略，确定多租户隔离方案
开发阶段
- 使用数据库函数封装搜索逻辑，避免客户端暴露表结构
- 实现批量 Embedding 生成，减少 API 调用次数
- 设计合理的分块策略（500-1000 tokens，适当重叠）
- 在 Edge Function 中处理敏感操作，使用 Service Role 绕过 RLS
- 添加错误处理和重试机制
部署阶段
- 生产环境使用 HNSW 索引，
ef_search设置为 80-200 - 配置连接池，避免连接数爆炸
- 启用 Supabase 的 Point-in-Time Recovery
- 设置监控告警（查询延迟、索引大小、缓存命中率）
- 使用 Supabase Dashboard 或 pgAdmin 监控数据库状态
维护阶段
- 定期执行
ANALYZE更新统计信息 - 监控索引膨胀，必要时重建索引
- 审查 RLS 策略，确保安全性
- 跟踪 pgvector 版本更新，及时升级
- 定期备份测试恢复流程
附录：常见问题
Q: 向量搜索结果不准确怎么办？
A: 首先检查 ef_search 参数是否足够大（默认 40 太小），建议设置为 100+。其次确认 Embedding 模型与查询文本的匹配度。最后检查 match_threshold 是否过于严格。
Q: 如何处理中文文本的 Embedding？
A: 推荐使用支持中文的 Embedding 模型，如 text-embedding-3-small（支持多语言）或 bge-large-zh-v1.5（中文专用）。分块时注意中文没有自然空格，建议按句子或段落分割。
Q: 数据量达到百万级后性能下降？
A: 确认已创建 HNSW 索引（而非 IVFFlat），调整 m 和 ef_construction 参数，考虑使用表分区，并确保有足够的 shared_buffers 和 work_mem。
Q: 如何降低 Embedding 生成成本？
A: 使用缓存机制（相同内容不重复生成），批量处理而非逐条调用，考虑使用开源模型（如 sentence-transformers）自托管，或使用 Cohere 等提供免费额度的服务。
本教程涵盖 Supabase pgvector 从入门到企业级部署的完整知识体系。建议按顺序学习，逐步实践。如有问题，欢迎在 Supabase 社区讨论。
