ライブウェブを見つめるLangChainエージェント：Scrapelessを使用したAIデータパイプラインの構築

主なポイント:

• ファーストパーティのLangChain統合。 PyPIのlangchain-scrapelessパッケージには、すぐに使える5つのツールが含まれています — ScrapelessDeepSerpGoogleSearchTool、ScrapelessDeepSerpGoogleTrendsTool、ScrapelessUniversalScrapingTool、ScrapelessCrawlerCrawlTool、およびScrapelessCrawlerScrapeTool — これらはLangChainまたはLangGraphエージェントに直接統合できます。サブプロセスの配管やカスタムCDPの配線は不要です。
• パイプラインは4つのステップで構成されます：発見 → レンダリング → 抽出 → 保存。 ディープSERPツールを使用してGoogleを検索し、ユニバーサルスクレイピングツールで任意のURLをレンダリング（内部ではスクレイピングブラウザを使用）、PydanticOutputParserで型付きレコードを抽出し、オプションでベクトルストアに埋め込んで下流のRAGに利用します。
• エージェントがスクレイピングのタイミングを決定します。 LangChainのcreate_agent（内部ではLangGraphランタイム）がLLMに過去のコンテキストから回答するかScrapelessツールを呼び出すかを選ばせます — 新しいデータが本当に必要な場合にのみクラウドブラウザが立ち上がるため、コストとレイテンシーは日常的な処理で制限が保持されます。
• 型付きレコード、ローHTMLではありません。 ScrapelessUniversalScrapingTool（マークダウンレスポンス）をPydanticスキーマと組み合わせることで、スクレイピングされたページが検証されたProduct / Article / JobListingレコードに変換され、下流のコードは追加のパースの接着剤なしで信頼できます。
• 検出防止クラウドブラウザ、195カ国以上の住宅プロキシ。 Scrapelessスクレイピングブラウザは、毎セッションでJavaScriptレンダリング、住宅プロキシの出口、フィンガープリンティングのランダム化（UA、タイムゾーン、WebGL、キャンバス）を処理するため、エージェントは回避ではなく推論に集中できます。

はじめに：ライブウェブを見るAIデータパイプライン

裸のLLMはトレーニングデータから回答します。実際に出荷されるほとんどのエージェントワークフロー — 競合価格インテリジェンス、リードリサーチ、市場監視、構造化ニュースの取り込み、ライブウェブ上のRAG — モデルには今すぐページを見る必要があります。トレーニングのカットオフ、ペイウォール、遅延読み込みSPA、パーソナライズされたレンダリングはすべて、回答をLLMが見たことのない領域に押し込み、応答は古い数値を引用するか丁寧に辞退します。

LangChainとクラウドブラウザが標準的な解決策です。モデルは推論し、ブラウザが取得し、エージェントがこの2つをつなぎます。ほとんどのチームが直面する摩擦点はエージェントの下にあります：住宅プロキシ、JavaScriptレンダリング、検出防止フィンガープリンティング、セッションライフサイクルは、エージェントが何か有用なことをする前に解決する必要があります。住宅VPN上の直接的なPlaywrightは単一のラップトップ実行には機能しますが、生産スケジュールには耐えられません。

Scrapelessスクレイピングブラウザは、プラットフォームレベルでこれら4つの懸念を処理し、langchain-scrapeless PyPIパッケージがそれらをネイティブのLangChainツールとして公開します。この投稿では、これらのツールを4ステップの発見 → レンダリング → 抽出 → 保存 AIデータパイプラインにまとめる方法について、競争リサーチの具体例、型付きPydantic出力、制限された同時実行性、観察性フックに関して説明します。異なるプロトコルで同じプリミティブについては、MCP統合投稿を参照してください。

あなたが構築できるもの

langchain-scrapelessによって提供される5つのツールは、最も一般的なAIデータパイプラインのパターンをカバーしています：

• 競合価格インテリジェンス。 カテゴリを検索し、トップの小売業者のページをレンダリングし、価格、評価、レビュー数を含む型付きProductレコードを抽出します。
• SERP監視。 glとhlでパラメータ化されたScrapelessDeepSerpGoogleSearchToolを使用して、地域ごとのキーワードランキングやスニペットの変化を追跡します。
• 市場トレンドの追跡。 カテゴリサイズを把握するために、ScrapelessDeepSerpGoogleTrendsToolを使用してinterest_over_timeや関連クエリを取得します（トレンドエンドポイントへのアクセスは、あなたのScrapelessプランのティアに依存します）。
• スケールでの製品詳細抽出。 URLのリストをScrapelessCrawlerScrapeToolに供給し、LLMエクストラクター用にマークダウン形式で返します。
• ディレクトリからのリード生成。 ScrapelessCrawlerCrawlToolを使用してビジネスリストサイトをクロールし、連絡先行を型付きレコードにパースし、ドメインで重複を排除します。
• RAGのための構造化ニュースの取り込み。 出版者ページをきれいなマークダウンにレンダリングし、Articleレコードを抽出し、LangChainベクトルストアに埋め込み、リトリーバル拡張チェーンを使用してクエリを実行します。

