AIエージェント向けWebスクレイピングのアクセス層と抽出層の設計方法

TL;DR

• Webアクセス層は、有効で使用可能なページスナップショットを取得します。データ抽出層は、そのスナップショットを候補レコードに変換します。
• アクセス側の境界で、レンダリングの継続、セッション状態、ネットワークの障害、およびサポートされているCAPTCHA処理を保持します。
• AIエージェントが使用する前に、抽出されたJSONの必須フィールド、ソースのサポート、完全性、および新鮮さを検証します。
• 一時的な読み取りエラーは、限られた予算内で再試行します。レート制限時は処理を延期し、認可の問題やチャレンジがある場合は停止して確認します。
• 解析に失敗した場合、ページを自動的に再要求する代わりに、保持されたスナップショットに対して再抽出を行います。
• Pythonの例は、合成HTMLでローカルに実行され、ウェブサイトにアクセスしたりモデルを呼び出したりせずに両方のレイヤーをテストします。

AIエージェントは、間違ったページからの有効な見た目のレコードを受け取ることがあります。ログイン画面には見出しがあるかもしれませんが、不完全なドキュメントは正常に解析される可能性があり、モデルは要求された事実が欠如している場合でもJSONを返すことがあります。AIエージェントのウェブスクレイピングアーキテクチャには、ソースに到達するための決定と、そのコンテンツを解釈するための決定が別々に必要です。

このチュートリアルでは、不変のスナップショットとバージョン付きレコード契約の周囲に境界を設計します。ワークフローは、認可された公開通知の収集、ドキュメントの更新、および他の許可されたウェブ情報に適用されます。CapSolverは、アクセスレイヤーでのサポートされているCAPTCHA処理機能として表示されます。実行可能な例では、取得結果を分類し、小さなレコードセットを抽出し、検証に失敗した場合に証拠を保持する方法が示されています。

アクセスおよび抽出レイヤーの契約を定義する

アクセスレイヤーは、使用可能なスナップショットまたは明示的な失敗を返す必要があります。抽出レイヤーは、そのスナップショットに関連する候補レコードを返す必要があります。ブラウザランタイム、パーサー、またはモデルの変更があっても、両方の契約を安定したままに保つ必要があります。

パイプラインは次の通りです：承認された収集要求 → アクセスアダプター → 保持されたスナップショット → 抽出アダプター → 検証 → 受け入れられたレコードストア → エージェント。この設計では、ブラウザを再開することなく、保存された証拠から抽出を再実行できます。

Webアクセスレイヤーはページ取得を担当します

アクセスアダプターに承認されたソース、タスクID、時間予算、および許可されたセッションコンテキストを提供します。その仕事には、必要な表現を選択し、関連するコンテンツを待機し、セッション所有権を保持し、障害を分類することが含まれます。ネットワーク構成とJavaScriptレンダリングは、HTTPまたはブラウザインフラストラクチャに属します。

出力は、要求されたおよび最終的なソースの場所、観測時間、表現タイプ、コンテンツダイジェスト、および準備状態の証拠を記録する必要があります。実際にロードされたページを隠す単一の「成功」フラグを避けてください。最終的なURLとHTTPステータスは役立ちますが、期待されるドキュメントのIDと必要なコンテンツ領域もチェックする必要があります。

データ抽出レイヤーは解釈を担当します

スナップショット参照、スキーマバージョン、およびフィールド定義を抽出器に提供します。暗黙的にナビゲートしたり、資格情報を変更したり、別のネットワークルートを選択したりしないでください。アクセスレイヤーに再試行を求める代わりに、欠落または曖昧なフィールドを明示的に返します。

AIウェブスクレイピング用語集では、AIを用いたウェブ情報の収集および解釈の広範な使用法が説明されています。この境界は、決定論的な抽出もサポートしています。安定した属性や文書化された構造化データは十分である場合があります。解釈が必要な場合にのみモデルを使用し、同じ下流の検証契約を保持してください。

フィールドの抽出前に正しい証拠を取得する

