企业官网建设流程全解析

先说说我为什么非要自己动手。每天刷 AI 资讯的姿势，我曾经是：早上看一遍科技媒体，中午刷 Hacker News，晚上再扫一圈 arXiv 和 GitHub Trending。听起来很自律，实际大部分时间花在重复阅读上——同一篇新闻换个标题出现五六次，真正有深度的长文被淹没在快讯里。所以我花了两周时间，自己搭了一个 AI 资讯聚合平台：RSS 和 API 自动抓取、正文抽取、去重、AI 摘要、自动打标签，最后统一在一个网页里浏览。这篇就把完整的架构设计、代码实现、部署和踩坑过程写清楚，适合有点编程基础、想重新夺回信息流控制权的人，也适合想给团队做 AI 情报系统的朋友抄作业。

1. 为什么自己搭：信息的“策展权”不能全交给别人

1.1 现成资讯服务的三个痛点

“聚合”这件事，市面上已经有太多产品，但用下来总有三个绕不开的痛点。

第一个是信息过载。平台为了覆盖足够多的用户，会把各种来源、各种质量的资讯涌到你面前，算法推荐还会不断投喂同质化内容。你看到的不是“今天 AI 领域发生了什么”，而是“平台希望你觉得 AI 领域发生了什么”。第二个是延迟和选择性。很多平台更新靠编辑搬运，热点出来两小时后才上架，如果你想抢在第一时间看到重要论文或开源项目，根本等不起。第三个是最致命的：你没法沉淀自己的筛选标准。这个源质量高、那个源偶尔有独家，你希望用某种规则把内容排序，希望把某些主题置顶，希望根据团队需求生成定制简报，这类需求在现成服务里几乎做不了。

1.2 自己搭的平台到底该长什么样

动手之前，先别想着做一个“今日头条”，而是想清楚“每天能省下多少时间”。我给自己定义的形态是：从固定信息源自动抓取内容，清洗后去重，再用大模型生成统一格式的摘要和标签，最后展示在一个极简的网页上，按时间倒序排列，支持标签和关键词筛选。

这套系统不追求实时，半小时更新一次就够；不追求海量，控制在几十个经过筛选的源；不追求复杂交互，能搜、能筛、能点开原文，就够了。换句话说，它更像是“给第二天的自己准备一份干净的早餐”，而不是“给你一个永远刷不完的瀑布流”。

1.3 谁适合参考这套方案

如果你刚接触编程，但已经会一点 Python，照着我下面的代码抄一遍，也能跑起来。如果你是产品经理或研发负责人，想给团队搭一套 AI 情报监控，可以先按最小版本跑通，再加上邮件或群机器人推送。如果你已经有一定开发经验，我更想分享的是其中的架构思路和踩坑经验，代码里的很多细节是我跑了半年之后才稳定下来的逻辑。

2. 整体设计：先画架构，再聊代码

2.1 核心数据流

整个系统可以拆成一条流水线：采集、清洗、存储、增强、分发。

采集层负责从 RSS、API、网页定时拉数据。清洗层做正文抽取、URL 去重、标题相似度去重。存储层用 SQLite 保存结构化条目。增强层调用大模型生成摘要、标签、重要度评分，顺便做文本向量化。分发层就是一个 Web 页面，再加可选的通知推送。

为什么把增强层放在存储层后面？因为原始抓下来的内容质量参差不齐，如果先做 AI 摘要，很多脏文本会浪费 token，还会重复计算。等入库、去重之后再增强，一条内容只处理一次，成本可控。

2.2 技术选型：为什么是 Python + FastAPI + SQLite

个人项目的第一原则是：怎么简单怎么来，怎么容易维护怎么来。

Python 生态做这件事太舒服了。RSS 解析有 feedparser，正文抽取有 trafilatura，Web 框架用 FastAPI，异步性能足够一台小水管服务器扛日更几千条资讯，数据库用 SQLite 单文件，备份直接 cp 一下就行。