全ての6つのパイプラインは、同じプリミティブ — 検索、レンダリング、抽出、保存 — で構成されており、以下の実例はフルーチェーンを端から端までカバーしています。

なぜScrapelessスクレイピングブラウザなのか

Scrapelessスクレイピングブラウザは、ウェブクローラーとAIエージェントのために設計されたカスタマイズ可能な検出防止クラウドブラウザです。特にLangChainエージェント向けに、以下のメリットを提供します：

• 195カ国以上の住宅プロキシ — 地理的に制約されたクエリは、ローカルユーザーが見るリスティングを返し、回転は毎セッションで自動的に行われます。
• クラウドサイドのJavaScriptレンダリング — ページが抽出前にハイドレーションされる完全なChromiumを使用しているため、SPA、無限スクロールフィード、およびレイジーロードパネルが一級のターゲットとなります。
• セッションごとの非検出フィンガープリンティング — UA、タイムゾーン、言語、画面解像度、WebGL、キャンバスがセッションごとにランダム化され、一貫性が重要な場合には固定アイデンティティ用のカスタムフィンガープリンティングAPIが用意されています。
• クラウドブラウザレイヤーでのセッション持続性 — sessionTTL（60-900秒）とsessionNameを介して利用可能で、WSSエンドポイントを直接操作する際に使用します。langchain-scrapelessツールの呼び出しは、invokeごとに新しいセッションを割り当てるため、研究パイプラインには最適なデフォルトです。
• ファーストパーティのLangChain統合 — pip install langchain-scrapelessにより、クラウドブラウザがネイティブLangChainツールとして公開されます。サブプロセスラッパーやCDPの配管、カスタムシリアライザーは不要です。

無料プランのAPIキーをScrapelessで取得してください。完全な統合については、github.com/scrapeless-ai/langchain-scrapelessにドキュメントがあります。

無料プランを取得してスクレイピングを始めましょう：

Scrapelessの活気あるコミュニティに参加して、$5-10の無料プランを取得し、他の革新者とつながりましょう：

Scrapeless公式Discordコミュニティ
Scrapeless公式Telegramコミュニティ

前提条件

• Python 3.10以降。
• ScrapelessアカウントとAPIキー — Scrapelessでサインアップし、設定 → APIキー管理からキーをコピーします。
• チャットモデルAPIキー — 下記の例ではOpenAI（OPENAI_API_KEY）を使用しています。同じエージェントコードは、langchain-anthropic、langchain-google-genai、langchain-ollama、または他のLangChainチャットモデルでもChatOpenAIの行を入れ替えるだけで動作します。
• pipとvenvに関する基本的な知識。

インストール

完全なセットアップは4つのサブステップからなります。各ステップは独立して確認できるため、進む前に一時停止して確認できます。

1. venvを作成し、パッケージをインストールする

python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install langchain langchain-scrapeless langchain-openai langgraph pydantic

langchain-scrapelessはlangchain-coreとScrapeless Python SDKを遷移依存関係として取り込みます。langchainメタパッケージは、現代的（非推奨ではない）なReActエージェントランタイムであるlangchain.agents.create_agentを提供します。langgraphは基盤となるCompiledStateGraphランタイムを提供し、langchain-openaiは例で使用されるチャットモデル提供者です。好みに応じてlangchain-anthropicまたは他の提供者に切り替えることができます。

2. APIキーを設定する

現在のシェルセッション用に両方のキーをエクスポートします：

export SCRAPELESS_API_KEY="your_api_token_here"
export OPENAI_API_KEY="your_openai_token_here"

永続的なインストールのためには、同じ行を~/.bashrc / ~/.zshrcに追加するか、.envローダー（python-dotenv）を使用してプロセス開始時にファイルを読み込みます。ScrapelessツールはSCRAPELESS_API_KEYを環境から自動的に読み取りますので、コンストラクタ引数として渡す必要はありません。

3. インストールを確認する

検索ツールをテストし、一行の結果を表示する短いスモークテストです。Scrapeless APIは、コールドスタート後の最初の呼び出しで一時的に400を返すことがあります。このテストは最大3回繰り返すので、1回の再試行でそのようなケースを救います：

# verify.py
import time
from langchain_scrapeless import ScrapelessDeepSerpGoogleSearchTool