使用可能なスナップショットには、要求されたフィールドに必要な証拠が含まれており、抽出器が理解できる表現である必要があります。その要件に応じて、HTML、レンダリングされたDOM、またはスクリーンショットを選択します。

HTMLとレンダリングされたDOM

レスポンスにすでに必要なコンテンツが含まれている場合、HTMLは適切です。必要フィールドがクライアントサイドの実行後にのみ表示される場合、タスク固有の準備チェック後にレンダリングされたDOMをキャプチャするブラウザアダプターを使用してください。準備を観測可能な条件（例：期待されるレコードコンテナと完了マーカー）として定義し、万能の固定スリープではなくしてください。

どの表現がキャプチャされたかを記録してください。レンダリングされたマーカップでテストされたパーサーは、明示的な契約変更なしに初期HTMLシェルを受け取ってはなりません。必要な領域が欠如している場合、抽出を試行する前にスナップショットを不完全と分類してください。

スクリーンショットと視覚的解釈

スクリーンショットは、特定のビューポートと時間におけるピクセルを提供します。スクリーンショットベースの抽出の場合、画像の寸法、キャプチャコンテキスト、および各抽出フィールドの領域参照を保持してください。値がキャプチャされたビューの外側にある場合、欠如として返してください。モデルが類似レイアウトに慣れていることは、その値の証拠ではありません。

視覚的な推定を正確な数値に変換する際には、不確実性を記録してください。DOMと視覚的証拠の両方が利用可能な場合、不一致をレビュー事例として使用してください。以下の例ではHTMLアダプターのみが実装されています。ビジョンアダプターには独自の証拠チェックと評価セットが必要です。

再試行を決定する前に障害を分類する

再試行の決定は、失敗したレイヤー、別の試行の予想される利益、および残りの予算を指定する必要があります。収集と解釈の再試行を分離して、抽出エラーが制御不能なトラフィックを生成しないようにしてください。

HTTPのセマンティクスは、1行目の実装において重要です。 RFC 9110の再試行と同一性ルールは、安全に繰り返せる操作と、効果が不明な操作を区別します。フォーム送信やその他の状態変更アクションには、読み込み再試行ループを再利用しないでください。

Retry-Afterヘッダーは遅延またはHTTP日付を表すことができます。値をスケジューリングに保持してください。ローカルワーカーは、サーバー要求の待機を短いバックオフに置き換えてはなりません。同じ許可された収集範囲を共有するワーカーは、クールダウン状態を共有する必要があります。

狭いアクセスアダプターを通じてCAPTCHA処理を追加する

ワークフローが許可、タスク互換性、および必要なセッションコンテキストを確立した後、CapSolverは文書化されたCAPTCHAタスクのみを処理する必要があります。403応答、空のページ、およびCAPTCHAウィジェットは異なる観察であり、これらをすべて解決要求にマッピングしないでください。

CapSolverのcreateTask契約は、適切なタスクオブジェクトを必要とします。非同期作業の場合、getTaskResultはタスクの状態と出力を返します。アクセスアダプターは、文書化された統合を適用し、その後に目的地をチェックすることに責任があります。

完了したタスクは、検証されたページスナップショットではありません。コンテンツを抽出に渡す前に、ドキュメントのIDと準備条件を再確認してください。チャレンジ予算を別途設定し、チャレンジがサポートされていない、認証が不明確な、または期待されるページが利用できない場合に停止してください。AIエージェントブラウザインフラストラクチャスタックは、ランタイム所有権とセッション証拠に関する関連するガイダンスを提供します。

CapSolverボーナスコードを引き換える

自動化予算を即座に増やす！
CapSolverアカウントにチャージする際にボーナスコード CAP26 を使用すると、すべてのチャージで 5%のボーナス を受け取れます — 制限はありません。
今すぐCapSolverダッシュボードで引き換えてください

異なる障害境界を持つPythonワークフローを実行する

以下のPythonワークフローは、合成アクセス応答を分類し、受け入れられたHTMLを保持し、公告フィールドを抽出し、構造化JSONを返します。これをpipeline_example.pyとして保存し、Python 3.9以降で実行してください。標準ライブラリのみを使用します。

