前端转 AI 实战 P0 地基：不写一行前端，先用 CLI 跑通 RAG 闭环

本文是系列第 1 篇。上一篇 总纲 讲了六阶段的整体路线，这一篇开始动手。

零、这篇要做出什么

P0 的目标不是做产品，是建立语感。所以这一阶段我刻意把 Web 层全部砍掉，只留两个命令行脚本：

# 把一份文档喂进知识库
uv run python scripts/ingest_cli.py --kb demo --file ../samples/hello.md# 提问
uv run python scripts/query_cli.py --kb demo --question "年假有多少天？"

跑完你会看到：

ANSWER:
根据资料，年假每年 10 天，逾期作废。CITATIONS:
- hello.md (a3f2...) score=0.83# 请假制度 员工请假需提前 3 天在 OA 提交申请。年假每年 10 天，逾期作废。

只要这两条命令跑通，RAG 你就算入门了——剩下的 P1～P5 全是在这条链路上做加法。

本篇要实现的链路：

  ┌──────────┐   read    ┌──────────┐  split_text  ┌─────────┐│ .md/.pdf │──────────▶│   text   │─────────────▶│ chunks  │└──────────┘           └──────────┘              └────┬────┘│ embedding▼┌───────────┐提问 ──embedding──▶ 相似度检索(top_k) ◀────────│  Chroma   ││                     └───────────┘▼top-k chunks ──▶ Prompt ──▶ LLM ──▶ 答案 + 引用

对应的代码结构（P0 结束时的样子）：

backend/app/config.py              # 全局配置（pydantic-settings）models/domain.py       # 领域模型：KnowledgeBase / DocumentRecordprompts/rag.py         # 系统提示词 + 用户提示词拼装services/chunking.py          # 纯函数：文本切分store.py             # JSON 元数据存储index_manager.py     # Chroma + Embedding 管理ingest.py            # 入库管线retrieve.py          # 检索chat.py              # 生成scripts/ingest_cli.py          # CLI：入库query_cli.py           # CLI：提问tests/                   # 不依赖网络的单元测试pyproject.toml.env.example
samples/hello.md                 # 测试文档

阅读约定：⚠️ 易错点 都是真实踩过的坑，紧跟的 ✅ 解决方案 可直接照抄。

步骤 0：环境准备

  • Python ≥ 3.12（我本机 3.13 也正常）
  • 包管理器 uv（比 pip 快很多，而且自带虚拟环境管理）
python --version   # 需要 >= 3.12
uv --version

⚠️ 易错点 1：uv: command not found。很多教程直接让你 uv sync，但 uv 不是 Python 自带的。
✅ 解决方案：装一下即可：

pip install uv
# 或官方脚本
curl -LsSf https://astral.sh/uv/install.sh | sh

Windows 上如果装完仍然找不到命令，多半是 %USERPROFILE%\.local\bin 没进 PATH，可以直接用绝对路径调用：C:\Users\你的用户名\.local\bin\uv.exe sync。

⚠️ 易错点 2：国内网络下 uv sync 卡在下载不动。
✅ 解决方案：换镜像源。

# bash
export UV_INDEX_URL=https://pypi.tuna.tsinghua.edu.cn/simple
# PowerShell
$env:UV_INDEX_URL="https://pypi.tuna.tsinghua.edu.cn/simple"

⚠️ 易错点 3：本机 Python 是 3.11 或更低，uv sync 报 requires-python >=3.12 冲突。
✅ 解决方案：不用去动系统 Python，让 uv 指定版本建虚拟环境即可：uv venv --python 3.12。

关于 API Key：这一阶段需要一个 OpenAI 兼容的服务，chat 和 embedding 都要用。后面步骤 2 会详细说怎么配，先准备好一个 Key（DeepSeek / 通义 DashScope / 硅基流动 / OpenAI 都行，但有个大坑，见步骤 2）。

步骤 1：项目骨架与依赖

mkdir -p backend/app/{models,prompts,services} backend/scripts backend/tests samples
cd backend

backend/pyproject.toml：

