从零搭建「AI 驱动的合规采集」流水线：Python + Ollama 本地大模型实战
全程本地运行、断网可用、零 API 费用。AI 负责选题与解析，爬虫守规矩干活。
本文 80% 是可直接复制的代码和命令，跟着敲完你就有一条完整流水线。
一、为什么是「AI + 合规爬虫」
传统爬虫的三大痛点：选题靠拍脑袋（不知道采什么）、解析靠硬编码（每个站写一套正则）、维护靠缘分（页面一改版全崩）。
把本地大模型塞进流水线后：
| 环节 | 传统做法 | AI 驱动做法 |
| 选题 | 人工想关键词 | 模型按方向生成选题 + 检索词 |
| 解析 | 每站写正则/CSS | 模型按统一 schema 抽字段 |
| 小结 | 无 | 模型对批次数据自动摘要 |
而「合规」不是道德口号，是工程约束：robots 预检、限速、止损、去重，全部代码化、默认开启。AI 负责脑力，不负责对抗——逆向、破解、轰炸，本文一概没有。
技术栈一句话：requests（请求）+ feedparser（RSS）+ Ollama（本地 LLM，OpenAI 兼容接口）+ 标准库。
话不多说，先上资源，边看边做链接：https://pan.quark.cn/s/e284d92f8ed5
二、环境准备（5 条命令）
# 1. 安装 Python 依赖（3.8+） pip install requests beautifulsoup4 lxml trafilatura feedparser # 2. 安装 Ollama（Windows：官网 ollama.com 下载 OllamaSetup.exe，约 1.5GB） # Linux 一条命令： curl -fsSL https://ollama.com/install.sh | sh # 3. 拉一个小模型先跑通（~1GB，7B/8B 模型约 5GB，建议内存 16GB+） ollama pull qwen2.5:1.5b # 4. 验证服务活着（11434 是 Ollama 默认端口） curl http://localhost:11434/api/tags # 5. 验证 OpenAI 兼容接口 curl http://localhost:11434/v1/chat/completions \ -H "Content-Type: application/json" \ -d '{"model":"qwen2.5:1.5b","messages":[{"role":"user","content":"说一句你好"}]}'
两个必设的环境变量（Windows：系统环境变量 → 新建，然后重启）：
OLLAMA_MODELS = D:\ollamaModels # 模型别塞 C 盘，动辄几个 GB
跑大任务的机器建议把间隔和重试也固化成环境变量，代码里读：
import os MIN_INTERVAL = float(os.getenv("CRAWL_INTERVAL", "2.0"))
三、架构总览
┌────────────────────────────────────────────────┐ │ config.ai.json ← 你唯一的配置文件（模型/限速） │ └───────────────────────┬────────────────────────┘ ▼ ┌──────────────┐ ┌─────────────────────────┐ │ ① AI 选题 │ → │ topics.json │ └──────────────┘ └───────────┬─────────────┘ ▼ ┌──────────────────────────────────────────────┐ │ ② 总控流水线 │ │ robots 预检 → 采集 → URL/MD5 双重去重 │ │ → AI 小结 → 导出 JSONL │ └───────────────────────┬──────────────────────┘ ▼ ┌──────────────┐ ┌─────────────────────────┐ │ ③ AI 解析 │ │ report.jsonl（结构化） │ └──────────────┘ └─────────────────────────┘
分工原则一句话：AI 出脑，爬虫出手，配置出权限。
四、核心代码 ①：合规请求封装（先立规矩）
写爬虫第一行代码前，先把"礼貌"写进基础设施。这个封装默认开启：2 秒限速、超时、重试上限、403/429 止损。
# polite.py —— 合规请求基础设施 import time import requests MIN_INTERVAL = 2.0 # 全局强制间隔（秒） MAX_RETRIES = 2 # 重试上限，防无限轰炸 STOP_STATUS = (403, 429) # 止损码：收到即放弃 HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; learning-bot/1.0)"} _last = [0.0] def _throttle(): elapsed = time.time() - _last[0] if elapsed < MIN_INTERVAL: time.sleep(MIN_INTERVAL - elapsed) _last[0] = time.time() def polite_get(url, params=None, retries=MAX_RETRIES, timeout=12): for attempt in range(1, retries + 2): _throttle() try: resp = requests.get(url, params=params, headers=HEADERS, timeout=timeout) if resp.status_code == 200: return resp if resp.status_code in STOP_STATUS: print(f"[止损] {url} 返回 {resp.status_code}，立即停止") return None # 绝不重试轰炸 print(f"[重试 {attempt}/{retries}] {url} → {resp.status_code}") except requests.RequestException as e: print(f"[重试 {attempt}/{retries}] {type(e).__name__}") return None
用起来和无脑 requests.get 一样：
from polite import polite_get resp = polite_get("https://hn.algolia.com/api/v1/search", params={"tags": "front_page", "hits_per_page": 5})
限速不是可选项。 收到 429 时读响应头的 Retry-After 等待，是服务器在告诉你"多久后可以再来"——听它的话：
if resp.status_code == 429: wait = int(resp.headers.get("Retry-After", 30)) print(f"超速，等 {wait}s") time.sleep(wait)
五、核心代码 ②：本地模型适配层（20 行接入任意模型）
Ollama 提供 OpenAI 兼容接口，所以一套代码通吃 Ollama / DeepSeek / 通义 / Kimi / 智谱——只换 api_base 和 model。
# ai_adapter.py —— 可插拔 LLM 适配层 import json, re, time, requests class AIAdapter: def __init__(self, api_base="http://localhost:11434/v1", api_key="ollama", model="qwen2.5:1.5b", timeout=60): self.api_base, self.api_key = api_base.rstrip("/"), api_key self.model, self.timeout = model, timeout self._last = 0.0 def chat(self, prompt, system=None, json_mode=False): messages = [] if system: messages.append({"role": "system", "content": system}) messages.append({"role": "user", "content": prompt}) payload = {"model": self.model, "messages": messages, "temperature": 0.3, "max_tokens": 1000} if json_mode: payload["response_format"] = {"type": "json_object"} try: resp = requests.post( f"{self.api_base}/chat/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json=payload, timeout=self.timeout) content = resp.json()["choices"][0]["message"]["content"] return self._extract_json(content) if json_mode else content except Exception as e: print(f"[AI降级] {type(e).__name__}") # 失败永不抛异常 return None @staticmethod def _extract_json(text): """模型输出的 JSON 容错提取：裸JSON → ```围栏 → 大括号扫描""" try: return json.loads(text) except Exception: pass m = re.search(r"\{[\s\S]*\}|\[[\s\S]*\]", text) try: return json.loads(m.group()) if m else None except Exception: return None
验证一下：
ai = AIAdapter() print(ai.chat("用一句话解释什么是robots.txt")) print(ai.chat('输出JSON {"ok":1}', json_mode=True)) # {'ok': 1}
经验：
json_mode依赖模型支持response_format；不支持的老模型就靠_extract_json兜底——永远做两层防御，因为模型总会在 JSON 前后说废话。
六、核心代码 ③：AI 选题（约束在合规源白名单内）
关键技巧：把数据源白名单写进提示词，模型只能在清单里选，从源头杜绝它"自由发挥"出不可控的目标。
# topic_agent.py from ai_adapter import AIAdapter WHITELIST = [ {"name": "Open-Meteo 天气", "url": "https://api.open-meteo.com", "适合": "天气/气候"}, {"name": "Frankfurter 汇率", "url": "https://api.frankfurter.dev", "适合": "金融/经济"}, {"name": "Hacker News", "url": "https://hn.algolia.com/api/v1/search", "适合": "技术热点"}, {"name": "NASA 图片库", "url": "https://images-api.nasa.gov/search", "适合": "科普/图片"}, {"name": "Gutendex 公版书", "url": "https://gutendex.com", "适合": "文学/文本分析"}, ] def generate_topics(ai: AIAdapter, direction: str, count=5): src_desc = "\n".join(f"- {s['name']}（{s['url']}，适合：{s['适合']}）" for s in WHITELIST) prompt = ( f"你是数据选题策划。方向：{direction}\n" f"从下面的合规数据源中挑选，生成{count}个采集选题。\n" f'严格输出JSON数组，字段：topic, source, keywords(数组), reason\n' f"可用数据源：\n{src_desc}\n" f"注意：只能用列表中的数据源。" ) topics = ai.chat(prompt, system="只输出JSON。", json_mode=True) if isinstance(topics, list): src_map = {s["name"]: s["url"] for s in WHITELIST} for t in topics: t["source_url"] = src_map.get(t["source"], "") return topics[:count] return [] # AI失败时上层降级为人工清单 if __name__ == "__main__": ai = AIAdapter() for t in generate_topics(ai, "数据分析练手", 3): print(t["topic"], "→", t["source"], t["keywords"])
本地 1.5B 小模型的输出质量有限，两个加固手段：
# 加固1：关键词清洗——中文选题词对英文接口无意义，过滤掉 kws = [k for k in t.get("keywords", []) if k and any(c.isascii() and c.isalpha() for c in k)] # 加固2：schema 校验——字段缺失直接丢弃该选题 valid = [t for t in topics if {"topic", "source"} <= set(t.keys())]
七、核心代码 ④：采集器（每个源一个函数，统一三字段）
所有采集器返回统一格式 {"title": ..., "text": ..., "url": ...}，下游就完全不用关心数据来自谁。
# crawlers.py import requests from polite import polite_get, HEADERS def crawl_hn(keywords, per_page=10): """Hacker News 公开检索接口""" q = " ".join(keywords) or "python" resp = polite_get("https://hn.algolia.com/api/v1/search", params={"query": q, "tags": "story", "hits_per_page": per_page}) if not resp: return [] return [{"title": h.get("title", ""), "text": f"{h.get('title','')} {h.get('story_text') or ''} 分值{h.get('points',0)}", "url": h.get("url") or f"https://news.ycombinator.com/item?id={h.get('objectID')}"} for h in resp.json().get("hits", [])] def crawl_weather(_keywords): """Open-Meteo 免费天气接口（无需密钥）""" resp = polite_get("https://api.open-meteo.com/v1/forecast", params={"latitude": 39.9, "longitude": 116.4, "current_weather": "true", "timezone": "auto"}) if not resp: return [] cw = resp.json().get("current_weather", {}) return [{"title": f"北京实时天气 {cw.get('temperature')}°C", "text": str(cw), "url": "https://api.open-meteo.com"}] def crawl_wiki(keywords, limit=10): """Wikipedia MediaWiki API：搜索 + 摘要""" kw = keywords[0] if keywords else "人工智能" resp = polite_get("https://zh.wikipedia.org/w/api.php", params={ "action": "query", "list": "search", "srsearch": kw, "srlimit": limit, "format": "json"}) if not resp: return [] import re out = [] for h in resp.json().get("query", {}).get("search", []): snippet = re.sub(r"<[^>]+>", "", h.get("snippet", "")) # 去高亮标签 out.append({"title": h.get("title", ""), "text": snippet, "url": "https://zh.wikipedia.org/wiki/" + h.get("title", "")}) return out ROUTES = {"Hacker News": crawl_hn, "Open-Meteo 天气": crawl_weather, "Wikipedia": crawl_wiki}
路由就一行字典查询，找不到就跳过——路由不存在时宁可放弃，不要现场发明采集方式。
八、核心代码 ⑤：robots 预检 + 双重去重（合规的牙齿）
# compliance.py import hashlib import urllib.parse import requests from polite import HEADERS def robots_allows(url): """robots.txt 预检：只看 User-agent: * 组是否全站 Disallow""" try: p = urllib.parse.urlparse(url) resp = requests.get(f"{p.scheme}://{p.netloc}/robots.txt", headers=HEADERS, timeout=8) if resp.status_code != 200: return True # 无robots → 允许（仍保持限速） in_star = False for line in resp.text.splitlines(): line = line.strip().lower() if line.startswith("user-agent:"): in_star = line.split(":", 1)[1].strip() == "*" elif in_star and line.startswith("disallow:"): if line.split(":", 1)[1].strip() == "/": return False # 全站禁爬 return True except Exception: return True def md5_of(text): return hashlib.md5(text.encode("utf-8")).hexdigest() def dedup(items, seen_urls, seen_md5): """URL + 内容MD5 双保险去重""" fresh = [] for it in items: if not it.get("title"): continue u, h = it.get("url", ""), md5_of(it["title"] + it.get("text", "")[:100]) if (u and u in seen_urls) or h in seen_md5: continue seen_urls.add(u); seen_md5.add(h) fresh.append(it) return fresh
实测里有个反直觉的发现：arXiv 的 export.arxiv.org/robots.txt 写的是全站 Disallow: /，而 Open-Meteo、Gutendex 这些纯 API 站点也这么写——因为 robots 是写给"爬网页"的，官方 API 提供方在文档里明确邀请程序化调用。所以预检要区分：登记在册的开放 API 按 API 条款调用；网页类目标严格过 robots 门。一刀切要么漏掉合法 API，要么放进不该爬的页面。
九、核心代码 ⑥：AI 解析器（一套 schema 吃遍所有源）
# parser_agent.py from ai_adapter import AIAdapter SCHEMA = { "主题分类": "technology|business|science|other 四选一", "一句话摘要": "不超过40字", "涉及技术或公司": "数组，没有则空数组", "热度评价": "low|medium|high 三选一", } def ai_extract(ai: AIAdapter, text): prompt = (f"按schema抽取信息：{json.dumps(SCHEMA, ensure_ascii=False)}\n" f"只输出JSON。\n\n内容：{text[:3000]}") return ai.chat(prompt, system="你是信息抽取助手，只输出JSON。", json_mode=True) def rule_extract(text, points=0): """AI不可用时的规则降级：关键词匹配""" tech = [k for k in ("python", "ai", "llm", "rust", "api") if k in text.lower()] heat = "high" if points >= 300 else "medium" if points >= 100 else "low" return {"主题分类": "technology" if tech else "other", "一句话摘要": text[:40], "涉及技术或公司": tech, "热度评价": heat}
为什么降级模式这么重要：本地模型会崩（显存不够）、会输出怪 JSON、会超时。规则模式保证流水线永远出得来数据，AI 只是锦上添花——这是"AI 参与关键路径"和"AI 挂在关键路径上"的区别。
十、核心代码 ⑦：总控流水线（串起来）
# main.py —— 完整流水线 import json, time from ai_adapter import AIAdapter from topic_agent import generate_topics from crawlers import ROUTES from compliance import robots_allows, dedup from parser_agent import ai_extract, rule_extract def run(direction="数据分析练手", count=3): ai = AIAdapter() topics = generate_topics(ai, direction, count) if not topics: # AI 降级：手工兜底选题 topics = [{"topic": "HN技术热点", "source": "Hacker News", "keywords": [], "source_url": "https://hn.algolia.com/api/v1/search"}] seen_urls, seen_md5, report = set(), set(), [] for i, t in enumerate(topics, 1): print(f"\n── 选题 {i}/{len(topics)}：{t['topic']} → {t['source']}") # 1) robots 预检（合规第一道门） if t.get("source_url") and not robots_allows(t["source_url"]): print(" [跳过] robots 全站禁爬") continue # 2) 路由采集 crawler = ROUTES.get(t["source"]) if not crawler: print(f" [跳过] 无 {t['source']} 的适配器") continue items = crawler(t.get("keywords", [])) time.sleep(2) # 3) 去重 fresh = dedup(items, seen_urls, seen_md5) print(f" [采集] {len(items)} 条 → 去重后 {len(fresh)} 条") # 4) AI 解析 for it in fresh: parsed = ai_extract(ai, it["title"] + " " + it["text"][:200]) it["ai"] = parsed if isinstance(parsed, dict) else rule_extract(it["title"]) report.append(it) # 5) 导出 JSONL（一行一条，方便流式读取和增量） with open("report.jsonl", "w", encoding="utf-8") as f: for it in report: f.write(json.dumps(it, ensure_ascii=False) + "\n") print(f"\n[完成] {len(report)} 条 → report.jsonl") if __name__ == "__main__": run()
实际运行输出长这样：
── 选题 1/3：USD兑CNY汇率季度波动观察 → Frankfurter 汇率 [采集] 1 条 → 去重后 1 条 ── 选题 2/3：技术圈「web scraping」热点趋势观察 → Hacker News [采集] 10 条 → 去重后 10 条 ── 选题 3/3：Wikipedia「网络爬虫」词条知识梳理 → Wikipedia [采集] 5 条 → 去重后 5 条 [完成] 16 条 → report.jsonl
十一、接入第三方 API（不改代码）
新源不想写采集器？做成注册表，把请求和响应映射写成配置：
{ "name": "我的第三方API", "enabled": true, "type": "custom_api", "request": { "method": "GET", "url": "https://api.example.com/v1/search", "params": {"q": "{keyword}", "page_size": 10}, "headers": {"Authorization": "Bearer 你的token"} }, "response": { "items_path": "data.items", "title_path": "item.name", "text_path": "item.description", "url_path": "item.link" } }
通用采集器只需一个点路径解析函数（约 20 行）：
import re def resolve_path(data, path): """支持 data.items / results[] / items[0].name / @常量""" if path.startswith("@"): return path[1:] cur = data for seg in path.split("."): m = re.match(r"^(\w+)?(\[\])?(?:\[(\d+)\])?$", seg) if not m or cur is None: return None name, wild, idx = m.groups() if name: cur = ([x.get(name) for x in cur if isinstance(x, dict)] if isinstance(cur, list) else cur.get(name)) if wild: cur = cur if isinstance(cur, list) else ([] if cur is None else [cur]) if idx is not None: cur = cur[int(idx)] if isinstance(cur, list) and int(idx) < len(cur) else None return cur
{keyword} 占位符在请求前替换，选题关键词自动注入第三方接口——选题、采集、解析整条链就通了。
十二、常见坑速查
| 现象 | 原因 | 解决 |
connection refused localhost:11434 |
Ollama 没启动 | 启动 Ollama / ollama serve |
| AI 输出全是废话 JSON 解析失败 | 小模型能力弱 | 加 response_format；_extract_json 兜底；换 7B+ 模型 |
| 中文关键词查英文接口 0 结果 | 检索词没清洗 | 过滤非 ASCII 词，置空让接口走默认值 |
| Windows 模型爆 C 盘 | 默认路径 | 环境变量 OLLAMA_MODELS |
| 429 越重试越多 | 重试无退避 | 读 Retry-After，最多重试 2 次 |
resp.json() 报错 |
接口返回 HTML | 先看 Content-Type；嗅探真实接口 |
| 磁盘里模型文件找不到 | 默认藏用户目录 | ollama list / OLLAMA_MODELS |
调试请求的第一站永远是 httpbin.org——它把你发的所有头、参数原样弹回来：
curl "https://httpbin.org/headers" -H "User-Agent: learning-bot/1.0"
十三、合规清单（发布前对照打勾）
- 目标全是官方开放 API / 公版内容 / 明确允许的练习站
- robots.txt 预检在代码里，默认开启
- 全局限速 ≥2 秒，Session 复用连接
- 403/429 止损，重试有上限
- URL + 内容双重去重
- AI 提示词约束在数据源白名单内
- 数据只用于学习研究，不商用不转售
- API Key 不进代码仓库（用环境变量）
十四、写在最后
这条流水线的精髓不是某个函数，而是分层降级的设计：AI 挂了走规则模式、接口挂了跳过、内容重复被去重、全站禁爬被预检拦下——每一层失败都有兜底，整条线永远出得来数据。
把 1.5B 小模型换成 7B，把三个采集器换成十个，把 JSONL 换成 DuckDB——这条线的上限取决于你，而不取决于某个网站改没改版。
完整工程化版本（含第三方 API 注册表、9 种内置源路由、断点续爬、词频统计）已整理成资源包，本文所有代码都在其中可以直接运行。