tool = ScrapelessDeepSerpGoogleSearchTool()
for attempt in range(3):
    try:
        result = tool.invoke({"q": "scrapeless scraping browser",
                              "hl": "en", "gl": "us"})
        print(str(result)[:300])
        break
    except ValueError as e:
        print(f"transient (attempt {attempt + 1}): {e}")
        time.sleep(3)

これを実行します：python verify.py。成功すれば、数秒以内に検索結果の文字列が表示されます。3回の試行中にすべてValueErrorが発生し、failed with status 401と表示された場合は、APIキーが欠落しているか不正です。シェル内でecho $SCRAPELESS_API_KEYを再確認してください。3回の試行後も400がPersistingする場合は、アカウントが要求されたエンドポイントへのアクセスを持っていない可能性があるため、一時的なエラーに関するFAQを参照してください。

4. （オプション）依存関係のバージョンを固定する

再現可能なビルドのためには、テストしたバージョンを固定してください。クリーンなPython 3.12環境でpip install langchain langchain-scrapeless langchain-openai langgraph pydanticによって解決された組み合わせは次のとおりです：

langchain==1.2.17
langchain-core==1.3.2
langchain-openai==1.2.1
langchain-scrapeless==0.1.3
langgraph==1.1.10
pydantic==2.13.3
scrapeless==1.1.1

注意: langchain-scrapeless 0.1.3のパッケージメタデータはlangchain-core <0.4.0を宣言していますが、langchain-openaiがそれを要求するため、pipはlangchain-core 1.3.2を解決します; ランタイムインポートはまだ機能します。リゾルバーの警告を完全に回避するには、langchain-scrapelessと、必要なチャットモデルプロバイダーのみをインストールしてください（例：langchain-openaiを削除し、代わりにChatAnthropicインスタンスを渡します）。LangGraph 1.0は2025年10月にGAされ、1.1.xシリーズが現在の安定版です。

実際にこれを使用する方法: エージェントにプロンプトを送信する

インストール後、ターゲットサイトがDOMを回転させるたびにCSSセレクタを再導出するのではなく、エージェントに話しかけることでパイプラインを構築します。エージェントは発見→レンダリング→抽出のループを所有し、LLMは各ターンに適したツールを選択します。

貼り付け可能なプロンプト

実例

あなたの入力:

150ドル未満の旅行用ポータブルエスプレッソメーカーのトップ3を見つけてください。それぞれについて、名前、価格、平均評価、レビュー数、3つの主な特徴を返してください。検索を米国のエグレスに固定してください。

エージェントの計画（平易な英語で）:

1. ScrapelessDeepSerpGoogleSearchToolをq="best portable espresso makers under 150", gl="us", hl="en", num=5で呼び出してオーガニック結果のURLを取得します。
2. トップ3の結果のURLに対して、各ページをクリーンなマークダウンとしてレンダリングするためにScrapelessUniversalScrapingToolをresponse_type="markdown"で呼び出します。
3. 各レンダリングされたページをProductスキーマにバインドされたPydanticOutputParserを用いてLLMに渡し、nameおよびpriceを抽出できないページは拒否します。
4. 三つのProductレコードをJSON配列に集約して返却します。

あなたが得るもの:

[
  {
    "name": "Wacaco Nanopresso",
    "price": 79.95,
    "rating": 4.7,
    "review_count": 12483,
    "key_features": [
      "手動ハンドポンプ操作",
      "最大18バーの抽出圧力",
      "アダプターを介して挽いたコーヒーまたはNSカプセルに対応"
    ],
    "url": "https://example.com/p/wacaco-nanopresso"
  },
  {
    "name": "Flair NEO Flex",
    "price": 119.00,
    "rating": 4.5,
    "review_count": 2104,
    "key_features": [
      "電力不要のレバー駆動",
      "ボトムレスポータフィルターでエスプレッソグレードの圧力",
      "旅行用に分離可能"
    ],
    "url": "https://example.com/p/flair-neo-flex"
  },
  {
    "name": "Outin Nano",
    "price": 129.99,
    "rating": 4.6,
    "review_count": 5871,
    "key_features": [
      "内蔵加熱要素",
      "セルフクリーニングサイクル",
      "USB-C充電、約3分の加熱"
    ],
    "url": "https://example.com/p/outin-nano"
  }
]
// スキーマはStep 4のパーサーが出力する内容を正確に反映しています。フィールド値は例示です。

プロンプトの形作り

以下のステップ1〜6は、背景資料です。発見→レンダリング→抽出→格納パターンがどのように構成されるかを見るために一度読んでから、エージェントがオペレーターが渡した任意のクエリにそれを適用することを信頼してください。

アーキテクチャ

┌──────────────────────────────────────────────────────────────────────┐
│  LangChain create_agent  (LangGraph runtime)                         │