有人会问为什么不用爬虫框架 Scrapy。答案是杀鸡用牛刀。我们的主要数据源是 RSS 和 API，不是需要复杂抓取的站点，Scrapy 的学习和维护成本对个人项目来说偏高。至于为什么不用 Celery，单机任务、API 调用也不重，APScheduler 就够了。

2.3 关键功能模块

模块化不是为了炫技，是为了出问题时知道去哪排查。我分成这几块：采集器负责从各个源拉数据并解析出结构化条目；清洗器负责正文抽取、去掉无效字符、统一时间格式；去重器负责 URL 和标题相似度判断；增强器调用大模型生成 AI 摘要、标签、重要度分数；查询与展示层负责搜索、筛选和页面渲染。模块之间通过数据库状态字段协作，不需要复杂的消息队列。

2.4 数据规模与升级路径

如果你的资讯量到了每天几万条，SQLite 可能不够看，多个进程同时写会出现锁繁忙。这时可以平滑迁移到 PostgreSQL，把数据库连接层换成 SQLAlchemy 即可。向量检索也建议从内存计算换成 pgvector 或专门的向量库。但起步阶段，SQLite 完全能扛住，别过度设计。

3. 信息源管理：聚合的第一步是把“源头”理清

3.1 信息源分类与推荐渠道

我把自己常用的信息源分成了四类：媒体类、社区类、论文类和项目类。媒体类比如机器之心、量子位、InfoQ 中文，它们有官方 RSS，一般在 /feed 或 /rss.xml 能找到。社区类比如 Hacker News 提供官方 API，Reddit 的 r/MachineLearning 用 .rss 后缀也能导出 feed，GitHub Trending 可以配合搜索 API 按时间排序。论文类就是 arXiv 的 cs.AI、cs.CL、cs.LG 分区，RSS 地址非常规整。项目类可以看 GitHub 的 releases 和 trending，也可以通过第三方 hub 生成订阅源。

找 RSS 有个笨但有效的方法：在网址后面试探 /feed、/rss、/atom.xml，很多系统都默认支持这几个路径。找不到了再用 RSSHub 这类工具生成，生态很大。

3.2 抓取与解析的坑

很多 RSS 源只给摘要，不给全文。这时候就需要正文抽取。我选 trafilatura 而不是 newspaper3k，因为它在边界页面上的稳定性更好。它的核心思路是看 DOM 节点里的文本密度和标点比例，找出最像正文的区块。局限也很明显：如果是纯 JavaScript 渲染的页面，直接抓 HTML 是抓不到的，得上无头浏览器，但这套复杂度我建议先别碰。

抓取频率控制也容易忽略。有些站点对 RSS 抓取不设限，但你每 10 分钟去拉一次，IP 很快会被封。我的做法是每个源设置不同的抓取间隔，重要源 30 分钟一次，普通源 6 小时一次，并且带着合法的 User-Agent 和联系邮箱，尽量不给对方服务器增加压力。

3.3 多级去重策略

如果只靠 URL 去重，同一个新闻在不同来源会被重复收录。所以我去重分三层：第一层是 URL 精确去重，入库前把 URL 去掉常见跟踪参数，再算 hash；第二层是标题规范化去重，把标题转小写、去掉末尾感叹号问号、去掉【】里的栏目名，然后算编辑距离或 simhash，相似度超过阈值就认为是同一条；第三层是内容 hash，如果正文能抽出来，就基于正文前 200 字做 hash。

阈值我一般设在 0.78 左右。为什么不是 0.8 也不是 0.7？太低容易误杀不同新闻，太高漏掉同题报道。这个值要根据你自己的信息源测试，多跑几次看结果再调。

3.4 字段设计与存储

我在 items 表里预留了这些字段：source_name、url、url_hash、title、content_text、content_html、summary、tags、ai_summary、ai_tags、importance、published_at、fetched_at、status。url_hash 要做唯一索引，这是去重的基础。content_html 不是必须的，但保留下来方便以后对接前端或做富文本展示。ai_summary 和 ai_tags 一开始可以先设空，等大模型处理完再回填。

一条完整的资讯，库里至少要能回答五个问题：来自哪、链接到哪、标题是什么、正文讲什么、什么时候发的。剩下的字段都是加分项。