[project]
name = "enterprise-rag"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = ["fastapi>=0.115.0","uvicorn[standard]>=0.32.0","python-multipart>=0.0.12","pydantic-settings>=2.6.0","llama-index-core>=0.12.0","llama-index-embeddings-openai>=0.3.0","llama-index-llms-openai>=0.3.0","llama-index-llms-openai-like>=0.3.0","llama-index-vector-stores-chroma>=0.4.0","chromadb>=0.5.0","pypdf>=5.0.0","httpx>=0.27.0",
][dependency-groups]
dev = ["pytest>=8.3.0","pytest-asyncio>=0.24.0",
][tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
pythonpath = ["."]

uv sync

几个值得说明的地方：

  • llama-index-llms-openai-like 不能省。 后面步骤 8 会讲为什么不用 OpenAI 类而用 OpenAILike——这是本篇最坑的一个点。
  • fastapi / uvicorn / python-multipart 这几个 P0 用不上，但一起装了，下一篇 P1 直接用，省得再改一次依赖。

⚠️ 易错点 4（很隐蔽）：pyproject.toml 里漏了 pythonpath = ["."]，跑 pytest 时所有 from app.services.xxx import ... 全部 ModuleNotFoundError: No module named 'app'。
✅ 解决方案：加上 [tool.pytest.ini_options] 里的 pythonpath = ["."]，让 pytest 把 backend/ 目录加入模块搜索路径。这行不写的话，你要么得把项目装成包，要么每次 PYTHONPATH=. pytest——都不如这一行省事。

⚠️ 易错点 5：依赖全写 latest 或不写版本，chromadb 和 llama-index-vector-stores-chroma 撞版本，一 import 就报 API 不匹配。
✅ 解决方案：像上面一样锁下限版本。这两个包的适配关系变得比较频繁，图省事用 latest 迟早出问题。

步骤 2：配置与环境变量

backend/app/config.py：

from pathlib import Path
from pydantic_settings import BaseSettings, SettingsConfigDictBACKEND_ROOT = Path(__file__).resolve().parents[1]
DATA_DIR = BACKEND_ROOT / "data"class Settings(BaseSettings):model_config = SettingsConfigDict(env_file=".env", extra="ignore")app_name: str = "Enterprise RAG"openai_api_key: str = ""openai_api_base: str = "https://api.openai.com/v1"llm_model: str = "gpt-4o-mini"llm_context_window: int = 128000embedding_model: str = "text-embedding-3-small"chroma_path: Path = DATA_DIR / "chroma"upload_dir: Path = DATA_DIR / "uploads"meta_path: Path = DATA_DIR / "meta.json"chunk_size: int = 600chunk_overlap: int = 120top_k: int = 5settings = Settings()

backend/.env.example：

OPENAI_API_KEY=sk-xxx
OPENAI_API_BASE=https://api.deepseek.com/v1
LLM_MODEL=deepseek-chat
# OpenAI: text-embedding-3-small；DashScope: text-embedding-v3 / v4
EMBEDDING_MODEL=text-embedding-3-small

cp .env.example .env    # Windows: copy .env.example .env
# 然后填入真实 Key
uv run python -c "from app.config import settings; print(settings.app_name)"
# 预期输出：Enterprise RAG

参数怎么定的

  • chunk_size = 600 / chunk_overlap = 120：注意这是「字符数」不是「token 数」（因为我们的切分是按字符切的，见步骤 3）。中文大约 1 字 ≈ 1 token 多一点，600 字符的 chunk 是个比较稳的起点：太小则语义被切碎，太大则检索精度下降、prompt 也贵。overlap 取 20% 保证跨 chunk 的句子不被割裂。
  • top_k = 5：召回 5 个片段拼进 prompt。P2 会讲怎么用评测把这个数调准，现在别纠结。
  • llm_context_window = 128000：必须显式给，原因见步骤 8。

⚠️ 易错点 6（本篇最高频）：chat 模型和 embedding 模型混为一谈。
上面 .env.example 里默认给的是 DeepSeek，因为它便宜好用——但DeepSeek 不提供 embedding 接口。如果你把 OPENAI_API_BASE 指向 DeepSeek，然后 EMBEDDING_MODEL 填个 text-embedding-3-small，入库时会直接 400 / 404 / 401，报错信息还很不直观。
✅ 解决方案：三选一：

  1. 最省事：全部用一家同时支持 chat 和 embedding 的服务。比如通义 DashScope 兼容模式：
    OPENAI_API_BASE=https://dashscope.aliyuncs.com/compatible-mode/v1
LLM_MODEL=qwen-plus
EMBEDDING_MODEL=text-embedding-v3
    硅基流动、OpenAI 同理。
  2. chat 用 DeepSeek，embedding 单独指一家——那就要把配置拆成两组 base / key（P2 会做，P0 先别加复杂度）。
  3. embedding 用本地模型（BAAI/bge-small-zh），零成本但要装 torch，P0 阶段不推荐。

⚠️ 易错点 7：忘了 cp .env.example .env，Settings 里 Key 是空串，一调 API 就 401，还以为是 Key 无效。
✅ 解决方案：.env 必须建。养成习惯：clone 下来第一件事就是复制 .env.example。另外 .env 记得进 .gitignore，.env.example 才是提交到仓库的那个。

⚠️ 易错点 8：.env 里写了 CHROMA_PATH=xxx 这类变量名，不确定能不能对上字段。
✅ 解决方案：pydantic-settings 对字段名大小写不敏感，openai_api_key 自动对应 OPENAI_API_KEY，chroma_path 对应 CHROMA_PATH。照着 config.py 的字段名全大写写就行。

⚠️ 易错点 9：.env 里多写了几个 Settings 里没定义的变量，启动直接报 validation error。
✅ 解决方案：SettingsConfigDict(extra="ignore") —— 上面代码已经加了。不加的话 pydantic 默认会对未知字段报错，团队协作时经常被这个卡住。

步骤 3：文本切分（从这里开始 TDD）

切分是 RAG 里唯一不依赖网络、又直接决定回答质量的环节，所以先写它，而且用 TDD 写——后面所有涉及 LLM 的部分都很难测，这里能测就一定要测。

先写测试 backend/tests/test_chunking.py：

from app.services.chunking import split_textdef test_split_text_respects_chunk_size():text = "你好世界" * 100  # 400 charschunks = split_text(text, chunk_size=50, chunk_overlap=10)assert len(chunks) > 1assert all(len(c) <= 50 for c in chunks)def test_split_text_empty():assert split_text("", chunk_size=50, chunk_overlap=0) == []def test_overlap_keeps_continuity():text = "abcdefghijklmnopqrstuvwxyz"chunks = split_text(text, chunk_size=10, chunk_overlap=3)assert chunks[0][-3:] == chunks[1][:3]

再写实现 backend/app/services/chunking.py：

def split_text(text: str, chunk_size: int, chunk_overlap: int) -> list[str]:text = text.strip()if not text:return []if chunk_size <= 0:raise ValueError("chunk_size must be positive")if chunk_overlap >= chunk_size:raise ValueError("chunk_overlap must be smaller than chunk_size")chunks: list[str] = []start = 0n = len(text)while start < n:end = min(start + chunk_size, n)chunks.append(text[start:end])if end == n:breakstart = end - chunk_overlapreturn chunks

uv run pytest tests/test_chunking.py -v

第三个测试 test_overlap_keeps_continuity 是关键：它验证相邻 chunk 确实有重叠——chunks[0] 的末 3 个字符必须等于 chunks[1] 的头 3 个字符。没有这条断言，overlap 写错方向（写成 start = end + overlap）也测不出来。

⚠️ 易错点 10：chunk_overlap >= chunk_size 时，start = end - chunk_overlap 会让 start 不前进甚至倒退，直接死循环，跑到内存爆掉。
✅ 解决方案：函数入口就校验并抛 ValueError（上面已加）。默认值 120 < 600 是安全的，但一旦允许用户在 API 里传这两个参数，这个校验就是救命的。

⚠️ 易错点 11：while start < n 循环里，如果不写 if end == n: break，最后一个 chunk 会因为 start = end - overlap 回退而被无限重复切出来。
✅ 解决方案：切到末尾立刻 break（上面已加）。这类边界问题正是必须写单测的原因。

⚠️ 易错点 12：按字符切分对中文其实是可以接受的，但很多人直接套英文教程按 token 切，然后拿 chunk_size=600 去理解成 600 token，导致对 prompt 长度和费用的估算全错。
✅ 解决方案：明确你的 chunk_size 单位。本实现是字符数。中文场景下按字符切简单直接、效果也不差；P2 会换成保留结构的切分（按标题 / 段落），那时才需要引入 token 计数。

步骤 4：领域模型与元数据存储

需要记住「有哪些知识库、每个库里有哪些文档、文档处理到哪一步了」。P0 阶段不上数据库，用一个 JSON 文件顶着。

backend/app/models/domain.py：

from datetime import datetime, timezone
from enum import Enum
from uuid import uuid4from pydantic import BaseModel, Fielddef utcnow() -> datetime:return datetime.now(timezone.utc)def new_id() -> str:return uuid4().hexclass DocStatus(str, Enum):pending = "pending"ready = "ready"failed = "failed"class KnowledgeBase(BaseModel):id: str = Field(default_factory=new_id)name: strdescription: str = ""created_at: datetime = Field(default_factory=utcnow)class DocumentRecord(BaseModel):id: str = Field(default_factory=new_id)kb_id: strfilename: strstatus: DocStatus = DocStatus.pendingerror: str | None = Nonecreated_at: datetime = Field(default_factory=utcnow)

backend/app/services/store.py（关键片段）：

class MetaStore:"""JSON-file metadata store for knowledge bases and documents."""def __init__(self, path: Path | None = None) -> None:self._path = path or settings.meta_pathself._lock = threading.Lock()self._ensure_file()def _write(self, data: dict) -> None:self._path.parent.mkdir(parents=True, exist_ok=True)with self._path.open("w", encoding="utf-8") as f:json.dump(data, f, ensure_ascii=False, indent=2, default=str)def create_kb(self, name: str, description: str = "") -> KnowledgeBase:kb = KnowledgeBase(name=name, description=description)with self._lock:data = self._read()data["knowledge_bases"].append(kb.model_dump(mode="json"))self._write(data)return kbdef update_document(self, doc: DocumentRecord) -> DocumentRecord:with self._lock:data = self._read()for i, item in enumerate(data["documents"]):if item["id"] == doc.id:data["documents"][i] = doc.model_dump(mode="json")self._write(data)return docraise KeyError(f"document not found: {doc.id}")store = MetaStore()

三个刻意的设计：

  1. DocStatus 用 str, Enum 双继承，这样 model_dump(mode="json") 能直接序列化成字符串，不用写自定义 encoder。
  2. 每个方法都在 self._lock 里读改写，虽然 P0 是单进程 CLI 用不上，但 P1 一上 FastAPI 就是多线程的，提前加成本几乎为零。
  3. MetaStore 是个类、路径可注入，P3 换 Postgres 时只需要写一个同接口的实现。

⚠️ 易错点 13：json.dump 不加 ensure_ascii=False，meta.json 里中文文件名全变成 \u4e2d\u6587，肉眼没法调试。
✅ 解决方案：ensure_ascii=False, indent=2（上面已加）。另外 default=str 用来兜底 datetime 序列化。

⚠️ 易错点 14：store = MetaStore() 是模块级单例，import 的瞬间就会去创建 data/meta.json。写测试时会污染真实数据文件。
✅ 解决方案：测试里用 MetaStore(path=tmp_path / "meta.json") 注入临时路径，不要用全局 store。这也是为什么构造函数要留 path 参数。

⚠️ 易错点 15：JSON 文件在多进程下（比如 uvicorn 多 worker）线程锁完全没用，照样丢数据。
✅ 解决方案：P0–P2 明确接受「单进程假设」，别自欺欺人地以为加了锁就安全了。P3 迁 Postgres 时一并解决。

步骤 5：Embedding 与向量库

到这里开始碰真正的「AI 部分」了。

backend/app/services/index_manager.py：

from __future__ import annotationsimport chromadb
from llama_index.core import StorageContext, VectorStoreIndex
from llama_index.core.embeddings import BaseEmbedding
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.chroma import ChromaVectorStorefrom app.config import Settings, settingsdef build_embed_model(cfg: Settings | None = None) -> OpenAIEmbedding:cfg = cfg or settings# model_name= 绕开 OpenAIEmbeddingModelType 枚举校验（DashScope 等需要）return OpenAIEmbedding(model_name=cfg.embedding_model,api_key=cfg.openai_api_key or "EMPTY",api_base=cfg.openai_api_base,)class IndexManager:"""Manage per-kb Chroma collections and LlamaIndex vector indexes."""def __init__(self,cfg: Settings | None = None,embed_model: BaseEmbedding | None = None,) -> None:self._cfg = cfg or settingsself._cfg.chroma_path.mkdir(parents=True, exist_ok=True)self._client = chromadb.PersistentClient(path=str(self._cfg.chroma_path))self._embed_model = embed_model or build_embed_model(self._cfg)def _collection_name(self, kb_id: str) -> str:# Chroma collection 名限制：3-63 字符、[a-zA-Z0-9._-]、首尾必须字母数字safe = "".join(c if c.isalnum() or c in "._-" else "_" for c in kb_id)return f"kb_{safe}"[:63]def get_or_create_index(self, kb_id: str) -> VectorStoreIndex:collection = self._client.get_or_create_collection(self._collection_name(kb_id))vector_store = ChromaVectorStore(chroma_collection=collection)storage_context = StorageContext.from_defaults(vector_store=vector_store)return VectorStoreIndex.from_vector_store(vector_store,storage_context=storage_context,embed_model=self._embed_model,)def delete_document_nodes(self, kb_id: str, document_id: str) -> None:collection = self._client.get_or_create_collection(self._collection_name(kb_id))result = collection.get(where={"document_id": document_id})ids = result.get("ids") or []if ids:collection.delete(ids=ids)def reset_kb(self, kb_id: str) -> None:name = self._collection_name(kb_id)try:self._client.delete_collection(name)except Exception:passself._client.get_or_create_collection(name)index_manager = IndexManager()

这个文件信息量很大，逐个说：

① 一个知识库 = 一个 Chroma collection。 不用一个大 collection 加 where 过滤，物理隔离更简单，删库也直接。

② embed_model 可注入。 IndexManager(embed_model=FakeEmbedding()) 就能在不联网的情况下测入库逻辑——这是让整条管线可测的关键设计。

③ delete_document_nodes 靠 metadata 里的 document_id 反查。 这是后面「删了文档就不该再被检索到」这条验收的实现基础。

⚠️ 易错点 16（卡了我半天）：OpenAIEmbedding(model="text-embedding-v3", ...) 直接抛异常，说这个模型名不合法。
✅ 解决方案：LlamaIndex 的 OpenAIEmbedding 里，model= 参数会走 OpenAIEmbeddingModelType 枚举校验，只认 OpenAI 官方那几个模型名。用国内的兼容网关（DashScope 的 text-embedding-v3、硅基流动的 BAAI/bge-m3）必然不在枚举里。
换成 model_name= 就绕过了枚举校验。一个下划线的差别，报错信息还完全不提示这一点。
我为这个专门留了一条回归测试，防止以后手滑改回去：

def test_build_embed_model_accepts_openai_compatible_custom_name():"""DashScope / 兼容网关的模型名不在 OpenAI 枚举内，须能构造。"""cfg = Settings(openai_api_key="test-key",openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",embedding_model="text-embedding-v3",)embed = build_embed_model(cfg)assert embed.model_name == "text-embedding-v3"

⚠️ 易错点 17：api_key 为空字符串时，OpenAI SDK 在构造阶段就抛错，导致连单元测试都跑不起来（测试里根本不需要真 Key）。
✅ 解决方案：api_key=cfg.openai_api_key or "EMPTY" —— 给个占位符，让对象能构造出来。真正调用时才会因为 Key 无效而失败，这时报错是明确的 401。

⚠️ 易错点 18：直接拿 kb_id 当 Chroma collection 名，报 Expected collection name that ... 3-63 characters。
✅ 解决方案：Chroma 的 collection 名有硬性限制：3–63 个字符、只能是 [a-zA-Z0-9._-]、首尾必须是字母或数字。所以要做两件事：把非法字符替换掉，再加一个 kb_ 前缀保证首字符合法、长度不会小于 3。上面 _collection_name 就干这个。如果你的 kb_id 用的是中文名，不处理必挂。

⚠️ 易错点 19：chromadb.PersistentClient(path=...) 的目录不存在时报错。
✅ 解决方案：构造函数里先 mkdir(parents=True, exist_ok=True)（上面已加）。另外记得把 backend/data/ 加进 .gitignore——向量库的二进制文件提交到 git 里是灾难。

⚠️ 易错点 20：换了 embedding 模型之后，检索结果突然全乱了。
✅ 解决方案：不同 embedding 模型的向量维度和语义空间完全不同，老数据是用旧模型编码的，新查询用新模型编码，算出来的相似度毫无意义（维度不一致时还会直接报错）。换模型后必须重建整个 collection——这就是 reset_kb 存在的意义。

步骤 6：入库管线

把前面几块拼起来：读文件 → 切分 → 造节点 → 写向量库 → 更新状态。

backend/app/services/ingest.py：

def read_file_text(path: Path) -> str:suffix = path.suffix.lower()if suffix in {".txt", ".md", ".markdown"}:return path.read_text(encoding="utf-8")if suffix == ".pdf":reader = PdfReader(str(path))parts = [(page.extract_text() or "") for page in reader.pages]return "\n".join(parts)raise ValueError(f"unsupported file type: {suffix}")def build_nodes_from_text(text: str,*,kb_id: str,document_id: str,filename: str,chunk_size: int,chunk_overlap: int,
) -> list[dict]:chunks = split_text(text, chunk_size=chunk_size, chunk_overlap=chunk_overlap)nodes: list[dict] = []for i, chunk in enumerate(chunks):nodes.append({"id": f"{document_id}_{i}","text": chunk,"metadata": {"kb_id": kb_id,"document_id": document_id,"filename": filename,"chunk_index": i,},})return nodesdef ingest_document(file_path: Path,doc: DocumentRecord,*,meta: MetaStore | None = None,manager: IndexManager | None = None,cfg: Settings | None = None,
) -> DocumentRecord:meta = meta or storemanager = manager or index_managercfg = cfg or settingstry:text = read_file_text(file_path)if not text.strip():# 扫描件 PDF / 空文档会提取出空文本，若不拦截会「入库成功但永远检索不到」raise ValueError("no text extracted from file; it may be a scanned PDF needing OCR")node_dicts = build_nodes_from_text(text,kb_id=doc.kb_id,document_id=doc.id,filename=doc.filename,chunk_size=cfg.chunk_size,chunk_overlap=cfg.chunk_overlap,)manager.delete_document_nodes(doc.kb_id, doc.id)   # 先删后插，保证幂等index = manager.get_or_create_index(doc.kb_id)nodes = [TextNode(text=n["text"], id_=n["id"], metadata=n["metadata"])for n in node_dicts]if nodes:index.insert_nodes(nodes)doc.status = DocStatus.readydoc.error = Noneexcept Exception as exc:doc.status = DocStatus.faileddoc.error = str(exc)meta.update_document(doc)return doc

设计要点：

① build_nodes_from_text 返回的是普通 dict 而不是 TextNode。 这样它就是一个纯函数，测试完全不需要 import LlamaIndex：

def test_build_nodes_from_text_metadata_and_count():text = "abcdefghij" * 10  # 100 charsnodes = build_nodes_from_text(text, kb_id="kb1", document_id="doc1", filename="note.md",chunk_size=40, chunk_overlap=5,)assert len(nodes) > 1for node in nodes:assert node["text"]assert node["metadata"]["document_id"] == "doc1"assert "chunk_index" in node["metadata"]def test_build_nodes_from_empty_text():nodes = build_nodes_from_text("   ", kb_id="kb1", document_id="doc1", filename="empty.txt",chunk_size=40, chunk_overlap=5,)assert nodes == []

② node id 用 f"{document_id}_{i}"。 确定性 ID，重复入库同一文档时 ID 一致，配合先删后插就是幂等的。

③ metadata 必须带 document_id。 这是删除的唯一抓手。

⚠️ 易错点 21（很常见，而且有两层）：改了一份文档重新上传，结果知识库里出现两份重复内容，检索时 top-5 全被同一段占满。

第一层：入库前没删旧向量，每次都是纯追加。
✅ 入库前先 manager.delete_document_nodes(doc.kb_id, doc.id)（上面已加）。注意顺序：先删旧向量，再插新的。

第二层（更隐蔽，我实测才发现）：即使加了先删后插，如果调用方每次都 uuid4() 生成一个全新的 doc_id，幂等依然不成立——因为删的是「新 id」对应的向量，而新 id 从来没入过库，等于删了个寂寞，旧向量原地不动。

我第一版 ingest_cli.py 就是这么写的，跑完两次入库、查询返回了两条一模一样的引用才发现。先删后插保证的是「同一 doc_id 重入幂等」，不是「同一文件重入幂等」——后者需要调用方主动按文件名复用 doc_id。步骤 9 会给出修正后的写法。

⚠️ 易错点 22：pypdf 对扫描件 PDF 提取出来是空字符串 → 切出 0 个 chunk → 入库"成功"但检索永远答不上来，还以为是检索代码写错了。
✅ 解决方案：page.extract_text() or "" 只能防 None，防不了空文档。所以上面在 read_file_text 之后加了一条 if not text.strip() 校验，直接抛异常让 status 变成 failed 并带明确原因，而不是静默成功：

status=failed
error=no text extracted from file; it may be a scanned PDF needing OCR

报错信息里写清「可能是扫描件、需要 OCR」，比一句干巴巴的 empty text 有用得多——用户看到就知道该去做什么。扫描件的 OCR 留到 P5。

⚠️ 易错点 23：except Exception 把异常吞掉写进 doc.error，调用方不看 status 就以为入库成功了。
✅ 解决方案：这里吞异常是有意的——单个文档失败不应该让整批入库崩掉，失败原因记在 doc.error 里前端可以展示。但调用方必须检查 status。步骤 9 的 CLI 里我就加了 if result.error: raise SystemExit(1)。

⚠️ 易错点 24：path.read_text() 不指定 encoding，Windows 上读中文 Markdown 直接 UnicodeDecodeError。
✅ 解决方案：永远显式写 encoding="utf-8"（上面已加）。Windows 的默认编码是 GBK，这个坑在跨平台协作时百分百会遇到。

给这两个坑配上回归测试

易错点 21 和 22 是改完之后很容易被后人改回去的那种，所以必须锁住。难点在于 ingest_document 会真的调向量库，单测里不能联网——用一个假的 IndexManager 替身就行：

class _FakeIndex:def __init__(self) -> None:self.inserted: list = []def insert_nodes(self, nodes) -> None:self.inserted.extend(nodes)class _FakeIndexManager:"""不联网的 IndexManager 替身，用于验证入库管线的控制流。"""def __init__(self) -> None:self.index = _FakeIndex()self.deleted: list[tuple[str, str]] = []def delete_document_nodes(self, kb_id: str, document_id: str) -> None:self.deleted.append((kb_id, document_id))def get_or_create_index(self, kb_id: str) -> _FakeIndex:return self.indexdef test_ingest_empty_file_marks_failed(tmp_path: Path):"""扫描件 PDF / 空文档不得静默成功，否则入库显示 ready 却永远检索不到。"""meta = MetaStore(path=tmp_path / "meta.json")kb = meta.create_kb(name="kb")doc = DocumentRecord(kb_id=kb.id, filename="empty.md")meta.add_document(doc)empty_file = tmp_path / "empty.md"empty_file.write_text("   \n\n  ", encoding="utf-8")result = ingest_document(empty_file, doc, meta=meta, manager=_FakeIndexManager())assert result.status is DocStatus.failedassert "no text extracted" in (result.error or "")def test_ingest_deletes_old_nodes_before_insert(tmp_path: Path):"""重复入库必须先按 document_id 删旧向量，否则检索结果出现重复片段。"""meta = MetaStore(path=tmp_path / "meta.json")manager = _FakeIndexManager()kb = meta.create_kb(name="kb")doc = DocumentRecord(kb_id=kb.id, filename="note.md")meta.add_document(doc)source = tmp_path / "note.md"source.write_text("年假每年 10 天，逾期作废。", encoding="utf-8")ingest_document(source, doc, meta=meta, manager=manager)ingest_document(source, doc, meta=meta, manager=manager)assert manager.deleted == [(kb.id, doc.id), (kb.id, doc.id)]assert doc.status is DocStatus.ready

这就是步骤 4 那个「路径可注入」设计的回报——MetaStore(path=tmp_path / "meta.json") 配合 manager= 参数注入，整个入库管线的控制流都能在不联网、不碰真实数据的前提下测干净。如果当初 MetaStore 把路径写死成 data/meta.json，这两个测试根本没法写。

步骤 7：检索

backend/app/services/retrieve.py：

@dataclass
class RetrievedChunk:document_id: strfilename: strsnippet: strscore: float | Nonedef retrieve(kb_id: str,query: str,*,top_k: int | None = None,manager: IndexManager | None = None,cfg: Settings | None = None,
) -> list[RetrievedChunk]:cfg = cfg or settingsmanager = manager or index_managerk = top_k or cfg.top_kindex = manager.get_or_create_index(kb_id)retriever = index.as_retriever(similarity_top_k=k)nodes = retriever.retrieve(query)results: list[RetrievedChunk] = []for node_with_score in nodes:node = node_with_score.nodemeta = node.metadata or {}results.append(RetrievedChunk(document_id=str(meta.get("document_id", "")),filename=str(meta.get("filename", "")),snippet=node.get_content(),score=float(node_with_score.score)if node_with_score.score is not Noneelse None,))return results

这一步代码最少，但有个概念要说清楚：retriever.retrieve(query) 内部会先把 query 做一次 embedding，然后在 Chroma 里算向量相似度。也就是说，提问也是要花 embedding 调用的——只是量很小。

⚠️ 易错点 25：直接用 node.text 取内容，某些节点类型下拿到空串。
✅ 解决方案：用 node.get_content()，它是 LlamaIndex 的标准取值方法，会正确处理 metadata 模板等情况。

⚠️ 易错点 26：node_with_score.score 直接 float() 转换，遇到 None 时 TypeError。
✅ 解决方案：某些向量库 / 检索模式下 score 可能是 None，做个判空（上面已加），并且把 RetrievedChunk.score 声明成 float | None。

⚠️ 易错点 27：top_k 一路调大，以为召回越多答得越准。
✅ 解决方案：top_k 太大会把不相关的片段也塞进 prompt，反而稀释了有效信息，还会拉高成本和延迟。5 是个稳妥的起点。想调它，等 P2 有了 Golden Set 再用数据说话。

步骤 8：Prompt 与生成

backend/app/prompts/rag.py：

SYSTEM_PROMPT = """你是企业知识库助手。只根据给定资料回答。
若资料不足以回答，明确说「根据现有资料无法回答」，不要编造。
回答时使用简体中文。"""def build_user_prompt(question: str, contexts: list[str]) -> str:joined = "\n\n---\n\n".join(contexts) if contexts else "（无检索结果）"return f"资料：\n{joined}\n\n问题：{question}"

backend/app/services/chat.py：

from llama_index.llms.openai_like import OpenAILikedef build_llm(cfg: Settings | None = None) -> OpenAILike:cfg = cfg or settings# OpenAILike 跳过 OpenAI 的模型名 / context window 枚举校验return OpenAILike(model=cfg.llm_model,api_key=cfg.openai_api_key or "EMPTY",api_base=cfg.openai_api_base,temperature=0.1,is_chat_model=True,context_window=cfg.llm_context_window,)def _format_history(history: list[dict[str, str]]) -> str:if not history:return ""recent = history[-8:]  # 最近 4 轮（user/assistant 成对）lines = []for item in recent:lines.append(f"{item.get('role', 'user')}: {item.get('content', '')}")return "\n".join(lines)def answer(kb_id: str,message: str,history: list[dict[str, str]] | None = None,*,manager: IndexManager | None = None,cfg: Settings | None = None,llm: OpenAILike | None = None,
) -> ChatResponse:cfg = cfg or settingsmanager = manager or index_managerllm = llm or build_llm(cfg)chunks = retrieve(kb_id, message, manager=manager, cfg=cfg)contexts = [c.snippet for c in chunks]user_prompt = build_user_prompt(message, contexts)hist = _format_history(history or [])if hist:user_prompt = f"对话历史：\n{hist}\n\n{user_prompt}"response = llm.complete(f"{SYSTEM_PROMPT}\n\n{user_prompt}")citations = [Citation(document_id=c.document_id,filename=c.filename,snippet=c.snippet[:500],score=c.score,)for c in chunks]return ChatResponse(answer=str(response), citations=citations)

⚠️ 易错点 28（本篇最坑，和易错点 16 是同一族）：用 from llama_index.llms.openai import OpenAI 配一个国内模型名（qwen-plus、deepseek-chat），构造的时候不报错，一访问 llm.metadata 或真正调用时才炸——报错还指向 context window 查表失败，完全看不出根因。
✅ 解决方案：用 OpenAILike 而不是 OpenAI（对应依赖 llama-index-llms-openai-like）。OpenAI 类内部维护了一张「模型名 → context window」的映射表，未知模型名会查表失败。OpenAILike 就是为兼容网关准备的，但要手动补两个参数：

  • is_chat_model=True：不写的话会走 completion 接口，很多国内网关只支持 /chat/completions，直接 404
  • context_window=...：不写会用一个很小的默认值，LlamaIndex 会据此悄悄截断你的 prompt，表现为「明明检索到了却答不上来」

同样留了回归测试：

def test_build_llm_accepts_openai_compatible_custom_model():"""DashScope 等自定义模型名不得在访问 metadata 时抛错。"""cfg = Settings(openai_api_key="test-key",openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",llm_model="qwen3.7-max",)llm = build_llm(cfg)assert llm.metadata.model_name == "qwen3.7-max"assert llm.metadata.is_chat_model is True

注意断言里访问了 llm.metadata——就是为了触发那个会炸的代码路径。只断言构造成功是测不出这个坑的。

⚠️ 易错点 29：没有拒答约束，模型「很会编」。测试时答得头头是道，一核对全是幻觉，而且因为语气笃定极具欺骗性。
✅ 解决方案：这是 RAG 防幻觉的第一道也是最重要的一道闸——系统提示词里明确写「只根据给定资料回答」「资料不足就说无法回答」。第二道闸是把引用片段返回给用户，让人能自己核对出处。两道闸都要有。

⚠️ 易错点 30：temperature 用默认值（通常 0.7～1.0），同一个问题问两次答案不一样，没法调试也没法做评测。
✅ 解决方案：RAG 场景要的是忠实复述资料，不是创意写作。temperature=0.1 让输出尽量稳定。P2 做评测时，这一点更是前提——temperature 高的话你根本分不清分数波动是策略变了还是采样随机。

⚠️ 易错点 31：把全部历史轮次都拼进 prompt，长对话直接超 context window，费用也线性上涨。
✅ 解决方案：只取最近 N 条（本实现 history[-8:]，即 4 轮 user/assistant 对话）。更进阶的做法是历史摘要，P2 再说。

⚠️ 易错点 32：检索结果为空时，prompt 里的资料部分是空字符串，模型看到一个「资料：」后面什么都没有，容易开始自由发挥。
✅ 解决方案：build_user_prompt 里给了兜底文案「（无检索结果）」（上面已加）。显式告诉模型「确实没检索到」，配合系统提示词的拒答约束，它才会老老实实说不知道。

步骤 9：两个 CLI 脚本

backend/scripts/ingest_cli.py：

from __future__ import annotationsimport argparse
import shutil
import sys
from pathlib import PathBACKEND_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND_ROOT))from app.config import settings
from app.models.domain import DocumentRecord
from app.services.ingest import ingest_document
from app.services.store import storedef main() -> None:parser = argparse.ArgumentParser(description="Ingest a document into a knowledge base")parser.add_argument("--kb", required=True, help="Knowledge base name (created if missing)")parser.add_argument("--file", required=True, type=Path, help="Path to .md/.txt/.pdf")args = parser.parse_args()file_path: Path = args.fileif not file_path.exists():raise SystemExit(f"file not found: {file_path}")kbs = [kb for kb in store.list_kbs() if kb.name == args.kb]kb = kbs[0] if kbs else store.create_kb(name=args.kb, description="created by ingest_cli")# 同名文件复用已有 doc_id：ingest_document 内部的「先删后插」以 document_id 为抓手，# 每次都新建 uuid 的话旧向量删不掉，重复入库会在检索结果里出现重复片段。existing = [d for d in store.list_documents(kb.id) if d.filename == file_path.name]reused = bool(existing)doc = existing[0] if reused else DocumentRecord(kb_id=kb.id, filename=file_path.name)dest_dir = settings.upload_dir / kb.iddest_dir.mkdir(parents=True, exist_ok=True)dest = dest_dir / f"{doc.id}_{file_path.name}"shutil.copy2(file_path, dest)if not reused:store.add_document(doc)result = ingest_document(dest, doc)action = "updated" if reused else "created"print(f"kb_id={kb.id} doc_id={result.id} status={result.status.value} ({action})")if result.error:print(f"error={result.error}")raise SystemExit(1)if __name__ == "__main__":main()

