LlamaIndexエージェントにおけるCAPTCHAの解決方法

# CapSolverパッケージ
pip install git+https://github.com/capsolver-ai/capsolver-core.git
pip install git+https://github.com/capsolver-ai/capsolver-agent.git

# LlamaIndex
pip install llama-index llama-index-llms-openai

export CAPSOLVER_API_KEY="your-capsolver-api-key"
export OPENAI_API_KEY="your-openai-api-key"

import asyncio
from llama_index.core.tools import FunctionTool
from capsolver_agent.schema import create_executor

executor = create_executor(api_key="YOUR_CAPSOLVER_API_KEY")

def solve_captcha(captcha_type: str, website_url: str, website_key: str) -> str:
    """CAPTCHAチャレンジを解決し、検証トークンを返します。
    
    ウェブリソースにアクセスする際には使用してください。
    
    引数:
        captcha_type: 'reCaptchaV2', 'reCaptchaV3', または 'cloudflare'
        website_url: CAPTCHAで保護されたページの完全なURL
        website_key: サイトキー（data-sitekey属性の値）
    
    戻り値:
        解決されたトークン文字列またはエラーメッセージ
    """
    result = asyncio.run(executor.execute("solve_captcha", {
        "captcha_type": captcha_type,
        "website_url": website_url,
        "website_key": website_key
    }))
    if result["success"]:
        return f"CAPTCHAが解決されました。トークン: {result['solution']['token']}"
    return f"失敗: {result['error']}"

# LlamaIndexのFunctionToolとしてラップ
captcha_tool = FunctionTool.from_defaults(
    fn=solve_captcha,
    name="solve_captcha",
    description="保護されたウェブページ上のreCAPTCHA v2/v3、Cloudflare TurnstileのCAPTCHAチャレンジを解決"
)

from llama_index.core.agent import ReActAgent
from llama_index.llms.openai import OpenAI

# LLMを作成
llm = OpenAI(model="gpt-4o", temperature=0)

# CAPTCHA解決ツールを持つエージェントを作成
agent = ReActAgent.from_tools(
    tools=[captcha_tool],
    llm=llm,
    verbose=True
)

# エージェントを実行
response = agent.chat(
    "https://example.com/dataにアクセスする必要があります。これはreCAPTCHA v2で保護されています。"
    "サイトキーは6LeIxAcTAAAAAJcZVRqyHh71UMIEGNQ_MXjiZKhIです。解決してトークンを教えてください。"
)
print(response)

from llama_index.core import VectorStoreIndex, Document

def fetch_protected_page(url: str, site_key: str) -> str:
    """最初に検証を解決してから、CAPTCHAで保護されたページを取得します。"""
    # CAPTCHAを解決
    token = solve_captcha("reCaptchaV2", url, site_key)
    if "Failed" in token:
        return ""
    
    # トークンを使用してページにアクセス
    import requests
    response = requests.post(url, data={"g-recaptcha-response": token.split("Token: ")[1]})
    return response.text

# 保護されたデータインジェスト用のカスタムツールを作成
protected_fetch_tool = FunctionTool.from_defaults(
    fn=fetch_protected_page,
    name="fetch_protected_page",
    description="最初に検証を解決してから、CAPTCHAで保護されたウェブページのコンテンツを取得"
)

# 解決と取得機能を持つエージェント
data_agent = ReActAgent.from_tools(
    tools=[captcha_tool, protected_fetch_tool],
    llm=llm,
    verbose=True
)

def solve_captcha_with_retry(captcha_type: str, website_url: str, website_key: str, max_retries: int = 3) -> str:
    """失敗時に再試行を行うプロダクショングレードのCAPTCHA解決。"""
    for attempt in range(max_retries):
        result = asyncio.run(executor.execute("solve_captcha", {
            "captcha_type": captcha_type,
            "website_url": website_url,
            "website_key": website_key
        }))
        if result["success"]:
            return f"解決済み (試行 {attempt+1}). トークン: {result['solution']['token']}"
        if attempt < max_retries - 1:
            import time
            time.sleep(3)
    return f"{max_retries}回の試行後に失敗: {result.get('error')}"

# 再試行機能付きのプロダクションツール
production_captcha_tool = FunctionTool.from_defaults(
    fn=solve_captcha_with_retry,
    name="solve_captcha",
    description="エラー時に自動的に再試行するCAPTCHA解決"
)