如何在LlamaIndex智能代理中解决验证码

# 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 验证保护的网络资源时使用。
    
    参数:
        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 "失败" in token:
        return ""
    
    # 使用令牌访问页面
    import requests
    response = requests.post(url, data={"g-recaptcha-response": token.split("令牌: ")[1]})
    return response.text

# 创建用于受保护数据摄取的自定义工具
protected_fetch_tool = FunctionTool.from_defaults(
    fn=fetch_protected_page,
    name="fetch_protected_page",
    description="通过先解决验证来获取受保护的网页内容"
)

# 具有解决和获取能力的代理
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 解决方案"
)