backend/scripts/query_cli.py（核心部分）：

def main() -> None:parser = argparse.ArgumentParser(description="Query a knowledge base")parser.add_argument("--kb", required=True, help="Knowledge base name or id")parser.add_argument("--question", required=True)args = parser.parse_args()kb = store.get_kb(args.kb)if kb is None:                                    # 支持按名字找，方便手敲matches = [item for item in store.list_kbs() if item.name == args.kb]if not matches:raise SystemExit(f"knowledge base not found: {args.kb}")kb = matches[0]result = answer(kb.id, args.question)print("ANSWER:")print(result.answer)print("\nCITATIONS:")for c in result.citations:print(f"- {c.filename} ({c.document_id}) score={c.score}")print(f"  {c.snippet[:200]}")

三个细节：

  • 上传的文件会被复制一份到 data/uploads/{kb_id}/{doc_id}_{filename}，加 doc_id 前缀是为了避免同名文件互相覆盖。留着原文件是为了后面「重建索引」能重新读。
  • --kb 同时支持传 id 和名字。CLI 阶段谁也不想手敲 32 位 uuid。
  • 同名文件复用已有 doc_id，输出里用 (created) / (updated) 区分。这就是易错点 21 第二层的解法，展开说一下。

⚠️ 易错点 33（幂等的最后一块拼图）：ingest.py 里明明写了先删后插，重复入库还是出现重复片段。

