# 用 FastAPI + pgvector 跑通最小可运行私有知识库
**作者**: Ando
**日期**: 2026-07-03T15:45:46.000Z
**来源**: [https://x.com/ando_w/status/2073979240962228582](https://x.com/ando_w/status/2073979240962228582)
---
上一篇把 RAG 容易翻车的地方讲清楚了。这一篇把那套架构落成一份真能跑的工程。
我会带你从 0 到 1 搭一套最小可运行版本：FastAPI 做接口，PostgreSQL + pgvector 做存储，一个简单的 Chat 前端，文档入库 → 向量检索 → 答案生成 → 引用返回，全部串通。
跑完这一篇，你就有一套能放进自己项目里的私有知识库底座。后面要换成 Qdrant、Milvus，或者把基础 RAG 升级成 Agentic RAG，结构都不用大改。
> **Ando@ando_w**: [原文链接](https://x.com/ando_w/status/2073070705789329561)
>
# 零、为什么用千问，不用 GPT 或 Claude
先说清楚，这篇教程的 LLM 用的是通义千问 Qwen3.7-Max，Embedding 用的是Qwen3-Embedding，全链路跑在阿里云百炼（DashScope）上。
不是标新立异。原因很实际：
1. OpenAI 最新的 GPT-5.6 系列（Sol/Terra/Luna）目前只向「可信合作伙伴」限量开放，API 公开版还停在 GPT-5.5。Claude Opus 4.8 前阵子被社区集体吐槽降智。你拿这两个模型写教程，企业第一反应是「我连模型都用不上」。
2. Qwen3.7-Max 5 月份发布的，对标 GPT-5.5 和 Opus 4.7，百万 Token 上下文，OpenAI 接口完全兼容。DashScope 的 API 地址换成 dashscope.aliyuncs.com/v1，代码一行不用改。
3. Qwen3-Embedding 是专门为检索和重排设计的，不是拿通用模型凑合。RAG 场景里，Embedding 质量直接决定召回率。
4. 成本。Qwen3.7-Max 的调用成本比 GPT-5.5 低一个量级。企业 RAG 跑起来每天几千次调用，这个差价是实打实的。
代码层面，你随时可以把 qwen3.7-max 换成 gpt-5.5 或者 claude-sonnet-5，改一个字符串的事。架构不动。
# 一、先定一个能跑通的小目标
别一上来就想着支持「公司所有文档」。挑一个窄场景，把 5 件事跑通：
1. 把 50 份 Markdown 文档读进来
2. 切分成 chunk，写进 PostgreSQL + pgvector
3. 用一个 HTTP 接口接收问题
4. 检索出最相关的 5 个 chunk
5. 调大模型生成答案，附上引用来源
跑通这 5 件事，你就已经超过 80% 的「RAG demo」了。它们不是 demo，是最小可运行工程。
下面所有的代码都可以直接复制运行，前提是：
- Python 3.11+
- PostgreSQL 16 + pgvector 扩展
- 一个阿里云百炼 API Key（去 https://bailian.console.aliyun.com 申请，免费额度够测试）
- 如果你更想用 OpenAI / DeepSeek / 智谱，改一下 config 里的 base_url 和 model 就行，代码完全兼容
# 二、目录结构
```
knowledge-base/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 入口
│ ├── config.py # 配置
│ ├── db.py # PostgreSQL 连接
│ ├── models.py # SQLAlchemy 模型
│ ├── ingest.py # 文档入库
│ ├── retrieve.py # 检索
│ ├── generate.py # 生成答案
│ └── schemas.py # Pydantic schema
├── data/
│ └── docs/ # 你的 Markdown 文档
├── requirements.txt
└── README.md
```
不要嫌这个结构啰嗦。它已经够小了，文件多了就分不清谁依赖谁。
# 三、依赖
requirements.txt：
```
fastapi==0.115.0
uvicorn[standard]==0.30.6
sqlalchemy==2.0.35
psycopg[binary]==3.2.3
pgvector==0.3.6
openai==1.54.0
pydantic==2.9.0
pydantic-settings==2.5.2
python-multipart==0.0.12
tiktoken==0.8.0
tenacity==9.0.0
```
pgvector 这个 Python 包是 SQLAlchemy 的方言适配器，让你能用 Column(Vector(1024)) 这种方式定义向量字段。
# 四、配置
app/config.py：
```
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):
# 数据库
database_url: str = "postgresql+psycopg://postgres:postgres@localhost:5432/kb"
# Embedding — 千问 Qwen3-Embedding，走 DashScope
embedding_base_url: str = "https://dashscope.aliyuncs.com/v1"
embedding_api_key: str = "sk-xxxxx" # 百炼 API Key
embedding_model: str = "text-embedding-v3"
embedding_dim: int = 1024 # Qwen3-Embedding 默认输出 1024 维
# LLM — 千问 Qwen3.7-Max
llm_base_url: str = "https://dashscope.aliyuncs.com/v1"
llm_api_key: str = "sk-xxxxx" # 同一个百炼 Key
llm_model: str = "qwen3.7-max"
# 切分参数
chunk_size: int = 600
chunk_overlap: int = 100
# 检索参数
top_k: int = 5
class Config:
env_file = ".env"
@lru_cache
def get_settings():
return Settings()
```
几个要点：
- text-embedding-v3 是百炼平台上的 Qwen3-Embedding 模型，输出 1024 维向量。如果你用 OpenAI 的 text-embedding-3-small，维度是 1536——记得同步改 embedding_dim 和数据库里的 Vector(1024)。
- qwen3.7-max 是千问旗舰模型。日常简单问答可以换 qwen3.6-plus（更便宜），复杂推理用 qwen3.7-max。
- DashScope 的 API 完全兼容 OpenAI SDK，所以 openai 这个 Python 包直接能用，不用装额外的 SDK。
- 想换回 OpenAI？把 base_url 改成 https://api.openai.com/v1，llm_model 改成 gpt-5.5，embedding_model 改成 text-embedding-3-small，embedding_dim 改成 1536。完事。
# 五、数据库模型
app/models.py：
```
from sqlalchemy import Column, String, Text, DateTime, Integer, JSON
from sqlalchemy.dialects.postgresql import ARRAY
from pgvector.sqlalchemy import Vector
from datetime import datetime
import uuid
from app.db import Base
def gen_id():
return str(uuid.uuid4())
class Document(Base):
__tablename__ = "documents"
id = Column(String, primary_key=True, default=gen_id)
title = Column(String, nullable=False)
source = Column(String, nullable=False) # 原始文件路径或 URL
url = Column(String, nullable=True)
updated_at = Column(DateTime, default=datetime.utcnow)
permission_group = Column(String, default="all")
raw_hash = Column(String, nullable=False) # 用于增量更新
class Chunk(Base):
__tablename__ = "chunks"
id = Column(String, primary_key=True, default=gen_id)
document_id = Column(String, nullable=False, index=True)
section_path = Column(ARRAY(String), default=list)
content = Column(Text, nullable=False)
metadata_ = Column("metadata", JSON, default=dict)
embedding = Column(Vector(1024)) # 维度跟 embedding 模型一致，Qwen3-Embedding 是 1024
```
metadata_ 这个名字是因为 metadata 是 SQLAlchemy 保留字，列名还是 metadata。
# 六、文档入库
app/ingest.py：
```
import hashlib
from pathlib import Path
from datetime import datetime
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.db import SessionLocal
from app.models import Document, Chunk
from app.config import get_settings
from app.retrieve import embed_text # 下面会写
settings = get_settings()
def file_hash(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def split_markdown(text: str, chunk_size: int, overlap: int) -> list[dict]:
"""按二级标题切分，每段控制在 chunk_size 字左右。"""
sections = []
current_h2 = None
current_buf: list[str] = []
for line in text.split("\n"):
if line.startswith("## "):
# 切出一个 section
if current_buf:
sections.append({"h2": current_h2, "text": "\n".join(current_buf).strip()})
current_h2 = line[3:].strip()
current_buf = [line]
else:
current_buf.append(line)
if current_buf:
sections.append({"h2": current_h2, "text": "\n".join(current_buf).strip()})
chunks = []
for sec in sections:
text = sec["text"]
if len(text) <= chunk_size:
chunks.append({"section_path": [sec["h2"]], "text": text})
else:
# 段内再做按段切分
buf = ""
for para in text.split("\n\n"):
if len(buf) + len(para) <= chunk_size:
buf += "\n\n" + para
else:
if buf:
chunks.append({"section_path": [sec["h2"]], "text": buf.strip()})
buf = para
if buf:
chunks.append({"section_path": [sec["h2"]], "text": buf.strip()})
return chunks
def ingest_file(path: Path, db: Session):
raw = path.read_text(encoding="utf-8")
h = file_hash(path)
# 检查是否需要更新
existing = db.execute(
select(Document).where(Document.source == str(path))
).scalar_one_or_none()
if existing and existing.raw_hash == h:
return {"status": "skipped", "reason": "hash unchanged"}
if existing:
# 删除旧 chunk
db.query(Chunk).filter(Chunk.document_id == existing.id).delete()
doc = existing
doc.raw_hash = h
doc.updated_at = datetime.utcnow()
else:
doc = Document(
title=path.stem,
source=str(path),
raw_hash=h,
)
db.add(doc)
db.flush()
# 切分 + embedding
pieces = split_markdown(raw, settings.chunk_size, settings.chunk_overlap)
for piece in pieces:
emb = embed_text(piece["text"])
chunk = Chunk(
document_id=doc.id,
section_path=piece["section_path"],
content=piece["text"],
metadata_={"title": doc.title, "source": doc.source},
embedding=emb,
)
db.add(chunk)
db.commit()
return {"status": "ingested", "chunks": len(pieces)}
def ingest_dir(dir_path: str):
db = SessionLocal()
results = []
for p in Path(dir_path).rglob("*.md"):
results.append({"file": str(p), **ingest_file(p, db)})
db.close()
return results
```
这段代码里有一处故意写得不完整：embed_text 还没实现，因为它跟检索是同一套调用。我会在下一节里把两个都写出来。
切分用了最朴素的「按二级标题分大段 + 大段内按段拼」的方式。上一篇说过，复杂切分（语义切分、父子切分）后面再加，先跑通主线。
# 七、Embedding 和检索
app/retrieve.py：
```
from openai import OpenAI
from sqlalchemy import select, text
from sqlalchemy.orm import Session
from app.db import SessionLocal
from app.models import Chunk
from app.config import get_settings
settings = get_settings()
_client = OpenAI(
base_url=settings.embedding_base_url,
api_key=settings.embedding_api_key,
)
def embed_text(text: str) -> list[float]:
"""单条文本 embedding。"""
text = text.replace("\n", " ")
resp = _client.embeddings.create(
model=settings.embedding_model,
input=[text],
)
return resp.data[0].embedding
def embed_query(text: str) -> list[float]:
return embed_text(text)
def search_similar(question: str, top_k: int = None, permission: str = "all") -> list[dict]:
"""向量相似度检索。"""
if top_k is None:
top_k = settings.top_k
q_emb = embed_query(question)
db: Session = SessionLocal()
try:
# pgvector 的余弦距离运算符：<=> 越小越相似
# 1 - (embedding <=> :q) 是相似度，越大越相关
stmt = text("""
SELECT
c.id,
c.document_id,
c.content,
c.metadata,
1 - (c.embedding <=> :q_vec) AS score
FROM chunks c
JOIN documents d ON d.id = c.document_id
WHERE d.permission_group = :perm
ORDER BY c.embedding <=> :q_vec
LIMIT :k
""")
rows = db.execute(stmt, {
"q_vec": q_emb,
"k": top_k,
"perm": permission,
}).fetchall()
finally:
db.close()
return [
{
"chunk_id": r.id,
"document_id": r.document_id,
"content": r.content,
"metadata": r.metadata,
"score": float(r.score),
}
for r in rows
]
```
pgvector 提供了几个距离运算符：
- <-> 欧氏距离
- <#> 内积（负数）
- <=> 余弦距离（推荐做相似度）
记得入库前建索引：
```
CREATE EXTENSION IF NOT EXISTS vector;
CREATE INDEX IF NOT EXISTS chunks_embedding_idx
ON chunks USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);
```
lists 的经验值是行数的平方根的几分之一。1 万行用 100 就够，10 万行用 300。
# 八、生成答案
app/generate.py：
```
from openai import OpenAI
from app.config import get_settings
settings = get_settings()
_llm = OpenAI(
base_url=settings.llm_base_url,
api_key=settings.llm_api_key,
)
SYSTEM_PROMPT = """你是一个企业私有知识库助手。
请严格遵守以下规则：
1. 只根据"参考资料"中的内容回答，不要使用你自己的知识。
2. 如果参考资料不足以回答问题，请直接说"资料中没有提到"并说明缺少什么。
3. 答案中每个关键结论都要在末尾用 [n] 标注来源编号。
4. 区分事实和推断。如果是推断，要明确写"这是基于上述资料的推断"。
5. 不要编造参考资料里没有的细节、数字、人名、链接。
输出格式：
<结论>
<依据摘要>
来源：[1] [3]
<补充说明或限制条件>"""
def generate_answer(question: str, retrieved: list[dict]) -> dict:
if not retrieved:
return {
"answer": "资料中没有找到与这个问题相关的内容。",
"sources": [],
}
# 拼接上下文
context_parts = []
for i, r in enumerate(retrieved, 1):
meta = r["metadata"] or {}
title = meta.get("title", "未命名文档")
source = meta.get("source", "")
context_parts.append(
f"[{i}] 文档: {title}\n来源: {source}\n内容: {r['content']}\n"
)
context = "\n".join(context_parts)
user_prompt = f"""用户问题：{question}
参考资料：
{context}
请按系统提示的格式回答。"""
resp = _llm.chat.completions.create(
model=settings.llm_model,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_prompt},
],
temperature=0.2, # 低温度，让答案更稳定
)
return {
"answer": resp.choices[0].message.content,
"sources": [
{
"index": i,
"title": r["metadata"].get("title", ""),
"source": r["metadata"].get("source", ""),
"score": r["score"],
}
for i, r in enumerate(retrieved, 1)
],
}
```
temperature=0.2 是经验值。RAG 场景你不想让模型「自由发挥」，但又不能完全 0（否则会复读机）。0.1~0.3 都行。
# 九、API 入口
app/main.py：
```
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from app.retrieve import search_similar
from app.generate import generate_answer
from app.ingest import ingest_dir
app = FastAPI(title="私有知识库", version="0.1.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
class AskRequest(BaseModel):
question: str
permission: str = "all"
top_k: int = 5
class AskResponse(BaseModel):
answer: str
sources: list
@app.post("/ask", response_model=AskResponse)
def ask(req: AskRequest):
retrieved = search_similar(req.question, top_k=req.top_k, permission=req.permission)
result = generate_answer(req.question, retrieved)
return result
class IngestRequest(BaseModel):
dir: str
@app.post("/ingest")
def ingest(req: IngestRequest):
try:
results = ingest_dir(req.dir)
except Exception as e:
raise HTTPException(500, str(e))
return {"results": results}
@app.get("/health")
def health():
return {"status": "ok"}
```
启动：
```
uvicorn app.main:app --reload --port 8000
```
调用：
```
curl -X POST http://localhost:8000/ask \
-H "Content-Type: application/json" \
-d '{"question": "新员工出差住酒店能报多少？"}'
```
返回长这样：
```
{
"answer": "根据公司差旅制度，新员工出差住宿标准为：\n\n- 一线城市：500 元/晚\n- 二线城市：400 元/晚\n- 三线及以下：300 元/晚\n\n具体城市划分参照《差旅报销标准表》。\n\n来源：[1] [2]",
"sources": [
{
"index": 1,
"title": "差旅报销制度",
"source": "data/docs/差旅报销制度.md",
"score": 0.83
}
]
}
```
这就跑通了。
# 十、一个能用的 Chat 前端
frontend/index.html（纯静态，不引框架）：
```
<!DOCTYPE html>
<html lang="zh">
<head>
<meta charset="UTF-8">
<title>私有知识库</title>
<style>
body { font-family: -apple-system, sans-serif; max-width: 720px; margin: 40px auto; padding: 0 20px; }
#chat { border: 1px solid #ddd; padding: 20px; height: 500px; overflow-y: auto; border-radius: 8px; }
.msg { margin: 12px 0; }
.user { color: #2563eb; }
.assistant { color: #111; }
.src { font-size: 12px; color: #666; margin-top: 4px; }
input { width: 100%; padding: 10px; margin-top: 12px; font-size: 14px; }
</style>
</head>
<body>
<h2>私有知识库</h2>
<div id="chat"></div>
<input id="q" placeholder="问一个问题…" autofocus>
<script>
const chat = document.getElementById("chat");
const input = document.getElementById("q");
function append(role, text, sources) {
const div = document.createElement("div");
div.className = "msg " + role;
div.innerHTML = "<b>" + role + "：</b>" + text.replace(/\n/g, "<br>");
if (sources && sources.length) {
const s = document.createElement("div");
s.className = "src";
s.textContent = "来源：" + sources.map(x => x.title).join("、");
div.appendChild(s);
}
chat.appendChild(div);
chat.scrollTop = chat.scrollHeight;
}
input.addEventListener("keydown", async (e) => {
if (e.key !== "Enter") return;
const q = input.value.trim();
if (!q) return;
input.value = "";
append("user", q);
const r = await fetch("/ask", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ question: q }),
});
const data = await r.json();
append("assistant", data.answer, data.sources);
});
</script>
</body>
</html>
```
打开浏览器访问 http://localhost:8000/，就能问问题了。
# 十一、踩过的 5 个坑（写给你省时间）
## 坑 1：embedding 维度对不上
vector(1024) 是 Qwen3-Embedding（text-embedding-v3）的维度。如果你换成了 OpenAI 的 text-embedding-3-small（1536 维）或者 text-embedding-3-large（3072 维），一定要同步改 Vector(N) 和 embedding_dim，不然写入直接报错。
## 坑 2：ivfflat 索引在小数据上反而慢
如果你只有几百行 chunk，先别建索引，pgvector 暴力扫描很快。等数据上 1 万再考虑 ivfflat 或者 hnsw。
## 坑 3：用户问题太短，向量检索失败
「报多少」这种问题，embedding 出来几乎是个平均向量，谁都像又谁都不像。
解法是在 prompt 之前先做一次「查询改写」，把短问题补全。但这是 Agentic RAG 的活，下一篇再讲。
## 坑 4：模型瞎编引用
temperature=0.2 不够。系统 prompt 里那句「只根据参考资料回答」必须写得明明白白，而且每个 chunk 都要在 prompt 里给编号。
## 坑 5：权限过滤不生效
这一版只在 SQL WHERE 里加了 permission_group = :perm。如果你的权限模型是「按文档组」或者「按部门」，要把 permission 字段拆成多对多。简单点可以先用 permission_group，复杂场景下一篇再展开。
# 十二、把它跑起来
按顺序执行：
```
# 1. 启动 PostgreSQL（假设已装好）
createdb kb
psql kb -c "CREATE EXTENSION vector;"
# 2. 安装依赖
pip install -r requirements.txt
# 3. 初始化表
python -c "from app.db import Base, engine; Base.metadata.create_all(engine)"
# 4. 准备文档
mkdir -p data/docs
# 把你自己的 Markdown 文档放进去
# 5. 启动
uvicorn app.main:app --reload
```
打开浏览器，开始问问题。
# 十三、下一步
这一版只是「能跑」。要变成上一篇说的「能生产」，还差：
- 查询改写（短问题补全）
- 混合检索（向量 + BM25）
- 评估集（50~200 个真实问题）
- 增量更新（hash + 异步任务）
- 权限模型（多对多）
- Agent 编排（多轮检索）
第三篇会把查询改写 + 混合检索 + 重排做进来，让 RAG 真的能扛日常使用。如果你想先看到哪一块，评论区扣一下，我加塞。
参考资料：
- pgvector 官方文档：https://github.com/pgvector/pgvector
- 阿里云百炼 Qwen3.7-Max 文档：https://help.aliyun.com/zh/model-studio/
- Qwen3-Embedding 技术报告：https://qwen.ai/blog?id=qwen3-embedding
- OpenAI Embeddings API 文档（如果换回 OpenAI）
上一篇：私有知识库从 demo 到生产，中间差了 8 层工程
## 相关链接
- [Ando](https://x.com/ando_w)
- [@ando_w](https://x.com/ando_w)
- [490](https://x.com/ando_w/status/2073979240962228582/analytics)
- [Jul 3](https://x.com/ando_w/status/2073070705789329561)
- [843](https://x.com/ando_w/status/2073070705789329561/analytics)
- [dashscope.aliyuncs.com/v1](https://dashscope.aliyuncs.com/v1)
- [https://bailian.console.aliyun.com](https://bailian.console.aliyun.com/)
- [https://api.openai.com/v1](https://api.openai.com/v1)
- [https://github.com/pgvector/pgvector](https://github.com/pgvector/pgvector)
- [https://help.aliyun.com/zh/model-studio/](https://help.aliyun.com/zh/model-studio/)
- [https://qwen.ai/blog?id=qwen3-embedding](https://qwen.ai/blog?id=qwen3-embedding)
- [私有知识库从 demo 到生产，中间差了 8 层工程](https://x.com/ando_w/status/2073070705789329561)
- [Upgrade to Premium](https://x.com/i/premium_sign_up)
- [11:55 AM · Jul 6, 2026](https://x.com/ando_w/status/2073979240962228582)
- [490 Views](https://x.com/ando_w/status/2073979240962228582/analytics)
---
*导出时间: 2026/7/8 15:53:55*
