1. 项目概述：当数据采集遇上AI Agent

最近在做一个挺有意思的探索，把Playwright这个自动化测试框架，用到了小红书的数据采集上，并且尝试和AI Agent进行集成。这听起来可能有点跨界，但实际跑下来，发现这个组合拳的潜力远超预期。简单来说，就是让程序像真人一样去浏览、抓取小红书上的公开数据（比如笔记内容、互动数据、话题趋势），然后把这些结构化的数据喂给AI Agent，让它去分析、总结，甚至生成新的内容策略。这不仅仅是写个爬虫那么简单，它涉及到如何稳定地模拟真人操作绕过平台反爬、如何高效地解析动态加载的内容、以及如何设计一个能与数据流无缝对接的智能体工作流。

这个项目适合谁呢？如果你是内容运营，想自动化监控竞品动态和热点趋势；如果你是数据分析师或研究者，需要批量获取社交媒体数据进行舆情或用户行为分析；或者你是个开发者，对RPA（机器人流程自动化）和AI应用集成感兴趣，想探索如何让AI更“接地气”地使用实时网络数据，那么接下来的内容应该能给你不少直接的参考。整个过程，我会把踩过的坑、试出来的有效方案，以及和AI Agent集成的关键接口设计，都掰开揉碎了讲清楚。

2. 核心思路与架构设计

2.1 为什么是Playwright，而不是Requests或Selenium？

提到网页自动化，很多人第一反应是Selenium，或者直接用Requests+BeautifulSoup。但对于小红书这样前端渲染复杂、反爬机制严密的现代单页应用（SPA），传统方法往往力不从心。Requests直接抓包难度大，需要逆向复杂的API接口；Selenium虽然能驱动浏览器，但指纹容易被识别，且运行效率相对较低。

Playwright在这里的优势就非常明显了。首先，它由微软开发，原生支持Chromium、Firefox和WebKit，能生成更接近真实用户的浏览器环境，对反爬的对抗能力更强。其次，它的API设计非常现代和强大，自动等待、网络拦截、文件下载、多页面上下文隔离等功能都是开箱即用，极大地简化了异步操作和资源管理的复杂度。最后，它的执行速度比传统Selenium快，并且可以无头模式运行，节省资源。对于需要模拟完整用户交互（如滚动、点击、输入）才能获取数据的小红书来说，Playwright几乎是当前最合适的技术选型。

2.2 整体架构与数据流设计

整个项目的目标不是一次性脚本，而是一个可持续、可扩展的数据管道。因此，架构设计上需要清晰的分层。

数据采集层（Playwright驱动） ：这是最底层，负责模拟浏览器行为，访问小红书网页版或移动端H5页面，执行登录（如需）、搜索、列表翻页、进入详情页等操作。这一层的核心输出是完整的HTML页面或通过网络拦截捕获的API响应数据。

数据解析与清洗层 ：采集到的原始HTML或JSON数据是杂乱的。这一层需要从中精准提取目标信息，如笔记ID（note_id）、用户ID（user_id）、笔记正文、点赞/收藏/评论数、发布时间、话题标签等。这里会用到像 parsel （结合XPath和CSS选择器）或 BeautifulSoup 这样的解析库。关键在于编写健壮的选择器，以应对小红书前端结构的微小变动。

数据存储层 ：清洗后的结构化数据需要持久化。根据数据量和后续使用场景，可以选择轻量级的SQLite、文件（JSON Lines格式），或者更专业的MySQL、PostgreSQL数据库。这一步要设计好表结构，方便后续查询和分析。

AI Agent集成层 ：这是项目的“大脑”。我们将采集到的数据通过API或直接读库的方式，提供给AI Agent。Agent可以根据预设的指令进行分析，例如：“总结过去一周‘露营装备’话题下点赞最高的10篇笔记的核心观点”、“分析某个竞品账号的发文频率和互动率变化趋势”、“根据热门话题生成5个内容创作方向建议”。这里的关键是设计清晰、结构化的Prompt，让AI能准确理解数据并执行任务。

调度与监控层 ：为了让整个流程自动化运行，我们需要一个调度器（如Linux的crontab，Python的APScheduler，或更复杂的Airflow）来定时触发采集任务。同时，加入简单的日志和报警机制（比如采集失败时发送通知），确保管道的可靠性。