fetch関数はインジェクトされた読み取り専用アダプターです。ここではメモリ内の固定値を提供し、デモではスリープを無効にしています。ウェブサイト、ブラウザサービス、モデル、またはCAPTCHA APIにアクセスしていません。challengeおよびreadyフィールドは、アクセスアダプターによって提供される観察を表しています。この例では、万能なチャレンジ検出器は実装されていません。

パーサーは、意図的に小さなマーカップ契約のためにPythonのHTMLParserコールバックを使用します。各articleには1つのh2、レコードID、および公開日が含まれます。これは汎用的なDOMパーサーまたは任意の不正なHTMLの検証器ではありません。

from dataclasses import dataclass
from datetime import date
from hashlib import sha256
from html.parser import HTMLParser
import json
import time


class PipelineError(Exception):
    def __init__(self, stage, reason, retry_after=""):
        self.stage, self.reason = stage, reason
        self.retry_after = retry_after
        super().__init__(f"{stage}:{reason}")


@dataclass(frozen=True)
class Reply:
    status: int
    body: str = ""
    content_type: str = "text/html"
    challenge: bool = False
    ready: bool = True
    retry_after: str = ""


def access(fetch, wait=time.sleep):
    # fetch is a read-only adapter; all values below are application policy.
    for attempt in range(2):
        try:
            reply = fetch()
        except TimeoutError:
            if attempt == 0:
                wait(0.5)
                continue
            raise PipelineError("access", "timeout_exhausted")
        if reply.status == 429:
            # Pass Retry-After to a shared scheduler; do not retry here.
            raise PipelineError("access", "defer_rate_limit", reply.retry_after)
        if reply.status in (401, 403):
            raise PipelineError("access", "authorization_review")
        if reply.challenge:
            raise PipelineError("access", "challenge_review")
        if reply.status == 503 and reply.retry_after:
            raise PipelineError("access", "defer_service", reply.retry_after)
        if reply.status in (502, 503, 504) and attempt == 0:
            wait(0.5)
            continue
        if reply.status != 200:
            raise PipelineError("access", "http_status")
        if reply.content_type.split(";")[0].strip().lower() != "text/html":
            raise PipelineError("access", "representation_mismatch")
        if not reply.ready or not reply.body.strip():
            raise PipelineError("access", "incomplete_snapshot")
        return reply.body
    raise PipelineError("access", "attempts_exhausted")


class BulletinParser(HTMLParser):
    # This small parser supports only the documented fixture markup.
    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.rows, self.current, self.in_title = [], None, False

    def handle_starttag(self, tag, attrs):
        attrs = dict(attrs)
        if tag == "article":
            if self.current is not None:
                raise PipelineError("extraction", "nested_record")
            self.current = {"id": attrs.get("data-id", ""),
                            "published": attrs.get("data-published", ""),
                            "title_parts": [], "title_count": 0}
        elif tag == "h2" and self.current is not None:
            self.current["title_count"] += 1
            self.in_title = True

    def handle_data(self, data):
        if self.current is not None and self.in_title:
            self.current["title_parts"].append(data)

    def handle_endtag(self, tag):
        if tag == "h2":
            self.in_title = False
        if tag == "article" and self.current is not None:
            self.rows.append(self.current)
            self.current, self.in_title = None, False


def extract(html):
    parser = BulletinParser()
    parser.feed(html)
    parser.close()
    if parser.current is not None or not parser.rows:
        raise PipelineError("extraction", "record_structure")
    records, seen = [], set()
    for row in parser.rows:
        title = " ".join("".join(row["title_parts"]).split())
        if not row["id"].strip() or not title or row["title_count"] != 1:
            raise PipelineError("extraction", "required_field")
        try:
            published = date.fromisoformat(row["published"]).isoformat()
        except ValueError:
            raise PipelineError("extraction", "invalid_date")
        if row["id"] in seen:
            raise PipelineError("extraction", "duplicate_id")
        seen.add(row["id"])
        records.append({"id": row["id"], "title": title,
                        "published": published})
    return records