│ │
│ ┌───────────────────────┐ ┌────────────────────────────────┐ │
│ │ チャットモデル │ ──► │ ツール (langchain-scrapeless) │ │
│ │ (OpenAI / Anthropic │ │ • DeepSerpGoogleSearch │ │
│ │ / Gemini / Ollama) │ ◄── │ • UniversalScraping │ │
│ └───────────────────────┘ │ • CrawlerCrawl │ │
│ ▲ │ • CrawlerScrape │ │
│ │ │ • DeepSerpGoogleTrends │ │
│ │ └──────────────┬─────────────────┘ │
│ │ │ │
│ │ PydanticOutputParser │ │
│ │ (型付きレコード) ▼ │
│ │ ┌──────────────────────────────────┐ │
│ │ │ スクラペレスクラウドブラウザ │ │
│ │ │ • 住宅用プロキシ (195以上) │ │
│ │ │ • 検出防止フィンガープリンティング │ │
│ │ │ • Chromium JSレンダリング │ │
│ │ │ • セッションTTL 60～900s │ │
│ │ └──────────────────────────────────┘ │
│ │ │ │
│ └──────────────────────────────────┘ │
│ 型付きレコードはエージェントに戻る │
└──────────────────────────────────────────────────────────────────────┘
│
▼ (オプション)
┌───────────────────────────┐
│ ベクトルストア (Chroma / │
│ PGVector / pgvector) │
└───────────────────────────┘

三層、クリーンな分離：LLMが会話に基づいて推論し、`langchain-scrapeless`ツールがネイティブのLangChainインターフェースを介してクラウドブラウザをラップし、クラウドブラウザが推論でないすべての懸念を処理します。各層は入れ替え可能です — チャットモデル、プロンプト、さらには基盤となるツール — 他の層を書き換えることなく。

---

## ステップ1 — 型付き出力スキーマの定義

Pydanticは、スクレイピングされたMarkdownを下流のコードが依存できるものに変える荷重支持のプリミティブです。ターゲットレコードを一度定義し、ステップ4でLLMエクストラクターにバインドします。

```python
# schema.py
from typing import Optional
from pydantic import BaseModel, Field, HttpUrl

class Product(BaseModel):
    name: str = Field(...,
        description="製品名。常に必要です — 明確な製品名がない場合はページタイトルまたはH1を使用してください。")
    price: Optional[float] = Field(None, description="米ドルでの数値価格; 不在の場合はnull")
    rating: Optional[float] = Field(None, description="平均評価、0〜5; 不在の場合はnull")
    review_count: Optional[int] = Field(None, description="レビュー数; 不在の場合はnull")
    key_features: list[str] = Field(default_factory=list, description="3〜5の短い特徴バレット")
    url: HttpUrl = Field(..., description="正準製品URL")

欠ける可能性のあるすべてのフィールドをOptionalとしてマークし、デフォルトリストは空に設定します — ボット対策のインタースティシャル、地域レイアウトの違い、遅延ハイドレートされたDOMノードは、ページが定期的に一つまたは二つのフィールドを省略することを意味し、非オプショナルなスキーマは、他の用途に有用な行を拒否します。nameを必須とし、その説明にフォールバック（ページタイトル、H1）を与えて、エクストラクターが必須フィールドに対してnullを返さないようにします。その1つのヒントが、スキーマにノイズの多いページを吸収させることができます。

ステップ2 — ScrapelessDeepSerpGoogleSearchToolでの発見

Deep-SERPツールは、クエリのためにオーガニックなGoogle結果を返します。言語（hl）、国（gl）、および結果数（num）によってパラメータ化されます。これは発見のプリミティブです — 検索は、ページごとのレンダーバジェットをコミットする前に、URLの宇宙を広げます。

# discover.py
from langchain_scrapeless import ScrapelessDeepSerpGoogleSearchTool

search = ScrapelessDeepSerpGoogleSearchTool()
results = search.invoke({
    "q": "最高のポータブルエスプレッソメーカー 2026年 150ドル以下",
    "hl": "ja",
    "gl": "jp",
    "num": 5,
})
print(results)

hlは結果の言語を制御し、glは出国を制御します — これが地域のレバーです。地域別のSERPモニタリングのために、異なるgl値（us、de、jp、br）で同じクエリを実行し、結果リストを比較します。クエリボリュームの高いところでの一時的なValueError（ツールによってラップされたHTTP 400/503）やTimeoutErrorの応答は正常です; スケーリングアウトする前に、ステップ6のリトライデコレーターでコールをラップします。

ステップ3 — ScrapelessUniversalScrapingToolでのレンダリング

ユニバーサルスクレイピングツールは、Scrapeless Scraping Browserを前面に出しています。これはURLを受け取り、レンダリングされたページをマークダウン（またはHTML、またはスクリーンショット）として返します。マークダウンはLLM抽出器に feed するための最も安価な形式です — 広告、ナビゲーションのクロム、およびインラインスタイルを取り除き、ページの実際の内容だけを残します。

```python
# render.py
from langchain_scrapeless import ScrapelessUniversalScrapingTool

