如何在CrewAI多智能体工作流程中解决CAPTCHA

# CapSolver 核心引擎（必需依赖）
pip install git+https://github.com/capsolver-ai/capsolver-core.git

# CapSolver 代理工具
pip install git+https://github.com/capsolver-ai/capsolver-agent.git

# CrewAI 框架
pip install crewai crewai-tools

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

import asyncio
from crewai import Agent, Task, Crew
from crewai.tools import tool
from capsolver_agent.schema import create_executor

# 创建 CapSolver 执行器
executor = create_executor(api_key="YOUR_CAPSOLVER_API_KEY")

@tool("解决 CAPTCHA")
def solve_captcha(captcha_type: str, website_url: str, website_key: str) -> str:
    """解决 CAPTCHA 挑战并返回令牌。
    在遇到网站上的 CAPTCHA 时使用此工具。
    
    参数:
        captcha_type: CAPTCHA 类型 - 'reCaptchaV2', 'reCaptchaV3' 或 'cloudflare'
        website_url: 包含 CAPTCHA 的页面完整 URL
        website_key: 网站密钥（data-sitekey 属性值）
    
    返回:
        用于表单提交的解决 CAPTCHA 令牌
    """
    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']}"
    else:
        return f"CAPTCHA 解决失败: {result['error']}"

from crewai import Agent

# 具有 CAPTCHA 解决能力的研究员
researcher = Agent(
    role="网络研究专家",
    goal="从网站收集数据，处理遇到的任何 CAPTCHA 挑战",
    backstory="""您是专家级网络研究员，从各种在线来源收集数据。
    当您在网站上遇到 CAPTCHA 挑战时，您使用 solve_captcha 工具清除它并继续研究。
    您知道如何识别 CAPTCHA 类型：reCaptchaV2（复选框或不可见）、reCaptchaV3（基于评分）和 cloudflare（Turnstile 小部件）。""",
    tools=[solve_captcha],
    verbose=True
)

# 数据收集代理
data_collector = Agent(
    role="数据收集专家",
    goal="从目标网站提取结构化数据",
    backstory="您从网络来源收集数据并使用 solve_captcha 工具处理 CAPTCHA 挑战。",
    tools=[solve_captcha],
    verbose=True
)

# 验证代理
verifier = Agent(
    role="数据验证专家", 
    goal="将收集的数据与权威来源进行核对",
    backstory="您通过检查官方来源验证数据准确性，在需要时解决 CAPTCHAs。",
    tools=[solve_captcha],
    verbose=True
)

from crewai import Task, Crew, Process

# 可能遇到 CAPTCHA 的任务
research_task = Task(
    description="""研究目标网站 {url}。
    如果遇到 CAPTCHA 挑战，请识别其类型和网站密钥，
    然后使用 solve_captcha 工具获取令牌。
    该网站使用 reCAPTCHA v2，网站密钥为：{site_key}。
    解决 CAPTCHA 后收集所需数据。""",
    expected_output="在需要时使用 CAPTCHA 令牌从目标网站收集数据",
    agent=researcher
)

# 分析任务（依赖研究结果）
analysis_task = Task(
    description="分析研究员收集的数据并生成摘要报告。",
    expected_output="收集数据的结构化分析报告",
    agent=data_collector
)

# 创建团队
crew = Crew(
    agents=[researcher, data_collector],
    tasks=[research_task, analysis_task],
    process=Process.sequential,
    verbose=True
)

# 运行团队
result = crew.kickoff(inputs={
    "url": "https://example.com/data",
    "site_key": "6LeIxAcTAAAAAJcZVRqyHh71UMIEGNQ_MXjiZKhI"
})
print(result)

@tool("解决 reCAPTCHA v3")
def solve_recaptcha_v3(website_url: str, website_key: str, page_action: str = "verify") -> str:
    """使用高评分令牌解决 reCAPTCHA v3 挑战。
    当网站使用不可见的 reCAPTCHA v3（基于评分，无可见复选框）时使用。
    
    参数:
        website_url: 页面的完整 URL
        website_key: reCAPTCHA 网站密钥
        page_action: v3 评分的操作名称（默认: 'verify'）
    """
    result = asyncio.run(executor.execute("solve_captcha", {
        "captcha_type": "reCaptchaV3",
        "website_url": website_url,
        "website_key": website_key,
        "page_action": page_action,
        "min_score": 0.7
    }))
    if result["success"]:
        return f"reCAPTCHA v3 已解决。令牌: {result['solution']['token']}"
    return f"失败: {result['error']}"

@tool("解决 Cloudflare Turnstile")
def solve_turnstile(website_url: str, website_key: str) -> str:
    """解决 Cloudflare Turnstile 挑战。
    当网站受 Cloudflare 保护并显示 Turnstile 小部件时使用。
    
    参数:
        website_url: 页面的完整 URL
        website_key: Turnstile 网站密钥（以 0x 开头）
    """
    result = asyncio.run(executor.execute("solve_captcha", {
        "captcha_type": "cloudflare",
        "website_url": website_url,
        "website_key": website_key
    }))
    if result["success"]:
        return f"Turnstile 已解决。令牌: {result['solution']['token']}"
    return f"失败: {result['error']}"

@tool("带重试的 CAPTCHA 解决")
def solve_captcha_robust(captcha_type: str, website_url: str, website_key: str) -> str:
    """带自动重试的 CAPTCHA 解决。
    最多尝试 3 次解决失败后报告失败。
    """
    for attempt in range(3):
        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 < 2:
            import time
            time.sleep(3)
    return f"3 次尝试后失败: {result.get('error', '未知错误')}"

@tool("检查 CAPTCHA 解决余额")
def check_balance() -> str:
    """检查 CAPTCHA 解决的剩余余额。"""
    result = asyncio.run(executor.execute("get_balance", {}))
    if result["success"]:
        return f"余额: ${result['balance']:.2f}"
    return "无法检查余额"