AI 代理审计发现：内容问题逐一报警 | 前端进阶之旅

Full translation

完整中文译文

每个数字营销机构都有一个人，其工作职责包括打开电子表格、访问每个客户 URL、检查标题标签、meta 描述和 H1、注意断链接，然后将所有内容粘贴到报告中。然后下周重新进行。

这项工作是确定性的。智能体可以完成它。

在本教程中，你将使用 Python、Browser Use 和 Claude API 从零开始构建本地 SEO 审计智能体。该智能体在可见的浏览器窗口中访问真实页面，使用 Claude 提取 SEO 信号，异步检查断链接，通过人工在环暂停处理边界情况，并编写结构化报告——如果中断，一切都可恢复。

完成后，你将拥有一个可以针对任何 URL 列表运行的工作智能体。每个 URL 的运行成本不到 $0.01。

一个七模块的 Python 智能体，可以：

从 CSV 文件读取 URL 列表

在真实 Chromium 浏览器中访问每个 URL（不是无头爬虫）

通过 Claude API 提取标题、meta 描述、H1 和规范标签

使用 httpx 异步检查断链接

检测边界情况（404、登录墙、重定向）并暂停等待人工输入

逐步写入结果到 report.json——可安全中断和恢复

完成时生成纯英文 report-summary.txt

完整代码在 GitHub：dannwaneri/seo-agent

Python 3.11 或更高版本

一个 Anthropic API 密钥（在 console.anthropic.com 获取）

Windows、macOS 或 Linux

基础 Python 和命令行知识

为什么选择 Browser Use 而不是爬虫

模块 1：状态管理

模块 2：浏览器集成

模块 3：Claude 提取层

模块 4：断链接检查器

模块 5：人工在环

模块 6：报告编写器

模块 7：主循环

机构使用的日程安排

结果看起来像什么

为什么选择 Browser Use 而不是爬虫

标准的 SEO 审计方法是使用 requests 获取页面 HTML，然后用 BeautifulSoup 解析。这对静态页面有效。但它在 JavaScript 渲染内容上会失败，错过动态注入的 meta 标签，并在身份验证页面上完全失败。

Browser Use（GitHub 84,000+ 星，MIT 许可证）采取了不同的方法。它控制真实的 Chromium 浏览器，在 JavaScript 执行后读取 DOM，并通过 Playwright 的辅助功能树暴露页面。智能体看到的是人类看到的。

实际差异：基于 requests 的爬虫可能会错过由 React 组件注入的 meta 描述。Browser Use 不会。

另一个值得提及的差异：Browser Use 语义化地读取页面。如果按钮的 CSS 类从 btn-primary 改为 button-main，Playwright 脚本就会破裂。Browser Use 能识别它仍然是"提交"按钮并相应采取行动。提取逻辑存在于 Claude 提示中，而不是脆弱的 CSS 选择器中。

seo-agent/
├── index.py          # Main audit loop
├── browser.py        # Browser Use / Playwright page driver
├── extractor.py      # Claude API extraction layer
├── linkchecker.py    # Async broken link checker
├── hitl.py           # Human-in-the-loop pause logic
├── reporter.py       # Report writer
├── state.py          # State persistence (resume on interrupt)
├── input.csv         # Your URL list
├── requirements.txt
├── .env.example
└── .gitignore

创建项目文件夹并安装依赖：

mkdir seo-agent && cd seo-agent
pip install browser-use anthropic playwright httpx
playwright install chromium

用你的 URL 创建 input.csv：

url
https://example.com
https://example.com/about
https://example.com/contact

ANTHROPIC_API_KEY=your-key-here

运行之前，将你的 API 密钥设置为环境变量：

# macOS/Linux
export ANTHROPIC_API_KEY="sk-ant-..."

# Windows PowerShell
$env:ANTHROPIC_API_KEY = "sk-ant-..."

state.json
report.json
report-summary.txt
.env
__pycache__/
*.pyc

模块 1：状态管理

智能体需要跟踪已经审计过的 URL。如果运行被中断——断电、键盘中断、网络错误——它应该从停止处恢复，而不是重新开始。

state.py 用扁平 JSON 文件处理这个问题：

import json
import os

STATE_FILE = os.path.join(os.path.dirname(__file__), "state.json")

_DEFAULT_STATE = {"audited": [], "pending": [], "needs_human": []}