我第一版是这么写的：

doc = DocumentRecord(kb_id=kb.id, filename=file_path.name)  # 每次都是新 uuid

DocumentRecord 的 id 默认 uuid4().hex，所以每跑一次就是一个全新的文档。ingest_document 里的 delete_document_nodes(kb_id, doc.id) 删的是这个刚出生的 id，向量库里根本没有对应记录，删除是空操作，然后新向量追加进去——旧的一条也没少。

实测现象很典型：同一个文件入库两次，问「年假有多少天」，CITATIONS 返回两条文本完全相同、只有 document_id 不同的引用。

✅ 解决方案：入库前先按文件名查一次，有就复用那条记录：

existing = [d for d in store.list_documents(kb.id) if d.filename == file_path.name]
reused = bool(existing)
doc = existing[0] if reused else DocumentRecord(kb_id=kb.id, filename=file_path.name)

注意 store.add_document(doc) 也要包在 if not reused 里，否则 meta 会多出一条重复记录。

更进一步（P1 会做）：用文件内容的 hash 而不是文件名做去重键，这样改名不会重复入库、改内容能正确触发更新。P0 阶段按文件名够用了。

⚠️ 易错点 34：直接 python scripts/ingest_cli.py 报 ModuleNotFoundError: No module named 'app'。
✅ 解决方案：脚本在 backend/scripts/ 下，而 app 包在 backend/ 下，Python 默认只把脚本所在目录加进 sys.path。所以脚本开头要手动加：