整个数据流可以概括为： Playwright采集 -> 解析清洗 -> 存储 -> AI Agent调用 -> 产出分析结果/决策建议 。这个架构保证了各模块职责单一，便于独立调试和扩展。

3. Playwright采集实战：从环境搭建到核心脚本

3.1 环境准备与关键配置

工欲善其事，必先利其器。首先确保你的Python环境（建议3.8以上）已经就绪。安装Playwright非常简单：

pip install playwright
# 安装Playwright所需的浏览器驱动（Chromium, Firefox, WebKit）
playwright install

这里我强烈推荐安装Chromium，因为它最常用，兼容性最好。安装驱动的过程可能会需要一点时间，因为它会下载完整的浏览器二进制文件。

接下来，一个容易被忽略但至关重要的步骤是： 配置浏览器启动参数以增强隐匿性 。小红书这类平台会检测自动化特征。我们可以通过传递一些启动参数来让浏览器环境看起来更“真人”一些。

from playwright.sync_api import sync_playwright

def create_stealth_browser_context(playwright):
    # 启动浏览器，添加一些反检测参数
    browser = playwright.chromium.launch(
        headless=False, # 调试时可设为False，实际运行建议True
        args=[
            '--disable-blink-features=AutomationControlled',
            '--disable-dev-shm-usage',
            '--no-sandbox',
            '--disable-web-security', # 谨慎使用，仅在某些复杂场景可能需要
            '--disable-features=IsolateOrigins,site-per-process', # 影响同源策略，按需
        ]
    )
    # 创建上下文，可以设置视窗大小、User-Agent等
    context = browser.new_context(
        viewport={'width': 1920, 'height': 1080},
        user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
        # 可以加载自定义的stealth插件或脚本（如果有）
        # ignore_https_errors=True # 忽略HTTPS错误，非必要不开
    )
    # 屏蔽某些不必要的资源加载，提升速度
    context.route("**/*.{png,jpg,jpeg,svg,gif,woff2}", lambda route: route.abort())
    return browser, context

注意 ： --disable-web-security 和 ignore_https_errors 这类参数会降低浏览器安全性，仅在调试解决特定跨域或证书问题时临时使用，生产环境应避免。 headless=False 在调试时非常有用，你可以亲眼看到浏览器在做什么。

3.2 核心采集策略：列表爬取与详情页深入

小红书的页面结构主要有两种： 列表页 （搜索页、用户主页、话题页）和 详情页 （单篇笔记）。我们的策略通常是先爬列表页获取一批笔记的ID或链接，再逐个进入详情页抓取完整信息。

1. 列表页爬取与翻页处理 列表页的最大挑战是 滚动加载 和 反爬限制 。不能爬太快，否则会触发验证码或IP封禁。

import time
import random
from playwright.sync_api import Page

def scrape_note_list(page: Page, keyword: str, max_scrolls: int = 10):
    """模拟搜索并滚动加载笔记列表"""
    search_url = f"https://www.xiaohongshu.com/search_result?keyword={keyword}"
    page.goto(search_url)
    page.wait_for_load_state('networkidle') # 等待网络基本空闲

    note_links = []
    for i in range(max_scrolls):
        # 1. 提取当前屏已加载的笔记链接
        current_links = page.locator('a[href*="/explore/"]').all() # 这是一个示例选择器，实际需根据页面结构调整
        for link in current_links:
            href = link.get_attribute('href')
            if href and href.startswith('/explore/') and href not in note_links:
                note_links.append(href)
        print(f"第{i+1}次滚动，已收集{len(note_links)}条链接")

        # 2. 模拟人类滚动：随机滚动距离和间隔
        scroll_height = random.randint(800, 1200)
        page.evaluate(f"window.scrollBy(0, {scroll_height})")
        time.sleep(random.uniform(1.5, 3.5)) # 随机等待，至关重要！

        # 3. 检查是否已到底部或出现异常（如验证码）
        if page.locator('text=没有更多了').count() > 0:
            print("已滚动到底部")
            break
        # 可以添加检测验证码的逻辑，如发现则暂停或触发处理流程

    return note_links

实操心得 ： time.sleep(random.uniform(1.5, 3.5)) 这种随机延迟是避免被识别为机器人的关键。更好的做法是监听页面内容变化（如新卡片出现）后再等待，但随机延时是最简单有效的基线策略。此外，列表页的选择器 'a[href*="/explore/"]' 需要你打开浏览器开发者工具，实际查看小红书列表页的HTML结构来确定，它可能会变。