def load_state() -> dict:
    if not os.path.exists(STATE_FILE):
        save_state(_DEFAULT_STATE.copy())
    with open(STATE_FILE, encoding="utf-8") as f:
        return json.load(f)

def save_state(state: dict) -> None:
    with open(STATE_FILE, "w", encoding="utf-8") as f:
        json.dump(state, f, indent=2)

def is_audited(url: str) -> bool:
    return url in load_state()["audited"]

def mark_audited(url: str) -> None:
    state = load_state()
    if url not in state["audited"]:
        state["audited"].append(url)
    save_state(state)

def add_to_needs_human(url: str) -> None:
    state = load_state()
    if url not in state["needs_human"]:
        state["needs_human"].append(url)
    save_state(state)

这个设计是有意的：mark_audited() 在 URL 被处理并写入报告后立即调用。如果智能体在运行中崩溃，最多损失一个 URL 的工作。

模块 2：浏览器集成

browser.py 执行实际的页面导航。它直接使用 Playwright（Browser Use 作为依赖安装）来打开可见的 Chromium 窗口，导航到 URL，捕获 HTTP 状态和重定向信息，并从 DOM 提取原始 SEO 信号。

关键设计决策：

可见浏览器，不是无头模式。设置 headless=False 以便你可以观看智能体工作。这对演示和调试都很重要。

通过响应监听器捕获状态。Playwright 在 4xx/5xx 响应上会抛出异常，但 on("response", ...) 处理程序在异常前触发。我们在那里捕获状态。

访问之间 2 秒延迟。防止在机构客户端网站上触发速率限制或机器人检测。

这是核心导航函数：

import asyncio
import sys
import time
from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout

TIMEOUT = 20_000  # 20 seconds

def fetch_page(url: str) -> dict:
    result = {
        "final_url": url,
        "status_code": None,
        "title": None,
        "meta_description": None,
        "h1s": [],
        "canonical": None,
        "raw_links": [],
    }

first_status = {"code": None}

with sync_playwright() as p:
        browser = p.chromium.launch(headless=False)
        page = browser.new_page()

def on_response(response):
            if first_status["code"] is None:
                first_status["code"] = response.status

page.on("response", on_response)

try:
            page.goto(url, wait_until="domcontentloaded", timeout=TIMEOUT)
            result["status_code"] = first_status["code"] or 200
            result["final_url"] = page.url

# Extract SEO signals from DOM
            result["title"] = page.title() or None
            result["meta_description"] = page.evaluate(
                "() => { const m = document.querySelector('meta[name=\"description\"]'); "
                "return m ? m.getAttribute('content') : null; }"
            )
            result["h1s"] = page.evaluate(
                "() => Array.from(document.querySelectorAll('h1')).map(h => h.innerText.trim())"
            )
            result["canonical"] = page.evaluate(
                "() => { const c = document.querySelector('link[rel=\"canonical\"]'); "
                "return c ? c.getAttribute('href') : null; }"
            )
            result["raw_links"] = page.evaluate(
                "() => Array.from(document.querySelectorAll('a[href]'))"
                ".map(a => a.href).filter(Boolean).slice(0, 100)"
            )

except PlaywrightTimeout:
            result["status_code"] = first_status["code"] or 408
        except Exception as exc:
            print(f"[browser] Error: {exc}", file=sys.stderr)
            result["status_code"] = first_status["code"]
        finally:
            browser.close()

time.sleep(2)
    return result

关键细节

raw_links 的 100 个上限是刻意设置的。DEV.to 个人资料页面有数百个链接——你不需要所有链接来进行损坏链接检测。

wait_until="domcontentloaded" 设置比 networkidle 更快，足以进行元标签提取。JavaScript 渲染内容只需要 DOM 准备好，不需要等待所有网络请求完成。

模块 3：Claude 提取层

extractor.py 从 browser.py 获取原始页面快照，并调用 Claude 来生成结构化的 SEO 审计结果。

这是大多数教程出错的地方。它们要么在 Python 中编写复杂的解析逻辑（易损坏），要么请求 Claude 提供自由形式的响应并尝试解析文本（不可靠）。正确的做法是：给 Claude 一个严格的 JSON 架构，并告诉它不要返回其他任何内容。

使其在生产环境可靠的提示工程设计：

import json
import os
import sys
from datetime import datetime, timezone
import anthropic

MODEL = "claude-sonnet-4-20250514"
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

