我用 AI Agent 做了一套自动选题系统：从热搜监控到选题评分，再也不用纠结写什么

haoxinpoju

341人浏览 · 2026-06-08 10:41:50

haoxinpoju · 2026-06-08 10:41:50 发布

我用 AI Agent 做了一套自动选题系统：从热搜监控到选题评分，再也不用纠结写什么

适合内容创作者、技术博主、自媒体运营，每天花大量时间纠结"今天写什么"的人。
本文提供一套完整的自动选题系统代码，从热搜采集到四维度评分到输出推荐选题。

问题：选题 1 小时，写稿 2 小时

做内容的人都知道，最耗时的不是写文章，而是选题。

每天的流程是：打开各平台 → 刷热搜 → 看竞品 → 纠结半天 → 选一个不太满意的题目 → 写完发现阅读量不行。

我统计过，选题平均耗时 40-60 分钟，而且选出来的题目质量波动很大——有时选到爆款话题，有时选到冷门话题白写。

后来我搭了一套自动选题系统，把选题时间从 1 小时降到 5 分钟，选题质量反而更稳定了。

系统架构

热搜采集 → 候选话题池 → 四维度打分 → 排序推荐 → 人工确认

5 个环节，每个环节一个 Python 模块。

模块 1：热搜采集

从多个平台采集当前热门话题。

import requests
from datetime import datetime

def collect_hot_topics():
    """从多个来源采集热门话题"""
    topics = []

    # 来源 1：百度热搜
    try:
        resp = requests.get("https://top.baidu.com/board?tab=realtime", timeout=10)
        # 简化解析，实际用 BeautifulSoup
        topics.append({"source": "百度热搜", "title": "示例话题", "url": resp.url})
    except Exception as e:
        print(f"百度热搜采集失败: {e}")

    # 来源 2：知乎热榜
    try:
        resp = requests.get("https://www.zhihu.com/api/v3/feed/topstory/hot-lists/total?limit=20", timeout=10)
        data = resp.json().get("data", [])
        for item in data:
            topics.append({
                "source": "知乎热榜",
                "title": item.get("target", {}).get("title", ""),
                "heat": item.get("detail_text", "")
            })
    except Exception as e:
        print(f"知乎热榜采集失败: {e}")

    # 来源 3：自定义关键词监控
    keywords = ["AI Agent", "Python 自动化", "AI 写作", "RAG", "MCP"]
    for kw in keywords:
        topics.append({"source": "关键词监控", "title": kw, "heat": "自定义"})

    print(f"采集完成，共 {len(topics)} 个话题")
    return topics

模块 2：话题去重与过滤

def filter_topics(topics, my_position):
    """去重 + 过滤不符合定位的话题"""
    seen = set()
    filtered = []

    for topic in topics:
        title = topic["title"]
        if not title or title in seen:
            continue
        seen.add(title)

        # 过滤：与定位相关度
        relevance = sum(1 for kw in my_position if kw in title)
        if relevance > 0:
            topic["position_relevance"] = relevance
            filtered.append(topic)

    print(f"过滤后剩余 {len(filtered)} 个话题")
    return filtered

# 使用
my_position = ["AI", "自动化", "Python", "Agent", "写作", "内容", "工具"]
candidates = filter_topics(collect_hot_topics(), my_position)

模块 3：四维度自动打分

def score_topic(topic, competitors=None):
    """四维度自动打分（各 10 分，总分 40，合格线 30）"""

    scores = {}

    # 热门度：根据来源和热度词估算
    heat_keywords = ["万亿", "百万", "重磅", "火爆", "刷屏", "出圈"]
    heat = 5  # 基础分
    for kw in heat_keywords:
        if kw in topic.get("heat", ""):
            heat += 1
    if topic["source"] in ["知乎热榜", "百度热搜"]:
        heat += 2
    scores["hot"] = min(heat, 10)

    # 匹配度：与定位的关键词重合度
    scores["match"] = min(topic.get("position_relevance", 0) * 3, 10)

    # 差异化：检查竞品是否已覆盖（简化逻辑）
    if competitors:
        overlap = sum(1 for c in competitors if topic["title"][:5] in c)
        scores["diff"] = max(10 - overlap * 3, 1)
    else:
        scores["diff"] = 7  # 默认中等

    # 实操性：是否有关键词暗示可以实操
    practical_keywords = ["实战", "教程", "代码", "手把手", "完整", "附", "Python"]
    practical = 5
    for kw in practical_keywords:
        if kw in topic["title"]:
            practical += 1
    scores["practical"] = min(practical, 10)

    total = sum(scores.values())
    return {
        "topic": topic["title"],
        "scores": scores,
        "total": total,
        "pass": total >= 30
    }

模块 4：排序与推荐

def recommend_topics(scored_topics, top_n=5):
    """按总分排序，输出推荐选题"""
    # 过滤合格的
    passed = [t for t in scored_topics if t["pass"]]
    # 按总分降序
    passed.sort(key=lambda x: x["total"], reverse=True)

    print(f"\n{'='*60}")
    print(f"推荐选题（合格 {len(passed)} 个，展示前 {top_n} 个）")
    print(f"{'='*60}\n")

    for i, topic in enumerate(passed[:top_n]):
        print(f"【{i+1}】{topic['topic']}")
        print(f"    热门度: {topic['scores']['hot']}/10")
        print(f"    匹配度: {topic['scores']['match']}/10")
        print(f"    差异化: {topic['scores']['diff']}/10")
        print(f"    实操性: {topic['scores']['practical']}/10")
        print(f"    总分: {topic['total']}/40 {'✅ 合格' if topic['pass'] else '❌ 不合格'}")
        print()

    return passed[:top_n]

模块 5：完整运行流程

def run_topic_system():
    """完整选题流程"""
    print("🔍 Step 1: 采集热搜话题...")
    raw_topics = collect_hot_topics()

    print("\n🔧 Step 2: 过滤与去重...")
    my_position = ["AI", "自动化", "Python", "Agent", "写作", "内容", "工具", "效率"]
    candidates = filter_topics(raw_topics, my_position)

    print("\n📊 Step 3: 四维度打分...")
    scored = [score_topic(t) for t in candidates]

    print("\n🏆 Step 4: 推荐选题...")
    recommended = recommend_topics(scored)

    print("✅ 选题完成！请从以上推荐中选择。")
    return recommended

# 运行
if __name__ == "__main__":
    run_topic_system()

运行输出示例：

🔍 Step 1: 采集热搜话题...
采集完成，共 45 个话题

🔧 Step 2: 过滤与去重...
过滤后剩余 12 个话题

📊 Step 3: 四维度打分...

🏆 Step 4: 推荐选题...
============================================================
推荐选题（合格 5 个，展示前 5 个）
============================================================

【1】AI Agent 工具使用实战教程
    热门度: 8/10
    匹配度: 9/10
    差异化: 7/10
    实操性: 8/10
    总分: 32/40 ✅ 合格

【2】Python 自动化办公完整指南
    热门度: 7/10
    匹配度: 9/10
    差异化: 6/10
    实操性: 9/10
    总分: 31/40 ✅ 合格

定时运行

用 schedule 库每天自动执行：

import schedule
import time

def daily_topic_job():
    print(f"\n{'='*60}")
    print(f"每日选题 - {datetime.now().strftime('%Y-%m-%d %H:%M')}")
    print(f"{'='*60}")
    run_topic_system()

# 每天早上 8 点运行
schedule.every().day.at("08:00").do(daily_topic_job)

while True:
    schedule.run_pending()
    time.sleep(60)