从零搭建「AI 驱动的合规采集」流水线：Python + Ollama 本地大模型实战

全程本地运行、断网可用、零 API 费用。AI 负责选题与解析，爬虫守规矩干活。
本文 80% 是可直接复制的代码和命令，跟着敲完你就有一条完整流水线。

一、为什么是「AI + 合规爬虫」

传统爬虫的三大痛点：选题靠拍脑袋（不知道采什么）、解析靠硬编码（每个站写一套正则）、维护靠缘分（页面一改版全崩）。

把本地大模型塞进流水线后：

而「合规」不是道德口号，是工程约束：robots 预检、限速、止损、去重，全部代码化、默认开启。AI 负责脑力，不负责对抗——逆向、破解、轰炸，本文一概没有。

技术栈一句话：requests（请求）+ feedparser（RSS）+ Ollama（本地 LLM，OpenAI 兼容接口）+ 标准库。

话不多说，先上资源，边看边做链接，请看这一篇文章：# 零搭建「AI 驱动的合规采集」流水线：Python + Ollama 本地大模型实战-CSDN博客

二、环境准备（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} 占位符在请求前替换，选题关键词自动注入第三方接口——选题、采集、解析整条链就通了。

十二、常见坑速查

调试请求的第一站永远是 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 种内置源路由、断点续爬、词频统计）已整理成资源包，本文所有代码都在其中可以直接运行。