Supabase 2026: 开源 Firebase 替代品，Postgres 向量搜索驱动 100 万+ AI 应用 — 完整部署指南

# 你的项目附带：
# - PostgreSQL 16 数据库
# - 自动生成的 REST API
# - 内置认证
# - 500 MB 数据库存储（免费层）
# - 1 GB 文件存储（免费层）
# - 2 GB 带宽（免费层）

# 安装 Supabase CLI
# macOS
brew install supabase/tap/supabase

# Linux
npm install -g supabase

# 验证安装
supabase --version
# 预期输出: 1.220.0

# 登录账户
supabase login

# 初始化新项目
mkdir my-ai-app && cd my-ai-app
supabase init

# 启动本地堆栈（需要 Docker）
supabase start

# 输出包括本地 API URL、anon key 和 service role key
# API URL: http://localhost:54321
# GraphQL URL: http://localhost:54321/graphql/v1
# anon key: eyJhbGciOiJIUzI1NiIs...

# 克隆官方自托管仓库
git clone https://github.com/supabase/supabase.git
cd supabase/docker

# 复制并编辑环境变量
cp .env.example .env

# 生成安全密钥
openssl rand -base64 32  # JWT 密钥
openssl rand -hex 32     # anon/service 密钥

# 编辑 .env 文件，设置域名、SMTP 和 S3 凭证
nano .env

# 启动堆栈
docker compose up -d

# 验证所有服务健康
docker compose ps

# 预期输出：
# NAME                STATUS
# supabase-db         healthy
# supabase-kong       healthy
# supabase-auth       healthy
# supabase-rest       healthy
# supabase-realtime   healthy
# supabase-storage    healthy
# supabase-meta       healthy
# supabase-studio     healthy

# 安装客户端库
npm install @supabase/supabase-js

# 或使用 Python
pip install supabase

// TypeScript / Next.js
import { createClient } from '@supabase/supabase-js'

const supabase = createClient(
  process.env.NEXT_PUBLIC_SUPABASE_URL!,
  process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY!
)

// 测试连接
const { data, error } = await supabase.from('test').select('*')
console.log(data)

# Python
from supabase import create_client

supabase = create_client(
    "https://your-project.supabase.co",
    "your-anon-key"
)

# 测试连接
response = supabase.table('test').select('*').execute()
print(response.data)

-- 在 Supabase SQL 编辑器或 psql 中
CREATE EXTENSION IF NOT EXISTS vector;

-- 验证扩展已安装
SELECT * FROM pg_extension WHERE extname = 'vector';

