# 把链接读取接进 RAG 管道：去重、缓存与分块
接口：链接读取（apiCode=3262）· 接入点：获取网页正文（3262-1）· 计费：50 厘/次 · 并发上限：2 次/秒 · 适用人群：要把公开网页正文批量灌进知识库的工程师 · 阅读时间：约 12 分钟 · 最后实测核对：2026-09-15
## 核心要点
- 读取前先按归一化 URL 去重，每次调用 0.05 元，去重直接减少实际调用次数。
- 读取失败的 URL 单独记入过滤表，与成功结果**分表存储**，避免对同一个 URL 重复调用。
- 并发要压在 2 次/秒以内，令牌桶比 `sleep` 好写，也不会拖慢整体。
## 管道的四个环节
一条能上生产的读取管道，比"写个循环调接口"多出四件事：
```
URL 列表 → ① 归一化去重 → ② 缓存查命中 → ③ 限流读取（2 次/秒）
↓
④ 清洗分块 → 向量库
↓
失败清单（不再重读）
```
下面按顺序拆。所有代码基于 Python，缓存用 SQLite（够用，换成 Redis 也行，键的设计一样）。
## 第一步：URL 归一化
同一个页面会有好几种写法，不去重会产生重复调用。
```python
from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
# 会被去掉的跟踪参数（按你实际数据源补充）
TRACKING = {"utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content",
"spm", "from", "share_token", "ref"}
def normalize(url: str) -> str:
s = urlsplit(url.strip())
scheme = (s.scheme or "https").lower()
netloc = s.netloc.lower()
if netloc.endswith(":80") and scheme == "http":
netloc = netloc[:-3]
if netloc.endswith(":443") and scheme == "https":
netloc = netloc[:-4]
path = s.path or "/"
if path != "/" and path.endswith("/"):
path = path[:-1] # 去掉尾部斜杠
qs = [(k, v) for k, v in parse_qsl(s.query, keep_blank_values=True) if k not in TRACKING]
return urlunsplit((scheme, netloc, path, urlencode(sorted(qs)), "")) # 丢掉 fragment
```
`#` 后面的 fragment 一定去掉——它对服务端不可见，留着只会让同一页面变成两条记录。
## 第二步：缓存表设计
两张表，别合并。这是把"读取不到"和"还没读过"分开的关键。
```python
import sqlite3, hashlib, time
def init_db(path="pages.db"):
conn = sqlite3.connect(path)
conn.executescript("""
-- 读取成功的正文
CREATE TABLE IF NOT EXISTS pages (
url_key TEXT PRIMARY KEY, -- sha256(归一化 URL)
url TEXT NOT NULL,
content TEXT NOT NULL,
fetched_at INTEGER NOT NULL
);
-- 读取不到的 URL。单独存，避免重复付费
CREATE TABLE IF NOT EXISTS blocked (
url_key TEXT PRIMARY KEY,
url TEXT NOT NULL,
reason TEXT NOT NULL, -- empty_output / http_500 / not_html ...
attempts INTEGER DEFAULT 1,
last_try INTEGER NOT NULL
);
""")
return conn
def key_of(url: str) -> str:
return hashlib.sha256(url.encode("utf-8")).hexdigest()
```
`blocked` 表要记 `reason` 和 `attempts`。原因决定了要不要重试，次数决定了要不要永久放弃。实测里 `empty_output` 这类原因是站点层面的，重试没意义；`timeout` 才是值得退避重试的。
命中缓存的判断，按内容类型给不同 TTL：新闻页可以放久一点，首页和榜单页最好短一些。
## 第三步：令牌桶限流
并发上限是 **2 次/秒**。最简单可靠的写法是令牌桶，读取线程自己算，不用全局锁。
```python
import threading, time
class TokenBucket:
"""按固定速率放行，平滑突发的关键。rate=2 表示每秒 2 次。"""
def __init__(self, rate: float, capacity: float = None):
self.rate = rate
self.capacity = capacity if capacity is not None else rate
self.tokens = self.capacity
self.updated = time.monotonic()
self.lock = threading.Lock()
def acquire(self):
while True:
with self.lock:
now = time.monotonic()
self.tokens = min(self.capacity, self.tokens + (now - self.updated) * self.rate)
self.updated = now
if self.tokens >= 1:
self.tokens -= 1
return
wait = (1 - self.tokens) / self.rate
time.sleep(wait)
```
留点余量，把 `rate` 设成 1.8 而不是 2.0。接口的读超时是 10 秒，服务端统计口径和你本地时钟不一定对齐，贴着上限跑没必要。
单线程串行时，每 0.5 秒一次调用就够安全，但在队列消费场景下通常会有多个 worker，令牌桶是更省心的选择。
## 第四步：读取 + 分类落库
失败分类必须走到落库这一步，否则重试策略没有依据。
```python
import requests
from requests.exceptions import Timeout, ConnectionError, HTTPError
API = "https://route.showapi.com/3262-1"
APPKEY = "YOUR_APPKEY"
def fetch(url: str, bucket: TokenBucket, retries: int = 3):
"""返回 (ok, payload_or_reason)。可重试的错误在这里退避，不可重试的直接返回。"""
# 不可重试的原因：站点层面问题，重试只是重复扣费
RETRYABLE = {"timeout", "conn_error"}
for attempt in range(retries):
bucket.acquire()
try:
r = requests.post(API, params={"appKey": APPKEY, "url": url}, timeout=(5, 10))
except Timeout:
reason = "timeout"
except ConnectionError:
reason = "conn_error"
else:
if r.status_code != 200:
# 如实测的 HTTP 500 + backend fail，不计费，可重试
reason = f"http_{r.status_code}"
else:
data = r.json()
if data.get("showapi_res_code") != 0:
# 参数错误，不可重试：这是代码 bug
return False, f"gateway:{data.get('showapi_res_error','')}"
output = (data.get("showapi_res_body") or {}).get("output") or ""
if not output:
# 读取失败但已计费，不可重试
return False, "empty_output"
return True, output
if reason not in RETRYABLE:
return False, reason
time.sleep(min(2 ** attempt, 8)) # 指数退避，封顶 8 秒
return False, reason
```
注意 `empty_output` 是直接 `return` 的，不进退避循环。实测里返回空的站点重复请求结果一致，退避重试只会多花几次钱。
消费端把结果分开写：
```python
def consume(urls, conn):
bucket = TokenBucket(rate=1.8) # 留余量
ok_cnt = blocked_cnt = 0
for raw in urls:
url = normalize(raw)
k = key_of(url)
hit = conn.execute("SELECT content FROM pages WHERE url_key=?", (k,)).fetchone()
if hit:
yield {"url": url, "md": hit[0], "from_cache": True}
continue
if conn.execute("SELECT 1 FROM blocked WHERE url_key=?", (k,)).fetchone():
continue # 已知读取不到，跳过，不花钱
ok, payload = fetch(url, bucket)
if ok:
conn.execute("INSERT OR REPLACE INTO pages VALUES (?,?,?,?)",
(k, url, payload, int(time.time())))
ok_cnt += 1
yield {"url": url, "md": payload, "from_cache": False}
else:
conn.execute("""INSERT INTO blocked VALUES (?,?,?,1,?)
ON CONFLICT(url_key) DO UPDATE SET
attempts = attempts + 1, last_try = excluded.last_try""",
(k, url, payload, int(time.time())))
blocked_cnt += 1
conn.commit()
```
## 成本怎么估
单价固定，50 厘/次，也就是 0.05 元/次。
```
总成本 ≈ 实际调用次数 × 0.05 元
实际调用次数 = 总 URL 数 − 缓存命中数 − 黑名单命中数
```
举个规模感受一下：一万个 URL，如果去重后剩 9,000 条、其中 2,000 条命中缓存、800 条在黑名单里，实际调用 6,200 次，成本 310 元。
折扣档位说明：9.90 元的档位只用本接入点可以调 **198 次**（198 × 0.05 = 9.90）。调用量大的场景建议直接选择更大的档位，按 9.9 元档位叠加多份并不划算。
时间成本：并发上限 2 次/秒，6,200 次调用的理论最短耗时是 6,200 ÷ 2 = 3,100 秒，约 52 分钟。这是下限，实际还要加上单页响应耗时和重试。本次未实测单页平均耗时，按上面的公式自己乘一下更准。
## 分块：这一步决定检索效果
正文清洗和分块的写法在第 3 篇里给了完整代码（[链接读取的 output 到底是什么格式](https://www.showapi.com/guides/link-read-markdown-output-3262)），这里只说管道层面的两个决定。
**按标题切，不按固定字数切。** `output` 里的 `#` / `##` 层级来自原页面，是免费拿到的语义边界。按字数硬切会把一段论述劈成两半。
**给每一块带上来源 URL 和标题。** 检索命中后要能回到原文，也要能在回答里标出处。`chunks` 里存 `{url, title, text}`，别只存文本。
## 监控该看什么
- **空值率**：`blocked` 表里 `empty_output` 占当天调用数的比例。这个数突然涨，通常是某个数据源改了前端。
- **重试率**：`timeout` 和 `http_5xx` 的次数。持续偏高就说明你的并发压太紧了。
- **缓存命中率**：低于预期说明 URL 归一化没做干净。
- **`showapi_res_id`**：读取异常时把它记下来，找客服时比自己描述现象有效得多。
## FAQ
**Q1：一天读取十万个页面现实吗？**
并发上限 2 次/秒是硬约束，十万次调用的理论最短耗时约 13.9 小时，成本约 5,000 元。这是个天花板，实际还要打折——所以去重和缓存不是优化项，是必需项。
**Q2：读取失败的 URL 要一直留在黑名单里吗？**
不用。`blocked` 表里有 `last_try`，可以定期做一次"复活"扫描，把超过 30 天的记录重新试一遍。站点改版、撤掉访问限制之后是能恢复的。复活窗口设得过短，会带来无谓的重复调用。
**Q3：能用多线程吗？**
可以，但所有 worker 必须共用同一个令牌桶。每个线程各建一个 `TokenBucket(rate=2)`，总速率就是线程数乘以 2，会立刻触发限流。
**Q4：空值要不要缓存？**
要存，但要存进 `blocked` 表而不是 `pages` 表。混在一起的话，你的读取逻辑拿到空字符串会当成"这个页面没内容"，而不是"这个页面读取不到"。
**Q5：超时该设多少？**
接口在 OpenAPI 文档里标了连接 5 秒、读取 10 秒，客户端按这个对齐就行：`requests` 里写 `timeout=(5, 10)`。客户端设得更短的话，慢页面会在本地先超时，服务端实际能返回内容的情况也会被记为失败。
## 下一步阅读
- 清洗与分块的代码 → [链接读取的 output 到底是什么格式](https://www.showapi.com/guides/link-read-markdown-output-3262)
- 哪些页面类型该提前过滤掉 → [链接读取返回结构说明与结果判断](https://www.showapi.com/guides/link-read-empty-output-3262)
- 自建读取和这个接口怎么选 → [读取公开网页正文的方案怎么选](https://www.showapi.com/guides/link-read-vs-selfbuilt-3262)
- **本系列共 6 篇**：查看[链接读取指南总目录](https://www.showapi.com/guides/link-read-guides-3262)