def _strip_fences(text: str) -> str:
    """Remove accidental markdown code fences from Claude's response."""
    text = text.strip()
    if text.startswith("```"):
        lines = text.splitlines()
        # Drop opening fence
        lines = lines[1:] if lines[0].startswith("```") else lines
        # Drop closing fence
        if lines and lines[-1].strip() == "```":
            lines = lines[:-1]
        text = "\n".join(lines).strip()
    return text

def extract(snapshot: dict) -> dict:
    if not os.environ.get("ANTHROPIC_API_KEY"):
        raise OSError("ANTHROPIC_API_KEY is not set.")

prompt = f"""You are an SEO auditor. Analyze this page snapshot and return ONLY a JSON object.
No prose. No explanation. No markdown fences. Raw JSON only.

Page data:
- URL: {snapshot.get('final_url')}
- Status code: {snapshot.get('status_code')}
- Title: {snapshot.get('title')}
- Meta description: {snapshot.get('meta_description')}
- H1 tags: {snapshot.get('h1s')}
- Canonical: {snapshot.get('canonical')}

Return this exact schema:
{{
  "url": "string",
  "final_url": "string",
  "status_code": number,
  "title": {{"value": "string or null", "length": number, "status": "PASS or FAIL"}},
  "description": {{"value": "string or null", "length": number, "status": "PASS or FAIL"}},
  "h1": {{"count": number, "value": "string or null", "status": "PASS or FAIL"}},
  "canonical": {{"value": "string or null", "status": "PASS or FAIL"}},
  "flags": ["array of strings describing specific issues"],
  "human_review": false,
  "audited_at": "ISO timestamp"
}}

PASS/FAIL rules:
- title: FAIL if null or length > 60 characters
- description: FAIL if null or length > 160 characters  
- h1: FAIL if count is 0 (missing) or count > 1 (multiple)
- canonical: FAIL if null
- flags: list every failing field with a clear description
- audited_at: use current UTC time in ISO 8601 format"""

response = client.messages.create(
        model=MODEL,
        max_tokens=1000,
        messages=[{"role": "user", "content": prompt}],
    )

raw = response.content[0].text
    clean = _strip_fences(raw)

try:
        return json.loads(clean)
    except json.JSONDecodeError as exc:
        print(f"[extractor] JSON parse error: {exc}", file=sys.stderr)
        return _error_result(snapshot, str(exc))

def _error_result(snapshot: dict, reason: str) -> dict:
    return {
        "url": snapshot.get("final_url", ""),
        "final_url": snapshot.get("final_url", ""),
        "status_code": snapshot.get("status_code"),
        "title": {"value": None, "length": 0, "status": "ERROR"},
        "description": {"value": None, "length": 0, "status": "ERROR"},
        "h1": {"count": 0, "value": None, "status": "ERROR"},
        "canonical": {"value": None, "status": "ERROR"},
        "flags": [f"Extraction error: {reason}"],
        "human_review": True,
        "audited_at": datetime.now(timezone.utc).isoformat(),
    }

有两点使其在生产环境中可靠：

