这篇能做出什么
走完这篇的流程，你会得到一个小工具：把几十上百篇 PDF 丢进一个目录，用一句自然语言提问，它返回的不是一段"看起来很像"的答案，而是每条结论都挂着出处的结果——文件名、页码、以及一段在原文里能逐字对上的原句。程序会自己检查"这句话到底有没有出现在 PDF 里"，对不上的直接标红。
举个具体场景。你手上有 60 篇关于某技术路线的论文，想知道"哪些文献报告过在长文本场景下检索精度下降"。传统做法是把 PDF 一个个打开、Ctrl+F 搜关键词、再肉眼扫上下文，一小时能过 5 篇。做完这套流程后，你输入问题，几秒内拿到 8 条候选结论，每条都指向具体页码和原句，你的工作从"找"变成"判断"。
关键在于：答案必须能落回原句。做不到这一点的工具，本质上是在帮你写读后感，不是在帮你查文献。
前置条件清单
- 一台能跑 Python 的机器，本地或服务器都行；纯 CPU 也能做，只是建索引慢一些
- Python 3.9 以上（以官方文档当前版本为准）
- 一个可调用的对话模型和嵌入模型接口，本地部署或云端 API 均可
- 若干篇带文本层的 PDF（能选中文字的那种）
- 目录结构先摆好：
```text
project/
├── corpus/
│ └── pdf/ # 把所有 PDF 放这里
├── index/ # 程序生成的索引，不用手动改
├── ingest.py
└── ask.py
```
依赖装这几个就够：
```bash
pip install pymupdf numpy openai
```
openai 这个包只是用来访问兼容 OpenAI 接口的服务，本地模型用 vLLM、Ollama 之类的兼容接口也能接，具体以对应项目的官方文档为准。
如果你暂时不想自己写代码，用现成的文献问答工具也完全可以，但请坚持上面那条纪律：答案里必须能看到原文片段和页码，并且你能在 PDF 里找到它。找不到的引用，一律当作模型自己编的。
分步骤
步骤 1：把 PDF 解析成"带页码坐标"的文本
大部分失败案例死在这一步。很多人直接把整篇 PDF 抽成一坨文本，页码信息丢了，后面再怎么优化都没法定位出处。正确做法是按页解析、逐页保留页码。
```python
ingest.py（片段）
import hashlib, json, pathlib, re
import fitz # PyMuPDF
import numpy as np
PDF_DIR = pathlib.Path("corpus/pdf")
IDX_DIR = pathlib.Path("index")
WINDOW, OVERLAP = 900, 180 # 分块长度与重叠，按实测调整
def doc_id_of(path: pathlib.Path) -> str:
"""用文件内容哈希做文档 ID，天然去重。"""
return hashlib.sha256(path.read_bytes()).hexdigest()[:12]
def extract_pages(pdf_path: pathlib.Path):
"""返回 [(页码, 该页文本), ...]，页码从 1 开始。"""
doc = fitz.open(pdf_path)
for i, page in enumerate(doc):
yield i + 1, page.get_text("text")
```
注意 page.get_text("text") 抽出来的是该页文本流，页码用 i + 1 表示物理页。物理页和 PDF 上印刷的页码常常差几页（封面、目录不算号），这个偏移在核验时要人工留意，后面"常见坑"里会再提。
步骤 2：切块，并记下每块的字符区间
切块的目的是让检索粒度足够小，小到一句话不会被无关内容淹没；同时块又不能太碎，否则语义不完整。经验区间是 500–1000 字符、重叠 10%–20%，具体在你自己的语料上试。
关键是每个块都要带上 chunk_id、page、char_start、char_end，这四个字段是后面核验的锚点。
```python
def build_chunks(doc_id: str, pages, window=WINDOW, overlap=OVERLAP):
chunks = []
for page_no, text in pages:
start = 0
while start < len(text):
end = min(start + window, len(text))
piece = text[start:end]
if len(piece.strip()) >= 40: # 丢掉空白页碎片
chunks.append({
"chunk_id": f"{doc_id}#p{page_no}#c{start}",
"doc_id": doc_id,
"page": page_no,
"char_start": start,
"char_end": end,
"text": piece,
})
if end >= len(text):
break
start = end - overlap
return chunks
```
char_start / char_end 是相对于同一套解析方式得到的该页文本而言的。这个前缀很重要：换个解析工具，偏移量就对不上了。所以核验时要用同一份文本，或者干脆直接做字符串匹配。
步骤 3：生成向量并落盘
嵌入这步没什么玄机，注意批量提交和归一化。归一化之后，AI 词典：余弦相似度">余弦相似度就等于点积，后面检索快很多。
```python
from openai import OpenAI
client = OpenAI() # 读 OPENAI_API_KEY / OPENAI_BASE_URL
EMBED_MODEL = "<你的嵌入模型名，以官方文档为准>"
def embed(texts):
resp = client.embeddings.create(model=EMBED_MODEL, input=texts)
v = np.array([d.embedding for d in resp.data], dtype="float32")
return v / np.linalg.norm(v, axis=1, keepdims=True)
def main():
IDX_DIR.mkdir(exist_ok=True)
all_chunks, seen = [], set()
for pdf in sorted(PDF_DIR.glob("*.pdf")):
did = doc_id_of(pdf)
if did in seen:
print(f"跳过重复文件：{pdf.name}")
continue
seen.add(did)
chunks = build_chunks(did, extract_pages(pdf))
for c in chunks:
c["source_file"] = pdf.name
all_chunks.extend(chunks)
print(f"{pdf.name}: {len(chunks)} 块")
with open(IDX_DIR / "chunks.jsonl", "w", encoding="utf-8") as f:
for c in all_chunks:
f.write(json.dumps(c, ensure_ascii=False) + "\n")
vecs = []
B = 64
for i in range(0, len(all_chunks), B):
batch = [c["text"] for c in all_chunks[i:i + B]]
vecs.append(embed(batch))
V = np.vstack(vecs)
np.save(IDX_DIR / "vectors.npy", V)
print(f"完成：{len(all_chunks)} 块，向量维度 {V.shape[1]}")
if __name__ == "__main__":
main()
```
跑一次：
```bash
python ingest.py
```
以后新增 PDF 就重跑，文件内容哈希会自动跳过重复的。
步骤 4：检索要开两条通道
这是"大海捞针"能不能成的分水岭。
只做向量检索是不够的。 嵌入模型擅长抓语义相近的段落，但对精确字符串很迟钝——基因编号、化学式、型号、特定数字、专有名词的细微拼写差异，向量很可能把它排在几十名开外。而你恰恰经常在找这些。
所以加一条字面检索通道，然后合并去重：
```python
import re
def load_chunks(path="index/chunks.jsonl"):
with open(path, encoding="utf-8") as f:
return [json.loads(line) for line in f]
def keyword_hits(chunks, query, max_hits=20):
"""朴素字面检索：查询按标点/空格切成词，统计出现次数。"""
terms = [t for t in re.split(r"[\s,，、;；:：]+", query) if len(t) >= 2]
if not terms:
return []
scored = []
for c in chunks:
low = c["text"].lower()
s = sum(low.count(t.lower()) for t in terms)
if s:
scored.append((s, c))
scored.sort(key=lambda x: -x[0])
return [c for _, c in scored[:max_hits]]
def dense_hits(qvec, V, chunks, top_k=8):
sims = V @ qvec
order = np.argsort(-sims)[:top_k]
return [(chunks[i], float(sims[i])) for i in order]
def merge(*lists, limit=12):
out, seen = [], set()
for lst in lists:
for item in lst:
c = item[0] if isinstance(item, tuple) else item
if c["chunk_id"] in seen:
continue
seen.add(c["chunk_id"])
out.append(c)
if len(out) >= limit:
return out
return out
```
语料规模到几万块以上时，字面检索可以换成 SQLite 的 FTS5 全文索引，接口一样，只是把扫描换成索引查询。
步骤 5：让模型"只许看着证据说话"
提示词要写死三条：只能用给定证据、每条结论必须带来源编号、证据不足就承认没有。把证据按编号拼进用户消息里。
```python
SYSTEM = """你是文献检索助手。规则：
1. 只能依据「证据」中的文字作答，不得引入外部知识。
2. 每条结论必须给出来源编号，格式 [n]。
3. 同时给出一条「原文引用」，必须是从证据里逐字复制的一句话，不得改写、不得拼接。
4. 证据不足以回答时，直接说「现有文献未覆盖」，不要猜测。
5. 数字、单位、专有名词原样保留。"""
def build_user_msg(question, evidences):
lines = [f"问题：{question}", "", "证据："]
for i, c in enumerate(evidences, 1):
lines.append(
f"[{i}] 文件={c['source_file']} 页码={c['page']} chunk_id={c['chunk_id']}\n{c['text']}"
)
lines.append("")
lines.append(
'请输出 JSON：{"answer": "整体回答", "items": [{"claim": "结论", '
'"chunk_id": "来源 chunk_id", "quote": "逐字原文引用"}]}'
)
return "\n".join(lines)
```
调用：
```python
CHAT_MODEL = "<你的对话模型名，以官方文档为准>"
def ask_model(question, evidences):
resp = client.chat.completions.create(
model=CHAT_MODEL,
messages=[
{"role": "system", "content": SYSTEM},
{"role": "user", "content": build_user_msg(question, evidences)},
],
response_format={"type": "json_object"}, # 部分服务不支持，见排错一节
)
return json.loads(resp.choices[0].message.content)
```
步骤 6：核验——引用必须是原文的子串
这一步是整个流程的价值所在。模型给出的 quote，程序去对应 chunk 的原文里做包含检查。为了避开换行、空格、连字符断词的干扰，先做一次规范化再比。
```python
def normalize(s: str) -> str:
s = s.replace("\u00ad", "").replace("\u2010", "-")
return re.sub(r"\s+", "", s)
def verify(items, chunk_map):
report = []
for it in items:
c = chunk_map.get(it.get("chunk_id", ""))
if c is None:
report.append({**it, "verified": False, "reason": "chunk_id 不存在"})
continue
ok = normalize(it.get("quote", "")) in normalize(c["text"])
report.append({
**it,
"verified": ok,
"page": c["page"],
"source_file": c["source_file"],
"reason": "" if ok else "引用在原文中找不到，可能被改写或拼接",
})
return report
```
规范化会带来一点误判风险（比如原文里其实是两行，被拼成一行也算通过）。所以核验通过只是"初筛"，最终还是要人看一眼原句。但这个筛子能拦下大部分凭空生成的引用。
步骤 7：把整条链串成一条命令
```python
ask.py
import sys, json, numpy as np
def main():
question = sys.argv[1]
chunks = load_chunks()
V = np.load("index/vectors.npy")
chunk_map = {c["chunk_id"]: c for c in chunks}
qvec = embed([question])[0]
hits = merge(dense_hits(qvec, V, chunks, top_k=10),
keyword_hits(chunks, question))
if not hits:
print("没有检索到相关内容，换个说法或补充关键词。")
return
result = ask_model(question, hits)
print("\n【回答】", result.get("answer", ""))
print("\n【出处核验】")
for i, r in enumerate(verify(result.get("items", []), chunk_map), 1):
flag = "已核验" if r["verified"] else "未通过"
print(f"\n[{i}] {flag} {r.get('source_file')} 第 {r.get('page')} 页")
print(f" 结论：{r.get('claim')}")
print(f" 原句：{r.get('quote')}")
if not r["verified"]:
print(f" 提示：{r.get('reason')}")
if __name__ == "__main__":
main()
```
使用：
```bash
python ask.py "哪些文献报告了长文本下检索精度下降？"
```
输出长这样：
```text
【回答】共有三篇文献涉及该问题，主要结论集中在……
【出处核验】
[1] 已核验 paper-b.pdf 第 7 页
结论：在序列长度超过训练窗口后，召回率出现明显下滑。
原句：当输入长度超过训练时的最大窗口，召回率随之下降。
```
看到"已核验"之后，打开那份 PDF 翻到第 7 页，Ctrl+F 搜原句的前几个字，能选中，就说明这条是实的。这就是"落回原句"。
常见坑与排错
扫描版 PDF 抽不出文字。 解析结果是空白或只有零星字符，说明这份 PDF 没有文本层，需要先做 OCR。OCR 会引入识别错误，核验时要更宽松，或者干脆对这类文档单独标注"OCR 来源，需人工确认"。
双栏排版顺序错乱。 得到的是文本块而不是文本流，可以改用按块解析（page.get_text("blocks")），拿到每个块的坐标后按列排序再拼接。公式多的 PDF 尤其容易出问题。
页码对不上。 程序记的是 PDF 物理页，PDF 上印的可能是另一套页码。先用两份已知文档测一下偏移量，如果固定，就在输出时做减法；如果不固定，就老实按物理页报，并说明"以阅读器页码为准"。
分块把一句话切断了。 重叠就是为了缓解这个，但没法完全消除。如果发现结论总是差半句，把 OVERLAP 调大，或改成按段落切分。
向量检索漏掉精确串。 前面已经说了，不要只开一条通道。数字、编号、罕见术语这一类查询，字面检索往往比向量更管用，两条通道必须都有。
模型不愿意说"没找到"。 它倾向于硬凑一个答案。除了提示词约束，还可以加一层阈值：当最高相似度低于某个值时，直接返回"未检索到相关段落"，不给模型发挥的机会。阈值需要在你自己的语料上试出来。
response_format 报错。 不是所有服务都支持强制 JSON 输出。不支持就把要求写进提示词，然后在代码里剥离 ```json 代码块再解析。
引文被改写。 模型常常会把两句合并、把"下降"换成"减少"。核验会拦下这类引用，这属于预期行为，不是 bug。你可以让它重试一次，提示"引用必须逐字复制"。
保密与版权。 把未公开稿件发给云端接口前，确认合规。敏感语料建议用本地模型，接口协议兼容的话，改一下 OPENAI_BASE_URL 就能切过去。
下一步建议
- 加一层重排。 先用双通道召回 30–50 条，再用一个重排模型精排到 8 条，命中率通常能提一档。
- 建自己的评测集。 从语料里挑 20 个问题，人工标好正确出处，每次改完参数跑一遍，看召回有没有变差。没有评测集的调参基本靠感觉。
- 把 OCR 做成独立通道。 扫描件走 OCR、原生 PDF 走文本层，两条路的结果都带出处标记，检索时合并。
- 建立"证据卡片"导出。 把核验通过的条目导成 Markdown 或表格，直接粘进综述草稿，引用信息已经在了。
- 定期重建索引。 语料更新后重跑
ingest.py，哈希去重会跳过没变的文件，增量成本不高。 - 关注解析质量而不是模型参数。 实践里，解析和分块带来的收益，往往大于换一个更大的对话模型。
整条链路上真正值钱的不是"AI 会答题"，而是每一步都留了可回溯的锚点：文档哈希、页码、字符区间、逐字引用。有了这些，AI 从"帮你写"变成"帮你定位"，你从"翻 PDF"变成"做判断"。