def run(fetch, archive, wait=time.sleep):
    html = access(fetch, wait)
    digest = sha256(html.encode("utf-8")).hexdigest()
    archive[digest] = html  # In-memory evidence retained even if parsing fails.
    records = extract(html)
    return {"schema_version": "bulletins.v1", "source_id": "fixture:bulletins",
            "snapshot_sha256": digest,
            "records": records}


if __name__ == "__main__":
    html = ('<article data-id="notice-1" data-published="2026-09-10">'
            '<h2>Maintenance window announced</h2></article>')
    replies = iter([Reply(503), Reply(200, html)])
    archive = {}
    output = run(lambda: next(replies), archive, wait=lambda seconds: None)
    print(json.dumps(output, indent=2))

デモは合成された503応答を受信し、その後有効なHTML応答を受け取ります。この結果を生成します：

{
  "schema_version": "bulletins.v1",
  "source_id": "fixture:bulletins",
  "snapshot_sha256": "6ed8df5a98ee53e2889feb5ef7ed4dd8d549dba82882580418d4ca9656b7d46b",
  "records": [
    {
      "id": "notice-1",
      "title": "Maintenance window announced",
      "published": "2026-09-10"
    }
  ]
}

例が検証する内容

各レコードにはID、1つ以上の空でない見出し、解析可能な日付が必要です。重複するIDはバッチを拒否します。スナップショットのデジストは出力を保持されたHTMLに関連付け、抽出エラーはそのHTMLをコールャー所有のアーカイブに残して再実行可能にします。

2回の再試行と0.5秒の遅延は例示的なアプリケーションのポリシーであり、プロバイダーの推奨ではありません。ループは429を即座に遅延させ、503のクールダウンを保持し、認証またはチャレンジレビューで停止します。extractでの失敗はfetchを再実行できません。

プロダクションアダプターが追加すべき内容

ソースの承認、リダイレクトチェック、サポートされるコンテンツのデコード、リクエストごとのタイムアウト、および実際のトランスポートに接続する前の全体的なデッドラインを実装します。同期的なfetchが常に返らない場合、試行回数カウンタで制限されません。残りのデッドラインをトランスポートに渡し、再試行の待機時間をその予算に含めます。

インメモリアーカイブを制御されたストレージに置き換え、メタデータに観測タイムスタンプ、実際のソースID、エクストラクターのバージョン、スキーマバージョンを追加します。デコードやパースの前にサイズ制限を適用します。失敗または不完全なスナップショットは診断的な証拠として有用な場合がありますが、抽出対象のスナップショットプールとは別に保持します。

エージェントが使用する前に構造化されたウェブデータを検証する

受け入れられたデータにはJSON構造に加えて意味とカバレッジのチェックが必要です。この例は小さな決定論的な契約を検証しますが、一般的な抽出サービスにはより豊富な受け入れポリシーが必要です。

フィールド定義から始めます。公開日付、更新日付、収集タイムスタンプは異なるイベントを表します。エージェントが必要とするものを指定し、代替を拒否します。モデル生成フィールドの場合、ソースのスパンまたは視覚的領域を添付し、証拠がフィールドの意味を支持しているか確認します。価格、在庫、または日付などのフィールドでは、ページ上のどこかに一致する単語は弱い証拠です。

収集レベルでの完全性をチェックします。空の配列は「レコードなし」、レイアウトの変更、不完全なページング、または抽出失敗を意味する可能性があります。ソースが明示的で検証された空状態を提供しない限り、空の結果を受け入れません。このfixtureは空のリストを拒否しますが、そのような契約がないためです。

タスクの新鮮さウィンドウを定義します。古いスナップショットを再抽出してもパーサーの問題を修正できますが、下位の観測を現在のものにすることはできません。スナップショットIDとエクストラクター/スキーマバージョンを再実行キーに含め、受け入れられたレコードをイドempotentなストレージ操作で公開します。拒否された候補は、診断的なレビューのために一時的に保持し、エージェントの作業データセットに混ぜ込まないでください。

