用 CloudBase PostgreSQL + pgvector 做 RAG

一句话定义：在 CloudBase PostgreSQL 上启用 pgvector 扩展，写两个 Web 云函数（ingest 和 retrieve），跑通"文档 → 切块 → 嵌入 → 入库 → 检索 → 拼 prompt → LLM 回答"的完整 RAG 流水线，并按 user_id 做多租户隔离。

预计耗时：60 分钟 | 难度：进阶

注：CloudBase PostgreSQL 与 pgvector 扩展的可用状态以控制台 PostgreSQL 数据库 实际显示为准。若该服务在你的环境暂未开放，请先提工单确认或自建 pgvector 实例后回到本篇 ingest / retrieve 函数的写法。

适用场景

• 想给 chatbot 接一个"基于自己文档"的回答能力（产品手册、客服 FAQ、内部 wiki）
• 已经在用 CloudBase PostgreSQL，不想再单独买一套向量数据库
• 数据敏感，需要数据落在自己环境内、向量索引也在自己环境内

不适用：

• 文档数量在百万级以上、要求亚秒级检索的——pgvector 能跑但要严肃做索引调优，本篇是入门级方案
• 想用 keyword 检索 + 向量检索混合（BM25 + ANN）——可以做但需要额外接 ElasticSearch 或在 PostgreSQL 里写 GIN 索引，本篇不覆盖

环境要求

依赖	版本
Node.js（云函数运行时）	≥ 18
@cloudbase/cli	latest
Web 云函数	HTTP 触发
CloudBase PostgreSQL	已开通的 PostgreSQL 版本环境
pg（Node 的 PostgreSQL 客户端）	^8.11.0
pgvector 扩展	是否预装/可装请以控制台「数据库 → 扩展管理」实际选项为准

需要前置：

• 一个能用的 LLM 网关（参考 云函数代理 LLM API）；后面会同时调 chat 和 embedding 两个 endpoint
• CloudBase PostgreSQL 数据库账号（在控制台「数据库 → PostgreSQL → 设置 → 账号管理」里创建，记下用户名 + 密码 + 内网/公网连接地址）

第一步：启用 pgvector 扩展，建表

进入 DMC 数据库管理工具，用刚才创建的账号登录。在 SQL 窗口 里执行：

-- 启用 pgvector 扩展
CREATE EXTENSION IF NOT EXISTS vector;

-- 建文档表(注意 vector(1536) 的维度要和你用的 embedding 模型对得上)
CREATE TABLE IF NOT EXISTS documents (
  id          BIGSERIAL PRIMARY KEY,
  user_id     TEXT NOT NULL,           -- 多租户隔离用,后续每次查询都要 WHERE user_id = $1
  source      TEXT,                    -- 文档来源(文件名 / URL 等),便于溯源
  chunk_index INT NOT NULL,            -- 当前块在原文档中的序号
  content     TEXT NOT NULL,           -- 块的原文
  embedding   vector(1536) NOT NULL,   -- OpenAI text-embedding-3-small 是 1536 维
  created_at  TIMESTAMPTZ DEFAULT NOW()
);

-- 多租户检索索引
CREATE INDEX IF NOT EXISTS documents_user_id_idx ON documents(user_id);

-- 向量近似最近邻索引(IVFFlat,适合中等数据量)
-- lists 一般取 sqrt(N),先用 100,数据量上来再调
CREATE INDEX IF NOT EXISTS documents_embedding_idx
  ON documents
  USING ivfflat (embedding vector_cosine_ops)
  WITH (lists = 100);

几个要点：

• CREATE EXTENSION vector 如果报权限错或"扩展不存在"，去控制台看「数据库 → 扩展管理」是否需要先在白名单里勾选；不同 CloudBase 版本和地域开放的扩展不一样，以控制台实际选项为准。
• vector(1536) 的维度必须和你的 embedding 模型对应。text-embedding-3-small 是 1536，text-embedding-3-large 是 3072，混元 embedding 是 1024。建表前先确定要用哪个，选错了需要 drop 重建。
• IVFFlat 索引在表为空时建会报警告但不会失败；推荐数据量到几千行后再 CREATE INDEX，否则索引质量很差。可以先建表灌数据再建索引。
• HNSW 索引（pgvector 0.5+ 支持）效果通常比 IVFFlat 好，但建索引慢、内存高，适合数据量大的场景。本篇用 IVFFlat 是因为最稳。