scrape = ScrapelessUniversalScrapingTool()
markdown = scrape.invoke({
    "url": "https://example.com/p/wacaco-nanopresso",
    "response_type": "markdown",
})
print(markdown[:600])

各 invoke は新しいクラウドブラウザーセッションを割り当てます。これは研究パイプラインに対して正しいデフォルトです — URLごとに新しいセッションを持つことは、シンプルかつセッションあたりのアンチボット状態に対してより耐性があります。(sessionName を介したクロスコールセッションの再利用はCDPレベルの機能です; あなたのワークフローがページ間での温かいクッキーとログイン状態を必要とする場合は、このLangChainツールを通じてではなく、WSSエンドポイントを介してクラウドブラウザーを直接駆動してください。) また、このツールは、独自のセレクターを実行する必要があるときに response_type="html" を、最も安価なLLMコンテキストのために response_type="plaintext" を、また視覚的回帰パイプラインのために response_type="png" / "jpeg" を受け取ります。

ステップ4 — PydanticOutputParser で抽出

ステップ1でバインドされたスキーマは、レンダリングされたマークダウンを受け取り、型付きの Product を返すLangChainエクスプレッション言語（LCEL) チェーンに直接接続されます。パーサーはJSONスキーマをプロンプトに注入し、LLMの応答をそれに対して検証します。

# extract.py
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

from schema import Product

parser = PydanticOutputParser(pydantic_object=Product)

prompt = ChatPromptTemplate.from_messages([
    ("system",
     "あなたはレンダリングされたウェブページから商品レコードを抽出します。\n"
     "出力はこのスキーマに厳密に一致します:\n{format_instructions}"),
    ("human",
     "ソースURL: {url}\n\nレンダリングされたマークダウン:\n{markdown}\n\n"
     "1つのプロダクトレコードを返してください。`name` フィールドは必須 — 明確な商品名がない場合は、ページのタイトルまたはH1を使用してください。存在しない場合は、オプショナルフィールド（価格、評価、レビュー数）のみをnullに設定してください。"),
]).partial(format_instructions=parser.get_format_instructions())

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
extract_chain = prompt | llm | parser

product = extract_chain.invoke({"url": "https://example.com/p/wacaco-nanopresso",
                                "markdown": "<ステップ3からのレンダリングされたマークダウン>"})
print(product.model_dump())

パーサーの失敗は、次の三つのことが成り立つときはまれです: (1) すべての不確かなフィールドが Optional[...] とマークされている、(2) プロンプトがモデルに 唯一 オプショナルフィールドのみが null にできると明示的に指示している、(3) すべての必須フィールドがその説明内でフォールバックを持っている（例：name はページのタイトルまたはH1を使用します）。これらの三つが整っていると、スキーマのnullable契約は欠落したオプショナルフィールドを処理し、プロンプトの指示がノイズの多いページでの必須フィールドをnullにするのを防ぎます — そのため、チェーンはクリーンに実行され、リトライラッパーなしで進行します。

ステップ5 — create_agent へコーディング

エージェントは三つのツールを結びつけ、LLMが任意のユーザーメッセージに対してどのツールを呼び出すかを決定できるようにします。langchain.agents.create_agent は langchain 1.2の標準的なランタイムです（これは非推奨となった langgraph.prebuilt.create_react_agent を置き換え、内部では同じLangGraph状態グラフを使用します）。

# agent.py
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_scrapeless import (
    ScrapelessDeepSerpGoogleSearchTool,
    ScrapelessUniversalScrapingTool,
    ScrapelessCrawlerCrawlTool,
)
from tenacity import (retry, stop_after_attempt,
                       wait_exponential, retry_if_exception_type)

# 基底のファーストパーティツール
_search   = ScrapelessDeepSerpGoogleSearchTool()
_scrape   = ScrapelessUniversalScrapingTool()
_crawl    = ScrapelessCrawlerCrawlTool()

# エージェントがすべてのツール呼び出しで見るリトライデコレータ。
# Scrapelessツールは一時的なAPIエラーをValueErrorとして浮き上がらせるため、それがフィルターになります。
_retry = retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=15),
    retry=retry_if_exception_type(ValueError),
)

def _check(payload: str) -> str:
    # クラウドブラウザーは時々HTTP 200を返しますが、埋め込まれたERR_ JSONがあります。
    # 上で定義した @_retry デコレータが発動するように、ValueErrorとして浮き上がらせます。
    if isinstance(payload, str) and payload.startswith('{"statusCode"') and "ERR_" in payload:
        raise ValueError(f"クラウドブラウザーエラー: {payload[:200]}")
    return payload