首先，_strip_fences() 处理 Claude 将响应包装在 ```json 围栏中的情况，尽管被明确告知不要这样做。这在 Sonnet 上偶尔发生，如果你不处理它，会持续破坏 json.loads()。

其次，_error_result() 回退机制意味着该智能体永远不会在错误的 Claude 响应上崩溃——它记录错误、将 URL 标记为需要人工审查，然后继续处理下一个 URL。

成本：Claude Sonnet 4 的定价为每百万输入令牌 $3，每百万输出令牌 $15。典型的页面快照约为 500 个输入令牌；结构化 JSON 响应约为 300 个输出令牌。这样每个 URL 的成本约为 $0.006——对于 20 个 URL 的审计来说约为 $0.12。

模块 4：损坏链接检查器

linkchecker.py 从浏览器快照中获取 raw_links 列表，并使用异步 HEAD 请求检查同域链接的损坏状态。

仅检查同域。检查页面上的每个外部链接会花费数分钟，也不是代理商客户需要的。只筛选与被审计页面相同域的链接。

HEAD 请求，而不是 GET。更快、带宽更低、足以进行状态码检测。

上限为 50 个链接。DEV.to 文章列表等页面有数百个内部链接。检查所有这些会主导运行时。

通过 asyncio 的并发请求。所有链接并行检查，而不是顺序检查。

import asyncio
import logging
from urllib.parse import urlparse
import httpx

CAP = 50
TIMEOUT = 5.0
logger = logging.getLogger(__name__)

def _same_domain(link: str, final_url: str) -> bool:
    if not link:
        return False
    lower = link.strip().lower()
    if lower.startswith(("#", "mailto:", "javascript:", "tel:", "data:")):
        return False
    try:
        page_host = urlparse(final_url).netloc.lower()
        parsed = urlparse(link)
        return parsed.scheme in ("http", "https") and parsed.netloc.lower() == page_host
    except Exception:
        return False

async def _check_link(client: httpx.AsyncClient, url: str) -> tuple[str, bool]:
    try:
        resp = await client.head(url, follow_redirects=True, timeout=TIMEOUT)
        return url, resp.status_code != 200
    except Exception:
        return url, True  # Timeout or connection error = broken

async def _run_checks(links: list[str]) -> list[str]:
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*[_check_link(client, url) for url in links])
    return [url for url, broken in results if broken]

def check_links(raw_links: list[str], final_url: str) -> dict:
    same_domain = [l for l in raw_links if _same_domain(l, final_url)]

capped = len(same_domain) > CAP
    if capped:
        logger.warning("Page has %d same-domain links — capping at %d.", len(same_domain), CAP)
        same_domain = same_domain[:CAP]

broken = asyncio.run(_run_checks(same_domain))

return {
        "broken": broken,
        "count": len(broken),
        "status": "FAIL" if broken else "PASS",
        "capped": capped,
    }

模块 5：人类在环

这是大多数自动化教程跳过的部分。当智能体遇到登录墙时会发生什么？返回 403 的页面？重定向到"订阅以继续阅读"页面的 URL？

大多数脚本要么崩溃，要么默默跳过。这两种情况在代理商环境中都是不可接受的。

hitl.py 通过两个函数处理这个问题：一个检测是否需要暂停，另一个处理暂停本身。

from state import add_to_needs_human

LOGIN_KEYWORDS = {"login", "sign in", "sign-in", "access denied", "log in", "unauthorized"}
REDIRECT_CODES = {301, 302, 307, 308}

def should_pause(snapshot: dict) -> bool:
    code = snapshot.get("status_code")

# Navigation failed entirely
    if code is None:
        return True

# Non-200, non-redirect
    if code != 200 and code not in REDIRECT_CODES:
        return True

# Login wall detection
    title = (snapshot.get("title") or "").lower()
    h1s = [h.lower() for h in (snapshot.get("h1s") or [])]

if any(kw in title for kw in LOGIN_KEYWORDS):
        return True
    if any(kw in h1 for kw in LOGIN_KEYWORDS for h1 in h1s):
        return True

return False

def pause_reason(snapshot: dict) -> str:
    code = snapshot.get("status_code")
    if code is None:
        return "Navigation failed (None status)"
    if code != 200 and code not in REDIRECT_CODES:
        return f"Unexpected status code: {code}"
    return "Possible login wall detected"

def pause_and_prompt(url: str, reason: str) -> str:
    print(f"\n⚠️  需要人工审查")
    print(f"   URL:    {url}")
    print(f"   原因: {reason}")
    print(f"   选项: [s] 跳过  [r] 重试  [q] 退出\n")

    while True:
        choice = input("你的选择: ").strip().lower()
        if choice in ("s", "r", "q"):
            return {"s": "skip", "r": "retry", "q": "quit"}[choice]
        print("   请输入 s、r 或 q。")

should_pause() 函数会捕获四种情况：导航失败、意外的 HTTP 状态、标题中出现登录关键词，以及 H1 标签中出现登录关键词。登录关键词检查的作用是捕获那些返回 200 但实际上无法访问的"请登录以继续"页面。

在 --auto 模式中（用于定时运行），主循环会跳过 pause_and_prompt() 调用，通过将 URL 记录到 state 中的 needs_human[] 列表，自动处理这些情况并继续执行。

模块 6：报告生成器

reporter.py 增量式写入结果。这很重要：结果在审计每个 URL 后立即写入，而不是在最后批量写入。如果运行被中断，你不会丢失已完成的工作。

import json
import os
from datetime import datetime, timezone

REPORT_JSON = os.path.join(os.path.dirname(__file__), "report.json")
REPORT_TXT = os.path.join(os.path.dirname(__file__), "report-summary.txt")

def _load_report() -> list:
    if not os.path.exists(REPORT_JSON):
        return []
    with open(REPORT_JSON, encoding="utf-8") as f:
        return json.load(f)

def write_result(result: dict) -> None:
    """追加或更新 report.json 中的结果。"""
    entries = _load_report()
    url = result.get("url", "")

    # 如果 URL 已存在则更新条目（处理重试情况）
    for i, entry in enumerate(entries):
        if entry.get("url") == url:
            entries[i] = result
            break
    else:
        entries.append(result)

    with open(REPORT_JSON, "w", encoding="utf-8") as f:
        json.dump(entries, f, indent=2, ensure_ascii=False)

def _is_overall_pass(result: dict) -> bool:
    fields = ["title", "description", "h1", "canonical"]
    for field in fields:
        if result.get(field, {}).get("status") not in ("PASS",):
            return False
    if result.get("broken_links", {}).get("status") == "FAIL":
        return False
    return True

def write_summary() -> None:
    entries = _load_report()
    passed = sum(1 for e in entries if _is_overall_pass(e))

    lines = []
    for entry in entries:
        overall = "PASS" if _is_overall_pass(entry) else "FAIL"
        failed_fields = [
            f for f in ["title", "description", "h1", "canonical", "broken_links"]
            if entry.get(f, {}).get("status") == "FAIL"
        ]
        suffix = f" [{', '.join(failed_fields)}]" if failed_fields else ""
        lines.append(f"{entry.get('url', 'unknown'):<60} | {overall}{suffix}")

    lines.append("")
    lines.append(f"{passed}/{len(entries)} URLs 通过")

    with open(REPORT_TXT, "w", encoding="utf-8") as f:
        f.write("\n".join(lines))

write_result() 中的重复数据消除逻辑可以优雅地处理重试。如果某个 URL 在人工审查登录墙并认证后进行重试，新结果会替换旧结果，而不是创建重复条目。

模块 7：主循环

index.py 将所有模块组织在一起。它读取 URL 列表、加载状态、跳过已审计的 URL，并运行审计循环。

import csv
import os
import sys
import time
import argparse

from state import load_state, is_audited, mark_audited, add_to_needs_human
from browser import fetch_page
from extractor import extract
from linkchecker import check_links
from hitl import should_pause, pause_reason, pause_and_prompt
from reporter import write_result, write_summary

INPUT_CSV = os.path.join(os.path.dirname(__file__), "input.csv")

def read_urls(path: str) -> list[str]:
    with open(path, newline="", encoding="utf-8") as f:
        return [row["url"].strip() for row in csv.DictReader(f) if row.get("url", "").strip()]

def run(auto: bool = False):
    if not os.environ.get("ANTHROPIC_API_KEY"):
        print("错误：未设置 ANTHROPIC_API_KEY 环境变量。")
        sys.exit(1)

    urls = read_urls(INPUT_CSV)
    pending = [u for u in urls if not is_audited(u)]

    print(f"启动审计：{len(pending)} 个待审计，{len(urls) - len(pending)} 个已完成。\n")

    total = len(urls)

    try:
        for i, url in enumerate(pending, start=1):
            position = urls.index(url) + 1
            print(f"[{position}/{total}] {url}", end=" -> ", flush=True)

            # 浏览器导航
            snapshot = fetch_page(url)

            # 人工审查检查
            if should_pause(snapshot):
                reason = pause_reason(snapshot)

                if auto:
                    print(f"自动跳过 ({reason})")
                    add_to_needs_human(url)
                    mark_audited(url)
                    continue

                action = pause_and_prompt(url, reason)
                if action == "quit":
                    print("退出。")
                    break
                elif action == "skip":
                    add_to_needs_human(url)
                    mark_audited(url)
                    continue
                # "retry" 继续下面的重新获取逻辑
                snapshot = fetch_page(url)

            # Claude 提取
            result = extract(snapshot)

            # 断链检查
            links = check_links(snapshot.get("raw_links", []), snapshot.get("final_url", url))
            result["broken_links"] = links

            # 立即写入结果
            write_result(result)
            mark_audited(url)

            overall = "PASS" if all(
                result.get(f, {}).get("status") == "PASS"
                for f in ["title", "description", "h1", "canonical"]
            ) and links["status"] == "PASS" else "FAIL"