第二步：写 ingest 云函数（入库）

mkdir rag-ingest && cd rag-ingest
npm init -y
npm install --save express pg

index.js：

const express = require('express');
const { Pool } = require('pg');

const app = express();
app.use(express.json({ limit: '20mb' }));

const pool = new Pool({
  host: process.env.PG_HOST,           // 内网连接地址
  port: parseInt(process.env.PG_PORT || '5432', 10),
  database: process.env.PG_DATABASE,
  user: process.env.PG_USER,
  password: process.env.PG_PASSWORD,
  max: 10,
  idleTimeoutMillis: 30000,
});

const LLM_PROXY_URL = process.env.LLM_PROXY_URL;     // https://xxx.service.tcloudbase.com/llm-proxy/v1
const LLM_PROXY_TOKEN = process.env.LLM_PROXY_TOKEN; // 第一篇里的 PROXY_ACCESS_TOKEN
const EMBED_MODEL = process.env.EMBED_MODEL || 'text-embedding-3-small';

// 简单切块:按段落 + 字符上限
function chunkText(text, maxChars = 1000, overlap = 100) {
  const paragraphs = text.split(/\n{2,}/).map((p) => p.trim()).filter(Boolean);
  const chunks = [];
  let buf = '';
  for (const p of paragraphs) {
    if (buf.length + p.length + 2 <= maxChars) {
      buf = buf ? buf + '\n\n' + p : p;
    } else {
      if (buf) chunks.push(buf);
      // 长段落超出 maxChars 时硬切
      if (p.length > maxChars) {
        for (let i = 0; i < p.length; i += maxChars - overlap) {
          chunks.push(p.slice(i, i + maxChars));
        }
        buf = '';
      } else {
        buf = p;
      }
    }
  }
  if (buf) chunks.push(buf);
  return chunks;
}

async function embed(text) {
  const resp = await fetch(`${LLM_PROXY_URL}/embeddings`, {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${LLM_PROXY_TOKEN}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({ model: EMBED_MODEL, input: text }),
  });
  if (!resp.ok) throw new Error(`embed failed: ${resp.status} ${await resp.text()}`);
  const json = await resp.json();
  return json.data[0].embedding;
}

app.post('/ingest', async (req, res) => {
  const { userId, source, text } = req.body || {};
  if (!userId || !text) {
    return res.status(400).json({ error: 'userId and text are required' });
  }

  const chunks = chunkText(text);
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    for (let i = 0; i < chunks.length; i++) {
      const vec = await embed(chunks[i]);
      // pgvector 接受 '[1,2,3]' 字符串字面量
      const vecLiteral = `[${vec.join(',')}]`;
      await client.query(
        `INSERT INTO documents (user_id, source, chunk_index, content, embedding)
         VALUES ($1, $2, $3, $4, $5::vector)`,
        [userId, source || '', i, chunks[i], vecLiteral]
      );
    }
    await client.query('COMMIT');
    res.json({ inserted: chunks.length });
  } catch (err) {
    await client.query('ROLLBACK').catch(() => {});
    console.error('ingest failed', err);
    res.status(500).json({ error: 'ingest_failed', message: err.message });
  } finally {
    client.release();
  }
});

app.listen(process.env.PORT || 9000);

切块策略说明：上面这个是最朴素的版本——按双换行分段、超长硬切、保留 100 字符 overlap。生产里常用 LangChain 的 RecursiveCharacterTextSplitter 或 tiktoken 按 token 数切，这里不展开。重点是切块大小要和 LLM 上下文窗口 + 检索 top-K 的乘积匹配——常见配置是 chunk 1000 字符 + top 5 = 5000 字符上下文。

第三步：写 retrieve 云函数（检索 + 回答）

mkdir rag-retrieve && cd rag-retrieve
npm init -y
npm install --save express pg

index.js：

const express = require('express');
const { Pool } = require('pg');

const app = express();
app.use(express.json({ limit: '5mb' }));

const pool = new Pool({
  host: process.env.PG_HOST,
  port: parseInt(process.env.PG_PORT || '5432', 10),
  database: process.env.PG_DATABASE,
  user: process.env.PG_USER,
  password: process.env.PG_PASSWORD,
  max: 10,
});

