用 pgEdge Cloud API 搭一个贴着 Postgres 跑的 RAG 服务

2026-07-03 39 预计阅读时间: 1 分钟
来源: postgr.es AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间：13 分钟

RAG 系统最容易失控的地方，不是“大模型会不会回答”，而是数据、切块、向量、检索、服务部署这几段胶水谁来维护。pgEdge Cloud 的思路很直接：把内容放进 Postgres，让数据库侧完成切块和向量化，再把 RAG 服务挂到数据库旁边。UI 里点几下就能完成，但用 API 做一遍，更适合自动化、复现和团队交付。

这篇文章按一个真实路径拆开：本地准备带向量的数据库，导入 pgEdge Cloud，用 API 创建集群和数据库，最后部署一个可以直接 HTTP 调用的 RAG pipeline。

关键变化：RAG 不再是数据库旁边的一堆脚本

传统 RAG 项目经常长这样：

  • 一个脚本负责从 PDF、Markdown 或网页里抽文本；
  • 一个脚本负责切块；
  • 一个 worker 调 embedding API；
  • 一个向量库负责相似度搜索；
  • 一个后端服务把检索结果拼进 prompt；
  • 另一个地方再处理权限、部署、重试和监控。

pgEdge 的方案把其中几件事收回到 Postgres 里：

  • pgvector 做向量相似度；
  • pgedge-vectorizer 做自动切块和 embedding；
  • vchord_bm25 与 pg_tokenizer 做关键词检索；
  • pg_cron 和后台 worker 处理异步任务；
  • Cloud RAG service 贴着数据库运行，负责查询 embedding、混合检索、上下文裁剪和最终回答。

这不是说你不用设计 RAG 了，而是你少维护了很多“胶水进程”。数据一旦进表，数据库会异步把它切成 chunk、调用 embedding provider、写回向量列，并在队列失败时重试。

第一步：先把内容变成 Postgres 里的可检索数据

源文里的示例数据是几十本桌面角色扮演规则书 PDF。你的场景可以换成内部知识库、客服文档、API 手册、事故复盘、合规制度。核心目标只有一个：把文本放进 Postgres 表，并让数据库生成 chunk 和向量。

可以这样实践：先在本地用 pgEdge Enterprise Postgres 容器启动一套带向量扩展的环境。下面示例假设你已经有 embedding provider 的 API key，并把它放在本地文件 ./secrets/vectorizer_api_key 中。

# docker-compose.yml
services:
  postgres:
    image: pgedge/enterprise-postgres:latest
    container_name: rag-postgres
    ports:
      - "5432:5432"
    environment:
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: ragdemo
    volumes:
      - ./pgdata:/var/lib/postgresql/data
      - ./secrets/vectorizer_api_key:/run/secrets/vectorizer_api_key:ro
    command:
      - "postgres"
      - "-c"
      - "shared_preload_libraries=pgedge_vectorizer,pg_cron"
      - "-c"
      - "pgedge_vectorizer.provider=voyage"
      - "-c"
      - "pgedge_vectorizer.model=voyage-3-large"
      - "-c"
      - "pgedge_vectorizer.api_key_path=/run/secrets/vectorizer_api_key"
      - "-c"
      - "pgedge_vectorizer.batch_size=32"
      - "-c"
      - "pgedge_vectorizer.max_workers=4"

启动后创建一张内容表。具体的 vectorizer 函数名和参数应以你使用的 pgEdge 镜像文档为准；这里的重点是表结构：原文、来源、向量列和索引要在同一个数据库里闭环。

CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS vchord_bm25;
CREATE EXTENSION IF NOT EXISTS pg_tokenizer;

CREATE TABLE docs (
  id bigserial PRIMARY KEY,
  source text NOT NULL,
  body text NOT NULL,
  embedding vector(1024)
);

CREATE INDEX docs_embedding_hnsw_idx
  ON docs USING hnsw (embedding vector_cosine_ops);

把文本导进去之后，pgedge-vectorizer 会通过触发器和后台 worker 处理 chunk 与 embedding。你真正需要盯的是队列是否排空。可以按实际扩展暴露的队列表或视图改造下面的检查语句：