@tool
@_retry
def google_search(q: str, hl: str = "en", gl: str = "us", num: int = 5) -> str:
    """Googleを検索し、上位のオーガニック結果をJSONとして返します。"""
    return _check(str(_search.invoke({"q": q, "hl": hl, "gl": gl, "num": num})))

@tool
@_retry
def render_page(url: str) -> str:
    """ScrapelessクラウドブラウザでURLをレンダリングし、クリーンなマークダウンを返します。"""
    return _check(_scrape.invoke({"url": url, "response_type": "markdown"}))

@tool
@_retry
def crawl_site(url: str, limit: int = 10) -> str:
    """サイトを巡回し、制限されたページ数に対してマークダウンを返します。"""
    return _check(str(_crawl.invoke({"url": url, "limit": limit})))

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

system_prompt = (
    "あなたは型付き製品データセットを構築するリサーチエージェントです。 "
    "カテゴリークエリに対して、あなたは: "
    "(1) google_searchを呼び出して、トップのオーガニックURLを取得し、 "
    "(2) 有望な各URLでrender_pageを呼び出し、 "
    "(3) スキーマを使用してページごとに製品レコードを抽出し、 "
    "(4) レコードのJSON配列を返します。 "
    "ユーザーが異なるように要求しない限り、gl='us'とhl='en'を設定します。 "
    "ページに価格がない場合、その行は最終的な配列から省略されます。"
)

agent = create_agent(llm,
                     [google_search, render_page, crawl_site],
                     system_prompt=system_prompt)

for chunk in agent.stream(
    {"messages": [("human",
                   "150ドル未満のポータブルエスプレッソメーカーのトップ3を見つけ、 "
                   "名前、価格、評価、レビュー数、主な特徴、URLを返してください。")]},
    stream_mode="values",
):
    chunk["messages"][-1].pretty_print()

@tool + @retryデコレータスタックは、プロダクションでの支えとなる要素です。langchain-scrapelessツールは、内部のScrapelessErrorをValueErrorとしてラップしてから発生させるため、エージェントのツール呼び出しループの中での一時的な400/503の応答はValueErrorとしてバブルアップします。リトライなしでは、一回の一時的な失敗が全エージェントの実行をクラッシュさせますが、デコレータスタックを使用すれば、各ツール呼び出しは最大三回、指数的ジッター・バックオフで再試行します（langchain-coreのRunnable.with_retry()はRunnableRetryを返しますが、create_agentは受け付けません — 上記の@toolデコレートされた関数がリトライシェル内に本物のStructuredToolを生成する道です）。

agent.stream(..., stream_mode="values")は、各ステップでの完全なメッセージ状態を発信するため、オペレーターはエージェントのツール呼び出しや中間的な推論をリアルタイムで監視できます。単一の最終回答のためには、agent.invoke(...)に切り替えます。ステップ4からの型付きProduct[]をエージェントを通じてパイプするには、抽出チェーンを別の@tool関数として公開し、リストに追加すれば — エージェントは各レンダリング後にそれを呼び出します。langchain-scrapelessのREADMEはまだfrom langgraph.prebuilt import create_react_agentを示していますが、そのパスは動作しますが、LangGraphDeprecatedSinceV10の警告を発します。したがって、上記の現代的なインポートが推奨されるパスです。

ステップ6 — プロダクションハーデニング

ノートブックで三つのURLで動作するリサーチスクリプトは、三万では生き残りません。四つのハーデニングパターンは、上記のパイプラインをスケジューラーが無人で実行できるものに変えます。

限界付き同時実行

# concurrent_render.py
import asyncio
from langchain_scrapeless import ScrapelessUniversalScrapingTool

scrape = ScrapelessUniversalScrapingTool()
SEM = asyncio.Semaphore(3)            # ホストごとに最大3つの同時レンダリングを制限

async def render(url: str) -> str:
    async with SEM:
        return await scrape.ainvoke({"url": url, "response_type": "markdown"})

async def render_all(urls: list[str]) -> list[str]:
    return await asyncio.gather(*(render(u) for u in urls))

ホストごとの同時レンダリングの三つは、甘いスポットです — セッションのウォームアップコストを分散できるほど高く、ほとんどのサイトのIPごとのレート制限を下回るほど低いです。グローバルに制限するのではなく、ホストレベルで制限します — 三つのワーカーを持つ十の異なるホストは大丈夫ですが、同じ小売業者に十のワーカーが集中するのは問題です。

一時的なエラーに対するリトライ

# retry.py
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from langchain_scrapeless import ScrapelessUniversalScrapingTool

scrape = ScrapelessUniversalScrapingTool()