4. AI 能力接入：用大模型给资讯做摘要和标签

4.1 为什么需要 AI 摘要而不是只取首段

我最初觉得 RSS 自带的 summary 字段够用，但很快发现两个问题：第一，很多源不提供 summary；第二，有的摘要就是文章第一段，完全看不出这篇到底推出了什么新东西。AI 摘要的价值不是“用高级模型重写一遍”，而是信息筛选。我给自己定的标准：80 字以内说清楚三件事——谁或哪个团队、做了什么、为什么重要。

实际写提示词的时候，我会把这三件事直接写进要求，让模型输出固定 JSON，里面包含 summary、tags 和 importance。有了统一格式，网页渲染和后续筛选都会方便很多。

4.2 两种接入方式：API 与本地模型

接入大模型有两条路，我在代码里做了同一个接口，用环境变量切换。

第一条是接大模型的 API，比如 DeepSeek、Kimi、通义千问这类服务，它们大多提供兼容接口。优点是零部署、效果好，模型理解能力强；缺点是长文本要计费，密钥需要保管好，内容也会发给第三方。第二条是本地跑 Ollama，拉一个 qwen2.5 7B 级别的模型，用 /api/generate 接口调用。优点是不花钱、数据不出门；缺点是 7B 模型的摘要质量弱于商业 API，速度也慢一些。

我个人的建议是：每天处理量在 300 条以内，直接用 API 更省心；如果完全不想把内容发给第三方，就上本地模型。两条路切换成本很低，只要把 base_url、model、api_key 配置化就行。

4.3 提示词设计与成本控制

提示词不需要写得像论文，但一定要结构化。我用的模板大概是：

你是一名资深 AI 领域编辑。请为下面的资讯写一段不超过 80 字的中文摘要，并给出 3 到 5 个标签和一个重要度分数（0 到 10）。只输出 JSON，格式为： {"summary": "...", "tags": ["..."], "importance": 8.5} 文章标题: {title} 文章内容: {content}

这里有两个关键点：一是输出格式必须用 JSON 约束，方便程序解析；二是内容截断取前 3000 字就够，多数新闻核心信息在前面。温度参数我控制在 0.2 以下，防止模型自由发挥。

成本粗算一下：按 4 个字符约 1 token 估算，3000 字中文大约 1500 token，加输出总共约 2000 token。300 条一天就是 60 万 token，按当前主流 API 价格一天大概几块钱，个人可接受。如果抓的文章量大，还可以加规则：只有标题命中关键词才做 AI 摘要，能省不少钱。

4.4 从标签推荐到多 AI 协作

单条资讯处理好了，下一步就是推荐。最简单的做法是标签匹配：用户点了某个标签，把相同 tags 但不同来源的文章优先展示。体验还行，但不够聪明。进阶做法是用 embedding 生成向量相似度，把文章编码成向量存起来，查询时找相近内容。

再往上走，就是我最近在折腾的 AI Agent 工作流。可以把整个聚合平台看成三个 Agent：采集 Agent 负责定时抓取和质量初筛，摘要 Agent 负责理解并输出结构化摘要，筛选 Agent 根据你的兴趣模型给每条内容评分。三个 Agent 之间通过数据库状态字段协作，这就是很多人说的“多 AI 协作”。实际落地时不需要很重的框架，用 Python + SQLite 状态机就能跑起来。

5. 实操实现：一个最小可跑的聚合服务

5.1 项目结构与依赖

我采用的目录结构非常简单，为了让你能直接抄：

ai-aggregator/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── feeds.py │ ├── ai_enhance.py │ ├── templates/ │ │ └── index.html ├── data/ ├── requirements.txt └── docker-compose.yml

依赖文件 requirements.txt：

fastapi==0.115.6 uvicorn[standard]==0.34.0 feedparser==6.0.11 httpx==0.28.1 jinja2==3.1.4 apscheduler==3.10.4 trafilatura==1.12.0

在开始写功能之前，先把 virtualenv 建好，然后安装依赖：

python -m venv venv source venv/bin/activate pip install -r requirements.txt