-- 按你的 vectorizer 队列表/视图名称调整
SELECT count(*) AS pending_chunks
FROM pgedge_vectorizer_queue
WHERE status IN ('pending', 'retrying');

当 pending 数量归零，说明 chunk 的向量已经生成，HNSW 索引也可以参与检索。接下来导出数据库，准备搬到 Cloud：

pg_dump \
  --format=custom \
  --file=ragdemo.dump \
  --dbname='postgresql://postgres:postgres@localhost:5432/ragdemo'

自定义格式的 pg_dump 适合跨网络恢复，schema、扩展、索引、内容和 embedding 都会一起带走，Cloud 侧不需要重新算一遍向量。

第二步：用 Cloud API 创建集群和数据库

UI 能点出来的东西，API 也应该能复现。这样做的好处是：开发、测试、生产环境可以用同一套脚本创建；出问题也能从命令历史和配置里查清楚。

先用 Cloud 控制台创建 API client，拿到 client ID 和 secret，再换取短期 bearer token。不要把 secret 直接写进 shell history，可以用环境变量或 secret manager。

export PGEDGE_CLIENT_ID='replace-with-client-id'
export PGEDGE_CLIENT_SECRET='replace-with-client-secret'

export PGEDGE_TOKEN=$(
  curl -sS -X POST 'https://api.pgedge.com/oauth/token' \
    -H 'content-type: application/json' \
    -d "{\"client_id\":\"$PGEDGE_CLIENT_ID\",\"client_secret\":\"$PGEDGE_CLIENT_SECRET\",\"grant_type\":\"client_credentials\"}" \
  | jq -r '.access_token'
)

创建单节点公开集群时，要考虑一个关键选择：public 还是 private。公开集群会有 TLS endpoint，适合只读知识库或非敏感数据；私有集群不暴露到公网，更适合内部文档、客户数据和受监管场景。

下面是一个可改造的公开集群创建请求。字段名请以你当前 pgEdge Cloud API 版本为准：

curl -sS -X POST 'https://api.pgedge.com/v1/clusters' \
  -H "authorization: Bearer $PGEDGE_TOKEN" \
  -H 'content-type: application/json' \
  -d '{
    "name": "rag-demo",
    "cloud": "aws",
    "region": "us-east-1",
    "nodes": [
      { "name": "n1", "location": "public" }
    ],
    "firewall_rules": [
      { "port": 5432, "source": "203.0.113.10/32" },
      { "port": 443, "source": "0.0.0.0/0" }
    ]
  }'

这里的 443 是给后面的 RAG endpoint 用的。如果你创建的是私有集群，就不应该开放公网 443，而是通过 ingress 或隧道访问服务。

创建数据库的调用同样可以脚本化。API 会帮数据库配置备份存储，返回 host、port、database ID 等连接信息。创建完成后，等待状态进入 ready，再恢复 dump：

pg_restore \
  --verbose \
  --clean \
  --if-exists \
  --dbname='postgresql://app_user:replace-password@replace-host:5432/ragdemo' \
  ragdemo.dump

如果是私有集群，可以用云厂商的隧道或 SSM port forwarding 连接到本地端口，再用 hostaddr=127.0.0.1 保持 TCP 走本地隧道，同时保留真实 host 供路由和证书使用。

第三步：把 RAG 服务挂到数据库上

RAG 服务不是另起一台你自己维护的应用服务器，而是 pgEdge Cloud 上附着到数据库的 service。你给它一组 pipeline 配置：查哪张表、哪个文本列、哪个向量列，用哪个 embedding 模型查相似度，用哪个 completion 模型生成回答，检索时如何混合向量和关键词结果。

一个典型 pipeline 里最容易踩坑的是 embedding 模型一致性。你用 voyage-3-large 生成文档向量，查询时也应该用同一类 embedding 模型；否则相似度比较就像拿两套坐标系硬算距离。

可以这样定义 RAG service：

export DATABASE_ID='replace-with-database-id'