def _raise_on_embedded_error(payload: str) -> str:
    # Scrapelessは時々、ブラウザ側の内部エラーを説明するJSON本体を伴うHTTP 200を返します。
    # ツールはそのエラーに対して例外を発生させません。それをValueErrorとして提示してリトライを行います。
    if payload.startswith('{"statusCode"') and "ERR_" in payload:
        raise ValueError(f"クラウドブラウザのエラー: {payload[:200]}")
    return payload

@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=2, max=20),
    retry=retry_if_exception_type((ValueError, TimeoutError)),
)
def render_with_retry(url: str) -> str:
    return _raise_on_embedded_error(
        scrape.invoke({"url": url, "response_type": "markdown"}))

スクレイピングブラウザセッションは、時折、二つの異なる形で失敗します：HTTPレイヤーの400/503（langchain-scrapelessラッパーがこれをValueErrorとして上げます）と、ブラウザ側のエラー（例：ERR_TUNNEL_CONNECTION_FAILED）を説明するJSONボディ付きのHTTP 200（ラッパーは例外を上げません — エラーJSONは文字列として返されます）。上記の_raise_on_embedded_errorガードは、二つ目の形をキャッチしてそれをValueErrorに変換するため、同じリトライポリシーが適用されます。この二つの形がカバーされていることで、指数バックオフと四回のリトライが高い90パーセンタイルをキャッチし、実際の失敗（ボット対策ブロック、404エラー）をリトライの下に埋めることを避けます。エージェント主導の呼び出しでは、ステップ5の@tool + @retryデコレーターのスタックを使用してください — そこにあるラッパー関数は返却前に同じ_raise_on_embedded_errorチェックを適用する必要があります。

LangSmithによる可視化

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY="your_langsmith_key"
export LANGCHAIN_PROJECT="scrapeless-research-agent"

これらの三つの環境変数が設定されていると、すべてのツール呼び出し、すべてのLLM呼び出し、およびすべてのパーサーの失敗がLangSmithにタイミング、コスト、モデルが見た正確なプロンプトとともに表示されます。プロダクションでの実行において、これは最高のレバレッジを持つ変更です — 「エージェントが何か変なことをした」という状態をクリック可能なトレースに変えます。

ベクターストアに保存する

# embed.py
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document

docs = [Document(page_content=p.model_dump_json(), metadata={"url": str(p.url)})
        for p in products]
vs = Chroma.from_documents(docs, OpenAIEmbeddings(), persist_directory=".chroma")

ベクターストアはパイプラインのオプショナルな第四のステージです。エージェントがデータセットに比して下流の質問に答える長期的なRAGサービスである場合、利点がありますが、一度きりのリサーチスクリプトには大げさです。操作の好みに基づいてストアを選択してください — ローカルファイル用にはlangchain_chroma、管理されたPostgres + pgvector用にはlangchain_postgres、ホスティングされたベクターデータベース用にはlangchain_pinecone。

受け取る結果

[
  {
    "name": "Wacaco Nanopresso",
    "price": 79.95,
    "rating": 4.7,
    "review_count": 12483,
    "key_features": [
      "手動ハンドポンプ操作",
      "最大18バーの抽出圧力",
      "アダプターを介して挽きコーヒーまたはNSカプセルに対応"
    ],
    "url": "https://example.com/p/wacaco-nanopresso"
  },
  {
    "name": "Flair NEO Flex",
    "price": 119.00,
    "rating": 4.5,
    "review_count": 2104,
    "key_features": [
      "レバー駆動、電気不要",
      "ボトムレスポルタフィルターによるエスプレッソグレードの圧力",
      "旅行用に取り外し可能"
    ],
    "url": "https://example.com/p/flair-neo-flex"
  },
  {
    "name": "Outin Nano",
    "price": 129.99,
    "rating": 4.6,
    "review_count": 5871,
    "key_features": [
      "内蔵ヒーティングエレメント",
      "自己洗浄サイクル",
      "USB-C充電、約3分の加熱"
    ],
    "url": "https://example.com/p/outin-nano"
  }
]
// スキーマはステップ4パーサーが出力するものと正確に一致します。フィールドの値は示例です。

この処理がライブウェブに対して実行される際に期待されるいくつかの正直な観察：