BACKEND_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND_ROOT))

注意这几行必须写在 from app.xxx import ... 之前——这会让 linter 报 E402（import 不在文件顶部），但这里是必要的，可以加 # noqa: E402 或在配置里忽略。

⚠️ 易错点 35：--file ../samples/hello.md 这种相对路径，换个目录跑就找不到文件。
✅ 解决方案：相对路径是相对当前工作目录而不是脚本位置。要么老老实实 cd backend 之后再跑，要么传绝对路径。脚本里已经加了 if not file_path.exists(): raise SystemExit(...)，至少报错是明确的。

步骤 10：端到端跑通

准备一份测试文档 samples/hello.md：

# 请假制度员工请假需提前 3 天在 OA 提交申请。
年假每年 10 天，逾期作废。

先跑单元测试（不需要网络和 Key）：

cd backend
uv run pytest -v

我这边跑出来是 14 passed。再跑真实链路（需要 .env 里有有效 Key）：

uv run python scripts/ingest_cli.py --kb demo --file ../samples/hello.md

kb_id=9cb5574406c341a49cef2bd466c5e8d4 doc_id=317ba583b9bc445e828632056cd19726 status=ready (created)

uv run python scripts/query_cli.py --kb demo --question "年假有多少天？"

ANSWER:
根据给定资料，年假每年有 10 天。CITATIONS:
- hello.md (317ba583b9bc445e828632056cd19726) score=0.4910987772091563# 请假制度员工请假需提前 3 天在 OA 提交申请。年假每年 10 天，逾期作废。

