这篇能做出什么
做完这套东西，你会得到三个可以复用的结果：
1. 一个定时同步任务：它只抓取你明确授权给集成（Integration）的那些 Notion 页面，把内容切成块后写入本地索引；每天只处理变更过的页面，因此不会重复消耗时间。
2. 一个检索接口：智能体在回答前先调用它，返回带出处（页面标题、小节路径、页面链接）的片段；没有权限的页面在检索阶段就会被过滤掉，根本进不了模型上下文。
3. 一套可度量的命中率：用 30~50 条真实问题做评测集，能看到改动分块策略或加一层重排之后，召回率有没有真的变好。
举个具体例子：客服同学在对话框里问“企业版退款要走谁审批、几个工作日到账”，智能体调用检索接口，命中《客服手册 / 计费与退款 / 退款审批流》这一节的三段文字，回答末尾附上 Notion 页面链接；而《薪酬制度》那页没有授权给这个集成，即便语义再接近也不会被召回。
---
前置条件清单
动手前先确认这几件事，缺一件都会在中途卡住：
| 项目 | 说明 |
|---|---|
| Notion 工作区权限 | 需要能创建内部集成。部分企业工作区由管理员统一管控集成创建，需先申请 |
| 集成令牌 | 在 Notion 的集成管理页面创建内部集成，拿到 Secret；能力只勾选读取内容 |
| Python 环境 | 3.10 以上即可，依赖 requests，向量库按需选择 |
| 存储 | 一张 SQLite 存同步状态与元数据；向量可选 sqlite-vec、pgvector 或任意向量库 |
| 环境变量管理 | 令牌只放环境变量或密钥管理服务，不进代码库 |
| 一份“哪些页面该被谁看到”的清单 | 这是权限边界设计的输入，比代码更重要 |
> Notion API 的版本头、速率限制、各接口字段以官方文档当前版本为准；本教程只固定“调用形态”，不固定具体数值。
---
第一步：创建集成并收窄能力
在 Notion 的集成管理页新建一个内部集成，命名要能表达用途，比如 kb-sync-customer-support。能力只勾读取内容，不要勾写入、不要勾用户信息。
创建后拿到 Secret，立刻放进环境变量：
```bash
export NOTION_TOKEN="你的集成密钥"
export NOTION_VERSION="按官方文档当前版本填写"
```
后续所有请求都带上这两个头：
```python
import os, time, requests
BASE = "https://api.notion.com/v1"
HEADERS = {
"Authorization": f"Bearer {os.environ['NOTION_TOKEN']}",
"Notion-Version": os.environ["NOTION_VERSION"],
"Content-Type": "application/json",
}
def call(method, path, **kwargs):
"""统一入口：带退避重试，避免被限流打断整批同步。"""
url = f"{BASE}{path}"
for attempt in range(5):
resp = requests.request(method, url, headers=HEADERS, timeout=30, **kwargs)
if resp.status_code == 429:
wait = float(resp.headers.get("Retry-After", 2 ** attempt))
time.sleep(wait)
continue
if resp.status_code >= 500:
time.sleep(2 ** attempt)
continue
resp.raise_for_status()
return resp.json()
raise RuntimeError(f"请求失败：{path}")
```
要点：所有同步流量都走这一个 call()，限流重试只有一处需要维护。
---
第二步：页面授权 —— 在 Notion 侧“连接”集成
这一步最容易出错，也最容易被误解：创建了集成不等于集成就看得到页面。必须在 Notion 页面里手动把它连上去。
操作路径：打开目标页面 → 右上角 ... → 连接 / Connections → 选择你的集成。
三条必须记住的授权规则：
1. 子页面继承父页面的授权。把《客服手册》根页面连给集成，它下面所有子页面自动可见。这是最省事的授权粒度。
2. 数据库需要单独连接。数据库本体和它里面的行，要在数据库页面上做一次连接操作；只连数据库所在父页面，不一定能读到库里的行。
3. 取消授权立即生效。反过来说，你本地索引里的旧副本不会自动消失，必须靠后面的“对账”逻辑清理。
权限边界怎么划
推荐按“知识域”拆集成，而不是全公司一个大集成：
kb-sync-customer-support→ 只连客服相关页面树kb-sync-engineering→ 只连研发文档页面树kb-sync-hr→ 只连制度类页面，且只给 HR 智能体用
每个集成的令牌只配给对应智能体的检索服务。这样即使某个智能体被越权调用，它能看到的也只是一棵子树。
把“集成 → 可见页面根节点”的映射写成配置，同步时按根节点递归，天然形成白名单：
```python
INTEGRATIONS = {
"customer_support": {
"roots": ["<客服手册根页面ID>", "<常见问题库页面ID>"],
"audience": ["support_agent", "cs_bot"],
},
"engineering": {
"roots": ["<研发文档根页面ID>"],
"audience": ["dev_agent"],
},
}
```
>audience 会写进索引元数据，检索时用来做过滤。权限判断发生在检索层，不要指望写在系统提示里让模型自觉遵守。
---
第三步：枚举可访问对象，建立同步清单
用 Search 接口把当前令牌可见的页面全列出来。注意它返回的是“已授权给这个集成的对象”，正好就是我们要的白名单。
```python
def search_all(object_type="page"):
results, cursor = [], None
while True:
body = {"page_size": 100,
"filter": {"property": "object", "value": object_type}}
if cursor:
body["start_cursor"] = cursor
data = call("POST", "/search", json=body)
results.extend(data["results"])
if not data.get("has_more"):
return results
cursor = data["next_cursor"]
```
拿到结果后，把每个对象的 id、last_edited_time、url、标题抽出来入库。标题位置随对象类型不同（页面在 properties 里，数据库在 title 里），写个兼容函数：
```python
def get_title(obj):
if obj["object"] == "database":
return "".join(t["plain_text"] for t in obj.get("title", []))
for prop in obj.get("properties", {}).values():
if prop.get("type") == "title":
return "".join(t["plain_text"] for t in prop.get("title", []))
return "未命名"
```
> 提示：Search 有索引延迟，刚创建的页面可能查不到；另外它只保证返回已授权的对象，不保证返回顺序稳定。增量同步靠 last_edited_time 排序，全量对账每天跑一次兜底。
---
第四步：拉取页面正文（递归展开子块）
Notion 的页面正文是块树，/blocks/{id}/children 一次只返回一层，且分页。两层循环都要写全，否则你会得到“只有标题没有内容”的索引。
```python
def fetch_children(block_id):
out, cursor = [], None
while True:
params = {"page_size": 100}
if cursor:
params["start_cursor"] = cursor
data = call("GET", f"/blocks/{block_id}/children", params=params)
out.extend(data["results"])
if not data.get("has_more"):
return out
cursor = data["next_cursor"]
```
把块转成带层级的纯文本，保留标题层级，后面切块时要用：
```python
def rt(rich):
return "".join(seg.get("plain_text", "") for seg in (rich or []))
HEAD_PREFIX = {"heading_1": "# ", "heading_2": "## ", "heading_3": "### "}
def block_to_text(block):
t = block["type"]
d = block.get(t, {})
if t in HEAD_PREFIX:
return HEAD_PREFIX[t] + rt(d.get("rich_text"))
if t in ("paragraph", "quote", "callout"):
return rt(d.get("rich_text"))
if t in ("bulleted_list_item", "numbered_list_item"):
return "- " + rt(d.get("rich_text"))
if t == "to_do":
return ("- [x] " if d.get("checked") else "- [ ] ") + rt(d.get("rich_text"))
if t == "code":
return "``\n" + rt(d.get("rich_text")) + "\n``"
if t == "table_row":
return " | ".join(rt(cell) for cell in d.get("cells", []))
return ""
递归：有 children 的块要展开，否则 toggle、column_list 里的内容会整段丢失
def flatten(block_id, depth=0, max_depth=8):
lines = []
if depth > max_depth:
return lines
for b in fetch_children(block_id):
text = block_to_text(b)
if text:
lines.append(text)
if b.get("has_children"):
lines.extend(flatten(b["id"], depth + 1, max_depth))
return lines
```
max_depth 是保险丝：同步块（synced_block）互相引用时可能成环，靠深度限制止血。
---
第五步：增量同步 —— 用水位线而不是全量比对
核心思路：记录上次同步成功的时间水位线（watermark），只处理水位线之后被编辑过的页面。
Search 接口支持按 last_edited_time 倒序，因此可以从最新往回翻，翻到比水位线旧就停，避免每次都遍历全部页面：
```python
def search_changed_since(watermark_iso):
changed, cursor = [], None
while True:
body = {
"filter": {"property": "object", "value": "page"},
"sort": {"direction": "descending", "timestamp": "last_edited_time"},
"page_size": 100,
}
if cursor:
body["start_cursor"] = cursor
data = call("POST", "/search", json=body)
stop = False
for page in data["results"]:
if page["last_edited_time"] <= watermark_iso:
stop = True
break
changed.append(page)
if stop or not data.get("has_more"):
return changed
cursor = data["next_cursor"]
```
状态表设计：
```sql
CREATE TABLE IF NOT EXISTS pages (
page_id TEXT PRIMARY KEY,
title TEXT,
url TEXT,
last_edited_time TEXT,
content_hash TEXT,
audience TEXT,
synced_at TEXT,
deleted INTEGER DEFAULT 0
);
CREATE TABLE IF NOT EXISTS sync_state (
key TEXT PRIMARY KEY,
value TEXT
);
```
同步流程：
1. 从 sync_state 读出 last_watermark。
2. 调用 search_changed_since 拿到候选页面。
3. 逐页拉正文，算 content_hash；哈希没变就跳过重建索引（有些页面只是被“打开过”或属性微调，正文其实没动）。
4. 写入分块和向量，更新 pages 行。
5. 把本次最大 last_edited_time 写回 sync_state。
关键陷阱：编辑子页面不会更新父页面的 last_edited_time。所以不能只同步“变了的页面”，还要维护页面树关系；更稳妥的做法是每天做一次全量枚举对账：
```python
def reconcile(all_visible_ids):
"""把已经不可见或已删除的页面标记掉，避免索引里残留过期内容。"""
known = {row[0] for row in conn.execute("SELECT page_id FROM pages WHERE deleted = 0")}
for pid in known - set(all_visible_ids):
conn.execute("UPDATE pages SET deleted = 1 WHERE page_id = ?", (pid,))
delete_chunks_of(pid) # 同时从向量库删除
conn.commit()
```
这一步是权限收敛的最后一道闸门：只要页面被取消授权，下一次对账就会把它的内容从索引里删掉。
---
第六步：分块与索引 —— 决定命中率的地方
不要按固定字数硬切。按标题层级切，并给每个块加上“面包屑前缀”：
```python
def build_chunks(page_title, lines, page_meta):
chunks, buf, path = [], [], [page_title]
def flush():
if not buf:
return
body = "\n".join(buf).strip()
if len(body) < 40: # 太短的块没有检索价值
buf.clear(); return
chunks.append({
"text": " > ".join(path) + "\n" + body,
"breadcrumb": " > ".join(path),
"page_id": page_meta["page_id"],
"url": page_meta["url"],
"audience": page_meta["audience"],
"last_edited_time": page_meta["last_edited_time"],
})
buf.clear()
for line in lines:
if line.startswith("## ") or line.startswith("### "):
flush()
level = 2 if line.startswith("## ") else 3
path = path[:level] + [line.lstrip("# ").strip()]
else:
buf.append(line)
if sum(len(x) for x in buf) > 700: # 上限控制，避免块过大稀释语义
flush()
flush()
return chunks
```
三个直接影响命中率的决定：
- 块大小：中文建议单块 300~800 字，长段落按段落边界切，不要从句子中间断开。
- 面包屑前缀：把
客服手册 > 计费与退款 > 退款审批流拼进块的正文再算向量。这一步通常能明显改善“短问题命中长文档小节”的效果，因为块本身可能没出现“退款”这个词，但路径里有。 - 元数据全带上：
page_id、audience、last_edited_time、url，后面过滤和回链都靠它。
---
第七步：检索接口 —— 权限过滤 + 混合召回
检索函数的第一件事是算权限，第二件事才是算相似度。
```python
def allowed_page_ids(audience):
rows = conn.execute(
"SELECT page_id FROM pages WHERE deleted = 0 AND audience LIKE ?",
(f"%{audience}%",)
).fetchall()
return [r[0] for r in rows]
def retrieve(query, audience, top_k=8):
allowed = allowed_page_ids(audience)
if not allowed:
return []
vec_hits = vector_search(query, limit=40, page_ids=allowed) # 向量召回
kw_hits = keyword_search(query, limit=40, page_ids=allowed) # 关键词/BM25 召回
RRF 融合：不依赖两路分数量纲，稳定好调
score = {}
for rank, hit in enumerate(vec_hits):
score[hit["chunk_id"]] = score.get(hit["chunk_id"], 0) + 1 / (60 + rank)
for rank, hit in enumerate(kw_hits):
score[hit["chunk_id"]] = score.get(hit["chunk_id"], 0) + 1 / (60 + rank)
merged = sorted(score.items(), key=lambda x: -x[1])[:top_k]
return [fetch_chunk(cid) for cid, _ in merged]
```
为什么必须做混合召回：Notion 里大量专有名词（产品代号、内部缩写、人名），纯向量容易“语义相近但答非所问”；而纯关键词又接不住“怎么退钱”和“退款流程”这种同义表达。两路融合后再交给模型，稳定得多。
返回给智能体的内容要带出处，方便人核对：
```json
{
"text": "客服手册 > 计费与退款 > 退款审批流\n企业版退款需……",
"source_url": "https://www.notion.so/...",
"breadcrumb": "客服手册 > 计费与退款 > 退款审批流",
"last_edited_time": "2025-xx-xxTxx:xx:xx.000Z"
}
```
---
第八步：命中率调优 —— 用评测集说话，别凭感觉改
准备 30~50 条真实问题，每条标注“应该命中的页面 ID 或小节”。跑批计算两个指标：
- Recall@8：正确来源有没有出现在前 8 条里。
- 来源正确率：模型最终引用的是不是那条来源。
```python
def evaluate(cases, audience):
hit, total = 0, 0
for q, gold_page_id in cases:
results = retrieve(q, audience, top_k=8)
total += 1
if any(r["page_id"] == gold_page_id for r in results):
hit += 1
return hit / total
```
调优顺序建议（从收益高到低）：
1. 先补面包屑前缀：很多“检索不到”其实是块里缺上下文。
2. 再调块大小：把 1500 字的大块拆到 600 字左右，看 Recall 变化。
3. 然后加标题权重：页面标题、小节标题命中的块，分数乘一个大于 1 的系数。
4. 再做查询改写：让模型把“怎么退钱”改写成“退款 流程 审批 周期”再检索，对口语化提问提升明显。
5. 最后考虑重排模型：成本最高，放在前四步做完之后再上。
每次只改一个变量，记录 Recall 变化。没有评测集的调优基本等于碰运气。
---
常见坑与排错
403 / object_not_found
九成不是令牌错，而是页面没连到集成。回到 Notion 页面 → ... → 连接 → 添加集成。数据库要单独连一次。
400 报错说版本头无效
Notion-Version 没带或写错。这个头是必填的，值以官方文档当前版本为准。
同步出来的内容只有标题
只取了一层 children，没处理 has_children 递归，也没翻页。has_more 为真时必须用 next_cursor 继续。
部分内容整段消失
toggle 折叠块、分栏（column_list）、同步块（synced_block）、表格行都在子块里。table 块本身没有文本，要展开到 table_row 才有内容。
频繁 429
并发开太高。串行 + 指数退避 + 读 Retry-After，单进程同步一般够用；确实要并发就把并发压到很低。
增量同步漏更新
两个典型原因：一是编辑子页面不改父页面 last_edited_time；二是把水位线设成“本次运行开始时间”，运行期间发生的编辑会被跳过。水位线要用本次拉到的最大 last_edited_time，不是当前时间。
取消授权后智能体还在引用旧内容
只做了增量、没做对账。每天跑一次全量枚举，把不可见的页面标记 deleted = 1 并从向量库删除。
检索总召回同一段无关内容
通常是某个超长块被反复切出相似向量。检查块大小上限，并把“目录型”页面（纯链接列表）单独降权或跳过。
中文检索效果差
确认关键词那一路用的是支持中文的分词方案（或直接用支持中文的 BM25 实现）；纯按空格切词对中文等于没有关键词召回。
令牌泄露风险
检查三处：代码库、日志、提示词。集成令牌绝不能出现在给模型的上下文里，它只属于同步服务。日志里打印请求时把 Authorization 头脱敏。
---
下一步建议
1. 接上 MCP 走一遍 PoC。Notion 官方在持续演进面向智能体的接入方式（连接器、MCP 服务等，具体名称与能力以官方文档当前版本为准）。如果你的智能体框架支持 MCP，可以先用它快速验证“能不能答对”，再决定是否自建索引。自建索引的价值在于：权限过滤可控、分块策略可调、命中率可度量。
2. 加引用回链。检索结果里已经带了 source_url，让智能体在回答末尾列出 1~3 条来源链接，人工核对成本会大幅下降。
3. 加权限审计日志。每次检索记录 audience、allowed_page_ids 数量、返回的 page_id 列表。出问题时能直接定位是授权配错还是召回跑偏。
4. 扩到数据库型内容。等页面同步稳定后，用数据库查询接口把任务库、FAQ 库这类结构化内容也接进来，并按属性做过滤（比如只同步“状态=已发布”的行），避免把草稿喂给智能体。
5. 把评测集接进 CI。每次改分块或检索参数，自动跑一遍 Recall，掉点就报警。这套东西一旦上线就会被依赖，靠人肉回归撑不住。