const LLM_PROXY_URL = process.env.LLM_PROXY_URL;
const LLM_PROXY_TOKEN = process.env.LLM_PROXY_TOKEN;
const EMBED_MODEL = process.env.EMBED_MODEL || 'text-embedding-3-small';
const CHAT_MODEL = process.env.CHAT_MODEL || 'gpt-4o-mini';

async function embed(text) {
  const resp = await fetch(`${LLM_PROXY_URL}/embeddings`, {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${LLM_PROXY_TOKEN}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({ model: EMBED_MODEL, input: text }),
  });
  if (!resp.ok) throw new Error(`embed failed: ${resp.status}`);
  const json = await resp.json();
  return json.data[0].embedding;
}

app.post('/ask', async (req, res) => {
  const { userId, question, topK = 5 } = req.body || {};
  if (!userId || !question) {
    return res.status(400).json({ error: 'userId and question are required' });
  }

  // 1. 嵌入问题
  const qVec = await embed(question);
  const qLiteral = `[${qVec.join(',')}]`;

  // 2. 向量近邻检索(注意 user_id 过滤,多租户隔离的关键)
  const { rows } = await pool.query(
    `SELECT id, source, chunk_index, content,
            1 - (embedding <=> $1::vector) AS similarity
     FROM documents
     WHERE user_id = $2
     ORDER BY embedding <=> $1::vector
     LIMIT $3`,
    [qLiteral, userId, topK]
  );

  if (rows.length === 0) {
    return res.json({
      answer: '没有找到相关文档,无法回答。请先用 /ingest 入库。',
      sources: [],
    });
  }

  // 3. 拼 prompt
  const context = rows
    .map((r, i) => `[片段 ${i + 1}|来源 ${r.source}]\n${r.content}`)
    .join('\n\n');

  const messages = [
    {
      role: 'system',
      content:
        '你是一个严谨的助手,严格基于「参考资料」回答问题。' +
        '如果资料里没有,直接说"参考资料中未找到相关信息",不要编造。' +
        '回答用简体中文,引用片段时标注来源编号。',
    },
    {
      role: 'user',
      content: `参考资料:\n\n${context}\n\n问题:${question}`,
    },
  ];

  // 4. 调 LLM
  const llmResp = await fetch(`${LLM_PROXY_URL}/chat/completions`, {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${LLM_PROXY_TOKEN}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({ model: CHAT_MODEL, messages }),
  });
  if (!llmResp.ok) {
    return res.status(502).json({ error: 'llm_failed', message: await llmResp.text() });
  }
  const json = await llmResp.json();
  const answer = json.choices?.[0]?.message?.content || '';

  res.json({
    answer,
    sources: rows.map((r) => ({
      source: r.source,
      chunkIndex: r.chunk_index,
      similarity: Number(r.similarity).toFixed(4),
      preview: r.content.slice(0, 120),
    })),
  });
});

app.listen(process.env.PORT || 9000);

注意几个细节：

• embedding <=> $1::vector 是 pgvector 的余弦距离操作符；越小越相似。1 - (embedding <=> ...) 转成"相似度分数"，越大越相似，方便阅读。
• WHERE user_id = $2 ORDER BY embedding <=> $1 是多租户隔离 + 向量检索的标准写法。如果 user_id 列上没建 B-tree 索引（前面建表 SQL 里有），数据量上来后会全表扫，性能很差。
• 没有命中相关文档时不要硬塞空 context 给 LLM——它会"基于自己的训练数据"瞎答，破坏 RAG 的事实一致性。直接告诉用户"没找到"更合适。

第四步：部署 + 配环境变量

两个云函数都用同样流程：

cd rag-ingest
tcb fn deploy rag-ingest --httpFn -e your-env-id

cd ../rag-retrieve
tcb fn deploy rag-retrieve --httpFn -e your-env-id

每个函数控制台 → 环境变量配：

Key	说明
PG_HOST	PostgreSQL 内网连接地址
PG_PORT	默认 5432
PG_DATABASE	数据库名
PG_USER	DMC 创建的账号名
PG_PASSWORD	该账号密码
LLM_PROXY_URL	第一篇里部署的代理 URL（带 /v1）
LLM_PROXY_TOKEN	第一篇里的 PROXY_ACCESS_TOKEN
EMBED_MODEL	默认 text-embedding-3-small（1536 维）
CHAT_MODEL	默认 gpt-4o-mini

