如何在AutoGen代理中解决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

# AutoGen框架
pip install autogen-agentchat

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

import asyncio
from autogen import AssistantAgent, UserProxyAgent, config_list_from_json
from capsolver_agent.schema import create_executor

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

def solve_captcha(
    captcha_type: str,
    website_url: str,
    website_key: str,
    page_action: str = "",
    min_score: float = 0.7
) -> str:
    """解决验证码挑战并返回验证令牌。
    
    Args:
        captcha_type: 验证码类型 - 'reCaptchaV2', 'reCaptchaV3' 或 'cloudflare'
        website_url: 包含验证码的页面完整URL
        website_key: 网站密钥（data-sitekey属性）
        page_action: reCAPTCHA v3的动作名称（可选）
        min_score: reCAPTCHA v3的最低分数（默认0.7）
    
    Returns:
        解决的验证码令牌或错误信息
    """
    params = {
        "captcha_type": captcha_type,
        "website_url": website_url,
        "website_key": website_key
    }
    if page_action and captcha_type == "reCaptchaV3":
        params["page_action"] = page_action
        params["min_score"] = min_score
    
    result = asyncio.run(executor.execute("solve_captcha", params))
    
    if result["success"]:
        token = result["solution"]["token"]
        return f"验证码成功解决。令牌（前80个字符）：{token[:80]}..."
    return f"验证码解决失败：{result['error']}"

# LLM配置
llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "YOUR_OPENAI_KEY"}],
    "temperature": 0
}

# 助理代理（决定做什么的AI）
assistant = AssistantAgent(
    name="web_assistant",
    system_message="""您是具有验证码解决能力的网络研究助理。
    当任务需要访问受验证码保护的网站时，请使用solve_captcha
    函数并提供以下参数：
    - captcha_type: 'reCaptchaV2', 'reCaptchaV3' 或 'cloudflare'
    - website_url: 页面完整URL
    - website_key: 页面中的网站密钥
    
    解决后，将令牌返回给用户以提交表单。""",
    llm_config=llm_config
)

# 用户代理（代表助理执行函数）
user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=5,
    code_execution_config={"work_dir": "workspace", "use_docker": False}
)

# 注册验证码解决函数
from autogen import register_function

register_function(
    solve_captcha,
    caller=assistant,      # 助理决定何时调用
    executor=user_proxy,   # 用户代理执行函数
    name="solve_captcha",
    description="解决验证码挑战（reCAPTCHA v2/v3 或 Cloudflare Turnstile）并返回验证令牌"
)

# 启动对话
user_proxy.initiate_chat(
    assistant,
    message="""我需要访问受reCAPTCHA v2保护的页面。
    URL: https://example.com/data-portal
    网站密钥: 6LeIxAcTAAAAAJcZVRqyHh71UMIEGNQ_MXjiZKhI
    
    请解决验证码并提供令牌。"""
)

from autogen import GroupChat, GroupChatManager

# 专业代理
researcher = AssistantAgent(
    name="researcher",
    system_message="您研究网站并收集数据。遇到验证挑战时使用solve_captcha。",
    llm_config=llm_config
)

analyst = AssistantAgent(
    name="analyst",
    system_message="您分析收集的数据并生成见解。",
    llm_config=llm_config
)

executor_agent = UserProxyAgent(
    name="executor",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "workspace", "use_docker": False}
)

# 为researcher注册验证码解决
register_function(
    solve_captcha,
    caller=researcher,
    executor=executor_agent,
    name="solve_captcha",
    description="解决网站上的验证码挑战"
)

# 创建群组聊天
group_chat = GroupChat(
    agents=[researcher, analyst, executor_agent],
    messages=[],
    max_round=10
)

manager = GroupChatManager(groupchat=group_chat, llm_config=llm_config)

# 启动群组研究任务
executor_agent.initiate_chat(
    manager,
    message="研究https://example.com/reports的数据。它受reCAPTCHA v2保护（密钥：6Le...）。解决验证码，收集数据，然后分析它。"
)

def solve_captcha_with_retry(
    captcha_type: str,
    website_url: str,
    website_key: str,
    max_retries: int = 3
) -> str:
    """在失败时自动重试的验证码解决。"""
    for attempt in range(max_retries):
        params = {
            "captcha_type": captcha_type,
            "website_url": website_url,
            "website_key": website_key
        }
        result = asyncio.run(executor.execute("solve_captcha", params))
        
        if result["success"]:
            return f"解决成功（尝试 {attempt+1}）。令牌：{result['solution']['token'][:80]}..."
        
        if attempt < max_retries - 1:
            import time
            time.sleep(3)
    
    return f"尝试{max_retries}次后失败：{result.get('error', '未知')}"

def check_captcha_balance() -> str:
    """检查剩余的验证码解决积分。"""
    result = asyncio.run(executor.execute("get_balance", {}))
    if result["success"]:
        return f"CapSolver余额：${result['balance']:.2f}"
    return "无法检索余额"

register_function(solve_captcha_with_retry, caller=assistant, executor=user_proxy,
    name="solve_captcha", description="带有重试逻辑的验证码解决")
register_function(check_captcha_balance, caller=assistant, executor=user_proxy,
    name="check_balance", description="检查验证码解决积分余额")