-- 创建带嵌入的文档表
CREATE TABLE documents (
    id          BIGSERIAL PRIMARY KEY,
    title       TEXT NOT NULL,
    content     TEXT NOT NULL,
    source_url  TEXT,
    embedding   VECTOR(1536),  -- 匹配 OpenAI text-embedding-3-large
    metadata    JSONB DEFAULT '{}',
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

-- 创建 HNSW 索引以实现快速相似性搜索
CREATE INDEX idx_documents_embedding
ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- 添加全文搜索索引用于混合搜索
CREATE INDEX idx_documents_fts ON documents
USING GIN (to_tsvector('english', content));

# Python：生成嵌入并插入 Supabase
from supabase import create_client
import openai

supabase = create_client(SUPABASE_URL, SUPABASE_KEY)
client = openai.OpenAI(api_key=OPENAI_API_KEY)

def insert_document(title: str, content: str, source_url: str = None):
    # 生成嵌入
    response = client.embeddings.create(
        input=content,
        model="text-embedding-3-large"
    )
    embedding = response.data[0].embedding

    # 插入 Supabase
    result = supabase.table('documents').insert({
        'title': title,
        'content': content,
        'source_url': source_url,
        'embedding': embedding,
        'metadata': {'word_count': len(content.split())}
    }).execute()
    return result

# 插入示例文档
insert_document(
    title="Docker Best Practices 2026",
    content="Use multi-stage builds to reduce image size...",
    source_url="https://docs.docker.com"
)

-- 纯 SQL：查找最相似的 5 个文档
SELECT
    id,
    title,
    content,
    1 - (embedding <=> :query_embedding::vector) AS similarity
FROM documents
ORDER BY embedding <=> :query_embedding::vector
LIMIT 5;

# Python：RAG 检索函数
async def search_similar_documents(query: str, top_k: int = 5):
    # 生成查询嵌入
    response = client.embeddings.create(
        input=query,
        model="text-embedding-3-large"
    )
    query_embedding = response.data[0].embedding

    # 查询 Supabase
    result = await supabase.rpc(
        'match_documents',
        {
            'query_embedding': query_embedding,
            'match_threshold': 0.7,
            'match_count': top_k
        }
    ).execute()
    return result.data

-- 创建用于文档检索的存储过程
CREATE OR REPLACE FUNCTION match_documents(
    query_embedding VECTOR(1536),
    match_threshold FLOAT,
    match_count INT
)
RETURNS TABLE (
    id BIGINT,
    title TEXT,
    content TEXT,
    similarity FLOAT
)
LANGUAGE plpgsql
AS $$
BEGIN
    RETURN QUERY
    SELECT
        d.id,
        d.title,
        d.content,
        1 - (d.embedding <=> query_embedding) AS similarity
    FROM documents d
    WHERE 1 - (d.embedding <=> query_embedding) > match_threshold
    ORDER BY d.embedding <=> query_embedding
    LIMIT match_count;
END;
$$;

# rag_pipeline.py
from supabase import create_client
from openai import OpenAI
import json

class SupabaseRAG:
    def __init__(self, supabase_url: str, supabase_key: str, openai_key: str):
        self.supabase = create_client(supabase_url, supabase_key)
        self.openai = OpenAI(api_key=openai_key)

    def embed_and_store(self, chunks: list[dict]):
        """存储带嵌入的文档块。"""
        for chunk in chunks:
            embedding = self.openai.embeddings.create(
                input=chunk['text'],
                model="text-embedding-3-large"
            ).data[0].embedding

            self.supabase.table('documents').insert({
                'title': chunk['title'],
                'content': chunk['text'],
                'embedding': embedding,
                'metadata': chunk.get('metadata', {})
            }).execute()

    def retrieve(self, query: str, top_k: int = 5) -> list[dict]:
        """使用向量搜索检索相关文档。"""
        query_embedding = self.openai.embeddings.create(
            input=query,
            model="text-embedding-3-large"
        ).data[0].embedding

        results = self.supabase.rpc(
            'match_documents',
            {
                'query_embedding': query_embedding,
                'match_threshold': 0.75,
                'match_count': top_k
            }
        ).execute()
        return results.data

    def generate(self, query: str, context: list[dict]) -> str:
        """使用检索到的上下文生成响应。"""
        context_text = "\n\n".join([
            f"[来源: {doc['title']}]\n{doc['content']}"
            for doc in context
        ])

        response = self.openai.chat.completions.create(
            model="gpt-4.1-mini",
            messages=[
                {
                    "role": "system",
                    "content": "你是一个有帮助的助手。仅基于提供的上下文回答。引用你的来源。"
                },
                {
                    "role": "user",
                    "content": f"上下文：\n{context_text}\n\n问题：{query}"
                }
            ],
            temperature=0.3,
            max_tokens=1000
        )
        return response.choices[0].message.content

    def chat(self, query: str) -> dict:
        """端到端 RAG 流水线。"""
        context = self.retrieve(query)
        answer = self.generate(query, context)
        return {
            'query': query,
            'answer': answer,
            'sources': [doc['title'] for doc in context],
            'similarity_scores': [doc['similarity'] for doc in context]
        }

# 使用
rag = SupabaseRAG(SUPABASE_URL, SUPABASE_KEY, OPENAI_KEY)
result = rag.chat("Docker 最佳实践是什么？")
print(result['answer'])

-- 启用行级安全
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;

-- 创建策略：用户只能读取自己的文档
CREATE POLICY "用户可读取自己的文档"
ON documents FOR SELECT
USING (auth.uid() = user_id);

-- 创建策略：用户可以插入自己的文档
CREATE POLICY "用户可插入自己的文档"
ON documents FOR INSERT
WITH CHECK (auth.uid() = user_id);

// 注册新用户
const { data: authData, error: authError } = await supabase.auth.signUp({
  email: 'user@example.com',
  password: 'secure-password-123'
})

// 登录
const { data: session } = await supabase.auth.signInWithPassword({
  email: 'user@example.com',
  password: 'secure-password-123'
})

// API 调用的访问令牌
const accessToken = session.session?.access_token

// 带认证上下文的查询（RLS 自动执行）
const { data } = await supabase
  .from('documents')
  .select('*')

# Python：服务端认证使用 service role key
supabase_admin = create_client(SUPABASE_URL, SERVICE_ROLE_KEY)

# 绕过 RLS 进行管理员操作
all_docs = supabase_admin.table('documents').select('*').execute()

// 订阅实时数据库变更
const channel = supabase
  .channel('documents-changes')
  .on(
    'postgres_changes',
    { event: 'INSERT', schema: 'public', table: 'documents' },
    (payload) => {
      console.log('新文档插入:', payload.new)
      // 触发重新索引、通知或 UI 更新
    }
  )
  .subscribe()

// 完成时取消订阅
supabase.removeChannel(channel)

# Python asyncio 版本
import asyncio

async def subscribe_to_changes():
    channel = supabase.channel('documents-changes')
    
    def handle_insert(payload):
        print(f"新文档: {payload['new']['title']}")
    
    channel.on(
        'postgres_changes',
        {'event': 'INSERT', 'schema': 'public', 'table': 'documents'},
        handle_insert
    ).subscribe()

asyncio.run(subscribe_to_changes())

# 初始化 edge 函数
supabase functions new ai-completion

# 编辑生成的文件
# supabase/functions/ai-completion/index.ts

// supabase/functions/ai-completion/index.ts
import { serve } from 'https://deno.land/std@0.224.0/http/server.ts'

serve(async (req) => {
  const { prompt } = await req.json()

  // 从边缘调用 OpenAI API
  const response = await fetch('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${Deno.env.get('OPENAI_API_KEY')}`,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      model: 'gpt-4.1-mini',
      messages: [{ role: 'user', content: prompt }],
      max_tokens: 500
    })
  })

  const data = await response.json()

  return new Response(
    JSON.stringify({ completion: data.choices[0].message.content }),
    { headers: { 'Content-Type': 'application/json' } }
  )
})