2. 详情页数据解析 获取到笔记链接（如 /explore/笔记ID ）后，进入详情页抓取丰富数据。这里的数据可能一部分在初始HTML中，另一部分通过后续API加载。

def scrape_note_detail(page: Page, note_path: str):
    """抓取单篇笔记详情"""
    detail_url = f"https://www.xiaohongshu.com{note_path}"
    page.goto(detail_url)
    # 等待关键内容加载
    page.wait_for_selector('[data-v-笔记容器]', state='attached') # 替换为实际选择器

    # 方法一：从页面HTML中解析
    html_content = page.content()
    # 使用parsel或BeautifulSoup解析html_content，提取标题、正文、图片等
    # 例如：title = selector.xpath('//h1/text()').get()

    # 方法二（更推荐）：拦截网络请求，直接获取JSON数据
    # 小红书的数据通常通过XHR请求加载，我们可以监听这些请求
    note_data = {}
    def handle_response(response):
        if '/api/sns/web/v1/note' in response.url: # 关键API端点，需自行抓包确认
            try:
                json_data = response.json()
                # 从json_data中提取结构化信息
                note_data.update({
                    'note_id': json_data.get('data', {}).get('id'),
                    'title': json_data.get('data', {}).get('title'),
                    'desc': json_data.get('data', {}).get('desc'),
                    'likes': json_data.get('data', {}).get('likes'),
                    'collected': json_data.get('data', {}).get('collected'),
                    'user_info': json_data.get('data', {}).get('user', {})
                })
            except:
                pass

    page.on('response', handle_response)
    # 重新加载页面以触发请求监听（或确保在goto前已监听）
    page.reload()
    page.wait_for_timeout(2000) # 给拦截请求一点时间

    # 如果note_data为空，则回退到方法一
    if not note_data:
        # ... 执行HTML解析逻辑 ...
        pass

    return note_data

重要提示 ：拦截API请求是最高效、最稳定的方式，但需要你事先通过浏览器开发者工具的“网络（Network）”面板，分析小红书页面加载时调用了哪些接口，找到包含笔记核心数据的那个请求URL（通常是 /api/sns/web/v1/note 或类似路径）。这个接口可能会随着时间推移而改变。

3.3 应对反爬：策略与技巧

没有任何一种反爬策略是万能的，但组合拳能大大提高成功率。

  1. 请求速率限制 ：这是底线。不要在短时间内发起大量请求。除了在代码中加随机延迟，还可以设计分布式爬虫，用多个IP轮询。
  2. User-Agent轮换 ：准备一个UA池，每次创建浏览器上下文时随机选择一个。
  3. 使用浏览器上下文隔离 ：为每个任务（或每个IP）创建独立的 browser context ，避免Cookie和缓存相互干扰。
  4. 代理IP池 ：如果采集量非常大，必须使用高质量的住宅代理或数据中心代理IP池，并在Playwright启动浏览器时通过 --proxy-server 参数设置。 
    browser = playwright.chromium.launch(args=[f'--proxy-server=http://your-proxy-ip:port'])
  5. 处理验证码 ：遇到验证码时，策略可以是：a) 自动识别（准确率低）；b) 打码平台（有成本）；c) 最实用的—— 暂停任务，发出警报，等待人工干预 。可以在代码中检测页面是否出现验证码元素，一旦发现就记录日志并停止后续操作。
  6. 模拟真人行为 ：除了随机滚动和等待，还可以随机模拟鼠标移动、点击非目标区域等。Playwright提供了 page.mouse.move(x, y) 等方法。

4. 数据解析、存储与清洗标准化

4.1 高效解析：XPath与CSS选择器实战

从HTML或JSON中提取数据，选择器的准确性决定了数据质量。以解析笔记正文为例：

from parsel import Selector

def parse_note_from_html(html: str):
    selector = Selector(text=html)
    data = {}

    # 使用XPath，通常更灵活
    data['title'] = selector.xpath('//h1[contains(@class, "title")]/text()').get(default='').strip()
    # 正文可能包含多行，用getall()
    data['content'] = '\n'.join(selector.xpath('//div[@class="note-content"]//text()').getall()).strip()

    # 使用CSS选择器，写起来更简洁
    data['like_count'] = selector.css('span.like-count::text').get()
    data['user_name'] = selector.css('a.user-name::text').get()

    # 处理可能存在的转义字符和多余空白
    import re
    if data['content']:
        data['content'] = re.sub(r'\s+', ' ', data['content'])

    return data