curl -sS -X POST "https://api.pgedge.com/v1/databases/$DATABASE_ID/services" \
  -H "authorization: Bearer $PGEDGE_TOKEN" \
  -H 'content-type: application/json' \
  -d '{
    "type": "rag",
    "name": "rules-rag",
    "embedding_provider": "voyage",
    "embedding_model": "voyage-3-large",
    "completion_provider": "anthropic",
    "completion_model": "claude-sonnet-4-6",
    "pipelines": [
      {
        "name": "ask",
        "table": "docs",
        "text_column": "body",
        "vector_column": "embedding",
        "system_prompt": "Answer only from the supplied context. Cite sources when possible. If the answer is not in the context, say so.",
        "retrieval": {
          "mode": "hybrid",
          "vector_weight": 0.5,
          "keyword_weight": 0.5,
          "top_k": 40,
          "context_token_budget": 6000
        }
      }
    ]
  }'

这个配置里有几个值得认真调的旋钮：

  • mode: hybrid：同时使用向量相似度和 BM25 关键词检索；
  • vector_weight / keyword_weight：决定语义匹配和精确词匹配谁更有话语权；
  • top_k：先捞多少候选 chunk，越大越可能覆盖答案，也越慢；
  • context_token_budget：控制塞给 completion model 的上下文上限，直接影响成本和延迟；
  • system_prompt：约束回答边界，尤其要明确“没有依据就说不知道”。

部署时数据库可能短暂进入更新状态。轮询 database 和 service 状态，等它们回到 ready，再调用 endpoint。

第四步：像调用普通 HTTP API 一样提问

公开集群通常会得到一个服务域名，pipeline 名就是路径的一部分。私有集群则通过 ingress 暴露入口；调用方式类似，只是 URL 来源不同。

export RAG_ENDPOINT='https://replace-service-domain.example.com'

curl -sS -X POST "$RAG_ENDPOINT/ask" \
  -H 'content-type: application/json' \
  -d '{
    "query": "What is the cost of Danger Sense, and where is it cited?"
  }' | jq

返回结果通常会包含整理后的回答和引用信息。你可以把这个 endpoint 接到 Slack bot、客服后台、内部搜索框，或者像源文那样接到一堆规则书上，用来快速终结饭桌旁的规则争论。

更工程化一点，可以把调用封装成一个小脚本，方便批量测试问题集：

# ask_rag.py
import os
import sys
import requests

endpoint = os.environ["RAG_ENDPOINT"].rstrip("/")
pipeline = os.environ.get("RAG_PIPELINE", "ask")
query = " ".join(sys.argv[1:]) or "Summarize the most important rule in this corpus."

response = requests.post(
    f"{endpoint}/{pipeline}",
    json={"query": query},
    timeout=60,
)
response.raise_for_status()
print(response.json())

运行：

export RAG_ENDPOINT='https://replace-service-domain.example.com'
python ask_rag.py 'How does Rapid Strike work, and what changes with Weapon Master?'

上线前的取舍清单

这套架构吸引人的地方是简单：Postgres 是内容库、向量库和检索核心，Cloud service 负责把检索和回答串起来。但上线前仍然要做几件严肃决定：

  • 敏感数据优先选 private cluster，并通过 ingress、VPN 或私有网络访问；
  • embedding 模型、维度和向量列必须一致，迁移模型时要计划重算；
  • 混合检索适合“语义 + 精确术语”场景，比如规则编号、产品型号、API 名称；
  • top_k 和 token budget 要用真实问题压测，不要只看 demo 效果；
  • system prompt 要明确引用、拒答和不确定性策略；
  • API key 不进命令历史、不进镜像、不进仓库。

如果你的知识主要已经在 Postgres 周边，pgEdge Cloud 的 RAG service 很适合做一条短路径：先把文档落表，把 chunk 和 embedding 留在数据库里，再用 API 把服务挂上去。它不替你解决内容质量问题，但能显著减少 RAG 基础设施的自建成本。

# AI # AWS # Go # LLM # OpenAI # PostgreSQL # 云原生 # 云平台 # 后端 # 工程效能 # 数据库 # 架构设计

相关推荐