# 设置密钥
supabase secrets set OPENAI_API_KEY=sk-...

# 部署函数
supabase functions deploy ai-completion

# 通过 HTTP 调用
supabase functions invoke ai-completion --data '{"prompt": "Explain RAG"}'

# docker-compose.prod.yml（摘录）
services:
  db:
    image: supabase/postgres:15.8.1.040
    environment:
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
      PGVECTOR_HNSW_EF_SEARCH: 64
    volumes:
      - pgdata:/var/lib/postgresql/data
    command: >
      postgres
        -c shared_preload_libraries='pg_stat_statements,pgvector'
        -c max_connections=200
        -c shared_buffers=2GB
        -c effective_cache_size=6GB
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 5s
      retries: 5

  kong:
    image: kong:3.7
    environment:
      KONG_DATABASE: "off"
      KONG_DECLARATIVE_CONFIG: /var/lib/kong/kong.yml
    ports:
      - "8000:8000"
    depends_on:
      - auth
      - rest
      - realtime

volumes:
  pgdata:

# 生产环境的 .env 文件
POSTGRES_PASSWORD=$(openssl rand -base64 32)
JWT_SECRET=$(openssl rand -base64 32)
ANON_KEY=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...
SERVICE_ROLE_KEY=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

# SMTP 用于认证邮件
SMTP_HOST=smtp.sendgrid.net
SMTP_PORT=587
SMTP_USER=apikey
SMTP_PASS=SG.xxx

# S3 兼容存储
STORAGE_S3_BUCKET=your-bucket
STORAGE_S3_ENDPOINT=s3.amazonaws.com
STORAGE_S3_ACCESS_KEY=AKIA...
STORAGE_S3_SECRET_KEY=...

# 使用 pg_dump 自动每日备份
0 2 * * * docker exec supabase-db pg_dump -U postgres -Fc postgres > /backups/supabase-$(date +\%Y\%m\%d).dump

# 或使用 Supabase 内置的时间点恢复（PITR）
# Pro 层及以上可用

# 创建新迁移
supabase migration new add_documents_table

# 编辑生成的 SQL 文件
# supabase/migrations/20260519000000_add_documents_table.sql

# 应用到本地实例
supabase db reset

# 部署到生产环境
supabase db push

# 从模式生成 TypeScript 类型
supabase gen types typescript --local > src/types/supabase.ts