注意事项 ：小红书的类名（如 note-content ）是示例， 实际必须通过查看网页源代码确定 。这些类名很可能经过混淆或定期变更，所以你的选择器需要一定的容错性，或者准备多套选择器方案。更好的做法是， 优先使用 data-v- 开头的数据属性 ，因为Vue.js框架生成的数据属性有时比类名更稳定。

4.2 数据存储方案选型与实现

采集到的数据需要持久化。这里给出两种常用方案。

方案一：SQLite（轻量级，适合入门和中小规模） 

import sqlite3
import json
from datetime import datetime

def init_db(db_path='xiaohongshu_data.db'):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS notes (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            note_id TEXT UNIQUE,
            title TEXT,
            content TEXT,
            likes INTEGER,
            collected INTEGER,
            user_id TEXT,
            user_name TEXT,
            publish_time TEXT,
            keywords TEXT,
            raw_data TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    conn.commit()
    return conn

def save_note_to_db(conn, note_data):
    cursor = conn.cursor()
    try:
        cursor.execute('''
            INSERT OR REPLACE INTO notes 
            (note_id, title, content, likes, collected, user_id, user_name, publish_time, keywords, raw_data)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        ''', (
            note_data.get('note_id'),
            note_data.get('title'),
            note_data.get('content'),
            note_data.get('likes'),
            note_data.get('collected'),
            note_data.get('user_id'),
            note_data.get('user_name'),
            note_data.get('publish_time'),
            ','.join(note_data.get('keywords', [])),
            json.dumps(note_data, ensure_ascii=False) # 保存原始JSON以备不时之需
        ))
        conn.commit()
        print(f"笔记 {note_data.get('note_id')} 保存成功")
    except sqlite3.Error as e:
        print(f"数据库错误: {e}")
        conn.rollback()

方案二：JSON Lines文件（简单，无需数据库） 对于快速原型或数据量不大时，每行一个JSON对象的文本文件非常方便。

import json

def save_note_to_jsonl(note_data, filename='notes.jsonl'):
    with open(filename, 'a', encoding='utf-8') as f:
        json_record = json.dumps(note_data, ensure_ascii=False)
        f.write(json_record + '\n')

选型建议 ：如果数据关系简单，主要是新增和查询，且量在百万条以内，SQLite完全够用，管理也方便。如果数据量极大或需要复杂关联分析，可考虑PostgreSQL。JSON Lines文件则适合作为中间缓存或日志式存储。

4.3 数据清洗与质量校验

原始数据往往存在脏乱、缺失、格式不一致等问题，清洗环节必不可少。

  1. 去重 ：基于 note_id 进行去重，避免同一篇笔记被多次存储。
  2. 处理缺失值 ：对于关键字段（如 note_id , content ）缺失的记录，可以考虑丢弃或标记为异常。对于数值字段（如 likes ），可以填充为0。
  3. 文本清洗 ： 
    import re

def clean_text(text):
    if not text:
        return ''
    # 移除多余空白字符（包括换行、制表符等）
    text = re.sub(r'\s+', ' ', text)
    # 移除不可见字符
    text = ''.join(char for char in text if char.isprintable())
    # 针对小红书特定内容：移除话题标签的‘#’号（根据需求决定）
    # text = re.sub(r'#(\w+)', r'\1', text)
    return text.strip()
  4. 时间格式化 ：小红书的时间格式可能是时间戳（如 1640995200000 ）或字符串（如 “2022-01-01 10:00:00” ）。需要统一转换为程序易于处理的datetime对象。 
    from datetime import datetime

def parse_publish_time(time_str):
    # 尝试多种可能的格式
    formats = ['%Y-%m-%d %H:%M:%S', '%Y/%m/%d %H:%M', '%Y年%m月%d日']
    for fmt in formats:
        try:
            return datetime.strptime(time_str, fmt)
        except ValueError:
            continue
    # 如果是时间戳（毫秒）
    try:
        timestamp = int(time_str) / 1000
        return datetime.fromtimestamp(timestamp)
    except:
        return None # 解析失败，返回None或默认时间
  5. 数据校验 ：在入库前，可以设置一些简单的规则进行校验，比如 likes 字段应该是非负整数， content 长度应在合理范围内等。不符合规则的记录可以放入一个 error_log 表供后续检查。

5. AI Agent集成：从数据到智能决策

采集和清洗数据只是第一步，让AI Agent利用这些数据产生价值才是目标。这里的关键是设计一个清晰的接口，让Agent能方便地“消费”数据。

5.1 为AI Agent提供数据服务

我们不需要让AI Agent直接操作数据库或文件。更好的做法是封装一个简单的数据查询服务（API）。这里用Flask快速搭建一个示例：

from flask import Flask, request, jsonify
import sqlite3
import json

app = Flask(__name__)

def get_db_connection():
    conn = sqlite3.connect('xiaohongshu_data.db')
    conn.row_factory = sqlite3.Row # 返回字典形式的行
    return conn

@app.route('/api/notes/recent', methods=['GET'])
def get_recent_notes():
    """获取最近N篇笔记"""
    limit = request.args.get('limit', default=10, type=int)
    keyword = request.args.get('keyword', default=None, type=str)

    conn = get_db_connection()
    query = 'SELECT * FROM notes WHERE 1=1'
    params = []
    if keyword:
        query += ' AND (title LIKE ? OR content LIKE ? OR keywords LIKE ?)'
        like_pattern = f'%{keyword}%'
        params.extend([like_pattern, like_pattern, like_pattern])
    query += ' ORDER BY created_at DESC LIMIT ?'
    params.append(limit)

    rows = conn.execute(query, params).fetchall()
    conn.close()
    # 将行对象转换为字典列表
    notes = [dict(row) for row in rows]
    # 处理raw_data字段，如果是JSON字符串就解析
    for note in notes:
        if note.get('raw_data'):
            try:
                note['raw_data'] = json.loads(note['raw_data'])
            except:
                pass
    return jsonify({'data': notes, 'count': len(notes)})

@app.route('/api/notes/stats', methods=['GET'])
def get_note_stats():
    """获取基础统计信息，如总笔记数、平均点赞等"""
    conn = get_db_connection()
    stats = conn.execute('''
        SELECT 
            COUNT(*) as total_notes,
            AVG(likes) as avg_likes,
            MAX(likes) as max_likes,
            COUNT(DISTINCT user_id) as unique_users
        FROM notes
    ''').fetchone()
    conn.close()
    return jsonify(dict(stats))

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

这样，AI Agent（无论是本地运行的脚本，还是云端服务）只需要向 http://localhost:5000/api/notes/recent?limit=5&keyword=露营 发送一个GET请求，就能拿到结构化的数据。

5.2 设计Agent的Prompt与工作流

有了数据接口，接下来就是设计AI Agent的“大脑”。这里以使用OpenAI API（或兼容API的本地大模型）为例，展示如何构建一个数据分析Agent。

核心思想 ：将数据作为上下文（Context）提供给大模型，并给出明确的指令（Prompt），让它执行分析任务。

import openai
import requests

# 1. 从我们的数据服务获取原始数据
def fetch_data_from_service(keyword='露营', limit=20):
    response = requests.get(f'http://localhost:5000/api/notes/recent?keyword={keyword}&limit={limit}')
    if response.status_code == 200:
        return response.json()['data']
    else:
        return []

# 2. 构建Prompt
def build_analysis_prompt(notes_data):
    # 将数据格式化成易于理解的文本
    data_context = ""
    for i, note in enumerate(notes_data[:10]): # 取前10条作为样例，避免token超限
        data_context += f"{i+1}. 标题：{note.get('title', 'N/A')}\n"
        data_context += f"   内容摘要：{note.get('content', 'N/A')[:100]}...\n"
        data_context += f"   点赞：{note.get('likes', 0)}， 收藏：{note.get('collected', 0)}\n"
        data_context += f"   用户：{note.get('user_name', 'N/A')}\n"
        data_context += "-"*40 + "\n"

    prompt = f"""
你是一个资深的小红书内容运营分析师。请根据以下近期采集的关于“露营”的笔记数据，完成分析任务。

【原始数据】
{data_context}

【分析任务】
1. **内容主题归纳**：请总结这些笔记主要围绕哪几个子话题展开（例如：装备推荐、营地选择、美食制作等）？
2. **高互动内容特征**：找出点赞或收藏数最高的3条笔记，简要分析它们为什么可能更受欢迎（从标题、内容切入点、呈现形式等方面）。
3. **内容建议**：基于以上分析，为计划创作“露营”相关内容的创作者提供3条具体的选题或内容形式建议。

请以清晰、有条理的格式输出你的分析结果。
"""
    return prompt

# 3. 调用大模型API进行分析
def analyze_with_ai(api_key, notes_data):
    openai.api_key = api_key
    prompt = build_analysis_prompt(notes_data)

    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo", # 或 "gpt-4", "claude-3-haiku"等
            messages=[
                {"role": "system", "content": "你是一个专业、严谨的数据分析师，擅长从社交媒体数据中提炼洞察。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.7, # 控制创造性，分析类任务可以调低
            max_tokens=1500
        )
        analysis_result = response.choices[0].message.content
        return analysis_result
    except Exception as e:
        return f"AI分析失败: {e}"

# 主流程
if __name__ == '__main__':
    # 获取数据
    notes = fetch_data_from_service(keyword='露营', limit=15)
    if notes:
        # 进行AI分析
        result = analyze_with_ai('your-api-key-here', notes)
        print("=== AI分析报告 ===")
        print(result)
        # 可以将result保存到文件或数据库
        with open('analysis_report.md', 'w', encoding='utf-8') as f:
            f.write(result)
    else:
        print("未获取到数据，请检查数据服务或采集任务。")

实操心得 ：Prompt工程是关键。你需要清晰地告诉AI：

  1. 角色 ：让它扮演什么专家（如“内容运营分析师”）。
  2. 背景/上下文 ：提供结构化的数据。
  3. 具体任务 ：分点列出要它做什么，指令越明确，输出越符合预期。
  4. 输出格式 ：指定它如何呈现结果（如“以清晰、有条理的格式”）。

此外，要注意上下文长度限制。如果数据很多，需要进行 摘要或筛选 后再喂给AI，或者使用支持更长上下文的模型。

5.3 构建自动化工作流：定时采集+智能分析

最后一步，将整个流程串联起来，实现全自动化。我们可以使用Python的 APScheduler 库来定时执行任务。

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
import logging
from your_crawler_module import main_crawler_function # 导入你写好的采集主函数
from your_ai_analysis_module import fetch_data_and_analyze # 导入AI分析函数

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def job_data_collection():
    """定时采集任务"""
    logger.info("开始执行小红书数据采集任务...")
    try:
        main_crawler_function(keywords=['露营', '徒步', '户外']) # 传入要爬取的关键词
        logger.info("数据采集任务完成。")
    except Exception as e:
        logger.error(f"数据采集任务失败: {e}")

def job_ai_analysis():
    """定时分析任务"""
    logger.info("开始执行AI分析任务...")
    try:
        report = fetch_data_and_analyze()
        # 可以将报告发送到钉钉、飞书、邮箱等
        # send_report_to_feishu(report)
        logger.info("AI分析任务完成，报告已生成。")
    except Exception as e:
        logger.error(f"AI分析任务失败: {e}")

if __name__ == '__main__':
    scheduler = BlockingScheduler()

    # 每天凌晨2点执行采集（网站负载较低）
    scheduler.add_job(job_data_collection, CronTrigger(hour=2, minute=0))
    # 每天上午10点执行分析，产出当日洞察
    scheduler.add_job(job_ai_analysis, CronTrigger(hour=10, minute=0))

    logger.info("调度器已启动，等待执行定时任务...")
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        logger.info("调度器已停止。")

这样，一个完整的、自动化的“数据采集 -> AI分析”管道就搭建完成了。它会在你设定的时间自动运行，无需人工干预。

6. 常见问题、排查技巧与优化方向

6.1 采集过程中的典型问题与解决

问题1：页面加载不全，数据抓不到。

  • 排查 ：检查选择器是否正确，使用 page.wait_for_selector() 确保元素加载完成。更常见的是数据通过JS异步加载，需要滚动或等待更长时间。
  • 解决 ：增加 page.wait_for_timeout() 或使用 page.wait_for_function() 监听特定JS变量或DOM变化。优先采用 拦截网络API 的方式。

问题2：出现验证码或访问被限制。

  • 现象 ：页面跳转到验证码界面，或返回空白、错误信息。
  • 解决 ：
    1. 立即降低频率 ：大幅增加请求间隔时间（如随机等待5-10秒）。
    2. 切换代理IP ：如果使用代理，换一个IP。
    3. 更换User-Agent和浏览器上下文 。
    4. 模拟更真人化的操作 ：在操作流程中加入随机移动鼠标、随机点击页面空白处等。
    5. 设置熔断机制 ：连续失败N次后，暂停任务一段时间（如1小时）。

问题3：选择器失效，解析不到数据。

  • 原因 ：小红书前端页面结构更新。
  • 解决 ：
    1. 定期维护 ：将选择器集中配置在配置文件（如 selectors.yaml ）中，便于统一更新。
    2. 使用更稳定的属性 ：优先选择 data-* 属性或 id ，而非易变的CSS类名。
    3. 多层容错 ：编写解析函数时，准备多套XPath或CSS选择器，依次尝试，直到有一个成功。
    4. 数据备份 ：始终保存原始HTML或API响应（ raw_data 字段），以便在解析逻辑更新后，能重新处理历史数据。

问题4：运行一段时间后内存占用过高或崩溃。

  • 原因 ：Playwright的浏览器实例、页面（Page）或上下文（Context）未正确关闭。
  • 解决 ：确保使用 try...finally 或 async with 上下文管理器，保证资源释放。 
    def safe_crawl():
    with sync_playwright() as p:
        browser = p.chromium.launch()
        context = browser.new_context()
        try:
            page = context.new_page()
            # ... 你的爬取逻辑 ...
        finally:
            # 按顺序关闭，先page，再context，最后browser
            page.close()
            context.close()
            browser.close()

6.2 AI Agent集成中的常见坑

问题1：Prompt效果不佳，AI回答笼统或偏离方向。

  • 解决 ：遵循“角色-背景-任务-格式”的Prompt结构。多迭代几次，根据输出调整指令。给AI提供 更具体、更结构化的数据 ，比如把“点赞数”直接给它，而不是让它从文本里猜。

问题2：数据量太大，超出模型上下文窗口。

  • 解决 ：
    1. 数据筛选与聚合 ：在提供给AI前，先进行预处理。例如，只取点赞最高的前20条笔记，或者按天聚合统计摘要（如“今日共新增100条笔记，平均点赞200”）。
    2. 分而治之 ：将大任务拆分成多个子任务，让AI分别分析，最后再汇总。例如，先让AI分析“标题特点”，再分析“内容结构”。
    3. 使用长上下文模型 ：如果成本允许，选用支持128K甚至更长上下文的模型。

问题3：分析结果不稳定，每次都不一样。

  • 解决 ：调整API调用时的 temperature 参数。对于需要稳定、可重复的分析任务，将 temperature 设低（如0.1或0.2）。对于需要创意的任务（如生成标题），可以调高（如0.8）。

6.3 项目优化与扩展方向

  1. 分布式爬虫 ：使用 Scrapy 框架结合 Playwright 的中间件，或者用 Celery 等任务队列分发采集任务到多台机器，可以极大提升采集效率和规模。
  2. 更智能的Agent ：当前的Agent主要是被动分析。可以将其升级为 自主智能体 ，让它不仅能分析数据，还能根据分析结果自动决策并执行动作。例如，发现某个话题热度飙升时，自动草拟一篇相关笔记的提纲；或者监控到竞品发布了爆文，自动发送警报给运营人员。
  3. 数据可视化 ：将AI分析出的关键指标（如热度趋势、话题分布、用户情感）用 ECharts 或 Plotly 做成仪表盘，直观呈现。
  4. 结合多平台 ：将这套方法扩展到抖音、微博、B站等其他平台，进行跨平台的数据对比和舆情监控。
  5. 模型微调 ：如果拥有大量标注好的小红书数据（例如，人工标记了“爆文”与“普通文”），可以尝试微调一个专属的小型开源大模型（如Qwen、ChatGLM），让它更精通小红书内容分析，降低API调用成本。

这个项目从技术实现到业务应用有一条很长的路可以走，但起点正是我们上面搭建的这个自动化闭环。最难的部分往往不是代码本身，而是应对目标网站的变化、设计稳定的数据管道以及让AI真正理解业务需求。保持代码的模块化和可配置性，定期维护和更新，这个工具就能持续为你产生价值。

Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

【花雕动手做】行空板 K10 系列实验之人工智能调用语音合成（TTS）模块

avatar AI Agent技术社区
cover

苏州企业AI Agent智能体从概念到落地：2026年开发者必须关注的技术范式与工程实践

avatar AI Agent技术社区
cover

GitHub 狂揽 4万+ Star！这个项目直接让你省下 60–95% 的 Token

avatar AI Agent技术社区