本文是系列第 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 | shWindows 上如果装完仍然找不到命令，多半是
%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，报错信息还很不直观。
✅ 解决方案：三选一：
- 最省事：全部用一家同时支持 chat 和 embedding 的服务。比如通义 DashScope 兼容模式：
硅基流动、OpenAI 同理。OPENAI_API_BASE=https://dashscope.aliyuncs.com/compatible-mode/v1 LLM_MODEL=qwen-plus EMBEDDING_MODEL=text-embedding-v3- chat 用 DeepSeek，embedding 单独指一家——那就要把配置拆成两组 base / key（P2 会做，P0 先别加复杂度）。
- 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()
三个刻意的设计：
DocStatus用str, Enum双继承，这样model_dump(mode="json")能直接序列化成字符串，不用写自定义 encoder。- 每个方法都在
self._lock里读改写，虽然 P0 是单进程 CLI 用不上，但 P1 一上 FastAPI 就是多线程的，提前加成本几乎为零。 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，直接 404context_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 一键起
其中有一条最容易翻车的验收：删掉文档之后再问，必须无法引用该内容。很多人做到这里才发现自己只删了元数据，向量还留在库里。
系列文章会陆续更新，有问题欢迎评论区交流。