接下来是三个反向验证，一个都别省。

① 幂等验证——把同一份文件再入库一次：

uv run python scripts/ingest_cli.py --kb demo --file ../samples/hello.md

kb_id=9cb5574406c341a49cef2bd466c5e8d4 doc_id=317ba583b9bc445e828632056cd19726 status=ready (updated)

关键看两点：doc_id 和第一次完全相同，标记从 (created) 变成 (updated)。然后再查一次，CITATIONS 必须还是只有一条。如果冒出两条内容一样、document_id 不同的引用，回去看易错点 33。

② 拒答验证——问一个资料里完全没有的问题：

uv run python scripts/query_cli.py --kb demo --question "公司的报销流程是什么？"

根据现有资料无法回答。

注意这时 CITATIONS 仍然会返回内容（我这次返回了 score=0.291 的那条请假制度）——向量检索总会给你最相近的几条，是模型判断了「这些资料答不了这个问题」。所以别用「有没有引用」来判断该不该拒答，得靠 prompt 约束。

③ 异常输入验证——空文档和不支持的类型：

printf '   \n\n  ' > empty.md && uv run python scripts/ingest_cli.py --kb demo --file empty.md

status=failed (created)
error=no text extracted from file; it may be a scanned PDF needing OCR

echo "x" > t.docx && uv run python scripts/ingest_cli.py --kb demo --file t.docx