ページコンテンツを信頼できない入力として扱います。OWASPのプロンプトインジェクションガイドラインは、外部コンテンツに埋め込まれた指示からのリスクを説明しています。抽出ツールを資格情報や重要なアクションから隔離し、ソーステキストがコレクションの範囲を変更したり、データを他に送信したりする権限を取得しないようにします。

ライブソースに接続する前に境界をテストする

境界テストは、返された結果と予期しない追加作業の不在の両方を検証する必要があります。成功したパーサーテストだけでは、アクセスレイヤーが正しいように停止していることを証明しません。

この例では、タイムアウト followed by 成功、繰り返しの一時的なエラー、チャレンジとしてマークされた200応答、クールダウン付きの429、サポートされていない表現、見出しの欠如、無効な日付、重複IDをテストします。アダプター呼び出しをカウント: チャレンジケースは1回の呼び出し後に停止し、パース失敗はスナップショットを保持して別のアクセス試行をしないようにします。

添付のローカルテストスイートは、21のテストを通過し、フィクスチャーパイプラインの再実行、クールダウンの保持、スナップショットの保持、決定論的な再実行を含みます。これらは合成ソフトウェアチェックであり、ライブソースの成功確率やモデル抽出ベンチマークではありません。

デプロイメント前に、小さな許可されたステージングソースを追加し、実際のレンダリング準備、リダイレクト処理、セッションの有効期限切れ、トランスポートのキャンセル、出力証拠をテストします。有効なスナップショットと受け入れられたレコードを別々に測定します。最終的な受け入れ率が変化したときに、取得または解釈を改善する必要があるかどうかを知るためです。

受け入れられたレコードを中心にエージェントを構築する

AIエージェントのウェブスクレイピングアーキテクチャは、各ステージに観測可能な出力と明確な所有者がいる場合、運用が容易になります。アクセス契約は有効なスナップショットに焦点を当て、抽出契約は候補フィールドに焦点を当て、検証は証拠、完全性、新鮮さに焦点を当てます。

ローカルワークフローから始め、ネガティブパスを実行し、アダプターの制限が明確になるまで許可されたソースに接続しないでください。文書化されたCAPTCHA処理が必要なワークフローでは、アクセス境界内でCapSolverを評価し、抽出が再開される前に目的地を検証してください。

FAQ

Q: ウェブアクセスレイヤーとデータ抽出レイヤーの違いは何ですか？
A: アクセスレイヤーは有効なページスナップショットを取得し、取得失敗を分類します。抽出レイヤーはそのスナップショットを候補フィールドに解釈します。受け入れチェックは、結果のレコードがエージェントに適切かどうかを決定します。

Q: AIモデルにHTML、DOMスナップショット、またはスクリーンショットを渡すべきですか？
A: 要求されたフィールドの証拠を含む表現を使用してください。HTMLはサーバー提供コンテンツに適し、レンダードムはクライアントサイドコンテンツをキャプチャし、スクリーンショットは視覚的解釈に領域参照と不確実性チェックをサポートします。

Q: 見つからないフィールドのために別のページリクエストが必要ですか？
A: 見つからないフィールドはまず保持されたスナップショットのレビューまたは再抽出をトリガーします。スナップショットが不完全または古いか、アクセスポリシーが別の試行を許可していることを証拠で示した場合にのみ、新しいページリクエストを行います。

Q: CapSolverはこのアーキテクチャでどこに位置しますか？
A: CapSolverはアクセスレイヤー内の許可されたCAPTCHAタスクインターフェースの後ろに配置されます。あなたのアプリケーションはタスクの資格、セッションコンテキスト、再試行予算、タスク完了後の目的地の検証を所有します。

Q: Pythonの例はライブAIウェブスクレイピングを行いますか？
A: いいえ。この例はローカルでHTMLのフィクスチャーパイプラインを実行し、その契約をテストします。ライブデプロイメントには許可されたアクセスアダプターを追加する必要があります。モデルベースまたは視覚的な抽出も独自の実装と証拠に基づく評価が必要です。