• 水分保持のタイミングはサイトによって異なる。 ユニバーサルスクレイピングツールはデフォルトでdomcontentloadedを待ちます；価格をセカンドXHRを介して水和するSPAの場合、マークダウンが価格が表示される前に到着することがあります。簡単な遅延を伴う再レンダリングを行うか、フィールドが一貫してnullである場合は、response_type="html"とカスタムセレクタにフォールバックします。
• オプションフィールドはオプションのまま。 一部の製品ページは、特に直接消費者向けサイトでは明示的な評価やレビュー件数を省略します。null値は失敗モードではなく情報として扱い、下流でフィルタリングします。
• LLM抽出がレイテンシを支配。 エンドツーエンドで、パイプラインは検索呼び出しに対して約1〜3秒、ページレンダリングに対して2〜4秒、およびLLM抽出に対して3〜5秒の時間がかかります。レンダリングと抽出のステージでの同時実行が最大のレバーです。
• ボット対策のインタースティシャルがValueErrorとして現れる。 サイトがクラウドブラウザが透過的に完了できないCloudflareまたはAkamaiのチャレンジを前倒しする場合、langchain-scrapelessラッパーはプレースホルダーページを静かに返すのではなくValueErrorを上げます。リトライデコレーターは一時的なケースをキャッチし、持続的なケースはフィンガープリンティングを広げたり、別のプロキシリージョンをピン留めする方が良いです。
• ベクターストアのステージはオプション。 下流の消費者に型付きレコードを返すリサーチパイプラインには、これを完全にスキップします。同じデータセットが時間をかけて複数の下流の質問に応える場合に追加します。

よくある質問

住宅用プロキシは必要ですか？
はい、意味のあるアンチボット保護を持つサイト、つまりほとんどの小売業者、マーケットプレイス、およびSERPエンドポイントについてはそうです。ScrapelessUniversalScrapingToolと深いSERPツールは、デフォルトでScrapelessの住宅用プロキシプールを経由します。検索ツールのglパラメータは、出口国を固定します。

400や503のような一時的なエラーについてはどうですか？
langchain-scrapelessツールは、一時的なAPIエラーをValueErrorとして表面化します（基になるScrapelessErrorは再発生の前にラップされます）。直接呼び出す場合は、ステップ6のtenacityデコレーターを使用し、retry_if_exception_type=(ValueError, TimeoutError)にします。create_agent内のエージェント駆動の呼び出しでは、ステップ5から@toolと@retryデコレーターのスタックで各ツールをラップします — これにより、エージェントが受け入れ、すべてのツール呼び出しにリトライポリシーを適用する本物のStructuredToolが生成されます。これがないと、単一の一時的な400エラーが全体のエージェント実行をクラッシュさせます。

サイトがアクセス拒否を返します。どうしますか？
まずはステップ6のデコレーターで再試行します。ページが持続的にブロックされる場合は、glを別の国に変更してセッションを広げるか、試行の間に短いawait asyncio.sleep(...)を追加してセッション状態をクールダウンさせます。IP層のブロックが一貫しているサイトについては、ブロックがアカウントレベルではなくプラットフォームレベルであることを確認するためにScrapelessサポートに連絡してください。

セレクタが壊れ続けます。DOMのローテーションにどう対処しますか？
ScrapelessUniversalScrapingToolにresponse_type="markdown"を使用することで、CSSセレクタでHTMLを解析するのではなくなります。Markdownはナビゲーションのクロームとほとんどのレイアウトのずれを崩すため、ステップ4のLLM抽出器は、基になるDOMがシフトしても安定したコンテンツの表現を確認できます。

ホストあたりの同時ワーカー数は何ですか？
安定した実行のための文書化された上限は3です。ホストレベルで制限します（ステップ6のasyncio.Semaphore(3)）。異なるホスト間のワーカーは独立して実行できます。

LangGraphなしで使用できますか？
はい。ScrapelessUniversalScrapingTool().invoke({...})は通常の呼び出し可能オブジェクトです — どのPythonスクリプト、FastAPIルート、またはCeleryタスクからでも呼び出すことができます。LangGraphはその上にエージェントループを追加しますが、ツール自体はフレームワークに依存しません。

OpenAIをClaude、Gemini、またはローカルモデルに置き換えることはできますか？
はい。ChatOpenAI(model="gpt-4o-mini")をChatAnthropic(model="claude-sonnet-4-6")、ChatGoogleGenerativeAI(model="gemini-2.5-pro")、ChatOllama(model="llama3.1")、または他のLangChainチャットモデルに置き換えます。toolsリスト、プロンプト、およびパーサは変更されません。

マルチターンメモリを追加するにはどうすればよいですか？
create_agent(llm, tools, checkpointer=MemorySaver())にMemorySaverチェックポイントを渡し、各invokeにthread_idを提供します。LangGraphはターンを跨いで会話状態を保持するので、エージェントは以前の検索を再実行することなく参照できます。

リクエストトレースはどこで見ることができますか？
LANGCHAIN_TRACING_V2=true、LANGCHAIN_API_KEY、およびLANGCHAIN_PROJECTを設定します（ステップ6）。すべてのツール呼び出し、LLM呼び出し、およびパーサの実行は、タイミング、コスト、および正確なプロンプトを伴ってLangSmithに表示されます — 本番環境での配備に対する最も高い観察可能性の変化です。