status=failed (created)
error=unsupported file type: .docx

两条都要明确失败 + 说清原因，而不是静默成功。

⚠️ 易错点 36（最容易被跳过的验收）：只测「能答对的问题」，不测「资料里没有的问题」。
✅ 解决方案：拒答能力和回答能力同样重要。如果问一个文档里完全没有的问题，模型开始编造，说明你的系统提示词没生效——检查 prompt 是不是真的拼进去了、检索结果为空时是不是给了兜底文案。这是 RAG 是否可信的分水岭。

⚠️ 易错点 37：单测全绿就宣布 P0 完成。
✅ 解决方案：单测只覆盖了不依赖网络的部分（切分、节点构造、模型构造）。真正的坑（易错点 16、28 那两个枚举问题）只有配上真 Key 跑一次才会暴露。每个阶段的验收都必须包含一次真实的端到端运行。

十一、P0 验收清单

跑完对照一下，全部打勾才算这一阶段过了：

最后一条最重要。如果讲不出来，说明前面是照着抄的，回去把每一步的输入输出想清楚。

十二、本篇踩坑速查表

# 坑 一句话解法
1–3 uv 未装 / 源慢 / Python 版本低 pip install uv；换清华源；uv venv --python 3.12
4 pytest 找不到 app 包 pyproject.toml 加 pythonpath = ["."]
5 chromadb 与 llama-index 适配包版本冲突 锁下限版本，别用 latest
6 DeepSeek 没有 embedding chat / embedding 是两个能力，用同时支持两者的服务
7–9 .env 没建 / 变量名对不上 / 多余变量报错 复制 .env.example；字段名大写即可；extra="ignore"
10–11 切分死循环 / 末尾 chunk 重复 校验 overlap < size；到末尾 break
12 chunk_size 单位搞混 本实现是字符数不是 token
13–15 JSON 中文转义 / 单例污染测试 / 多进程丢写 ensure_ascii=False；路径可注入；P3 迁 Postgres
16 Embedding 模型名不在 OpenAI 枚举 用 model_name= 而非 model=
17 空 api_key 构造即报错 api_key or "EMPTY" 兜底
18 Chroma collection 命名非法 清洗非法字符 + kb_ 前缀 + 截断 63
19–20 持久化目录不存在 / 换模型后检索乱 先 mkdir；换 embedding 模型必须重建索引
21 重复入库产生重复向量（服务层） 先 delete_document_nodes 再 insert
22 扫描件 PDF 提取空文本 校验文本非空，否则标记 failed
23–24 异常被吞 / Windows 编码报错 调用方检查 status；显式 encoding="utf-8"
25–27 node.text 取空 / score 为 None / top_k 越大越好 get_content()；判空；top_k=5 起步
28 LLM 模型名 / context window 枚举炸 用 OpenAILike + is_chat_model + context_window
29–32 幻觉 / 输出不稳定 / 历史超长 / 空检索 拒答提示词 + 引用；temperature=0.1；history[-8:]；空结果兜底文案
33 加了先删后插仍然重复（调用层） 调用方每次新建 uuid 等于白删，按文件名复用 doc_id
34–35 脚本 import 不到 app / 相对路径找不到文件 sys.path.insert；cd backend 或用绝对路径
36–37 不测拒答 / 只跑单测就验收 必须做反向验证 + 真 Key 端到端

下一篇预告

P0 结束时你手上是两个 CLI 脚本，能跑但没法给人看。[第 02 篇：P1 MVP] 会把它包装成真正能演示的产品：

  • FastAPI 项目结构、依赖注入、错误模型
  • SSE 流式输出（前端同学的主场，也是坑最多的地方——「为什么我的流式不流式」）
  • Next.js 管理台：上传、文档列表、删除、对话 + 引用展示
  • CORS、NEXT_PUBLIC_* 构建期变量这些前后端联调必踩的坑
  • Docker Compose 一键起

其中有一条最容易翻车的验收：删掉文档之后再问，必须无法引用该内容。很多人做到这里才发现自己只删了元数据，向量还留在库里。

系列文章会陆续更新，有问题欢迎评论区交流。