由于云函数要访问 PostgreSQL，确认控制台「网络配置」里把云函数和数据库挂在同一个 VPC，否则连不通。

运行验证

灌一段文档：

curl -X POST 'https://your-env.service.tcloudbase.com/rag-ingest/ingest' \
  -H 'Content-Type: application/json' \
  -d '{
    "userId": "u-001",
    "source": "company-handbook.md",
    "text": "公司年假政策:入职满 1 年可享 5 天带薪年假,满 5 年享 10 天,满 10 年享 15 天。\n\n报销流程:发票需在 30 天内提交,超过期限不予报销。"
  }'
# 预期: { "inserted": 2 }

提问：

curl -X POST 'https://your-env.service.tcloudbase.com/rag-retrieve/ask' \
  -H 'Content-Type: application/json' \
  -d '{
    "userId": "u-001",
    "question": "我入职 6 年,有几天年假?"
  }'

预期返回：

{
  "answer": "根据参考资料,入职满 5 年可享 10 天带薪年假...[片段 1]",
  "sources": [
    { "source": "company-handbook.md", "chunkIndex": 0, "similarity": "0.7821", "preview": "公司年假政策:..." }
  ]
}

如果换一个 userId（比如 u-999）问同样问题，应该返回"没有找到相关文档"——这就是 user_id 隔离生效的证据。

多租户隔离要点

风险	防御
跨租户查询	每条 SQL 必须 WHERE user_id = $X，写到 ORM 的 base scope 里更稳
用户冒充	userId 不能从前端 body 直接信任，必须从 CloudBase 鉴权后的会话里拿
租户数据导出	只能 SELECT ... WHERE user_id = ?，不能 SELECT * FROM documents
删除租户	写一个 DELETE FROM documents WHERE user_id = ? 的清理函数，跑 GDPR 流程时用

常见错误

错误信息	原因	修复
extension "vector" is not available	pgvector 没在白名单	控制台「数据库 → 扩展管理」勾选 vector，或联系 CloudBase 工单
dimensions mismatch: column has 1536 but value has 3072	灌库用了 large、查询用了 small（或反过来）	全链路统一一个 embedding 模型；改了模型就 drop 表重灌
检索结果一片 0 相似度	embedding 列里灌进去的不是真向量，是字符串	检查 INSERT 时 $5::vector 的 cast；用 SELECT pg_typeof(embedding) 看实际类型
查询很慢（几秒一次）	没建 IVFFlat 索引；或者 lists 设置太低	数据量 > 1 万行后建索引；lists = sqrt(rows) 经验值；可以试 HNSW
password authentication failed for user	PostgreSQL 账号密码错；或 Public IP / 内网混淆	重置 DMC 账号密码；确认云函数和 DB 在同 VPC，用内网地址
ingest 灌大文件超时	单次 HTTP 请求超过函数 30 秒上限	客户端分批调；或者 ingest 改成「先存原始文档→定时函数异步嵌入」两步
答案明显在编造	没命中相关 chunk，但 LLM 仍然作答	system prompt 里强调"资料里没有就说没有"；rows.length === 0 时直接 short-circuit 返回
pgvector 距离操作符报错 <=> does not exist	扩展未启用	CREATE EXTENSION vector;

相关文档

• CloudBase PostgreSQL 概述 — 平台数据库能力
• DMC 数据库管理 — 直接执行 SQL 的入口
• PostgreSQL FAQ — pg 协议直连 — 用原生 PostgreSQL 客户端的连接说明
• HTTP 云函数 — Web 云函数基础
• pgvector 官方文档 — 索引调优、距离函数选型

下一步

RAG 跑通后建议接着做：

• add-vercel-ai-sdk-streaming-chatbot — 把 retrieve 函数接到流式 chat 界面，用户体验更顺
• connect-openai-api-cloud-function — 如果还没做代理层，建议先做，统一 embedding 和 chat 的鉴权
• secure-secrets-in-cloud-function — PG_PASSWORD 等数据库密码的环境分层管理