はじめに
語彙力なくてすみません、
browser-use は、「AI エージェントがウェブブラウザを操作できるようにする」ためのライブラリです。
プロンプトで与えられた指示どおりに動き、ほかの技術と比較しても精度が抜群に高いです。
早速試してみます。
実践
複数のECサイトから特定の商品価格を取得することを目標とする。
Python は 3.11 以上が必要です。
mkdir browser_use_project
cd browser_use_project
python -m venv .venv
source .venv/bin/activate
pip install browser-use
playwright install
プロジェクト配下に.envファイルを作って、下記のどれかのキーを入れます。
.env
OPENAI_API_KEY=
or
ANTHROPIC_API_KEY=
一旦下記のように、出力だけ見てみます。
index.py
from langchain_openai import ChatOpenAI
from browser_use import Agent
import asyncio
async def main():
agent = Agent(
task="""
あなたは価格監視のエージェントです。
与えられたURLから商品の監視をしてください
対象商品は: ロイヤルカナン 犬用 消化器サポート 低脂肪 小型犬用S 3kgx1
- Sundrug-online url: https://sundrug-online.com/products/3182550925792
- Rakuten url: https://item.rakuten.co.jp/sundrug/3182550925792/
- yodobashi url: https://www.yodobashi.com/product/100000001008730001/
下記の形式でデータを教えてほしい
- 価格
- 送料(なけれな0円)
- クーポン(なけれな0円)
- ポイント(なけれな0円)
- ショップ名
""",
llm=ChatOpenAI(model="gpt-4o"),
)
result = await agent.run()
print(result)
asyncio.run(main())
実行中にブラウザが起動し、ターゲットとなっているURLをそれぞれのタブで開いて、情報を取得していました。
最終的に下記のようなログが出力され、期待していた結果が得られました。
log
...
INFO [agent] 📄 Result: Here are the extracted details for the product 'ロイヤルカナン 犬用 消化器サポート 低脂肪 小型犬用S 3kgx1':
1. **サンドラッグ (Sundrug-online)**
- 価格: ¥9,498
- 送料: ¥418
- クーポン: ¥0
- ポイント: ¥0
- ショップ名: サンドラッグ Online Store
2. **楽天 (Rakuten)**
- 価格: ¥9,498
- 送料: ¥0
- クーポン: ¥0
- ポイント: ¥0
- ショップ名: サンドラッグe-shop
3. **ヨドバシカメラ (Yodobashi)**
- 価格: ¥8,960
- 送料: ¥0
- クーポン: ¥0
- ポイント: ¥896
- ショップ名: ヨドバシカメラ
INFO [agent] ✅ Task completed successfully
...
カスタムアクションの登録とエージェントの並列化
browser-useで取得されたデータを保存するというユースケースの実現で下記の二つの機能を確認します。
カスタムアクション
そもそもカスタムアクションとは何なのでしょうか。
私の理解では、Amazon Bedrock Agentのアクショングループと似た機能です。
プロンプトとメソッドを与えるとエージェントがそれを理解し、特定のタイミングで実行してくれるものです。
でも必ず実行される既存のアクションも存在します。
logを確認しますと、doneという名前のアクションを確認できます。
INFO [agent] 🛠️ Action 1/1: {"done":{"....
下記のように書けば、doneの実行タイミングで実行されるメソッドをカスタマイズできます。
controllErexamples.py
@controller.action('use this action in done', param_model=HogeHoge, requires_browser=True)
async def done(params: HogeHoge, browser: Browser):
print(params)
ただし、基本的には別のカスタマイズメソッドを作ったほうがよいと思います。
カスタマイズメソッド使用してデータ保存する場合下記のように書けばできます。
index.py
from typing import List
import asyncio
from pydantic import BaseModel
from langchain_openai import ChatOpenAI
from browser_use import Agent
from browser_use.controller.service import Controller
controller = Controller()
class StoreResult(BaseModel):
price: str
postage: str
coupon: str
points: str
store: str
ecName: str
time: str
class StoreResults(BaseModel):
storeResults: List[StoreResult]
@controller.action('Save Price', param_model=StoreResults)
def save_price(params: StoreResults):
with open('price.txt', 'a') as f:
for storeResult in params.storeResults:
f.write(f'price:{storeResult.price},postage:{storeResult.postage},coupon:{storeResult.coupon},points:{storeResult.points},store:{storeResult.store},ecName:{storeResult.ecName},time:{storeResult.time}\n')
async def main():
agent = Agent(
task="""
あなたは価格監視のエージェントです。
与えられたURLから商品の監視をしてください
対象商品は: hegehoge商品
- ecName: Sundrug-online url: https://sundrug-online.com/hegehoge
- ecName: Rakuten url: https://item.rakuten.co.jp/sundrug/hegehoge
- ecName: yodobashi url: https://www.yodobashi.com/product/hegehoge
下記の形式でデータを教えてほしい
- 価格
- 送料(なけれな0円)
- クーポン(なけれな0円)
- ポイント(なけれな0円)
- ショップ名
- ECサイトの名称
""",
llm=ChatOpenAI(model="gpt-4o"),
controller=controller
)
result = await agent.run()
print(result)
asyncio.run(main())
エージェントの並列化
公式のサンプルは下記の通りです。
実際実行される場合、ブラウザが立ち上げて、ループしてタブを開き、タスクをこなしていくイメージです。
index.py
browser = Browser()
for i in range(10):
# This create a new context and automatically closes it after the agent finishes (with `__aexit__`)
async with browser.new_context() as context:
agent = Agent(task=f"Task {i}", llm=model, browser_context=context)
カスタムアクションとの違いとしては、1エージェントには独自のライフサイクルを持ってるため、よりコントロールしやすいです。
下記がエージェントの並列化を利用したデータ保存用のコードです。
index.py
import asyncio
from pydantic import BaseModel
import re
import json
from datetime import datetime
from langchain_openai import ChatOpenAI
from browser_use import Agent, Browser
class StoreResult(BaseModel):
price: str
postage: str
coupon: str
points: str
store: str
time: str
def parse_store_data(text: str) -> dict:
"""単一店舗の情報をパースする"""
price_match = re.search(r'\*\*Price\*\*:\s*([^\n]+)', text)
postage_match = re.search(r'\*\*Postage\*\*:\s*([^\n]+)', text)
coupon_match = re.search(r'\*\*Coupon\*\*:\s*([^\n]+)', text)
points_match = re.search(r'\*\*Points\*\*:\s*([^\n]+)', text)
store_match = re.search(r'\*\*Store\*\*:\s*([^\n]+)', text)
time_match = re.search(r'\*\*Time\*\*:\s*([^\n]+)', text)
return {
"price": price_match.group(1).strip() if price_match else "0円",
"postage": postage_match.group(1).strip() if postage_match else "0円",
"coupon": coupon_match.group(1).strip() if coupon_match else "0円",
"points": points_match.group(1).strip() if points_match else "0円",
"store": store_match.group(1).strip() if store_match else "不明",
"time": time_match.group(1).strip() if time_match else datetime.now().strftime("%Y-%m-%d %H:%M")
}
async def create_store_agent(url: str, store_name: str, context, model):
"""店舗ごとのエージェントを作成"""
return Agent(
task=f"""
商品の価格を監視してください。たまに最初広告表示する、それを閉じてから操作してください。
対象URL: {url}
対象商品: ロイヤルカナン 犬用 消化器サポート 低脂肪 小型犬用S 3kgx1
下記の形式でデータを返してください:
#### {store_name}
- **Price**: (価格)
- **Postage**: (送料、なければ0円)
- **Coupon**: (クーポン、なければ0円)
- **Points**: (ポイント、なければ0円)
- **Store**: {store_name}
- **Time**: (現在時間)
""",
llm=model,
browser_context=context,
)
async def save_store_data(store_data: dict, store_name: str):
"""店舗ごとのデータを個別のJSONファイルに保存"""
output_data = {
"timestamp": datetime.now().isoformat(),
"store": store_data
}
filename = f'price_monitor_{store_name}_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json'
with open(filename, 'w', encoding='utf-8') as f:
json.dump(output_data, f, ensure_ascii=False, indent=4)
print(f"\nデータを {filename} に保存しました")
async def main():
# 店舗情報の定義
stores = [
{"name": "Sundrug", "url": "https://sundrug-online.com/hegehoge"},
{"name": "Rakuten", "url": "https://item.rakuten.co.jp/hegehoge"},
{"name": "Yodobashi", "url": "https://www.yodobashi.com/hegehoge"}
]
browser = Browser()
async with await browser.new_context() as context:
model = ChatOpenAI(model="gpt-4o")
for store in stores:
try:
# 店舗ごとのエージェントを作成
agent = await create_store_agent(
url=store["url"],
store_name=store["name"],
context=context,
model=model
)
history = await agent.run()
result = history.final_result()
if result:
store_data = parse_store_data(result)
await save_store_data(store_data, store["name"])
except Exception as e:
print(f"{store['name']}の処理中にエラーが発生しました: {e}")
if __name__ == "__main__":
asyncio.run(main())
ポイントとなってるのは最後のresult = history.final_result()です、
取得されるデータはfinal_result()使って取得できます、デフォルトでは文字列になってます、適切なパース処理が必要です。
最後
browser-useは使いやすく、X-Pathを使ったクローリングよりもはるかに手軽ですが、その分トークンの消費量が多いです。正確に計算したわけではありませんが、通常のモデルを呼び出してチャットする場合の4～5倍程度のコストがかかっているようです。
クローリングだけが目的であれば、まずは通常の手法を試し、困難な場合にbrowser-useを使用するのが正しいユースケースだと思います。
また、周囲に迷惑をかけないよう、規約などをきちんと確認した上でクローリングを行ってください