5.2 采集入库：feedparser 加 SQLite

下面这段代码实现了三个最基础的动作：初始化数据库、解析 RSS、批量入库。先建表，再抓取，最后用 INSERT OR IGNORE 保证 URL 唯一。

import calendar import hashlib import sqlite3 import time from datetime import datetime, timezone import feedparser def init_db(db_path: str): conn = sqlite3.connect(db_path) conn.execute(""" CREATE TABLE IF NOT EXISTS items ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_name TEXT NOT NULL, url TEXT NOT NULL, url_hash TEXT UNIQUE NOT NULL, title TEXT NOT NULL, content_text TEXT, content_html TEXT, summary TEXT, tags TEXT, ai_summary TEXT, ai_tags TEXT, importance REAL DEFAULT 0, published_at TEXT, fetched_at TEXT, status TEXT DEFAULT 'new' ) """) conn.commit() conn.close() def _published_to_iso(entry): parsed = entry.get("published_parsed") or entry.get("updated_parsed") if parsed: return datetime.fromtimestamp(calendar.timegm(parsed), tz=timezone.utc).isoformat() return datetime.now(timezone.utc).isoformat() def fetch_rss(feed_url: str) -> list[dict]: data = feedparser.parse(feed_url) entries = [] for e in data.entries: entries.append({ "source_name": data.feed.get("title", feed_url), "title": e.get("title", "").strip(), "link": e.get("link", "").strip(), "summary": e.get("summary", ""), "content": e.get("content", [{}])[0].get("value", "") if e.get("content") else "", "published": _published_to_iso(e), }) return entries def url_hash(url: str) -> str: return hashlib.sha256(url.strip().encode("utf-8")).hexdigest() def save_entries(db_path: str, entries: list[dict]): conn = sqlite3.connect(db_path) for item in entries: h = url_hash(item["link"]) conn.execute(""" INSERT OR IGNORE INTO items (source_name, url, url_hash, title, content_text, content_html, summary, published_at, fetched_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( item["source_name"], item["link"], h, item["title"], item["content"] or item["summary"], item["content"], item["summary"], item["published"], datetime.now(timezone.utc).isoformat(), )) conn.commit() conn.close()

入库前把 URL 的跟踪参数去掉很重要，否则一个带 utm_source、一个不带，会当成两条。我在正式代码里会先做一次 clean_url，把 ?utm、?from、?ref 这些参数删掉再算 hash。

5.3 AI 增强：调用本地 Ollama 或兼容 API

增量处理未做摘要的条目，是保证系统长期运行的关键。下面这段代码封装了统一的调用入口，同时兼容 Ollama 和远程 API。

import json import os import sqlite3 from concurrent.futures import ThreadPoolExecutor import httpx OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434/api/generate") LLM_API_URL = os.getenv("LLM_API_URL", "") LLM_API_KEY = os.getenv("LLM_API_KEY", "") LLM_MODEL = os.getenv("LLM_MODEL", "qwen2.5:7b") PROMPT = """你是一名资深AI领域编辑。请为下面的资讯写一段不超过80字的中文摘要， 并给出3到5个标签和一个重要度分数(0到10)。只输出JSON，格式为: {"summary": "...", "tags": ["..."], "importance": 8.5} 文章标题: {title} 文章内容: {content[:3000]} """ def call_llm(title: str, content: str) -> dict: user_content = PROMPT.format(title=title, content=content) if LLM_API_URL: headers = {"Authorization": f"Bearer {LLM_API_KEY}"} if LLM_API_KEY else {} resp = httpx.post( f"{LLM_API_URL}/chat/completions", json={"model": LLM_MODEL, "messages": [{"role": "user", "content": user_content}], "temperature": 0.2}, headers=headers, timeout=60, ) resp.raise_for_status() return json.loads(resp.json()["choices"][0]["message"]["content"]) else: resp = httpx.post( OLLAMA_URL, json={"model": LLM_MODEL, "prompt": user_content, "stream": False, "format": "json"}, timeout=120, ) resp.raise_for_status() return json.loads(resp.json()["response"]) def enhance_pending_items(db_path: str, limit: int = 50): conn = sqlite3.connect(db_path) rows = conn.execute( "SELECT id, title, content_text FROM items WHERE ai_summary IS NULL AND status='new' " "ORDER BY published_at DESC LIMIT ?", (limit,) ).fetchall() def worker(row): rid, title, content_text = row try: result = call_llm(title, content_text or title) except Exception: return conn.execute( "UPDATE items SET ai_summary=?, ai_tags=?, importance=?, status='done' WHERE id=?", (result.get("summary", ""), ",".join(result.get("tags", [])), float(result.get("importance", 0)), rid), ) conn.commit() # 控制在 3 个并发，避免触发 API 限流 with ThreadPoolExecutor(max_workers=3) as pool: pool.map(worker, rows) conn.close()

强调一下并发数。第一次跑的时候我直接开 10 个并发，结果一半请求超时。后来把 max_workers 降到 3，再加上简单的失败重试，才稳定下来。

5.4 Web 展示层：FastAPI 加 Jinja2

后端只需要一个首页路由，读数据库后把数据传给模板。

from pathlib import Path import sqlite3 from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates from .feeds import init_db BASE_DIR = Path(__file__).resolve().parent DB_PATH = Path(__file__).resolve().parent.parent / "data" / "aggregator.db" app = FastAPI() templates = Jinja2Templates(directory=str(BASE_DIR / "templates")) @app.on_event("startup") def startup(): init_db(DB_PATH) def query_items(search: str = "", limit: int = 50): conn = sqlite3.connect(DB_PATH) sql = "SELECT * FROM items WHERE 1=1" params = [] if search: sql += " AND (title LIKE ? OR ai_summary LIKE ? OR tags LIKE ?)" params.extend([f"%{search}%"] * 3) sql += " ORDER BY published_at DESC LIMIT ?" params.append(limit) rows = conn.execute(sql, params).fetchall() cols = [d[0] for d in conn.execute("SELECT * FROM items LIMIT 0").description] conn.close() return [dict(zip(cols, r)) for r in rows] @app.get("/", response_class=HTMLResponse) def home(request: Request, q: str = ""): items = query_items(search=q.strip()) return templates.TemplateResponse("index.html", {"request": request, "items": items, "q": q.strip()})

模板 index.html 很长，我只留下最核心的部分，样式可以按自己审美重写：

<!DOCTYPE html> <html lang="zh"> <head> <meta charset="utf-8"> <meta name="viewport" content="width=device-width, initial-scale=1"> <title>AI 资讯聚合</title> </head> <body> <form method="get"> <input type="text" name="q" value="{{ q }}" placeholder="搜索标题 / 摘要 / 标签"> <button type="submit">搜索</button> </form> {% for item in items %} <div class="item"> <h3><a href="{{ item['url'] }}" target="_blank">{{ item['title'] }}</a></h3> <div class="meta">{{ item['source_name'] }} · {{ item['published_at'] }} · {{ item['ai_tags'] }}</div> <p>{{ item['ai_summary'] or item['summary'] }}</p> </div> {% else %} <p>暂无内容，请稍后回来。</p> {% endfor %} </body> </html>

启动方式是uvicorn app.main:app --host 0.0.0.0 --port 8000，浏览器打开http://localhost:8000就能看到聚合列表。

5.5 定时任务与自动更新

页面跑起来之后，还需要定时触发采集和 AI 增强。我用 APScheduler 在 FastAPI 启动时挂一个后台任务：

from apscheduler.schedulers.background import BackgroundScheduler scheduler = BackgroundScheduler(timezone="Asia/Shanghai") SOURCES = [ ("机器之心", "https://www.jiqizhixin.com/rss"), ("量子位", "https://www.qbitai.com/feed"), ("InfoQ 中文", "https://www.infoq.cn/feed"), ("Hacker News", "https://news.ycombinator.com/rss"), ("Arxiv AI", "http://export.arxiv.org/rss/cs.AI"), ] def pipeline(): for name, url in SOURCES: try: entries = fetch_rss(url) save_entries(DB_PATH, entries) except Exception: continue enhance_pending_items(DB_PATH) def start_scheduler(): scheduler.add_job(pipeline, "interval", minutes=30, id="aggregate_pipeline") scheduler.start()

在 startup 事件里调用 start_scheduler() 就行。如果你不想让进程常驻，也可以写一个独立脚本，让 cron 每半小时跑一次，两种方式效果差不多。

5.6 Docker 部署

部署这块我直接上 Docker Compose，原因是迁移方便，服务器上不用装 Python 环境。Dockerfile：

FROM python:3.12-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY app ./app RUN mkdir -p /data CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

docker-compose.yml：

services: aggregator: build: . ports: - "8000:8000" volumes: - ./data:/data environment: - DATA_PATH=/data/aggregator.db - OLLAMA_URL=http://host.docker.internal:11434/api/generate

注意这里把数据库目录挂载到宿主机，容器重建不会丢数据。OLLAMA_URL 用了 host.docker.internal，是为了让容器能访问宿主机上的 Ollama。如果你用的是远程 API，就把 OLLAMA_URL 换成你自己的服务地址。

6. 常见问题排查与避坑实录

6.1 抓取下来乱码怎么办

RSS 大多是 XML，feedparser 会按文档声明的编码解析，一般不会出问题。但有些站点会返回压缩后的响应，如果你直接用 requests 抓，要设置 Accept-Encoding。还有一次我遇到某博客的 charset 声明和实际内容不一致，页面上看正常，抓回来全是乱码，后来强制用 UTF-8 解码才解决。遇到乱码先别改代码，先用 curl 看一眼响应头里的 content-type，很多问题一眼就能定位。

6.2 明明有新文章却不入库

最可能的原因是 INSERT OR IGNORE 加 url_hash 把内容拦住了。同一个 URL 第一次入库后，后面所有重复请求都会被忽略。这对去重是好事，但如果源更新了同一篇的标题或正文，你想抓最新版本就难了。解法是保留一个 updated_at 字段，相同 url_hash 但标题变化超过一定比例时做 UPDATE。

另一种情况是 URL 带了 utm 跟踪参数，导致同一篇文章每次 URL 都不同。入库前要把常见跟踪参数删掉再算 hash。

6.3 AI 摘要调用频繁报错

常见原因有三个：并发太高触发限流、超时时间太短、模型上下文不够。我遇到过 API 同时并发 8 个请求，一半超时。后来改成线程池 max_workers=3，加失败重试两次，平稳很多。

还有一个很容易被忽略的坑：模型返回的内容不一定每次都是合法 JSON，它可能在 JSON 外面包一层代码块标记。解析的时候要先用正则把代码块标记去掉，再做 json.loads，否则会直接抛异常。

6.4 搜索和过滤慢

当数据量到几千条之后，LIKE '%关键词%' 会开始变慢。SQLite 可以用 FTS5 全文索引解决。建一张虚拟表，把 title、content_text、ai_summary 放进去，用 MATCH 查询。如果不想动表结构，也可以在写入时把整理好的标签字符串存在一个字段里，过滤时用 tags LIKE 也能凑合，但数据量大了还是建议上 FTS5。

6.5 噪音还是太多

AI 资讯圈的信息噪点特别多，我的经验是三层过滤。第一层是源级别白名单，只加真正持续输出高质量内容的源。第二层是内容规则，排除纯转载、标题带“重磅”“震惊”但没有实质内容、视频转文字的低质量稿件。第三层才是模型打分，importance 低于 5 的内容在列表里默认置灰，只在标签页里出现。把规则下沉到采集层，比在展示层手动屏蔽要省心得多。

6.6 避坑速查表

这套系统我自己跑了半年多，最大的感受是：自己搭资讯平台的重点不是“我有多少信息源”，而是“我如何定义什么值得看”。AI 摘要和自动化只是帮你省时间，真正的筛选规则必须沉淀成代码和评分模型。你如果也想做，别一上来就想得多完整，先用 5 个 RSS 源加几十行代码把最小闭环跑通，再加 AI 摘要，再加推荐。跑顺了，后面加能力都是水到渠成的事。